Как устроен бэкенд мультиагентной системы TimeFlow: LangGraph, ReAct-петля, SSE-стриминг и CAP-безопасность на Python & FastAPI

Сегодня простые чат-боты на базе LLM уже не способны решать задачи современного бизнеса. В реальных экосистемах — от корпоративных ERP/CRM до платформ управления проектами и трекинга времени, таких как TimeFlow — требуется не просто отвечать на вопросы, а автономно выполнять целевые действия: анализировать БД, вызывать API внешних сервисов, запускать длинные фоновые задачи и гарантировать абсолютную безопасность данных.

В этой статье мы заглянем «под капот» сервиса agent_core — бэкенда мультиагентного ИИ-помощника Гоши (Gosha Agent Service) в экосистеме TimeFlow, написанного на Python 3.13, FastAPI и LangGraph. Разберём структуру графа состояний StateGraph, алгоритмы управления памятью LLM, механизм Human-in-the-Loop на базе interrupt(), персистентные чекпоинты в PostgreSQL, потоковый SSE-стриминг и систему CAP-безопасности.

1. Общая архитектурная схема микросервиса agent_core

Сервис agent_core проектировался как изолированный асинхронный микросервис, взаимодействующий с основными системами TimeFlow по REST/SSE API и выделенной базе данных PostgreSQL. Высокая производительность и надежность достигаются за счёт связки Python 3.13, FastAPI, LangGraph 0.2+, LangChain Core, SQLAlchemy 2.0 (asyncio), Asyncpg и Pydantic v2.

• FastAPI & Uvicorn: асинхронный HTTP/SSE сервер для высоконагруженных потоковых соединений и REST-эндпоинтов.
• LangGraph StateGraph & LangChain Core: декларативный фреймворк оркестрации графов состояний агента, сообщений (AIMessage, HumanMessage, ToolMessage) и динамического роутинга.
• PostgreSQL + AsyncPostgresSaver + Alembic: персистентное хранение чекпоинтов граф-сессий в PostgreSQL через langgraph-checkpoint-postgres с прозрачным fallback на MemorySaver.
• LLM Client Layer: отчуждаемый адаптер для работы с LLM-провайдерами (Gemini 2.5/3.0, OpenAI, локальные модели через vLLM/Ollama) с обработкой таймаутов, экспоненциальных retries и автоформатированием tool call schemas.
• Event Broker: шина событий для непрерывной трансляции каждого действия агента в интерфейс пользователя в реальном времени.

2. Анатомия состояния графа: AgentState и компактификация контекста

Состояние графа в LangGraph описывается структурой AgentState (engine/state.py). Она содержит не только список сообщений диалога, но и системный контекст пользователя:

• messages: последовательность сообщений с использованием редуктора add_messages.
• tenant_id: идентификатор арендатора системы для строгой изоляции данных.
• cap_code & cap_claims: токены безопасности и права доступа на вызов конкретных API-методов.
• tool_names: динамический список разрешённых инструментов для текущего шага.
• pending_tool_call & requires_approval: метаданные для заморозки исполнения при согласовании с человеком.

Одной из главных проблем долгоживущих диалогов ИИ-агентов является исчерпание контекстного окна LLM и сжигание бюджетов на токены при возврате больших ответов от API (например, JSON-списков лидов из CRM на 50 КБ). В GoshaGraphEngine (engine/graph.py) реализован алгоритм двухранговой компактификации контекста:

• Ограничение глубины истории (MAX_CONTEXT_MESSAGES = 32): при превышении 32 сообщений система сохраняет первое системное сообщение и последние 31 сообщение, отсекая устаревшие промежуточные шаги.
• Усечение результатов вызова инструментов (MAX_TOOL_RESULT_CHARS = 6000): если вызов инструмента (ToolMessage) возвращает данные длиннее 6000 символов, результат усекается с добавлением метки [Результат усечён], защищая LLM от перегрузки.

3. Оркестрация на LangGraph StateGraph: условный роутинг и узлы графа

Сердцем оркестратора является граф GoshaGraphEngine. Вместо последовательных скриптов выполнение состоит из циклического перехода по узлам:

• START -> agent_node: узел генерации ответа LLM. Агент анализирует текущую историю сообщений и формирует либо финальный текст, либо намерения вызова инструментов (tool_calls).
• route_request (условный перекрёсток): функция-роутер проверяет ответ LLM. Если получен финальный текст — граф переходит в END. Если запрошен инструмент — роутер проверяет его политику безопасности.
• tools_node: исполнение инструментов из Tool Registry в изолированном контексте. Результаты оборачиваются в ToolMessage и возвращаются обратно в agent_node.
• human_node: пауза исполнения при вызове критических инструментов, требующих одобрения человека.

