Создайте маршрутизатор с несколькими агентами с использованием LangGraph и агента-надзирателя

Если поставить одного агента для работы с реальным продуктом, он быстро начинает делать слишком много. Если попросить одного агента заниматься обслуживанием клиентов, отслеживанием заказов и возвратами средств, он окажется посредственным во всем трёх аспектах. Инструкции для его работы превращаются в 2000-словный клубок формулировок вроде «если пользователь спрашивает о X, делай Y, но если речь также идёт о Z, тогда...». Решением не является создание более умного агента — это больше мелких агентов плюс один руководитель, который координирует их действия. По сути, это самый распространённый подход в искусственном интеллекте с агентами на данный момент, причём рабочая версия такого решения помещается в небольшую программу на LangGraph.

Проблемы с одним гигантским агентом

Представьте, что вы создаёте ассистента, который отвечает на общие вопросы, выполняет арифметические операции и ищет актуальную информацию в Интернете. Привлекательным решением кажется использование одного агента, одного большого промпта и всех необходимых инструментов. Однако по мере накопления инструментов и инструкций модель теряет уверенность в выборе подходящего инструмента, а её логика становится менее чёткой. Отладка превращается в настоящий кошмар, поскольку невозможно выяснить, почему модель выбрала неправильный инструмент, если всё находится в одном промпте.

Решением является разделение обязанностей: один агент для математики, другой — для поиска информации, а также руководитель, который выступает в роли маршрутизатора и определяет, какой специалист будет заниматься каждой частью запроса.

Паттерн руководителя

Представьте колл-центр. Вы звоните с вопросом, но вас не соединяют напрямую с экспертом. Сначала разговаривает рецепционист, который определяет, нужна ли вам служба биллинга, техническая поддержка или продажи, и затем переключает вас. Рецепционист является руководителем, а три группы — специалистами-агентами. Руководитель никогда не отвечает на вопрос сам, его единственная задача — выбрать того, кто должен это сделать.

Визуально это выглядит так:

                    ┌─────────────┐
                    │    User     │
                    │   Request   │
                    └──────┬──────┘
                           │
                           ▼
                   ┌───────────────┐
                   │   Supervisor  │
                   │  (routes task)│
                   └───────┬───────┘
                    ┌──────┴──────┐
                    ▼             ▼
             ┌────────────┐ ┌────────────┐
             │  Research  │ │    Math    │
             │   Agent    │ │   Agent    │
             └──────┬─────┘ └──────┬─────┘
                    │              │
                    └──────┬───────┘
                           ▼
                   ┌───────────────┐
                   │   Supervisor  │
                   │ (final answer)│
                   └───────────────┘

Надзиратель берет задачу, передает её соответствующему специалисту, получает результат, а затем решает, следует ли снова направить задачу или передать пользователю окончательный ответ.

Рекомендуемый способ реализации, шаг за шагом

Важный момент, который стоит знать перед написанием кода: команда LangChain теперь рекомендует создавать структуру надзирателя непосредственно на основе вызовов инструментов, а не полагаться только на старую вспомогательную библиотеку langgraph-supervisor, поскольку вызовы инструментов позволяют лучше контролировать формирование контекста. Сначала вы увидите готовую версию, а затем ручную реализацию, чтобы было понятно, что происходит на уровне кода.

Шаг 1: установка пакетов

pip install langgraph langchain-openai langgraph-supervisor

Шаг 2: создание специалистов

Начните с импорта:

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

Первый специалист занимается математикой, а её инструментами являются две простые функции. Сначала идёт сложение:

