LangGraph Многоподразделенческие системы: надзиратели, подграфы и отправка

Предпосылки: в примерах предполагается, что вы уже умеете создавать граф LangGraph с одним агентом с использованием соответствующих инструментов и механизмов хранения данных, а также знаете, как приостанавливать и возобновлять выполнение задач с помощью функции interrupt(). Всё, что описано далее, основано на этом, начиная с вопроса о необходимости использования нескольких агентов и заканчивая созданием полноценного многопроцессного исследовательского помощника.

Почему разделять одного агента на несколько

Предположим, что помощник с одним агентом работает хорошо. Тогда пользователи начинают просить его проверять электронную почту, исследовать темы, писать отчеты и управлять задачами — всё это в рамках одного диалога. Обычным решением является добавление новых инструментов и включение большего количества инструкций в системный промпт. Вскоре количество инструментов достигает 15, а длина промпта — 2000 слов, в результате чего качество работы снижается. Модель начинает путать задачи: использует инструмент для работы с электронной почтой во время исследовательской задачи и наоборот.

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

• Когнитивная перегрузка. Один модель должен справляться с слишком большим количеством нерелевантных задач. Специализированные агенты, каждый из которых хорош в одной области, снижают эту нагрузку.
• Последовательные узкие места. Работа, которая могла бы выполняться одновременно, происходит пошагово. Параллельные агенты сокращают время ожидания.
• Сложность. Никто не может анализировать граф из 40 узлов. Меньшие подграфы можно понимать и тестировать по одному.

Каждый из приведенных ниже подходов направлен на решение одной из этих трех проблем.

Пять уровней архитектуры

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

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

Уровень 2: параллелизм. Независимые задачи делятся и выполняются одновременно, затем объединяются. Идея заключается в расширении и сборе информации: несколько ветвей выполняются параллельно и передают результаты на следующий этап. Это подходит для повторяющейся работы с нескольких источников или подзадач.

Уровень 3: подграфы. Вместо одного огромного рабочего процесса каждая ветвь превращается в более мелкий граф с собственными внутренними шагами и схемой вывода. Таким образом, сложные компоненты становятся проще для создания, тестирования и понимания.

Уровень 4: динамический параллелизм. С помощью шаблона Send, основанного на принципе map-reduce, количество ветвей не известно заранее. Рабочие процессы создаются во время выполнения программы — по одному на каждый элемент, тему или аналитика. Агенты не предопределяются заранее, все они работают одновременно, причем каждый выполняет конкретную задачу.

Уровень 5: сочетание всего вышеперечисленного. Реалистичная система использует все перечисленные методы одновременно, как показано на примере исследовательского ассистента в конце.

Краткий способ запомнить: на уровне 1 задачи обрабатываются через одного руководителя, на уровне 2 работает фиксированный набор рабочих процессов, на уровне 3 каждая ветвь представляет собой полную сеть, а на уровне 4 рабочие процессы создаются по мере необходимости.

Уровень 1: шаблон руководителя

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

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

Поток действий выглядит следующим образом:

User Input
    ↓
[Supervisor Node]  ← decides who handles what
    ↓         ↓
[Specialist A] [Specialist B]   ← do the actual work
    ↓         ↓
[Supervisor Node]  ← reviews results, responds or routes again
    ↓
Response

Состояние

Ключевым решением при проектировании является поле next_agent. В него записывает супервайзер, а читает его маршрутизатор.

# ── MODULE 2: STATE ─────────────────────────────────────────
from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState

class SupervisorState(MessagesState):
    # The supervisor writes this field to indicate who should act next.
    # "researcher", "writer", "FINISH" are the possible values.
    next_agent: str

Инструменты специалистов

Каждый специалист имеет свои собственные специализированные инструменты, которые не пересекаются между собой.

# ── MODULE 3: TOOLS ─────────────────────────────────────────
from langchain_core.tools import tool

# Researcher's tools
@tool
def search_web(query: str) -> str:
    """Search the web for factual information, news, or research."""
    return f"[Web results for: {query}]"
@tool
def search_academic(query: str) -> str:
    """Search academic papers and journals for scholarly sources."""
    return f"[Academic results for: {query}]"
# Writer's tools
@tool
def format_as_report(content: str, title: str) -> str:
    """Format raw content into a structured report with headings."""
    return f"# {title}\n\n{content}"
@tool
def check_grammar(text: str) -> str:
    """Check and correct grammar in the provided text."""
    return f"[Grammar-checked]: {text}"
