Как работает координация нескольких агентов в LangGraph — как подграфы взаимодействуют друг с другом?

Координация агентов через подграфы в LangGraph

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

Базовая архитектура: supervisor + subgraphs

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

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Literal

# --- State подграфа (researcher) ---
class ResearchState(TypedDict):
    query: str
    findings: str

def search_web(state: ResearchState) -> ResearchState:
    # имитация поиска
    return {"findings": f"Results for: {state['query']}"}

def summarize(state: ResearchState) -> ResearchState:
    return {"findings": state["findings"] + " [summarized]"}

research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_web)
research_builder.add_node("summarize", summarize)
research_builder.set_entry_point("search")
research_builder.add_edge("search", "summarize")
research_builder.add_edge("summarize", END)
research_graph = research_builder.compile()

# --- State подграфа (writer) ---
class WriterState(TypedDict):
    research: str
    draft: str

def write_draft(state: WriterState) -> WriterState:
    return {"draft": f"Draft based on: {state['research']}"}

writer_builder = StateGraph(WriterState)
writer_builder.add_node("write", write_draft)
writer_builder.set_entry_point("write")
writer_builder.add_edge("write", END)
writer_graph = writer_builder.compile()

# --- Общий state супервизора ---
class SupervisorState(TypedDict):
    task: str
    findings: str
    draft: str
    next: Literal["researcher", "writer", END]

def supervisor(state: SupervisorState) -> SupervisorState:
    if not state.get("findings"):
        return {"next": "researcher"}
    if not state.get("draft"):
        return {"next": "writer"}
    return {"next": END}

# Адаптеры: преобразование state супервизора в state подграфа
def call_researcher(state: SupervisorState) -> SupervisorState:
    result = research_graph.invoke({"query": state["task"], "findings": ""})
    return {"findings": result["findings"]}

def call_writer(state: SupervisorState) -> SupervisorState:
    result = writer_graph.invoke({"research": state["findings"], "draft": ""})
    return {"draft": result["draft"]}

supervisor_builder = StateGraph(SupervisorState)
supervisor_builder.add_node("supervisor", supervisor)
supervisor_builder.add_node("researcher", call_researcher)
supervisor_builder.add_node("writer", call_writer)
supervisor_builder.set_entry_point("supervisor")
supervisor_builder.add_conditional_edges(
    "supervisor",
    lambda s: s["next"],
    {"researcher": "researcher", "writer": "writer", END: END}
)
supervisor_builder.add_edge("researcher", "supervisor")
supervisor_builder.add_edge("writer", "supervisor")

memory = MemorySaver()
graph = supervisor_builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "task-1"}}
result = graph.invoke(
    {"task": "quantum computing trends", "findings": "", "draft": "", "next": "supervisor"},
    config=config
)
print(result["draft"])

Прямое встраивание подграфа как узла

Если state подграфа содержит ключи, совместимые с родительским state (пересечение ключей), можно вставить скомпилированный граф напрямую как узел без адаптера:

parent_builder = StateGraph(SupervisorState)
# Прямая вставка — LangGraph автоматически передаёт
# пересекающиеся ключи в подграф
parent_builder.add_node("research_agent", research_graph)

Паттерн параллельного запуска подграфов

Для параллельного выполнения нескольких агентов используется Send API — он создаёт несколько параллельных ветвей:

from langgraph.types import Send

def dispatch_agents(state: SupervisorState):
    return [
        Send("researcher", {"query": state["task"], "findings": ""}),
        Send("researcher", {"query": state["task"] + " advanced", "findings": ""})
    ]

supervisor_builder.add_conditional_edges(
    "supervisor",
    dispatch_agents
)

Подводные камни

• Несовместимость state: родительский и дочерний граф имеют разные TypedDict — при прямой вставке без адаптера LangGraph передаёт только пересекающиеся ключи, остальные молча игнорируются. Это частая причина потери данных.
• Чекпоинты подграфов: по умолчанию подграф, вставленный как узел, не использует checkpointer родителя. Для сохранения истории подграфа нужно явно передавать checkpointer при компиляции подграфа или использовать subgraphs=True в конфиге.
• thread_id изоляция: при вызове подграфа через invoke внутри узла thread_id не передаётся автоматически — нужно явно прокидывать config.
• Рекурсия: граф с циклами (агент вызывает другой агент, который возвращается к первому) может уйти в бесконечный цикл. LangGraph ограничивает глубину через recursion_limit (по умолчанию 25).
• Отладка вложенных графов: стандартные print-логи не показывают иерархию вызовов. Используйте LangSmith или stream_mode="values" для полного трейсинга.
• Производительность Send: параллельные ветки через Send выполняются псевдопараллельно (Python asyncio), а не истинно параллельно. Для CPU-heavy задач это не даст выигрыша без ProcessPoolExecutor.

• Объяснять multi agent subgraphs только синтаксисом без shape, dtype, состояния или режима выполнения.
• Игнорировать leakage, воспроизводимость, пустые входы и скрытые копии данных.
• Не проверять production-симптомы: latency, память, ретраи, дрейф качества и несовпадение версий.

• Может ли связать multi agent subgraphs с реальным контрактом входов и выходов.
• Упоминает ли тесты, метрики, reproducibility и диагностику ошибок.
• Видит ли различие между demo-кодом в ноутбуке и production-пайплайном.
• Предлагает ли observability, rollback, ограничения стоимости и стратегию incident replay.