# Specialist 1: handles math
def add(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b

Затем умножение:

def multiply(a: float, b: float) -> float:
    """Multiply two numbers together."""
    return a * b

Загрузите их в агента ReAct с именем и промптом, которые ограничивают его роль:

math_agent = create_react_agent(
    model="openai:gpt-4o",
    tools=[add, multiply],
    name="math_expert",
    prompt="You are a math specialist. Solve numeric problems precisely."
)

Второй специалист занимается исследованиями. Одним из его инструментов является функция поиска, показанная здесь в упрощенном виде:

# Specialist 2: handles research
def web_search(query: str) -> str:
    """Search the web for current information."""
    # In production, replace this with a real search API call
    return f"Search results for '{query}': [your real search integration here]"

Оберните это тоже в агента:

research_agent = create_react_agent(
    model="openai:gpt-4o",
    tools=[web_search],
    name="research_expert",
    prompt="You are a research specialist. Find accurate, current information."
)

У каждого агента есть ровно одна задача и только те инструменты, которые требуются для выполнения этой задачи, поэтому не возникает путаницы и проблем с перегрузкой запроса. В примерах используются модели OpenAI, хотя подойдут и бесплатные альтернативы, такие как Groq или OpenRouter.

Шаг 3: создание надзорщика

Импортируем вспомогательный инструмент:

from langgraph_supervisor import create_supervisor

Затем предоставьте двух специалистов: модель и инструкции, указывающие, когда использовать каждого из них:

workflow = create_supervisor(
    [research_agent, math_agent],
    model=ChatOpenAI(model="gpt-4o"),
    prompt=(
        "You are a team supervisor managing a research expert and a math expert. "
        "For questions requiring current information or web lookups, use research_expert. "
        "For anything involving calculation, use math_expert. "
        "Route each request to exactly the right specialist."
    )
)

Скомпилируйте рабочий процесс в запускаемое приложение:

app = workflow.compile()

Шаг 4: запустить

Отправьте математический вопрос:

result = app.invoke({
    "messages": [
        {"role": "user", "content": "What's 1,847 multiplied by 293?"}
    ]
})

И выведите окончательное сообщение:

print(result["messages"][-1].content)

Вот что происходит внутри:

1. Надзиратель читает запрос.
2. Он видит расчёт, а не исследовательский вопрос.
3. Он передаёт задание math_expert.
4. math_expert вычисляет ответ с помощью своего инструмента multiply.
5. Надзиратель снова берёт управление и возвращает ответ.

Для исследовательского вопроса, например о текущем численном составе крупных технологических компаний, тот же надзиратель направляет его к research_expert. Код остаётся идентичным, меняется только решение о направлении, причём ЯЗЫК сам принимает это решение.

Что на самом деле представляют собой инструменты передачи

Эта часть устраняет ощущение магии. Супервайзер не использует какой-то скрытый механизм LangGraph для передачи задач — он делает это с помощью вызовов инструментов, точно как любой обычный агент, использующий инструменты. На фоне происходит создание функции передачи задач для каждого специалиста с помощью функции create_supervisor; названия таких функций могут быть вроде transfer_to_research_expert и transfer_to_math_expert. Модель супервайзера рассматривает их как обычные инструменты, не отличающиеся от функций поиска погоды; поэтому, когда она определяет необходимость проведения исследования, она просто вызывает соответствующую функцию передачи. Это придает уверенности: супервайзер вовсе не является новым понятием — это просто стандартный цикл работы агента, в котором некоторые «инструменты» на самом деле представляют собой другие агенты, а не вызовы API.

Создание вручную

Если вы хотите полный контроль над маршрутизацией, что рекомендуется на этапе перехода к производственной среде, вот ручная версия, основанная на необработанном StateGraph. Начните с импортов:

from typing import Annotated, Sequence, TypedDict
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END

Определите общее состояние, которое хранит список сообщений и имя следующего узла:

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next: str

Узел-надзиратель просматривает последнее сообщение:

def supervisor_node(state: AgentState):
    """Decide which specialist should act next."""
    last_message = state["messages"][-1].content

В реальных условиях решение принимается на основе вызова большой языковой модели. Здесь же это сводится к простой проверке ключевых слов, чтобы логика маршрутизации оставалась понятной:

# In production, this decision comes from an LLM call —
    # simplified here to show the routing logic clearly
    if any(word in last_message.lower() for word in ["calculate", "multiply", "add", "sum"]):
        return {"next": "math_agent"}
    else:
        return {"next": "research_agent"}

Каждый специалист получает узел, который запускает его агента, а затем возвращает управление надзирателю. Сначала математика:

def math_node(state: AgentState):
    result = math_agent.invoke(state)
    return {"messages": [result], "next": "supervisor"}

Затем проведите исследование:

def research_node(state: AgentState):
    result = research_agent.invoke(state)
    return {"messages": [result], "next": "supervisor"}

Далее соедините граф, зарегистрировав узлы:

# Wire the graph together
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("math_agent", math_node)
workflow.add_node("research_agent", research_node)

Установите точку входа:

workflow.set_entry_point("supervisor")

Добавьте условные рёбра, соответствующие решению руководителя:

workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next"],
    {"math_agent": "math_agent", "research_agent": "research_agent", "end": END}
)

