Друзья, приветствую!
Как вы поняли по названию этой статьи — сегодня мы продолжаем погружаться в тему разработки личного ИИ-ассистента на основе собственных данных.
Краткое напоминание о первой части
Напоминаю, что в предыдущей части мы:
Разобрались с понятием векторной базы данных
Научились собирать данные и реализовывать такую базу
Изучили механизм "умного поиска"
Познакомились с мощным Python-инструментом Langchain
Обеспечили интеграцию нейросетей ChatGPT и Deepseek
Создали консольный вариант диалога с нейросетями на основе собственной базы знаний (документации сервиса Amvera Cloud)
О сегодняшней части
Сегодня мы прокачаем наш проект — превратим его в полноценный веб-сервис с авторизацией и удобным веб-чатом для взаимодействия с нейросетями. Получится некое подобие веб-интерфейса сайта https://chat.openai.com/chat, но с возможностью выбора между Deepseek и ChatGPT, работающими на основе нашей собственной базы знаний.
Для более детального понимания того, что мы будем создавать, я подготовил короткий видео-обзор с комментариями: Видео-демонстрация проекта
Необходимые предварительные условия
Для продолжения вам потребуются:
Готовая векторная база данных на собственных данных (не обязательно документация Amvera Cloud)
API-токены для ChatGPT и Deepseek
Базовое понимание принципов создания бэкенда на FastAPI и основ фронтенд-разработки
План работы
В ходе этой части мы:
Разработаем класс для удобного управления соединением с векторной базой Chroma (с возможностью однократного подключения)
Создадим класс для интеграции нейросетей Deepseek и ChatGPT в FastAPI-приложение с помощью Langchain
Разработаем API проекта для авторизации и отправки запросов к нейросетям
Создадим стильный веб-интерфейс чата со страницами входа и самого чата
Объединим фронтенд с бэкендом в формате полноценного защищенного веб-сервиса
Выполним деплой проекта на платформе 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 есть несколько ключевых моментов:
Отложенная инициализация — соединение с базой данных устанавливается только при явном вызове метода
init(), что позволяет контролировать момент подключения в жизненном цикле приложения.Асинхронность — все методы являются асинхронными (
async), что позволяет эффективно работать с базой данных в контексте асинхронного веб-приложения FastAPI.Использо��ание GPU — если доступно CUDA-совместимое устройство, модель эмбеддингов будет использовать его для ускорения вычислений.
Логирование — все важные события и ошибки записываются в лог с помощью библиотеки
loguru, что упрощает отладку и мониторинг.Обработка ошибок — все операции обернуты в блоки try-except для корректной обработки ошибок.
Данная реализация обеспечивает эффективное использование соединения с базой данных Chroma в рамках всего жизненного цикла приложения FastAPI.
К этому файлу мы ещё вернемся сегодня, но пока переключимся на реализацию класса для интеграции нейросетей Deepseek и ChatGPT в наш сервис.
Класс для интеграции нейросетей
Описание задачи
Прежде чем приступить к реализации класса, напомню о функциональных требованиях к нашему механизму взаимодействия с нейросетями:
Принимать текстовый запрос от пользователя
На основании запроса возвращать релевантные документы из базы данных Chroma
Формировать контекст из полученных данных Chroma, пользовательского запроса и системного промпта
Передавать этот контекст выбранной нейросети (ChatGPT или Deepseek)
Выдавать ответ не целым куском, а в формате потокового вывода (стрима)
Весь этот процесс должен работать асинхронно для обеспечения эффективной работы веб-приложения.
Реализация класса 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 содержит несколько ключевых особенностей, которые стоит отметить:
Поддержка нескольких провайдеров — класс позволяет выбирать между ChatGPT от OpenAI и Deepseek, что дает гибкость в использовании различных языковых моделей в зависимости от потребностей и доступности.
Асинхронный потоковый вывод — метод
astream_responseреализован как асинхронный генератор, что позволяет постепенно передавать части ответа клиенту по мере их генерации моделью, без необходимости ожидать полный ответ.Форматирование запроса — для каждого запроса формируется специальный системный промпт, определяющий роль и контекст работы модели, что помогает получать более релевантные и структурированные ответы.
Расширенное логирование — все ключевые этапы работы класса сопровождаются детальным логированием, что упрощает отладку и мониторинг работы системы.
Обработка ошибок — все потенциально опасные операции обернуты в блоки try-except, что обеспечивает устойчивость работы приложения даже при возникновении ошибок взаимодействия с API нейросетей.
Реализуем систему авторизации
Так как в нашем сервисе будут задей��твованы платные токены — очевидно, что мы захотим защищать свой сервис от использования его сторонними людьми и в немеренных количествах.
Логика авторизации
Логика авторизации будет следующей:
Для пользования сервисом необходимо будет ввести логин и пароль в форму входа
После корректного ввода данных мы будем выдавать пользователю специальный JWT-токен, который мы будем помещать в куки сессию. У этого токена будет свой срок жизни (1 час). То есть после того как этот час пройдет — мы сделаем так, чтоб токен был недействительным. Недействительный токен равняется выходу из системы и полному закрытию функционала.
Если логин или пароль не корректны либо токен истек — доступа к системе у пользователя не будет.
Создание базы данных пользователей
Для экономии времени я решил использовать простой 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— для подключения к Chromaget_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": "Ничего не найдено"}
Что здесь происходит:
Получаем документы из базы Chroma
Формируем AI-контекст — объединяем содержимое документов в одну строку, разделённую переносами
Инициализируем
ChatWithAIс выбранным провайдером (deepseekилиchatgpt)Создаем генератор
stream_response, который будет по кусочкам отдавать результат от нейросетиВозвращаем
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('Произошла ошибка при входе') } })
Здесь реализован стандартный подход:
Перехватываем отправку формы.
Забираем логин и пароль.
Отправляем
POST-запрос на/api/login.Если получили
Logged in, то редиректим пользователя на главную.В случае ошибки — показываем
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
Вот что происходит при отправке запроса:
Проверка, ввел ли пользователь текст.
Добавляется пользовательское сообщение в чат.
Показывается индикатор загрузки.
Отправляется запрос на эндпоинт
/api/ask_with_ai.Ответ читается потоком (stream).
Полученные символы по одному добавляются в чат с задержкой в 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()
📌 Что здесь происходит:
Инициализируем ChromaDB — вызываем
init()для подключения к базе.Подключаем API-роуты — с префиксом
/api(чтобы все API-запросы были по/api/...).Подключаем маршруты для рендеринга HTML — без префикса.
Подключаем статику — все файлы из
app/staticбудут доступны как/static/....Закрываем 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 участников — присоединяйтесь!
На этом всё. Увидимся в следующих проектах.
