Шматагентныя сістэмы ў LangGraph: супервізары, падграфы і Send

Пярэхутныя абавязкі: у прыкладах прыпускаецца, што вы ужо можаце стварыць граф 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, які тут прадстаўлены.

Функцыя: корыстнік задае тэму даследжэння. Сістэма стварае групу персонажаў-аналітыкаў AI, а чалавек можа іх затвердзіць або дапрацаваць. Кожны аналітык працуе як падагент для паралельных адпытак, ставячы запытанні „эксперту“ AI за дапамогою пошуку ў інтэрнэце і ствараючы адзін раздзіл атласу. У падсумку вузел-пісар складае этыя раздзілы ў готовы дакумент.

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-го рангу максымальна набліжаецца да таго, што мае база кода для калькуляцый з удзелам калькulatorаў: гэта цыклы затверджэння ад людзя, паралельныя спецыялісты, структураваны выхід модэлю, дынамічнае стварэнне працоўнікаў і фінальны крок сынтэзы. Разам з канонічная структурой праекта, калектаваннем памяці і мерамі безпекі з удзелам людзя, гэтыя патэрны пакрываюць архітэктуру большасці рэальных прыемакоў LangGraph.