Отчет по лабораторной работе №3

Выполнил: Шафиков Максим Азатович

Факультет: ПИН (ИКТ)

Группа: К3339

Преподаватель: Говоров Антон Игоревич

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

🔹 1. Цель работы

Целью данной лабораторной работы является проектирование и реализация серверной части веб-приложения "Edulytica" с использованием микросервисной архитектуры. Необходимо разработать набор отказоустойчивых, масштабируемых и безопасных сервисов для обработки запросов от клиентской части, управления данными пользователей, взаимодействия с моделями машинного обучения и выполнения сложных асинхронных задач по обработке документов.

🔹 2. Задание

Спроектировать и реализовать микросервисную архитектуру приложения.
Реализовать сервис аутентификации и авторизации пользователей с использованием JWT.
Создать основной API-сервис для управления бизнес-логикой: создание "тикетов" на обработку документов, управление мероприятиями (events), получение результатов.
Разработать сервис-оркестратор для управления сложными, многоэтапными задачами (пайплайнами), которые включают в себя вызовы различных LLM-моделей.
Реализовать сервисы-воркеры для моделей машинного обучения, которые получают задачи через брокер сообщений Kafka.
Интегрировать сервис RAG (Retrieval-Augmented Generation) для обогащения запросов к моделям контекстом из базы знаний.
Настроить API Gateway (Nginx) для маршрутизации внешних запросов к соответствующим внутренним сервисам.
Обеспечить взаимодействие между сервисами с помощью асинхронных HTTP-запросов и брокера сообщений Kafka.

🔹 3. Используемые технологии

Язык программирования: Python 3.11+
Фреймворк: FastAPI
База данных: PostgreSQL
ORM и миграции: SQLAlchemy, Alembic
Валидация данных: Pydantic
Аутентификация: JWT (JSON Web Tokens)
Асинхронные задачи и очереди: Kafka
Кэширование и управление состоянием: Redis
Векторная база данных (для RAG): ChromaDB
API Gateway: Nginx
Контейнеризация: Docker, Docker Compose

🔹 4. Структура проекта

Проект имеет микросервисную архитектуру. Основные компоненты расположены в директории src/:

auth/: Сервис аутентификации. Отвечает за регистрацию, вход, выход из системы и управление токенами.
edulytica_api/: Основной API-сервис. Является главной точкой входа для клиентского приложения. Обрабатывает запросы, связанные с действиями пользователя (создание тикетов, загрузка файлов, получение результатов), и взаимодействует с другими сервисами.
orchestration/: Сервис-оркестратор. Управляет выполнением сложных пайплайнов обработки документов. Он разбивает задачу на подзадачи, управляет их зависимостями и состоянием, формирует промпты и отправляет их в Kafka.
models/: Сервисы-воркеры для LLM. Каждый такой сервис "слушает" свою тему в Kafka, получает задачу, обрабатывает ее с помощью конкретной языковой модели (например, Qwen или Vikhr) и отправляет результат в общую тему для результатов.
rag/: Сервис Retrieval-Augmented Generation. Предоставляет API для обогащения промптов контекстом. Он управляет векторной базой данных ChromaDB, куда предварительно загружаются и индексируются тексты.
gateway/: Конфигурация Nginx, который выступает в роли обратного прокси. Он принимает все входящие HTTP-запросы и маршрутизирует их на соответствующие сервисы (auth, edulytica_api).
common/: Общий код, используемый в нескольких сервисах. Включает в себя модели SQLAlchemy, CRUD-операции для работы с БД, утилиты для аутентификации, конфигурацию и т.д.
llm/: Абстракции и реализации для работы с различными языковыми моделями.

🔹 5. Реализованный функционал

Аутентификация и управление пользователями

Реализован полный цикл аутентификации.

Регистрация: Пользователь регистрируется с помощью email и пароля. На почту отправляется код подтверждения.
Логин: После успешного ввода логина и пароля пользователь получает access_token и refresh_token. refresh_token устанавливается в HttpOnly cookie для безопасности.
Обновление токена: access_token имеет короткий срок жизни. Когда он истекает, фронтенд может запросить новый, используя refresh_token.
Управление профилем: Пользователь может изменять свои данные (имя, организацию) и пароль.

Пример эндпоинта для входа в систему из src/auth/routers/auth.py:

