LangGraph багатоагентні системи: керівники, підграфи та надсилання

Попередні вимоги: у прикладах передбачається, що ви вже вмієте створювати граф LangGraph з одним агентом за допомогою інструментів та пам’яті, а також знаєте, як призупиняти та відновлювати роботу за допомогою interrupt(). Усе, що йде далі, ґрунтується на цьому, починаючи з питання про те, чому може знадобитися кілька агентів, та закінчуючи створенням повноцінного багатоагентного дослідницького асистента.

Чому розділяти одного агента на багато

Припустимо, що асистент з одним агентом працює добре. Тоді користувачі починають просити його перевіряти електронну пошту, досліджувати теми, писати звіти та керувати завданнями — усе це під час однієї розмови. Зазвичай вирішенням є додавання нових інструментів та більшої кількості інструкцій у системний запит. Незабаром кількість інструментів сягає 15, а довжина запиту — 2 000 слів, і якість роботи поступово погіршується. Модель починає плутати завдання: під час дослідження використовує інструмент для роботи з електронною поштою, а під час обробки пошти — інструмент для досліджень.

Це і є когнітивне перевантаження, і саме це є основною причиною існування систем з багатьма агентами. Рішенням є те, на що спираються програмісти вже десятиліттями: розподіл обов’язків. У 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.