# Each specialist gets only its own tools
researcher_tools = [search_web, search_academic]
writer_tools = [format_as_report, check_grammar]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
researcher_llm = llm.bind_tools(researcher_tools)
writer_llm     = llm.bind_tools(writer_tools)

Узлы

Далее идут узел-контролер и специализированные узлы. Ответ узла-контролера в виде обычного текста сам по себе является решением о маршрутизации.

# ── MODULE 4: NODES ─────────────────────────────────────────
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import ToolNode

# ── Supervisor ────────────────────────────────────────────
SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of specialists.
Given the conversation, decide who should act next.
Your team:
- researcher: finds information, searches web and academic sources
- writer: formats content, writes reports, checks grammar
- FINISH: use this when the task is complete and you have a final answer
Respond with ONLY the name of the next agent: researcher, writer, or FINISH."""
def supervisor_node(state: SupervisorState) -> dict:
    """Reads the conversation history and routes to the right specialist."""
    messages = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    response = llm.invoke(messages)
    # The supervisor's plain text response IS the routing decision
    return {"next_agent": response.content.strip()}

# ── Researcher Specialist ──────────────────────────────────
def researcher_node(state: SupervisorState) -> dict:
    """Specialist that handles all research and information-gathering tasks."""
    messages = [
        SystemMessage(content="You are a research specialist. Use your search tools to find accurate, thorough information.")
    ] + state["messages"]
    response = researcher_llm.invoke(messages)
    return {"messages": [response]}

# ── Writer Specialist ──────────────────────────────────────
def writer_node(state: SupervisorState) -> dict:
    """Specialist that handles all writing, formatting, and editing tasks."""
    messages = [
        SystemMessage(content="You are a writing specialist. Use your tools to produce well-structured, polished output.")
    ] + state["messages"]
    response = writer_llm.invoke(messages)
    return {"messages": [response]}

# Tool execution nodes for each specialist
researcher_tool_node = ToolNode(researcher_tools)
writer_tool_node     = ToolNode(writer_tools)

Маршрутизация

# ── MODULE 5: ROUTING ────────────────────────────────────────
def route_after_supervisor(state: SupervisorState) -> Literal["researcher", "writer", "__end__"]:
    """Read the supervisor's decision and route accordingly."""
    decision = state["next_agent"]
    if decision == "FINISH":
        return "__end__"
    return decision  # "researcher" or "writer"

def route_researcher(state: SupervisorState) -> Literal["researcher_tools", "supervisor"]:
    """After researcher acts: did it call a tool, or produce a final answer?"""
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "researcher_tools"
    return "supervisor"  # Research done - report back to supervisor
def route_writer(state: SupervisorState) -> Literal["writer_tools", "supervisor"]:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "writer_tools"
    return "supervisor"

Сборка графа

# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