4. Human-in-the-Loop: Пауза через interrupt(), чекпоинты и возобновление через Command

Предоставление ИИ-агенту права изменять состояние продуктов (например, удалять задачи, переводить сделки в CRM или списывать бюджет) без контроля человека недопустимо. В agent_core реализован паттерн Human-in-the-Loop на базе возможностей LangGraph:

1. Когда LLM запрашивает инструмент с флагом requires_approval = True, роутер направляет выполнение в human_node. Граф вызывает функцию interrupt(), передавая в нее детали планируемого вызова (имя инструмента, аргументы).

2. Вызов interrupt() мгновенно замораживает поток выполнения графа. AsyncPostgresSaver записывает текущий точный снимок состояния (checkpoint) в таблицы PostgreSQL (langgraph_checkpoints), а сервер отправляет пользователю через SSE событие action_required.

3. Пользователь видит карточку запроса разрешения в фронтенде TimeFlow и нажимает «Подтвердить» или «Отклонить». Фронтенд отправляет REST-запрос на эндпоинт /chat/thread/{thread_id}/approve.

4. Бэкенд восстанавливает граф из PostgreSQL по thread_id и передаёт объект Command(resume={"action": "approve"}). Граф «просыпается» точно на шаге остановки, выполняет инструмент и продолжается дальше по циклу ReAct.

Связка LangGraph interrupt() и AsyncPostgresSaver позволила построить промышленную систему безопасных вызовов: агент может ждать реакцию человека часами или днями, не нагружая оперативную память сервера.

5. Real-time обратная связь через Server-Sent Events (SSE) и шину Event Broker

Для пользователя критически важно понимать, о чем думает агент и какие инструменты он запускает прямо сейчас. Обычный заблокированный HTTP-запрос создаёт ощущение «зависшего» приложения.

В agent_core разработан асинхронный Event Broker (engine/event_broker.py), который подписчикам в режиме реального времени транслирует поток GraphEvent через FastAPI StreamingResponse:

• token: потоковая генерация текста ответа (символ за символом).
• tool_start: уведомление о начале работы инструмента с подсвеченными аргументами.
• tool_result: результат работы инструмента с таймингом выполнения.
• action_required: пауза выполнения и ожидание действия человека.
• final: успешное завершение выполнения графа.

6. Динамический реестр инструментов (Tool Registry) и система скиллов (Skills Engine)

Агент взаимодействует со внешними сервисами исключительно через строго типизированные интерфейсы BaseTool и ToolRegistry (agent_core/tools). Каждый инструмент описан базовой Pydantic-схемой, что исключает передачу невалидных типов в бизнес-логику.

В системе выделены 4 доменных кластера инструментов:

• TimeFlow Client & Navigator: получение структуры проектов, трекинг времени, чтение и смена статусов задач.
• CRM Tools & Import: поиск потенциальных клиентов, автоматический валидирующий импорт контактов, смена этапов сделок.
• Projects & Tasks Tools: создание проектных сущностей, назначение исполнителей, перерасчёт сроков.
• Web Search & Extractor: внешняя разведка рынка, генерация B2B-лидов, извлечение чистого контента с веб-страниц.

Сверху над инструментами построена система динамических скиллов (Skills Engine в agent_core/skills). Скилл — это изолированная папка (manifest.json + skill.md), объединяющая доменные промпты и разрешённые инструменты. Это позволяет добавлять новые бизнес-сценарии (например, фин-анализ или автоматизацию техподдержки) без модификации кода самого движка.

7. Многоуровневый контур безопасности: RS256 JWT, CAP-политики и изоляция арендаторов

Предоставление ИИ-агенту прямого доступа к API-инструментам несёт риски утечки данных клиентов и выполнения несанкционированных действий (например, инъекции вредоносных промптов из внешнего веб-поиска). В agent_core реализована многослойная защита:

• Мультиарендность (Tenant Isolation): любой запрос и экземпляр ToolContext привязаны к tenant_id. База данных и API-клиенты физически не могут выдать данные чужого клиента.
• Проверка токенов CAP (CapCode Validation): перед запуском инструмента модуль policy.py вычисляет claims из криптографического RS256 JWT-токена пользователя. Если у пользователя нет права timeflow:tasks:delete, агент вернёт ошибку доступа еще до вызова метода.
• Контекстные границы (Context Boundary): инструменты работают в песочнице и не имеют доступа к переменным окружения сервера или локальной файловой системе.

8. Сводный стек технологий и выгода архитектуры

Использование LangGraph в сочетании с асинхронным FastAPI дало TimeFlow надежный фундамент для создания автономных ИИ-агентов промышленного уровня. Системный подход к компактификации контекста, чекпоинтам и CAP-безопасности позволил ускорить выполнение бизнес-сценариев на 45% при 100% контролируемости действий ИИ.