Как стать автором
Поиск
Написать публикацию
Обновить
118.29
Рейтинг
Amvera
Amvera — облако для хостинга IT-приложений

Личный ИИ-ассистент на ваших данных. Часть 2: Веб-интерфейс, авторизация и стриминг ответов от ИИ

Время на прочтение27 мин
Количество просмотров2.5K
Блог компании AmveraВеб-разработка*Python*Программирование*Искусственный интеллект
Туториал

Друзья, приветствую!

Как вы поняли по названию этой статьи — сегодня мы продолжаем погружаться в тему разработки личного ИИ-ассистента на основе собственных данных.

Краткое напоминание о первой части

Напоминаю, что в предыдущей части мы:

  • Разобрались с понятием векторной базы данных

  • Научились собирать данные и реализовывать такую базу

  • Изучили механизм "умного поиска"

  • Познакомились с мощным Python-инструментом Langchain

  • Обеспечили интеграцию нейросетей ChatGPT и Deepseek

  • Создали консольный вариант диалога с нейросетями на основе собственной базы знаний (документации сервиса Amvera Cloud)

О сегодняшней части

Сегодня мы прокачаем наш проект — превратим его в полноценный веб-сервис с авторизацией и удобным веб-чатом для взаимодействия с нейросетями. Получится некое подобие веб-интерфейса сайта https://chat.openai.com/chat, но с возможностью выбора между Deepseek и ChatGPT, работающими на основе нашей собственной базы знаний.

Для более детального понимания того, что мы будем создавать, я подготовил короткий видео-обзор с комментариями: Видео-демонстрация проекта

Необходимые предварительные условия

Для продолжения вам потребуются:

  • Готовая векторная база данных на собственных данных (не обязательно документация Amvera Cloud)

  • API-токены для ChatGPT и Deepseek

  • Базовое понимание принципов создания бэкенда на FastAPI и основ фронтенд-разработки

План работы

В ходе этой части мы:

  1. Разработаем класс для удобного управления соединением с векторной базой Chroma (с возможностью однократного подключения)

  2. Создадим класс для интеграции нейросетей Deepseek и ChatGPT в FastAPI-приложение с помощью Langchain

  3. Разработаем API проекта для авторизации и отправки запросов к нейросетям

  4. Создадим стильный веб-интерфейс чата со страницами входа и самого чата

  5. Объединим фронтенд с бэкендом в формате полноценного защищенного веб-сервиса

  6. Выполним деплой проекта на платформе Amvera Cloud

На выходе, после деплоя, мы получим такой результат:

Важное замечание о доступе к нейросетям

Отдельно хочу остановиться на особенностях удаленного запуска проекта. В России некоторые нейросети, такие как ChatGPT, официально недоступны даже через API-интеграцию, что обычно требует использования VPN.

Однако с сервисом Amvera Cloud такой проблемы не возникнет — вы сможете запускать собственные проекты с ChatGPT, Claude и другими нейросетями, недоступными в РФ. Кроме того, процесс деплоя даже такого сложного сервиса займет всего пару минут. Платформа предоставляет бесплатное доменное имя с HTTPS, что делает выбор этого сервиса очевидным.

Технический стек

В рамках статьи мы будем использовать два языка программирования:

Python (серверная логика):

  • langchain-openai — для интеграции с ChatGPT

  • langchain-deepseek — для интеграции с Deepseek

  • FastAPI — фреймворк для создания бэкенда

  • langchain-chroma — для взаимодействия с базой данных Chroma

  • Jinja2 — шаблонизатор фронтенда

  • PyJWT — для работы с JWT-токенами

  • И другие Python-инструменты

JavaScript (клиентская логика):

  • Чистый JavaScript без фреймворков

  • Несколько вспомогательных библиотек, интегрированных непосредственно в проект

Для стилизации будем частично писать собственные стили и частично использовать Bootstrap.

Итак, начнем разработку!

Подготовка проекта

Настройка окружения

Так как для рендеринга HTML страниц мы будем использовать Jinja2, всю логику мы опишем в рамках одного FastAPI приложения.

Начнем с создания виртуального окружения в вашем любимом IDE.

Создание конфигурационных файлов

Файл .env

Создадим файл .env и заполним его следующими данными:

DEEPSEEK_API_KEY=sk-123456  # токен Deepseek
OPENAI_API_KEY=sk-proj-S_12345  # токен ChatGPT
SECRET_KEY=super_secret_key  # секретный ключ для генерации JWT-токена (сами придумываем)
ALGORITHM=HS256  # алгоритм шифрования для JWT

Файл requirements.txt

Создадим файл requirements.txt со следующим содержимым:

langchain-huggingface==0.1.2
torch==2.6.0
loguru==0.7.3
chromadb==0.6.3
sentence-transformers==3.4.1
langchain-chroma==0.2.2
pydantic-settings==2.8.1
fastapi==0.115.12
uvicorn==0.34.0
Jinja2==3.1.6
aiofiles==24.1.0
PyJWT==2.10.1
langchain-deepseek==0.1.3
langchain-openai==0.3.11

Установим зависимости:

pip install -r requirements.txt

Структура проекта

Создадим папку app и внутри нее файл config.py. Заполним его:

import os
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Config(BaseSettings):
    DEEPSEEK_API_KEY: SecretStr
    BASE_DIR: str = os.path.abspath(os.path.join(os.path.dirname(__file__)))
    SECRET_KEY: str
    USERS: str = os.path.join(BASE_DIR, "..", "users.json")
    ALGORITHM: str
    AMVERA_CHROMA_PATH: str = os.path.join(BASE_DIR, "chroma_db")
    AMVERA_COLLECTION_NAME: str = "amvera_docs"
    LM_MODEL_NAME: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
    DEEPSEEK_MODEL_NAME: str = "deepseek-chat"
    OPENAI_MODEL_NAME: str = "gpt-3.5-turbo"
    OPENAI_API_KEY: SecretStr

    model_config = SettingsConfigDict(env_file=f"{BASE_DIR}/../.env")