graph_builder = StateGraph(SupervisorState)
# Register all nodes
graph_builder.add_node("supervisor",        supervisor_node)
graph_builder.add_node("researcher",        researcher_node)
graph_builder.add_node("researcher_tools",  researcher_tool_node)
graph_builder.add_node("writer",            writer_node)
graph_builder.add_node("writer_tools",      writer_tool_node)
# Entry: always start at supervisor
graph_builder.add_edge(START, "supervisor")
# Supervisor routes to specialist or ends
graph_builder.add_conditional_edges(
    "supervisor",
    route_after_supervisor,
    {"researcher": "researcher", "writer": "writer", "__end__": END}
)
# Researcher loop: tools → researcher → supervisor
graph_builder.add_conditional_edges(
    "researcher", route_researcher,
    {"researcher_tools": "researcher_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("researcher_tools", "researcher")
# Writer loop: tools → writer → supervisor
graph_builder.add_conditional_edges(
    "writer", route_writer,
    {"writer_tools": "writer_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("writer_tools", "writer")
graph = graph_builder.compile(checkpointer=MemorySaver())

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

Уровень 2: параллелизм с распространением и сбором данных

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

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

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

Существует одно строгое правило: когда несколько узлов в одном супершаге записывают данные в одно и то же поле состояния, LangGraph должен знать, как объединить эти записи. Без специального функционала для объединения возникает ошибка. С использованием operator.add результаты безопасно объединяются между собой.

# ── MODULE 2: STATE FOR PARALLEL PATTERN ────────────────────
import operator
from typing import Annotated

class ResearchState(TypedDict):
    query: str
    # Without Annotated + operator.add, parallel writes to 'context' would CRASH.
    # With it, LangGraph concatenates results from all parallel nodes.
    context: Annotated[list[str], operator.add]
    final_answer: str

Пример параллельных исследований использует два источника:

# ── MODULE 4: PARALLEL WORKER NODES ─────────────────────────

def web_search_node(state: ResearchState) -> dict:
    """Runs in parallel with academic_search_node."""
    result = f"[WEB] Results for: {state['query']}"
    return {"context": [result]}   # Returns a LIST - operator.add will concatenate
def academic_search_node(state: ResearchState) -> dict:
    """Runs in parallel with web_search_node."""
    result = f"[ACADEMIC] Papers about: {state['query']}"
    return {"context": [result]}   # Also a LIST
def synthesize_node(state: ResearchState) -> dict:
    """Runs AFTER both parallel nodes complete (fan-in point).
    state['context'] already contains results from BOTH parallel nodes."""
    combined = "\n".join(state["context"])
    prompt = [
        SystemMessage(content="Synthesize the following research into a clear answer."),
        HumanMessage(content=f"Research:\n{combined}\n\nQuery: {state['query']}")
    ]
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# ── MODULE 6: PARALLEL GRAPH ASSEMBLY ───────────────────────
from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(ResearchState)
graph_builder.add_node("web_search",      web_search_node)
graph_builder.add_node("academic_search", academic_search_node)
graph_builder.add_node("synthesize",      synthesize_node)
# Fan-out: START branches to BOTH search nodes simultaneously
graph_builder.add_edge(START,       "web_search")
graph_builder.add_edge(START,       "academic_search")
# Fan-in: BOTH search nodes must complete before synthesize runs
graph_builder.add_edge("web_search",      "synthesize")
graph_builder.add_edge("academic_search", "synthesize")
graph_builder.add_edge("synthesize", END)
parallel_graph = graph_builder.compile()

Распространение данных заключается просто в двух ребрах, исходящих из одного и того же источника — START. LangGraph обнаруживает это и одновременно обрабатывает оба пункта назначения. Вход данных также происходит автоматически, поскольку функция synthesize запускается только тогда, когда оба входящих ребра обработаны, то есть когда оба параллельных узла завершили работу.

Уровень 3: субграфы для инкапсуляции

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

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

Родительский граф и подграф взаимодействуют через общие ключи состояния. Любой ключ, присутствующий как в состоянии родительского графа, так и в состоянии подграфа, передаётся при запуске подграфа и возвращается при его завершении.

Parent State: {topic, cleaned_data, report_A, report_B, processed_ids}
                         ↓
Subgraph A State: {cleaned_data, internal_step_1, report_A, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_A' and 'processed_ids' back to parent

Subgraph B State: {cleaned_data, internal_step_2, report_B, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_B' and 'processed_ids' back to parent

Ключи, существующие только в подграфе, такие как internal_step_1 выше, остаются приватными, и родительский узел никогда их не видит.

Ниже приведена полная реализация:

# ── MODULE 2: STATE SCHEMAS (Three Levels) ──────────────────

# Parent state - what the orchestrating graph tracks
class ParentState(TypedDict):
    raw_input: str
    cleaned_data: str                               # passed INTO both subgraphs
    failure_summary: str                            # written back by subgraph A
    performance_report: str                         # written back by subgraph B
    processed_ids: Annotated[list[int], operator.add]  # BOTH subgraphs write this

# Subgraph A's internal state - includes its private fields + shared keys
class FailureAnalysisState(TypedDict):
    cleaned_data: str        # shared input from parent
    failures: list[str]      # private intermediate state - parent never sees this
    failure_summary: str     # shared output to parent
    processed_ids: Annotated[list[int], operator.add]  # shared output (accumulated)

# Subgraph A's output schema - filters which keys return to parent
# Without this, ALL keys return, which can cause conflicts
class FailureAnalysisOutput(TypedDict):
    failure_summary: str     # only these two keys go back to parent
    processed_ids: Annotated[list[int], operator.add]

# Subgraph B's internal state
class PerformanceAnalysisState(TypedDict):
    cleaned_data: str
    metrics: list[str]          # private
    performance_report: str     # shared output
    processed_ids: Annotated[list[int], operator.add]

class PerformanceAnalysisOutput(TypedDict):
    performance_report: str
    processed_ids: Annotated[list[int], operator.add]

# ── SUBGRAPH A: Failure Analysis ─────────────────────────────
def extract_failures(state: FailureAnalysisState) -> dict:
    failures = [f"Error in {state['cleaned_data']}: timeout"]
    return {"failures": failures, "processed_ids": [1]}
def summarize_failures(state: FailureAnalysisState) -> dict:
    summary = f"Found {len(state['failures'])} failure(s): {state['failures']}"
    return {"failure_summary": summary}
failure_builder = StateGraph(FailureAnalysisState, output=FailureAnalysisOutput)
failure_builder.add_node("extract_failures",  extract_failures)
failure_builder.add_node("summarize_failures", summarize_failures)
failure_builder.add_edge(START, "extract_failures")
failure_builder.add_edge("extract_failures", "summarize_failures")
failure_builder.add_edge("summarize_failures", END)
failure_subgraph = failure_builder.compile()  # compiled → can be used as a node

# ── SUBGRAPH B: Performance Analysis ────────────────────────
def collect_metrics(state: PerformanceAnalysisState) -> dict:
    metrics = [f"Latency: 120ms", "Uptime: 99.8%"]
    return {"metrics": metrics, "processed_ids": [2]}
def write_report(state: PerformanceAnalysisState) -> dict:
    report = f"Performance Report:\n" + "\n".join(state["metrics"])
    return {"performance_report": report}
performance_builder = StateGraph(PerformanceAnalysisState, output=PerformanceAnalysisOutput)
performance_builder.add_node("collect_metrics", collect_metrics)
performance_builder.add_node("write_report",    write_report)
performance_builder.add_edge(START, "collect_metrics")
performance_builder.add_edge("collect_metrics", "write_report")
performance_builder.add_edge("write_report", END)
performance_subgraph = performance_builder.compile()

# ── PARENT GRAPH ─────────────────────────────────────────────
def clean_data_node(state: ParentState) -> dict:
    return {"cleaned_data": state["raw_input"].strip().lower()}
def combine_reports_node(state: ParentState) -> dict:
    combined = f"{state['failure_summary']}\n\n{state['performance_report']}"
    print(f"Final combined report:\n{combined}")
    return {}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("clean_data",           clean_data_node)
parent_builder.add_node("failure_analysis",     failure_subgraph)    # subgraph as node
parent_builder.add_node("performance_analysis", performance_subgraph) # subgraph as node
parent_builder.add_node("combine_reports",      combine_reports_node)
parent_builder.add_edge(START, "clean_data")
# Fan-out to both subgraphs (parallel execution)
parent_builder.add_edge("clean_data",           "failure_analysis")
parent_builder.add_edge("clean_data",           "performance_analysis")
# Fan-in: combine only after both subgraphs complete
parent_builder.add_edge("failure_analysis",     "combine_reports")
parent_builder.add_edge("performance_analysis", "combine_reports")
parent_builder.add_edge("combine_reports", END)
parent_graph = parent_builder.compile()

Схема вывода FailureAnalysisOutput и PerformanceAnalysisOutput в данном примере — это та часть, которую часто пропускают в учебниках, и именно поэтому возникают конфликты ключей состояния. Она определяет, какие именно поля подграф предоставляет своему родителю, а всё, чего в них нет, считается приватным внутренним состоянием.

Уровень 4: API Send и подход map-reduce

Параллельные ветви и подграфы подходят для случаев, когда количество ветвей известно на этапе проектирования. Но что, если пользователь хочет исследовать 5 тем сегодня и 20 — завтра? Жесткое задание 20 ветвей не является решением. Для этого используется подход map-reduce, реализуемый с помощью Send.

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

Важны три ключевых момента:

• Send(node_name, state), импортируемый из langgraph.constants, создаёт новый экземпляр узла с собственным набором данных состояния. Функция маршрутизации, возвращающая список таких узлов, заставляет LangGraph запустить их все сразу, причём у каждого будет своё независимое состояние.
• operator.add для поля сбора данных представляет собой шаг суммирования. По мере завершения работы сотрудников их результаты накапливаются в растущем списке в родительском состоянии.

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

Вот полная реализация алгоритма map-reduce:

# ── MODULE 2: MAP-REDUCE STATE ──────────────────────────────
from langgraph.constants import Send

class OverallState(TypedDict):
    topic: str
    subjects: list[str]                         # populated by the map step
    jokes: Annotated[list[str], operator.add]   # accumulated by the reduce step
    best_joke: str
class JokeState(TypedDict):
    """The private state each parallel worker gets.
    Notice: this does NOT need to match OverallState."""
    subject: str

# ── MODULE 4: MAP-REDUCE NODES ──────────────────────────────
def generate_subjects(state: OverallState) -> dict:
    """MAP step 1: Expands the topic into a list of subjects to process."""
    prompt = [
        SystemMessage(content="Generate a list of 3 sub-topics related to the given topic. Return as comma-separated values only."),
        HumanMessage(content=state["topic"])
    ]
    response = llm.invoke(prompt)
    subjects = [s.strip() for s in response.content.split(",")]
    return {"subjects": subjects}

def generate_joke(state: JokeState) -> dict:
    """MAP step 2: Worker node. Each instance handles exactly ONE subject.
    Receives its own isolated JokeState - not the full OverallState."""
    prompt = [HumanMessage(content=f"Write a short, funny joke about: {state['subject']}")]
    response = llm.invoke(prompt)
    # Returns a LIST - operator.add in OverallState will accumulate these
    return {"jokes": [response.content]}

def pick_best_joke(state: OverallState) -> dict:
    """REDUCE step: All jokes are now in state['jokes']. Pick the winner."""
    jokes_text = "\n".join([f"{i}. {j}" for i, j in enumerate(state["jokes"])])
    prompt = [
        SystemMessage(content="You are a comedy judge. Pick the funniest joke from the list below. Reply with ONLY the number."),
        HumanMessage(content=jokes_text)
    ]
    response = llm.invoke(prompt)
    winning_index = int(response.content.strip())
    return {"best_joke": state["jokes"][winning_index]}

# ── MODULE 5: THE SEND ROUTER ────────────────────────────────
def continue_to_jokes(state: OverallState):
    """The routing function that LAUNCHES parallel workers.

    Returning a LIST of Send objects (not a string) is what triggers map-reduce.
    Each Send creates an independent, parallel execution of 'generate_joke'
    with its own state payload. The number of parallel workers is determined
    at runtime by how many subjects exist in state.
    """
    return [
        Send("generate_joke", {"subject": subject})
        for subject in state["subjects"]
    ]

# ── MODULE 6: MAP-REDUCE GRAPH ASSEMBLY ─────────────────────
graph_builder = StateGraph(OverallState)
graph_builder.add_node("generate_subjects", generate_subjects)
graph_builder.add_node("generate_joke",     generate_joke)
graph_builder.add_node("pick_best_joke",    pick_best_joke)
graph_builder.add_edge(START, "generate_subjects")
# This conditional edge uses the Send router - it fans out dynamically
graph_builder.add_conditional_edges(
    "generate_subjects",
    continue_to_jokes,   # returns a LIST of Send objects
    ["generate_joke"]    # list of possible destination nodes (for graph validation)
)
# Fan-in: pick_best_joke runs after ALL parallel generate_joke instances complete
graph_builder.add_edge("generate_joke", "pick_best_joke")
graph_builder.add_edge("pick_best_joke", END)
map_reduce_graph = graph_builder.compile()

Секрет заключается в функции continue_to_jokes. Она возвращает список объектов Send вместо строки, и когда LangGraph получает такой список от функции маршрутизации, он немедленно запускает каждый из них. Три элемента в state["subjects"] означают три параллельных узла generate_joke, а двадцать элементов — двадцать узлов. Граф масштабируется в зависимости от количества операций, выполняемых во время работы.

Уровень 5: многопроцессный исследовательский ассистент

Теперь всё объединяется: оркестрация в стиле руководителя, утверждение человеком, параллельные подграфы и API Send — всё это в одной системе производственного типа. Это финальный проект курса LangGraph Academy, описанный здесь.

Принцип работы: пользователь указывает тему исследования. Система генерирует группу персонажей-аналитиков ИИ, которые могут быть утверждены или доработаны человеком. Каждый аналитик затем действует в качестве отдельного агента, проводя параллельные интервью с «экспертами» ИИ с помощью поиска в интернете и составляя один раздел отчета. В конце узел-писатель объединяет все разделы в готовый документ.

User provides topic
        ↓
[create_analysts] → LLM generates N analyst personas
        ↓
[human_feedback]  ← INTERRUPT: human approves or sends feedback
        ↓         ↑
[should_continue] → if feedback, loop back to create_analysts
        ↓
[initiate_research] → Send API: spawns N parallel interview subgraphs
        ↓↓↓ (N parallel workers running simultaneously)
[interview_subgraph × N] → each analyst interviews an "expert" via web search
        ↓
[write_report]    → LLM compiles all sections into a final report

Состояние на трёх уровнях

from pydantic import BaseModel, Field

# ── Analyst persona - a Pydantic model for structured LLM output ──
class Analyst(BaseModel):
    name: str = Field(description="The analyst's full name")
    role: str = Field(description="Their professional role (e.g. 'Financial Analyst')")
    focus: str = Field(description="Their specific analytical focus area")

    @property
    def persona(self) -> str:
        return f"Name: {self.name}\nRole: {self.role}\nFocus: {self.focus}"
class Perspectives(BaseModel):
    analysts: list[Analyst]

# ── Outer graph state ─────────────────────────────────────────
class ResearchGraphState(TypedDict):
    topic: str
    max_analysts: int
    human_analyst_feedback: str      # human writes here during HITL pause
    analysts: list[Analyst]
    sections: Annotated[list, operator.add]  # accumulated from all parallel interviews
    final_report: str

# ── Interview subgraph state ──────────────────────────────────
class InterviewState(MessagesState):
    # inherits: messages: Annotated[list[BaseMessage], add_messages]
    max_num_turns: int
    context: Annotated[list, operator.add]  # search results accumulate here
    analyst: Analyst
    interview: str       # formatted transcript
    sections: list       # completed section (sent back to outer graph)

Создание аналитиков с проверкой человеком

# ── ANALYST GENERATION NODE ───────────────────────────────────
ANALYST_PROMPT = """You are generating a team of {max_analysts} research analysts
for the topic: {topic}.
{feedback_section}
Each analyst should have a unique perspective and area of focus.
Generate analysts with diverse viewpoints - financial, technical, social, etc."""
def create_analysts(state: ResearchGraphState) -> dict:
    """Generates analyst personas using structured LLM output."""
    feedback = state.get("human_analyst_feedback", "")
    feedback_section = (
        f"User feedback to incorporate:\n{feedback}" if feedback
        else "No feedback yet - generate your best initial set."
    )

    prompt = ANALYST_PROMPT.format(
        max_analysts=state["max_analysts"],
        topic=state["topic"],
        feedback_section=feedback_section
    )

    # with_structured_output: forces LLM to return a Pydantic model
    # instead of free-form text - ensures clean, usable data
    structured_llm = llm.with_structured_output(Perspectives)
    result = structured_llm.invoke([SystemMessage(content=prompt)])
    return {"analysts": result.analysts}

def human_feedback_node(state: ResearchGraphState) -> dict:
    """No-op node. Its only purpose is to be a pause point.
    The graph is compiled with interrupt_before=['human_feedback'].
    See Part 3 for the full HITL pattern."""
    pass  # Does nothing - the pause happens BEFORE this node via interrupt_before

def should_continue_to_research(state: ResearchGraphState) -> Literal["create_analysts", "initiate_research"]:
    """After human feedback: loop back to refine, or proceed to interviews."""
    if state.get("human_analyst_feedback"):
        return "create_analysts"
    return "initiate_research"

Подграф интервью

Каждый аналитик работает как отдельный подграф — небольшой агент ReAct, который проводит интервью с «экспертом» (другим LLM, выполняющим эту роль) и ищет информацию в Интернете.

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def web_search_tool(query: str) -> str:
    """Search the web for information on a topic."""
    return f"[Search results for '{query}']"  # Replace with real search API
interview_tools = [web_search_tool]
interview_llm = llm.bind_tools(interview_tools)

def generate_question(state: InterviewState) -> dict:
    """The analyst node: generates the next interview question."""
    analyst = state["analyst"]
    messages = [
        SystemMessage(content=(
            f"You are {analyst.name}, a {analyst.role} focused on {analyst.focus}. "
            f"You are interviewing an expert about the research topic. "
            f"Ask insightful questions that align with your analytical focus. "
            f"When you have enough information, say 'Thank you, that is all I needed.'"
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def generate_answer(state: InterviewState) -> dict:
    """The expert node: answers the analyst's question, using web search."""
    messages = [
        SystemMessage(content=(
            "You are an expert being interviewed. Answer thoroughly and factually. "
            "Use the web search tool when you need current data."
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def save_interview(state: InterviewState) -> dict:
    """Formats the full Q&A exchange into a clean transcript string."""
    transcript = []
    for msg in state["messages"]:
        if hasattr(msg, "content"):
            role = "Analyst" if type(msg).__name__ == "HumanMessage" else "Expert"
            transcript.append(f"{role}: {msg.content}")
    return {"interview": "\n\n".join(transcript)}

def write_section(state: InterviewState) -> dict:
    """The final node: uses the interview transcript to write a report section."""
    analyst = state["analyst"]
    prompt = [
        SystemMessage(content=(
            f"Based on the following interview conducted by {analyst.name} ({analyst.role}), "
            f"write a structured report section focused on {analyst.focus}. "
            f"Be concise, factual, and cite specific points from the interview."
        )),
        HumanMessage(content=state["interview"])
    ]
    response = llm.invoke(prompt)
    # sections is the key that Send passes back to the outer graph
    return {"sections": [response.content]}

def route_interview(state: InterviewState) -> Literal["generate_answer", "save_interview"]:
    """Continue interviewing OR wrap up when analyst is satisfied or max turns hit."""
    last_message = state["messages"][-1]
    if (
        "thank you, that is all" in last_message.content.lower()
        or len(state["messages"]) >= state.get("max_num_turns", 6) * 2
    ):
        return "save_interview"
    # Check if analyst made a tool call (web search)
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "generate_answer"
    return "generate_answer"

# ── INTERVIEW SUBGRAPH ASSEMBLY ───────────────────────────────
interview_builder = StateGraph(InterviewState)
interview_tool_node = ToolNode(interview_tools)
interview_builder.add_node("generate_question", generate_question)
interview_builder.add_node("generate_answer",   generate_answer)
interview_builder.add_node("interview_tools",   interview_tool_node)
interview_builder.add_node("save_interview",    save_interview)
interview_builder.add_node("write_section",     write_section)
interview_builder.add_edge(START, "generate_question")
interview_builder.add_conditional_edges(
    "generate_question",
    route_interview,
    {"generate_answer": "generate_answer", "save_interview": "save_interview"}
)
interview_builder.add_conditional_edges(
    "generate_answer",
    lambda s: "interview_tools" if (hasattr(s["messages"][-1], "tool_calls") and s["messages"][-1].tool_calls) else "generate_question",
    {"interview_tools": "interview_tools", "generate_question": "generate_question"}
)
interview_builder.add_edge("interview_tools", "generate_answer")
interview_builder.add_edge("save_interview",  "write_section")
interview_builder.add_edge("write_section",   END)
interview_graph = interview_builder.compile()

Граф внешней оркестрации

# ── SEND ROUTER: Spawn one interview per analyst ─────────────
def initiate_research(state: ResearchGraphState):
    """Launches N parallel interview subgraphs via Send — one per analyst.
    This is the map-reduce Send pattern from Level 4, applied to subgraphs."""
    return [
        Send("interview", {
            "analyst": analyst,
            "messages": [HumanMessage(content=f"Research topic: {state['topic']}")],
            "context": [],
            "max_num_turns": 3,
            "interview": "",
            "sections": []
        })
        for analyst in state["analysts"]
    ]
# ── REPORT COMPILATION NODE ───────────────────────────────────
def write_report(state: ResearchGraphState) -> dict:
    """Fan-in point: all sections have arrived from parallel interviews.
    The LLM compiles them into a final report."""
    sections_text = "\n\n---\n\n".join(state["sections"])
    prompt = [
        SystemMessage(content=(
            "You are a senior editor. Compile the following report sections written by "
            "different analysts into one cohesive, well-structured final report. "
            "Add a proper introduction and conclusion. Do not repeat content."
        )),
        HumanMessage(content=f"Topic: {state['topic']}\n\nSections:\n{sections_text}")
    ]
    response = llm.invoke(prompt)
    return {"final_report": response.content}

# ── OUTER GRAPH ASSEMBLY ─────────────────────────────────────
outer_builder = StateGraph(ResearchGraphState)
outer_builder.add_node("create_analysts",   create_analysts)
outer_builder.add_node("human_feedback",    human_feedback_node)
outer_builder.add_node("interview",         interview_graph)   # subgraph as node
outer_builder.add_node("write_report",      write_report)
outer_builder.add_edge(START, "create_analysts")
outer_builder.add_edge("create_analysts", "human_feedback")
outer_builder.add_conditional_edges(
    "human_feedback",
    should_continue_to_research,
    {"create_analysts": "create_analysts", "initiate_research": "initiate_research"}
)
# initiate_research is a ROUTING FUNCTION (returns Send objects), not a node
outer_builder.add_conditional_edges(
    "human_feedback",
    initiate_research,   # returns list of Send objects
    ["interview"]        # validation: lists possible destinations
)
outer_builder.add_edge("interview",    "write_report")
outer_builder.add_edge("write_report", END)
research_graph = outer_builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_feedback"]  # HITL pause for analyst approval
)

Запуск всего процесса

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "research-001"}}

    # Step 1: Generate analysts
    result = research_graph.invoke(
        {
            "topic": "The impact of AI on the future of software engineering",
            "max_analysts": 3,
            "human_analyst_feedback": "",
            "analysts": [],
            "sections": [],
            "final_report": ""
        },
        config=config
    )

    # Step 2: HITL — graph paused at human_feedback node
    snapshot = research_graph.get_state(config)
    analysts = snapshot.values["analysts"]

    print("Generated analysts:")
    for a in analysts:
        print(f"  - {a.name} ({a.role}): {a.focus}")

    feedback = input("\nProvide feedback to refine analysts (or press Enter to approve): ")

    if feedback.strip():
        # Human has feedback — inject it and let supervisor regenerate
        research_graph.update_state(
            config,
            {"human_analyst_feedback": feedback},
            as_node="human_feedback"
        )
    else:
        # Approved — clear feedback and proceed to interviews
        research_graph.update_state(
            config,
            {"human_analyst_feedback": None},
            as_node="human_feedback"
        )

    # Step 3: Resume — N parallel interviews launch via Send, then report is compiled
    print("\nRunning parallel interviews...")
    final_result = research_graph.invoke(None, config=config)

    print("\n" + "="*60)
    print("FINAL RESEARCH REPORT")
    print("="*60)
    print(final_result["final_report"])

Выбор шаблона

• Один агент, который путается между задачами, требует использования шаблона надзирателя. Надзиратель направляет задачи, а специалисты их выполняют.
• Независимые, медленные задачи требуют применения структур типа fan-out и fan-in. Выполняйте их одновременно и используйте редюсеры operator.add для общих полей.
• Параллельные ветви, сами по себе сложные, требуют использования подграфов. Скомпилируйте каждую ветвь как отдельный граф, встроите её в виде узла и используйте схемы вывода для контроля того, что возвращается к родительскому узлу.
• Количество рабочих, определяемое во время выполнения, требует использования механизма Send. Маршрутизатор генерирует список объектов Send, и каждый из них создаёт независимого рабочего с собственным состоянием.
• Настоящая производственная система использует всё вышеперечисленное. Ассистент уровня 5 объединяет все схемы, поскольку реальная задача требует их всех.

Справочник ключевых терминов

Структура надзирателя. supervisor_node является оркестратором. Он считывает состояние, задаёт значение next_agent и никогда не выполняет специализированную работу. specialist_node обрабатывает одну конкретную задачу с использованием собственных инструментов и инструкций, затем возвращает управление надзирателю. next_agent: str — это стандартное поле состояния для маршрутизации: надзиратель записывает имя агента, а маршрутизатор его считывает.

Параллелизм. Расширение сети означает несколько вызовов add_edge из одного источникового узла, которые LangGraph выполняет одновременно. Сжатие сети означает несколько рёбер, ведущих в один конечный узел, который ждёт завершения работы всех источников. Функция Annotated[list, operator.add] обязательна для любого поля, в которое записываются данные нескольких параллельных узлов, поскольку без неё параллельные записи приводят к сбоям, а operator.add является стандартным функцией объединения, сопоставляющей списки данных из одновременно обрабатываемых узлов.

Подграфы. Функция StateGraph(InternalState, output=OutputSchema) представляет собой версию с двумя аргументами, при которой второй аргумент определяет, какие ключи будут возвращаться к родительскому узлу. Схема вывода — это TypedDict, содержащий только те ключи, которые должны быть обнаружены снаружи, а все остальное остается приватным. Перекрывающиеся ключи служат каналом связи между родительским и дочерним узлами, поскольку любые ключи, присутствующие в обеих схемах, автоматически становятся общими. Метод subgraph.compile() преобразует подграф в функцию, которая затем передается в parent.add_node("name", compiled_subgraph). Вызов graph.get_graph(xray=1).draw_mermaid() позволяет отобразить внутреннюю структуру подграфа в диаграмме.

Map-reduce. Send(node_name, state_dict) означает один вызов параллельного рабочего процесса, и его данные не обязаны совпадать с состоянием родительского узла. Возврат [Send(...), Send(...)] из маршрутизатора запускает их все. При использовании Send третий аргумент функции add_conditional_edges представляет собой список, а не словарь, с именами допустимых пунктов назначения, используемый исключительно для проверки графа. Наконец, конструкция with_structured_output(PydanticModel) применяется после вызова модели для гарантии формирования JSON-данных, соответствующих схеме Pydantic. Это стандартный способ получения надежных решений от системы управления и списков для аналитиков.

Навык композиции

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

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

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