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

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

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

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

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

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

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

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

• Познакомились с мощным 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 участников — присоединяйтесь!

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