settings = Config()

Важное примечание: Обратите внимание на строку from pydantic import SecretStr. Для корректной интеграции через Langchain нейросетей Deepseek и ChatGPT необходимо описывать переменные окружения как DEEPSEEK_API_KEY: SecretStr. В остальном — это просто удобное представление переменных окружения в одном классе при помощи библиотеки pydantic-settings.

Теперь нам достаточно будет импортировать settings и везде по проекту вызывать необходимые переменные через точку: settings.OPENAI_API_KEY

Создание структуры директорий

В рамках подготовки давайте в папке app создадим дополнительные директории:

  • api: опишем тут всю API-логику приложения

  • chroma_client: опишем класс для работы с Chroma и класс интеграции с нейросетями

  • chroma_db: тут необходимо поместить сгенерированную ранее базу данных Chroma (предполагаю, что база у вас уже есть, если это не так — можете брать мою. Базу можно найти в полном исходном коде проекта)

  • pages: тут опишем FastAPI эндпоинты для рендеринга HTML-страниц

  • static: папка со статическими файлами (стили и скрипты JS)

  • templates: папка с HTML-шаблонами

И сразу создадим в папке app пустой файл main.py. В нем мы будем собирать наше приложение.

На этом подготовка завершена, и мы можем приступать к написанию кода.

Класс для управления соединением с базой Chroma

Обоснование подхода

Напоминаю, что наша задача состоит в том, чтобы при запуске FastAPI приложения мы только один раз открывали соединение с базой данных Chroma, и после все запросы для всех пользователей должны использовать эту открытую сессию.

Это необходимо для:

  • Максимально быстрого получения контекста из Chroma для наших нейросетей

  • Корректной работы системы при одновременных запросах

  • Обеспечения гибкости даже при больших нагрузках

Реализация класса ChromaVectorStore

Создадим файл app/chroma_client/chroma_store.py и начнем с необходимых импортов:

import torch
from langchain_chroma import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from loguru import logger
from app.config import settings

Теперь реализуем сам класс для управления соединением с базой:

class ChromaVectorStore:
    def __init__(self):
        """
        Инициализирует пустой экземпляр хранилища векторов.
        Соединение с базой данных будет установлено позже с помощью метода init().
        """
        self._store: Chroma | None = None

    async def init(self):
        """
        Асинхронный метод для инициализации соединения с базой данных Chroma.
        Создает embeddings на основе модели из настроек, используя CUDA если доступно.
        """
        logger.info("🧠 Инициализация ChromaVectorStore...")
        try:
            # Определяем устройство для вычислений: GPU если доступен, иначе CPU
            device = "cuda" if torch.cuda.is_available() else "cpu"
            logger.info(f"🚀 Используем устройство для эмбеддингов: {device}")

            # Создаем модель эмбеддингов с указанными параметрами
            embeddings = HuggingFaceEmbeddings(
                model_name=settings.LM_MODEL_NAME,
                model_kwargs={"device": device},
                encode_kwargs={"normalize_embeddings": True},
            )

            # Инициализируем соединение с базой данных Chroma
            self._store = Chroma(
                persist_directory=settings.AMVERA_CHROMA_PATH,
                embedding_function=embeddings,
                collection_name=settings.AMVERA_COLLECTION_NAME,
            )

            logger.success(
                f"✅ ChromaVectorStore успешно подключен к коллекции "
                f"'{settings.AMVERA_COLLECTION_NAME}' в '{settings.AMVERA_CHROMA_PATH}'"
            )
        except Exception as e:
            logger.exception(f"❌ Ошибка при инициализации ChromaVectorStore: {e}")
            raise

    async def asimilarity_search(self, query: str, with_score: bool, k: int = 3):
        """
        Асинхронный метод для поиска похожих документов в базе данных Chroma.

        Args:
            query (str): Текстовый запрос для поиска
            with_score (bool): Включать ли оценку релевантности в результаты
            k (int): Количество возвращаемых результатов

        Returns:
            list: Список найденных документов, возможно с оценками если with_score=True

        Raises:
            RuntimeError: Если хранилище не инициализировано
        """
        if not self._store:
            raise RuntimeError("ChromaVectorStore is not initialized.")

        logger.info(f"🔍 Поиск похожих документов по запросу: «{query}», top_k={k}")
        try:
            if with_score:
                results = await self._store.asimilarity_search_with_score(
                    query=query, k=k
                )
            else:
                results = await self._store.asimilarity_search(query=query, k=k)

            logger.debug(f"📄 Найдено {len(results)} результатов.")
            return results
        except Exception as e:
            logger.exception(f"❌ Ошибка при поиске: {e}")
            raise

    async def close(self):
        """
        Асинхронный метод для закрытия соединения с базой данных Chroma.
        В текущей реализации Chroma не требует явного закрытия,
        но метод добавлен для полноты API и возможных будущих изменений.
        """
        logger.info("🔌 Отключение ChromaVectorStore...")
        # Пока Chroma не требует явного закрытия, но в будущем может понадобиться
        # self._store.close() или подобный метод
        pass

Особенности реализации

В нашей реализации ChromaVectorStore есть несколько ключевых моментов:

  1. Отложенная инициализация — соединение с базой данных устанавливается только при явном вызове метода init(), что позволяет контролировать момент подключения в жизненном цикле приложения.

  2. Асинхронность — все методы являются асинхронными (async), что позволяет эффективно работать с базой данных в контексте асинхронного веб-приложения FastAPI.

  3. Использование GPU — если доступно CUDA-совместимое устройство, модель эмбеддингов будет использовать его для ускорения вычислений.

  4. Логирование — все важные события и ошибки записываются в лог с помощью библиотеки loguru, что упрощает отладку и мониторинг.

  5. Обработка ошибок — все операции обернуты в блоки try-except для корректной обработки ошибок.