@api_logs(auth_router.post('/login'))
async def login_handler(
        response: Response,
        login: str = Body(...),
        password: str = Body(...),
        session: AsyncSession = Depends(get_session)
):
    # ... проверка пользователя и пароля ...
    user = await UserCrud.get_active_user(session=session, login=login)
    if not user or not verify_password(password, user.password_hash):
        raise HTTPException(
            status_code=HTTP_401_UNAUTHORIZED,
            detail='Credentials are incorrect'
        )

    # ... создание токенов ...
    access_token = create_access_token(user.id)
    checker = uuid.uuid4()
    refresh_token = create_refresh_token(subject=user.id, checker=checker)

    # ... сохранение refresh_token в БД и установка в cookie ...
    response.set_cookie(
        key="refresh_token",
        value=f"Bearer {refresh_token}",
        httponly=True,
        expires=get_expiry(REFRESH_TOKEN_EXPIRE_MINUTES)
    )
    return {'detail': 'Credentials are correct', 'access_token': access_token}

Обработка документов и запуск пайплайнов Это основная бизнес-логика приложения. 1. Загрузка файла: Пользователь через edulytica_api загружает документ (.pdf, .docx) и выбирает тип задачи (mega_task_id). 2. Создание тикета: В базе данных создается "тикет", связывающий пользователя, документ и задачу. 3. Запуск оркестрации: edulytica_api отправляет асинхронный запрос в сервис orchestration, передавая ID тикета, текст документа и ID задачи. Это происходит в фоновом режиме, чтобы не блокировать пользователя. 4. Работа оркестратора: - Оркестратор получает запрос и на основе mega_task_id определяет граф зависимостей подзадач. - Он инициализирует состояние тикета в Redis. - Начинается выполнение подзадач, у которых нет зависимостей. - Для каждой подзадачи формируется промпт. Если подзадача требует RAG, оркестратор обращается к rag-сервису для обогащения промпта. - Задача с промптом отправляется в нужную тему Kafka (например, llm_tasks.qwen). 5. Обработка моделью: - Сервис-воркер (models) получает задачу из Kafka. - Он выполняет инференс модели и получает результат. - Результат отправляется в общую тему llm_tasks.result. 6. Обновление состояния: - Оркестратор слушает тему llm_tasks.result. Получив результат, он обновляет состояние подзадачи в Redis (например, на COMPLETED) и сохраняет результат. - Затем он проверяет, можно ли запустить следующие подзадачи, зависимости которых теперь выполнены. 7. Завершение: Когда все подзадачи выполнены, оркестратор собирает итоговый отчет и через внутренний API-вызов к edulytica_api сохраняет его в базу данных, помечая тикет как "завершенный".

Пример запуска пайплайна в оркестраторе из src/orchestration/orchestrator.py:

class Orchestrator:
    # ...
    async def run_pipeline(
            self, ticket_id: Union[str, uuid.UUID], document_text: str
    ):
        dependencies = self.TASKS.get(self.mega_task_id)
        # ...
        await self.state_manager.create_ticket(
            ticket_id=ticket_id,
            # ...
        )

        initial_subtasks_to_run = []
        for task_id, subtasks in dependencies.items():
            for subtask_id, details in subtasks.items():
                if not details.get("dependencies"):
                    initial_subtasks_to_run.append(subtask_id)

        execution_tasks = [
            self._execute_subtask(ticket_id, subtask_id, document_text)
            for subtask_id in initial_subtasks_to_run
        ]

        if execution_tasks:
            await asyncio.gather(*execution_tasks)

🔹 6. Вывод

В ходе выполнения лабораторной работы была успешно спроектирована и реализована серверная часть приложения на основе микросервисной архитектуры с использованием FastAPI. Были созданы ключевые сервисы, отвечающие за аутентификацию, обработку бизнес-логики, асинхронную оркестрацию сложных задач и взаимодействие с LLM-моделями через брокер сообщений Kafka. Все было развернуто с помощью docker-compose для упрощения управления и масштабирования. Применение микросервисного подхода и асинхронных технологий позволило создать гибкую и масштабируемую систему, способную эффективно обрабатывать ресурсоемкие задачи анализа документов без блокировки пользовательского интерфейса. Реализованная архитектура закладывает прочный фундамент для дальнейшего развития функционала платформы.