Интеграция мульти-агентного ядра: связка LangGraph, LangChain и локальной инфраструктуры Ollama
Интеграция мульти-агентного ядра: связка LangGraph, LangChain и локальной инфраструктуры Ollama
В корпоративной среде 90% ИИ-прототипов успешно справляются с демонстрационными (happy path) сценариями, но рассыпаются при столкновении с реальными бизнес-процессами. Линейные RAG-пайплайны, какими бы сложными они ни были, не способны к саморефлексии: если база данных возвращает пустой ответ, линейная система просто транслирует этот отказ пользователю. Переход от хрупкого прототипа к отказоустойчивому MVP требует смены парадигмы — от конвейеров к конечным автоматам. На этом этапе мы синтезируем ранее изученные компоненты: вычислительную мощь локальных моделей Ollama, интеграционные абстракции LangChain и графовую оркестрацию LangGraph в единое, автономное ядро, способное принимать решения, ошибаться и самостоятельно исправлять свои ошибки до возврата ответа пользователю.
Архитектурная триада: Роли компонентов в синтезированном ядре
В мульти-агентной системе границы между библиотеками стираются, но для корректного проектирования необходимо жестко разграничить их зоны ответственности. Ошибка многих разработчиков заключается в попытке реализовать логику ветвления внутри промптов LLM или, наоборот, в написании сотен строк императивного Python-кода для парсинга ответов.
В нашем корпоративном MVP распределение ролей выглядит следующим образом:
- Ollama (Двигатель вычислений): Отвечает исключительно за вероятностную трансформацию текста. Локальная Llama 3, запущенная через Ollama, не знает о существовании базы данных или веб-поиска. Её задача — получив контекст и схему доступных инструментов, сгенерировать валидный JSON-объект с решением или аргументами для вызова функции.
- LangChain (Сенсорный и моторный аппарат): Предоставляет стандартизированные контракты (интерфейсы
Runnable). LangChain оборачивает вызовы к Qdrant, форматирует промпты, валидирует Pydantic-схемы и выполняет фактический вызов Python-функций (инструментов), когда модель запрашивает действие. - LangGraph (Префронтальная кора): Хранит глобальное состояние запроса и управляет потоком управления. Это конечный автомат, который решает, какой узел должен активироваться следующим. Если LangChain сообщает, что инструмент вернул ошибку, именно LangGraph направляет поток обратно в Ollama с требованием исправить аргументы.
Такое разделение позволяет нам менять локальную модель на облачную, или PostgreSQL на MongoDB, не переписывая логику принятия решений.
Проектирование глобального состояния (Global State)
Кровеносной системой LangGraph является объект State. В линейных цепочках данные передаются от шага к шагу, мутируя в процессе. В графах состояние глобально, и каждый узел либо читает из него, либо возвращает дельту (изменения), которые сливаются с текущим состоянием через редукторы.
Для корпоративного MVP, объединяющего RAG и вызов инструментов, состояние должно быть достаточно емким, чтобы хранить контекст, но структурированным, чтобы не переполнить окно контекста модели.
from typing import Annotated, List, Sequence, TypedDict
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
sender: str
documents: List[dict]
error_count: int
requires_human: bool
Разберем анатомию этого состояния:
- Поле
messagesиспользует редукторoperator.add. Это означает, что когда узел возвращает новый словарь{"messages": [AIMessage(...)]}, LangGraph не перезаписывает историю, а добавляет новое сообщение в конец списка. Это критически важно для сохранения истории вызовов инструментов (Tool Calls). - Поле
senderхранит идентификатор последнего активного агента (например,supervisor,rag_agent,sql_agent). У него нет редуктора, поэтому возврат нового значения полностью перезапишет предыдущее. - Поле
documentsрезервируется для хранения сырых результатов из Qdrant. Выделение документов в отдельное поле, а не внедрение их сразу вmessages, позволяет узлам-оценщикам (Grader Nodes) анализировать релевантность фактов независимо от сгенерированного ответа. error_count— механизм защиты от бесконечных циклов (Graceful Degradation), о котором мы поговорим ниже.
Адаптация локальной Llama 3 для детерминированного Tool Calling
Главный вызов при использовании локальных моделей (в отличие от коммерческих API) — их склонность к галлюцинациям в структуре JSON. Даже Llama 3 8B может забыть закрыть скобку или придумать аргумент, не описанный в Pydantic-схеме инструмента.
Для надежной интеграции с LangChain мы используем метод bind_tools, но усиливаем его на уровне системного промпта и парсеров.
from langchain_community.chat_models import ChatOllama
from langchain_core.utils.function_calling import convert_to_openai_tool
# Инициализация модели с жесткими параметрами
llm = ChatOllama(
model="llama3.1:8b",
temperature=0.0,
format="json" # Принудительный JSON-режим Ollama
)
# Допустим, у нас есть инструмент поиска по Qdrant
tools = [corporate_rag_search, sql_database_query]
# Привязка инструментов
agent_llm = llm.bind_tools(tools)
Включение format="json" на уровне Ollama заставляет движок llama.cpp отбрасывать любые токены, которые нарушают синтаксис JSON. Однако это не спасает от семантических ошибок (например, передачи строки вместо числа).
Поэтому в мульти-агентном ядре мы не полагаемся на то, что модель ответит правильно с первого раза. Мы закладываем ожидание ошибки в саму топологию графа.
Паттерн Supervisor: Маршрутизация на основе намерений
Вместо того чтобы создавать одного «бога-агента» с доступом ко всем базам данных и API, корпоративная архитектура требует паттерна Supervisor (Супервизор). Супервизор — это легковесный узел, который не выполняет работу сам, а только анализирует запрос и решает, какому специализированному агенту передать управление.
Для реализации Супервизора мы используем структурированный вывод LangChain (Structured Output). Мы заставляем модель вернуть строго один из предопределенных вариантов.
from pydantic import BaseModel, Field
from typing import Literal
class RouteDecision(BaseModel):
"""Схема для маршрутизатора"""
next_agent: Literal["rag_agent", "sql_agent", "FINISH"] = Field(
description="Выбор следующего агента для выполнения задачи."
)
supervisor_prompt = """Ты — главный координатор.
Проанализируй запрос и историю диалога.
Если вопрос касается регламентов, отпусков или текстовых документов -> rag_agent.
Если вопрос касается метрик, транзакций или табличных данных -> sql_agent.
Если задача полностью решена -> FINISH."""
supervisor_chain = prompt | llm.with_structured_output(RouteDecision)
В графе функция маршрутизации будет выглядеть так:
def supervisor_node(state: AgentState):
decision = supervisor_chain.invoke({"messages": state["messages"]})
return {"sender": decision.next_agent}
Условное ребро (Conditional Edge) в LangGraph прочитает поле state["sender"] и физически перенаправит поток выполнения на соответствующий узел.
Циклы самокоррекции и изящная деградация (Graceful Degradation)
Самая мощная часть синтезированного ядра — способность исправлять собственные ошибки. Рассмотрим сценарий: Супервизор направил задачу в sql_agent. Агент сгенерировал SQL-запрос, но ошибся в названии таблицы. База данных PostgreSQL вернула ошибку relation "users_data" does not exist.
В линейной архитектуре пользователь получил бы сырую ошибку БД. В LangGraph мы создаем цикл самокоррекции.
Логика узла, выполняющего инструменты (Tool Node), оборачивается в блок try/except. Если возникает ошибка, узел не падает, а формирует специальное сообщение ToolMessage с текстом ошибки и увеличивает счетчик error_count.
def tool_execution_node(state: AgentState):
last_message = state["messages"][-1]
# Защита от бесконечного цикла
if state.get("error_count", 0) >= 3:
return {
"messages": [AIMessage(content="Я не смог выполнить задачу из-за технических проблем.")],
"sender": "FINISH",
"requires_human": True
}
try:
# Попытка выполнить запрошенный инструмент
result = execute_tool(last_message.tool_calls[0])
return {
"messages": [ToolMessage(content=str(result), tool_call_id=last_message.tool_calls[0]["id"])],
"error_count": 0 # Сброс счетчика при успехе
}
except Exception as e:
# Возврат ошибки обратно в модель
error_msg = f"Ошибка выполнения: {str(e)}. Исправь аргументы и попробуй снова."
return {
"messages": [ToolMessage(content=error_msg, tool_call_id=last_message.tool_calls[0]["id"])],
"error_count": state.get("error_count", 0) + 1
}
Граф устроен так, что после tool_execution_node поток всегда возвращается к агенту, который этот инструмент вызвал. Агент видит в истории ToolMessage с ошибкой, анализирует её (понимает, что таблицы users_data нет), генерирует новый ToolCall с исправленным запросом, и цикл повторяется.
Если локальная модель недостаточно умна и продолжает ошибаться, срабатывает условие error_count >= 3. Это реализация паттерна Graceful Degradation (Изящная деградация). Система прекращает попытки, честно сообщает о сбое и устанавливает флаг requires_human = True, который в будущем позволит перехватить эту сессию оператору (Human-in-the-Loop).
Математика задержек в циклических графах
Синтез мульти-агентной системы неизбежно влияет на производительность. В линейном API мы боролись за оптимизацию TTFT (Time-to-First-Token). В графовой архитектуре с локальными моделями мы должны учитывать кумулятивную задержку системы.
Полная задержка мульти-агентного ответа описывается следующим образом:
Где:
- — количество итераций (циклов) в графе до достижения узла
FINISH. - — время инференса локальной модели на -й итерации, которое зависит от количества входных токенов (растущего с каждым шагом из-за накопления истории) и выходных токенов .
- — время выполнения фактического инструмента (например, векторного поиска в Qdrant) на -й итерации.
- — накладные расходы LangGraph на маршрутизацию и сериализацию состояния.
Из этой формулы вытекает важное архитектурное правило: каждый возврат по циклу (ошибка инструмента или рефлексия) увеличивает для следующего шага. Поскольку локальная Llama 3 обрабатывает контекст медленнее коммерческих API, длинные циклы самокоррекции могут привести к неприемлемому времени ожидания (Abandonment Rate). Именно поэтому жесткое ограничение error_count и использование легковесного Супервизора критически важны для сохранения юнит-экономики локального MVP.
Сборка графа: Компиляция конечного автомата
Завершающим этапом синтеза ядра является физическая сборка узлов и ребер в единый граф.
from langgraph.graph import StateGraph, END
# Инициализация графа с нашей структурой состояния
workflow = StateGraph(AgentState)
# Добавление узлов (вычислительных блоков)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("rag_agent", rag_agent_node)
workflow.add_node("sql_agent", sql_agent_node)
workflow.add_node("tools", tool_execution_node)
# Установка стартовой точки
workflow.set_entry_point("supervisor")
# Добавление условных ребер от Супервизора
workflow.add_conditional_edges(
"supervisor",
lambda state: state["sender"],
{
"rag_agent": "rag_agent",
"sql_agent": "sql_agent",
"FINISH": END
}
)
# Маршрутизация от агентов к инструментам или обратно к супервизору
workflow.add_conditional_edges(
"rag_agent",
lambda state: "tools" if state["messages"][-1].tool_calls else "supervisor"
)
# Инструменты всегда возвращают управление тому агенту, который их вызвал
workflow.add_conditional_edges(
"tools",
lambda state: state["sender"]
)
# Компиляция графа
app = workflow.compile()
Скомпилированный app представляет собой объект Runnable, который полностью совместим с экосистемой LangChain. Мы можем вызывать его через app.invoke(), передавая начальное состояние с запросом пользователя. Внутри этого черного ящика агенты будут переговариваться, вызывать инструменты, ошибаться, исправляться и, в конечном итоге, выдавать проверенный результат.
Однако на данном этапе наше глобальное состояние AgentState живет исключительно в оперативной памяти (RAM) процесса Python. Если сервер перезагрузится во время выполнения долгого цикла или если мы захотим масштабировать систему на несколько воркеров Celery, контекст диалога и промежуточные шаги графа будут безвозвратно потеряны. Для превращения этого мощного вычислительного ядра в полноценный корпоративный бэкенд, состояние графа необходимо сделать персистентным, связав его с надежным хранилищем.