Данная реализация обеспечивает эффективное использование соединения с базой данных Chroma в рамках всего жизненного цикла приложения FastAPI.

К этому файлу мы ещё вернемся сегодня, но пока переключимся на реализацию класса для интеграции нейросетей Deepseek и ChatGPT в наш сервис.

Класс для интеграции нейросетей

Описание задачи

Прежде чем приступить к реализации класса, напомню о функциональных требованиях к нашему механизму взаимодействия с нейросетями:

  1. Принимать текстовый запрос от пользователя

  2. На основании запроса возвращать релевантные документы из базы данных Chroma

  3. Формировать контекст из полученных данных Chroma, пользовательского запроса и системного промпта

  4. Передавать этот контекст выбранной нейросети (ChatGPT или Deepseek)

  5. Выдавать ответ не целым куском, а в формате потокового вывода (стрима)

Весь этот процесс должен работать асинхронно для обеспечения эффективной работы веб-приложения.

Реализация класса ChatWithAI

Создадим файл app/chroma_client/ai_store.py и начнем с необходимых импортов:

from typing import AsyncGenerator, Literal
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI
from loguru import logger
from app.config import settings

Теперь реализуем класс для интеграции с нейросетями:

class ChatWithAI:
    """
    Класс для взаимодействия с различными языковыми моделями (LLM).
    Поддерживает работу с ChatGPT и Deepseek, с возможностью потокового вывода ответов.
    """

    def __init__(self, provider: Literal["deepseek", "chatgpt"] = "deepseek"):
        """
        Инициализирует экземпляр класса с выбранным провайдером LLM.

        Args:
            provider (str): Провайдер языковой модели ('deepseek' или 'chatgpt')

        Raises:
            ValueError: Если указан неподдерживаемый провайдер
        """
        self.provider = provider

        if provider == "deepseek":
            logger.info(f"🤖 Инициализация Deepseek модели: {settings.DEEPSEEK_MODEL_NAME}")
            self.llm = ChatDeepSeek(
                api_key=settings.DEEPSEEK_API_KEY,
                model=settings.DEEPSEEK_MODEL_NAME,
                temperature=0.7,  # Настройка творческой свободы модели
            )
        elif provider == "chatgpt":
            logger.info(f"🤖 Инициализация ChatGPT модели: {settings.OPENAI_MODEL_NAME}")
            self.llm = ChatOpenAI(
                api_key=settings.OPENAI_API_KEY,
                model=settings.OPENAI_MODEL_NAME,
                temperature=0.7,  # Настройка творческой свободы модели
            )
        else:
            logger.error(f"❌ Неподдерживаемый провайдер: {provider}")
            raise ValueError(f"Неподдерживаемый провайдер: {provider}")

        logger.success(f"✅ Модель {provider} успешно инициализирована")

    async def astream_response(
        self, formatted_context: str, query: str
    ) -> AsyncGenerator[str, None]:
        """
        Асинхронно генерирует потоковый ответ от выбранной языковой модели.

        Args:
            formatted_context (str): Контекст из базы знаний, форматированный для запроса
            query (str): Пользовательский запрос

        Yields:
            str: Фрагменты ответа в потоковом режиме

        Notes:
            Использует системный промпт для задания контекста и роли модели
        """
        try:
            # Формируем системный промпт, определяющий роль и контекст работы модели
            system_message = SystemMessage(
                content="""Ты — внутренний менеджер компании Amvera Cloud.
                Отвечаешь по делу без лишних вступлений.
                Свой ответ, в первую очередь, ориентируй на переданный контекст.
                Если информации недостаточно - пробуй получить ответы из своей базы знаний."""
            )

            # Формируем пользовательский запрос с включенным контекстом
            human_message = HumanMessage(
                content=f"Вопрос: {query}\nКонтекст: {formatted_context}. Ответ форматируй в markdown!"
            )

            logger.info(f"🔄 Начинаем стриминг ответа для запроса: «{query}»")

            # Асинхронно получаем фрагменты ответа
            async for chunk in self.llm.astream([system_message, human_message]):
                if chunk.content:  # Пропускаем пустые фрагменты
                    logger.debug(f"📝 Получен фрагмент: {chunk.content[:50]}...")
                    yield chunk.content

            logger.info("✅ Стриминг ответа успешно завершен")

        except Exception as e:
            logger.error(f"❌ Ошибка при стриминге ответа: {e}")
            yield f"Произошла ошибка при обработке запроса: {str(e)}"

Особенности реализации

Класс ChatWithAI содержит несколько ключевых особенностей, которые стоит отметить:

  1. Поддержка нескольких провайдеров — класс позволяет выбирать между ChatGPT от OpenAI и Deepseek, что дает гибкость в использовании различных языковых моделей в зависимости от потребностей и доступности.

  2. Асинхронный потоковый вывод — метод astream_response реализован как асинхронный генератор, что позволяет постепенно передавать части ответа клиенту по мере их генерации моделью, без необходимости ожидать полный ответ.

  3. Форматирование запроса — для каждого запроса формируется специальный системный промпт, определяющий роль и контекст работы модели, что помогает получать более релевантные и структурированные ответы.

  4. Расширенное логирование — все ключевые этапы работы класса сопровождаются детальным логированием, что упрощает отладку и мониторинг работы системы.

  5. Обработка ошибок — все потенциально опасные операции обернуты в блоки try-except, что обеспечивает устойчивость работы приложения даже при возникновении ошибок взаимодействия с API нейросетей.