После этого отправьте каждого специалиста обратно к руководителю:

workflow.add_edge("math_agent", "supervisor")
workflow.add_edge("research_agent", "supervisor")

И скомпилируйте:

app = workflow.compile()

Это более развернуто, но ничего не скрыто. Вы можете увидеть, как изменяется состояние, как принимаются решения по маршрутизации и где добавить собственную логику, такую как ведение журнала событий, ограничения или пользовательские условия завершения.

Выбор структуры состояния, который быстрее всего устаревает

Предупреждение, основанное на практике использования в производстве: схема состояния — это то, изменение которой позже бывает самым сложным. Соблазнно с самого начала поместить все возможные поля в один плоский словарь, охватывающий сообщения, контекст пользователя и промежуточные результаты. Это подходит для систем с двумя агентами. Но когда восемь или десять узлов одновременно читают и записывают в одно общее состояние, это становится проблематичным, поскольку проверка типов в LangGraph строгая, и чтобы разобраться с перегруженным состоянием, приходится вмешиваться в работу каждого узла. С самого первого дня необходимо тщательно спланировать структуру схемы, даже для системы из двух агентов. Определите, что именно каждому агенту нужно читать и записывать, и не включайте всё остальное в общее состояние.

Проблема, с которой вы столкнетесь

Системы управления, вводимые в эксплуатацию, часто демонстрируют такую закономерность. Представьте, что система управления поручает задание специалисту. Специалист выполняет его, но система считает результат недостаточным и возвращает его обратно. Специалист вносит исправления, но новая версия оказывается ещё хуже первой. Система снова остается недовольна и отправляет её обратно, и если не принять мер предосторожности, это может продолжаться бесконечно.

Решением является ограничение количества таких циклов, что можно сделать с помощью простого счётчика в состоянии системы вместе с чётким, детерминированным условием остановки. Мнение большой языковой модели относительно того, достаточно ли что-то хорошо, не является надёжным единственным критерием остановки.

Когда не следует его использовать

Будьте честны относительно сложности задачи. Если работа заключается в одном поиске информации с последующей генерацией ответа, как в обычном вопросе-ответе в стиле RAG, наличие надзорного агента является избыточным. Дополнительное управление состоянием и нагрузка на маршрутизацию бесполезны, когда нет реальной необходимости в принятии решений о направлении действий. Обращайтесь к паттерну надзорного агента только тогда, когда у вас действительно есть отдельные специалисты и реальный выбор маршрута, и не применяйте его по умолчанию.

Итоги

• Один перегруженный агент теряет эффективность при росте сложности.
• Надзорный агент — это агент-маршрутизатор, который делегирует задачи специалистам.
• create_supervisor — это быстрый готовый способ маршрутизации.
• В основе всего лежат обычные вызовы инструментов для передачи задач.
• Ручное управление StateGraph обеспечивает полный контроль над маршрутизацией.
• Два важных урока из практики: заранее спроектируйте схему состояний и ограничьте количество циклов корректировок.