Реализуем систему авторизации

Так как в нашем сервисе будут задействованы платные токены — очевидно, что мы захотим защищать свой сервис от использования его сторонними людьми и в немеренных количествах.

Логика авторизации

Логика авторизации будет следующей:

  1. Для пользования сервисом необходимо будет ввести логин и пароль в форму входа

  2. После корректного ввода данных мы будем выдавать пользователю специальный JWT-токен, который мы будем помещать в куки сессию. У этого токена будет свой срок жизни (1 час). То есть после того как этот час пройдет — мы сделаем так, чтоб токен был недействительным. Недействительный токен равняется выходу из системы и полному закрытию функционала.

  3. Если логин или пароль не корректны либо токен истек — доступа к системе у пользователя не будет.

Создание базы данных пользователей

Для экономии времени я решил использовать простой JSON в качестве базы данных пользователей - файл, в который поместил информацию о пользователях. Конечно, в боевом проекте — это должна быть реляционная база данных, в которой будет храниться полная информация о пользователе и его хэшированный пароль. Подробно об этом я рассказывал в своей статье: «Создание собственного API на Python (FastAPI): Авторизация, Аутентификация и роли пользователей» (https://habr.com/ru/articles/829742/)

В корне проекта (на одном уровне с папкой app) создаём файл users.json и заполняем его следующим образом:

[
  {
    "user_id": 1,
    "login": "user1",
    "password": "pass1"
  },
  {
    "user_id": 2,
    "login": "user2",
    "password": "pass2"
  }
]

Если решите следовать этому примеру — позаботьтесь о надежности пароля. И ещё раз подчеркиваю — такой подход только для экономии времени!

Реализация утилит для авторизации

Теперь в папке app/api создаем файл utils.py, там опишем несколько утилит, которые позволят нам организовать логику авторизации в проекте.

Импорты:

import json
from datetime import datetime, timedelta
from typing import Optional

import aiofiles
import jwt
from fastapi import Cookie, HTTPException

from app.config import settings

Тут я использовал библиотеку aiofiles для асинхронного взаимодействия с JSON-файлом.

Метод для получения всех пользователей:

async def get_all_users():
    async with aiofiles.open(settings.USERS) as f:
        content = await f.read()
        users = json.loads(content)
        return users

Метод для проверки корректности логина и пароля:

async def authenticate_user(login: str, password: str):
    users = await get_all_users()
    for user in users:
        if user["login"] == login and user["password"] == password:
            return user
    return None

Метод для создания JWT токена:

async def create_jwt_token(user_id: int):
    expire = datetime.now() + timedelta(hours=1)
    payload = {
        "sub": str(user_id),
        "exp": expire.timestamp(),
    }
    token = jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return token

На вход принимаем айди пользователя и генерируем срок жизни токена в 1 час.

Метод для верификации JWT-токена:

async def verify_jwt_token(token: str):
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
        )
        return payload["sub"]
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

Тут мы на вход будем принимать токен и будем проверять валидность токена и его срок жизни. Только если токен валидный и его срок жизни не подошел к концу — мы будем возвращать айди пользователя, иначе выбрасываем ошибку.

Метод для проверки авторизации пользователя:

async def get_current_user(
    access_token: Optional[str] = Cookie(default=None),
):
    if not access_token:
        raise HTTPException(status_code=401, detail="Missing token")
    user_id = await verify_jwt_token(access_token)
    return user_id

Тут мы пытаемся извлечь из куки значение ключа access_token. Если его нет — сразу выбрасываем ошибку, что токен не найден.

Затем выполняем проверку валидности и срока жизни токена, и только если все ок — возвращаем айди пользователя, что будет равняться тому, что пользователь авторизован.

Опциональная проверка авторизации:

async def get_optional_current_user(
    access_token: Optional[str] = Cookie(default=None),
) -> Optional[int]:
    if not access_token:
        return None
    try:
        user_id = await verify_jwt_token(access_token)
        return user_id
    except Exception:
        return None

Данный метод от get_current_user отличается только тем, что в случае если пользователь не авторизован — мы будем возвращать None, а не ошибку. Этот метод нам пригодится, когда мы приступим к написанию веб-интерфейса.

Определение Pydantic-моделей

Теперь в папке api создаем файл schemas.py.

Там мы опишем pydantic-модели для валидации при авторизации пользователя:

from pydantic import BaseModel

class SUserAuth(BaseModel):
    login: str
    password: str

К этому файлу мы ещё вернемся.

Реализация API-роутов

Теперь в папке api создаем файл router.py и в нем опишем логику авторизации и проверки.

Импорты:

from fastapi import APIRouter, Depends, HTTPException, Response
from app.api.schemas import SUserAuth
from app.api.utils import authenticate_user, create_jwt_token, get_current_user

Инициализация роутера:

router = APIRouter()

Метод для входа:

@router.post("/login")
async def login(
    response: Response,
    user_data: SUserAuth,
):
    user = await authenticate_user(user_data.login, user_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    token = await create_jwt_token(user["user_id"])
    response.set_cookie(
        key="access_token",
        value=token,
        httponly=True,
        secure=False,  # Включите True в проде (HTTPS)
        samesite="lax",
        max_age=3600,
        path="/",
    )
    return {"message": "Logged in"}

На вход принимаем логин с паролем. Если все проверки пройдены, то создаем JWT-токен, который помещается в куки-сессию.

Метод для выхода:

@router.post("/logout")
async def logout(response: Response):
    response.delete_cookie("access_token")
    return {"message": "Logged out"}

Тут мы просто из кук удаляем ключ со значением access_token.

Метод для проверки авторизации:

@router.get("/protected")
async def protected_route(user_id: int = Depends(get_current_user)):
    return {"message": f"Привет, пользователь {user_id}!"}

Тут я использовал механизм зависимостей FastAPI. Подробнее о нем рассказывал в своих предыдущих статьях, посвященных FastAPI. Если коротко, то зависимость — это функция, которая выполняется до основной функции.

В нашем случае мы пытаемся получить айди пользователя. Если это удастся — значит пользователь авторизован. Иначе — мы выбросим ошибку. Для лучшего понимания продемонстрирую это через документацию:

Без авторизации (ошибка)
Без авторизации (ошибка)
С авторизацией (приветствие)
С авторизацией (приветствие)

На этом с блоком авторизации все, и мы можем приступать к реализации остальной API логики.

Пишем API-эндпоинты

Сейчас нам предстоит реализовать два API-эндпоинта.

  • Первый — для прямого взаимодействия с базой данных ChromaDB. Он нужен исключительно для тестирования: чтобы понять, как она работает и с какой скоростью.

  • Второй — уже полноценный, с подключением нейросети: он будет возвращать результат на основе данных из Chroma и ответа от AI.

Интеграция ChromaDB с FastAPI

Для корректной интеграции ChromaDB в наш FastAPI-сервис необходимо создать глобальный инстанс на основе класса ChromaVectorStore, а также функцию-зависимость, которую мы будем использовать в эндпоинтах.

Сделаем это в файле:app/chroma_client/chroma_store.py

# Глобальный инстанс
chroma_vectorstore = ChromaVectorStore()

# Зависимость
def get_vectorstore() -> ChromaVectorStore:
    return chroma_vectorstore

Pydantic-схемы

Теперь опишем модели для запросов и ответов в app/api/schemas.py:

from pydantic import BaseModel
from typing import Literal


class AskResponse(BaseModel):
    response: str

    
class AskWithAIResponse(BaseModel):
    response: str
    provider: Literal["deepseek", "chatgpt"] = "deepseek"

Обновляем импорты

Переходим в app/api/router.py и актуализируем импорты:

from fastapi import APIRouter, Depends, HTTPException, Response
from fastapi.responses import StreamingResponse

from app.api.schemas import AskResponse, AskWithAIResponse, SUserAuth
from app.api.utils import authenticate_user, create_jwt_token, get_current_user
from app.chroma_client.ai_store import ChatWithAI
from app.chroma_client.chroma_store import ChromaVectorStore, get_vectorstore

Здесь нам понадобится StreamingResponse из FastAPI — мы будем использовать его во втором эндпоинте.

Эндпоинт №1: запрос к Chroma напрямую

Создаем эндпоинт для получения результатов из ChromaDB без подключения нейросети:

@router.post("/ask")
async def ask(
    query: AskResponse,
    vectorstore: ChromaVectorStore = Depends(get_vectorstore),
    user_id: int = Depends(get_current_user),
):
    results = await vectorstore.asimilarity_search(
        query=query.response, with_score=True, k=5
    )
    formatted_results = []
    for doc, score in results:
        formatted_results.append({
            "text": doc.page_content,
            "metadata": doc.metadata,
            "similarity_score": score,
        })
    return {"results": formatted_results}

На вход принимаем запрос пользователя. Используем две зависимости:

  • get_vectorstore — для подключения к Chroma

  • get_current_user — для проверки авторизации пользователя

Проверим работу:

Как видите — всё прекрасно работает 🎉

Эндпоинт №2: Chroma + нейросеть

Теперь создаем второй эндпоинт: Chroma + нейросеть.

@router.post("/ask_with_ai")
async def ask_with_ai(
    query: AskWithAIResponse,
    vectorstore: ChromaVectorStore = Depends(get_vectorstore),
    user_id: int = Depends(get_current_user),
):
    results = await vectorstore.asimilarity_search(
        query=query.response, with_score=True, k=5
    )

    if results:
        ai_context = "\n".join([doc.page_content for doc, _ in results])
        ai_store = ChatWithAI(provider=query.provider)

        async def stream_response():
            async for chunk in ai_store.astream_response(ai_context, query.response):
                yield chunk

        return StreamingResponse(
            stream_response(),
            media_type="text/plain",
            headers={
                "Content-Type": "text/plain",
                "Transfer-Encoding": "chunked",
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
            },
        )
    else:
        return {"response": "Ничего не найдено"}

Что здесь происходит:

  1. Получаем документы из базы Chroma

  2. Формируем AI-контекст — объединяем содержимое документов в одну строку, разделённую переносами

  3. Инициализируем ChatWithAI с выбранным провайдером (deepseek или chatgpt)

  4. Создаем генератор stream_response, который будет по кусочкам отдавать результат от нейросети

  5. Возвращаем StreamingResponse с соответствующими заголовками

Важное замечание:

return StreamingResponse(
    stream_response(),
    media_type="text/plain",
    headers={
        "Content-Type": "text/plain",
        "Transfer-Encoding": "chunked",
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
    },
)

Swagger не умеет обрабатывать стриминг, так что протестировать этот эндпоинт получится только через фронт — когда реализуем обработку стрима.

На этом с API-логикой всё. Дальше — разработка веб-интерфейса чата. Поехали! 🚀

Реализация веб-интерфейса

На этом этапе мы реализуем два HTML-шаблона — страницу входа и чат. Добавим к ним стили, JS-логику и создадим два FastAPI-эндпоинта, которые будут рендерить эти страницы.

1. Страница входа

Создаем HTML-шаблон login.html по пути:
app/templates/login.html

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Вход</title>
    <link
      href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css"
      rel="stylesheet"
    />
    <link rel="stylesheet" href="/static/form_style.css" />
  </head>
  <body>
    <div class="login-form">
      <h2 class="text-center mb-4">Вход в систему</h2>
      <form id="loginForm">
        <div class="form-floating mb-3">
          <input
            type="text"
            class="form-control"
            id="login"
            name="login"
            placeholder="Логин"
            required
          />
          <label for="login">Логин</label>
        </div>
        <div class="form-floating mb-4">
          <input
            type="password"
            class="form-control"
            id="password"
            name="password"
            placeholder="Пароль"
            required
          />
          <label for="password">Пароль</label>
        </div>
        <button type="submit" class="btn btn-primary w-100">Войти</button>
      </form>
    </div>

    <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
    <script src="/static/form_script.js"></script>
  </body>
</html>

📌 Это простая форма входа с использованием Bootstrap. Я также добавил кастомные стили (form_style.css), но на них подробно останавливаться не будем — при желании их можно посмотреть в исходниках проекта (см. мой Telegram-канал «Легкий путь в Python»).

JavaScript-логика для входа

JS-код для формы мы размещаем в:
app/static/form_script.js

document.getElementById('loginForm').addEventListener('submit', async (e) => {
  e.preventDefault()

  const login = document.getElementById('login').value
  const password = document.getElementById('password').value

  try {
    const response = await fetch('/api/login', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ login, password }),
      credentials: 'include',
    })

    const data = await response.json()

    if (data.message === 'Logged in') {
      window.location.href = '/'
    } else {
      alert('Ошибка входа')
    }
  } catch (error) {
    console.error('Error:', error)
    alert('Произошла ошибка при входе')
  }
})

Здесь реализован стандартный подход:

  1. Перехватываем отправку формы.

  2. Забираем логин и пароль.

  3. Отправляем POST-запрос на /api/login.

  4. Если получили Logged in, то редиректим пользователя на главную.

  5. В случае ошибки — показываем alert.

2. Страница чата

Теперь создаем HTML-шаблон для чата:
app/templates/index.html

<!DOCTYPE html>
<html lang="ru">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Amvera Cloud Chat</title>
    <link
      href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/css/bootstrap.min.css"
      rel="stylesheet"
    />
    <link
      href="https://cdn.jsdelivr.net/npm/highlight.js@11.9.0/styles/github.min.css"
      rel="stylesheet"
    />
    <link
      href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.1/font/bootstrap-icons.css"
      rel="stylesheet"
    />
    <link rel="stylesheet" href="/static/styles.css" />
  </head>
  <body>
    <div class="chat-container">
      <div class="chat-header">
        <h1>Amvera Cloud Chat</h1>
        <div class="header-controls">
          <select class="provider-select" id="provider">
            <option value="deepseek">DeepSeek</option>
            <option value="chatgpt">ChatGPT</option>
          </select>
          <button class="btn btn-outline-danger" id="logoutButton">
            <i class="bi bi-box-arrow-right"></i>
            Выйти
          </button>
        </div>
      </div>
      <div class="chat-output" id="chatOutput"></div>
      <div class="input-container">
        <input
          type="text"
          class="form-control"
          id="queryInput"
          placeholder="Введите ваш запрос..."
          aria-label="Query"
        />
        <button class="btn btn-primary" type="button" id="sendButton">
          <i class="bi bi-send"></i>
          Отправить
        </button>
      </div>
    </div>

    <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/js/bootstrap.bundle.min.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/marked/marked.min.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/highlight.js@11.9.0/lib/highlight.min.js"></script>
    <script src="/static/script.js"></script>
  </body>
</html>

Здесь мы дополнительно подключили:

  • bootstrap-icons — для иконок

  • highlight.js — для подсветки синтаксиса

  • marked.js — для обработки markdown-ответов от нейросети

Следующим шагом мы реализуем основную JS-логику чата — в файле static/script.js. Именно там будет обрабатываться ввод запроса, отправка на API, получение и отображение стриминга ответа.

Логика клиентской части: JavaScript для страницы чата

Теперь разберем ключевую часть клиентской логики — JavaScript-код, который обрабатывает взаимодействие пользователя с веб-интерфейсом чата.

Файл со скриптом: app/static/script.js

Настройка markdown и подсветки кода

Для корректного отображения ответов, особенно если это код, мы используем библиотеку marked и подключаем highlight.js для подсветки.

marked.setOptions({
  highlight: function (code, lang) {
    if (lang && hljs.getLanguage(lang)) {
      return hljs.highlight(code, { language: lang }).value
    }
    return hljs.highlightAuto(code).value
  },
  breaks: false,
  gfm: true,
  headerIds: true,
  mangle: false,
})

💡 Поддержка GFM (GitHub Flavored Markdown) включена, чтобы красиво отображались списки, заголовки и т.д. Подсветка работает как по указанному языку, так и в автоматическом режиме.

Вспомогательные функции: индикатор загрузки и отображение сообщений

Создание анимации загрузки:

function createLoadingIndicator() {
  const loadingDiv = document.createElement('div')
  loadingDiv.className = 'loading'
  loadingDiv.innerHTML = `
    <div class="loading-dot"></div>
    <div class="loading-dot"></div>
    <div class="loading-dot"></div>
  `
  return loadingDiv
}

Добавление сообщения в чат:

function addMessage(content, isUser = false) {
  const chatOutput = document.getElementById('chatOutput')
  const messageDiv = document.createElement('div')
  messageDiv.className = `message ${isUser ? 'user' : 'assistant'}`

Сюда же добавляется кнопка копирования с иконкой. При клике — содержимое копируется в буфер обмена:

const copyButton = document.createElement('button')
copyButton.className = 'copy-button'
copyButton.innerHTML = '<i class="bi bi-clipboard"></i>'

Также добавляется блок с сообщением:

const contentDiv = document.createElement('div')
contentDiv.className = 'message-content'
contentDiv.innerHTML = marked.parse(content)

И в конце все элементы добавляются в DOM:

messageDiv.appendChild(contentDiv)
messageDiv.appendChild(copyButton)
chatOutput.appendChild(messageDiv)
chatOutput.scrollTop = chatOutput.scrollHeight

Обновление уже существующего сообщения

Если мы получаем ответ стримингом, то сообщение должно обновляться по мере поступления данных:

function updateMessage(messageDiv, content) {
  const contentDiv = messageDiv.querySelector('.message-content') || document.createElement('div')
  contentDiv.className = 'message-content'
  contentDiv.innerHTML = marked.parse(content)
  ...
}

Отправка запроса

Теперь обработаем событие нажатия Enter или кнопку "Отправить":

document
  .getElementById('queryInput')
  .addEventListener('keypress', function (e) {
    if (e.key === 'Enter' && !e.shiftKey) {
      e.preventDefault()
      document.getElementById('sendButton').click()
    }
  })

Основная логика запроса к API

Вот что происходит при отправке запроса:

  1. Проверка, ввел ли пользователь текст.

  2. Добавляется пользовательское сообщение в чат.

  3. Показывается индикатор загрузки.

  4. Отправляется запрос на эндпоинт /api/ask_with_ai.

  5. Ответ читается потоком (stream).

  6. Полученные символы по одному добавляются в чат с задержкой в 50 мс.

const reader = response.body.getReader()
const decoder = new TextDecoder('utf-8')
let buffer = ''
let currentText = ''

Затем происходит поочередное "считывание" символов и постепенное обновление текста ответа:

while (buffer.length > 0) {
  const char = buffer[0]
  buffer = buffer.slice(1)
  currentText += char
  await new Promise((resolve) => setTimeout(resolve, 50))
  updateMessage(assistantMessage, currentText)
}

Обработка ошибок и выход из аккаунта

Если сервер вернул 401 — редиректим на страницу логина:

if (response.status === 401) {
  window.location.href = '/login'
}

А при нажатии кнопки "Выйти":

document.getElementById('logoutButton').addEventListener('click', async () => {
  await fetch('/api/logout', { method: 'POST', credentials: 'include' })
  window.location.href = '/login'
})

На этом этапе у нас уже есть полноценный фронтенд:

  • Красивый интерфейс с Bootstrap и markdown

  • Поддержка стриминга ответов

  • Кнопка копирования

  • Обработка ошибок

  • Логика выхода

В следующем разделе мы создадим FastAPI-эндпоинты, которые будут рендерить эти HTML-шаблоны.

Эндпоинты для рендеринга HTML-страниц

На этом этапе мы подключаем HTML-шаблоны к FastAPI-приложению — опишем маршруты, которые будут отвечать за отображение страницы входа и страницы чата.

Создаем router.py

В папке app/pages создаем новый файл:

app/pages/router.py

Импорты

Выполним все необходимые импорты:

from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates

from app.api.utils import get_optional_current_user

💡 Здесь мы используем Jinja2Templates, чтобы рендерить HTML-шаблоны. А через get_optional_current_user проверяем, авторизован ли пользователь и выбрасываем None если это не так.

Инициализация роутера и шаблонизатора

router = APIRouter()
templates = Jinja2Templates(directory="app/templates")

Теперь templates.TemplateResponse(...) будет искать шаблоны в папке app/templates.

Эндпоинт: Страница входа

Добавим маршрут, обрабатывающий /login:

@router.get("/login", response_class=HTMLResponse)
async def login_page(
    request: Request,
    user_id: int = Depends(get_optional_current_user),
):
    if user_id:
        return RedirectResponse(url="/")
    else:
        return templates.TemplateResponse("login.html", {"request": request})

Логика:

  • Если user_id найден (то есть пользователь уже авторизован) — выполняем редирект на главную.

  • Иначе — возвращаем HTML-шаблон страницы входа (login.html).

Эндпоинт: Главная (чат)

Теперь опишем обработку главной страницы (/), где будет отображаться наш чат:

@router.get("/", response_class=HTMLResponse)
async def chat_page(
    request: Request,
    user_id: int = Depends(get_optional_current_user),
):
    if user_id:
        return templates.TemplateResponse("index.html", {"request": request})
    else:
        return RedirectResponse(url="/login")

Здесь всё наоборот:

  • Если пользователь авторизован — рендерим index.html со страницей чата.

  • Если нет — редиректим обратно на /login.

Теперь остается описать main-файл прилжения и можно делать запуск проекта!

Финальный штрих — main.py

На этом этапе мы соберём всё, что сделали ранее, в единый рабочий проект. Главный файл main.py будет отвечать за запуск FastAPI-приложения, подключение всех маршрутов, инициализацию базы и статики.

Работаем в файле:

app/main.py

Импорты

Подключим всё необходимое:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from app.api.router import router as api_router
from app.chroma_client.chroma_store import chroma_vectorstore
from app.pages.router import router as page_router

Здесь всё просто: мы импортируем роутеры, клиент для ChromaDB и необходимую утилиту для статики. Обратите внимание — я дал алиасы (as api_router, as page_router), чтобы избежать путаницы.

Жизненный цикл приложения

Теперь определим lifespan — это специальная функция, которая выполняется при старте и завершении приложения:

@asynccontextmanager
async def lifespan(app: FastAPI):
    await chroma_vectorstore.init()
    app.include_router(api_router, prefix="/api", tags=["API"])
    app.include_router(page_router, tags=["ФРОНТ"])
    app.mount("/static", StaticFiles(directory="app/static"), name="static")
    yield
    await chroma_vectorstore.close()

📌 Что здесь происходит:

  1. Инициализируем ChromaDB — вызываем init() для подключения к базе.

  2. Подключаем API-роуты — с префиксом /api (чтобы все API-запросы были по /api/...).

  3. Подключаем маршруты для рендеринга HTML — без префикса.

  4. Подключаем статику — все файлы из app/static будут доступны как /static/....

  5. Закрываем ChromaDB при завершении работы.

Инициализация FastAPI-приложения

Теперь создаем само приложение и указываем ему функцию жизненного цикла:

app = FastAPI(lifespan=lifespan)

Запуск проекта

Открываем терминал и запускаем сервер командой:

uvicorn app.main:app --port 8000 --host 0.0.0.0

🔥 После запуска переходите в браузере на http://localhost:8000 — и проверяйте работу проекта!

Остается только запустить наш проект удаленно и можно будет считать, что проект полностью завершен.

Деплой проекта на Amvera Cloud

Теперь, когда наш проект полностью готов к работе, самое время выкатить его в продакшн — чтобы любой, у кого есть ссылка и доступ (логин с паролем), мог им пользоваться. Для этого мы воспользуемся Amvera Cloud — удобной платформой для быстрого хостинга Python и прочих приложений. Тем более, что у нас уже есть всё необходимое.

Шаги по запуску проекта:

  • Регистрируемся на Amvera Cloud — если аккаунта ещё нет. Новичкам начисляется бонус на баланс, что особенно приятно.

  • Заходим во вкладку «Приложения»:https://cloud.amvera.ru/projects/applications

  • Жмём «Создать приложение».

  • Указываем имя проекта — только латиница, без пробелов и спецсимволов.

  • Выбираем тариф — не ниже "Начальный плюс", так как у нас используется векторная база (ChromaDB), и нужен минимум 1 ГБ ОЗУ.

  • На этапе выбора способа загрузки файлов:

    • Можно перетащить файлы проекта вручную в окно (без venv, он не нужен).

    • Или подключить репозиторий через Git — в интерфейсе будут подробные инструкции.

  • Блок с переменными можно пропустить — ведь мы заранее подготовили файл .env.

  • На финальном окне заполним инструкции как на скрине ниже:

Подключаем домен

После загрузки файлов и создания проекта:

  • Заходим в сам проект → вкладка «Домены»

  • Выбираем:

    • Бесплатный домен от Amvera (быстро и удобно)

    • или подключаем свой собственный домен

Пример домена от Amvera: https://amveraai-yakvenalex.amvera.io

Терпение — залог успеха

Из-за объёма зависимостей (и ChromaDB в том числе) первый запуск может занять 15–20 минут. Не переживай — это нормально.

После загрузки:

  • При переходе по привязанному домену вы увидите форму входа в чат

  • А в интерфейсе Amvera будет статус «Проект запущен»

Если не применился домен - кликните на кнопку "пересобрать проект"
Если не применился домен - кликните на кнопку "пересобрать проект"

Проверим:

Всё готово!

Поздравляю — мы написали и задеплоили полноценное AI-приложение! Оно теперь живёт в интернете и готово к работе с пользователями.

Заключение

Позади — насыщенное и увлекательное путешествие: от первой идеи до полноценного AI-приложения с векторной памятью и удобным веб-интерфейсом. Вместе мы прошли все ключевые этапы создания современной интеллектуальной системы.

Вот что удалось реализовать:

  • Построили чистую архитектуру на базе FastAPI, чётко разделив логику API и фронтенда;

  • Настроили рендеринг HTML-страниц, внедрили авторизацию и реализовали удобный чат-интерфейс;

  • Познакомились с тем, что такое векторная база данных, и научились использовать ChromaDB для хранения и поиска информации по векторным embeddings;

  • Связали базу данных с нейросетевой моделью, чтобы реализовать умный, контекстно-зависимый чат;

  • Настроили асинхронную обработку сообщений, добавили интерфейс с подсветкой кода, возможностью копирования и аккуратным UI;

  • Объединили всё в единый main.py, оформив маршруты и инициализацию всех компонентов проекта;

  • И в финале — развернули проект на Amvera Cloud, сделав его доступным по ссылке для всех желающих.

Особенность этого проекта — прозрачность, простота архитектуры и минимум зависимостей. Всё сделано пошагово, чтобы вам было легко не только разобраться, но и развивать решение дальше.

Вы держите в руках не просто pet-проект, а реальную платформу: её можно масштабировать, адаптировать под заказчика или превратить в полноценный SaaS-сервис.

Если материал оказался полезным — поддержите статью лайком или комментарием. Полный исходный код, а также эксклюзивные материалы, которые не публикуются на Хабре, вы найдёте в моем бесплатном Telegram-канале «Лёгкий путь в Python». Нас там уже более 3300 участников — присоединяйтесь!

На этом всё. Увидимся в следующих проектах.

Только зарегистрированные пользователи могут участвовать в опросе. Войдите, пожалуйста.
Хотите ещё контента по теме ИИ?
63.64% Конечно7
9.09% Возможно1
27.27% Нет3
Проголосовали 11 пользователей. Воздержавшихся нет.
Теги:
Хабы:
Если эта публикация вас вдохновила и вы хотите поддержать автора — не стесняйтесь нажать на кнопку
+5
Комментарии1
Amvera
Amvera — облако для хостинга IT-приложений
Сайт
74
Карма
62
Рейтинг
Алексей Яковенко @yakvenalex

Опытный python разработчик с многолетним стажем.

Отправить сообщение

Публикации

Информация

Сайт
amvera.ru
Дата регистрации
Численность
11–30 человек
Местоположение
Россия
Представитель
Кирилл Косолапов

Ваш аккаунт

Разделы

Информация

Услуги

Техническая поддержка
© 2006–2025, Habr