第17章 多 Agent 模式实战

17.1 引言

单个 Agent 能力有限——它只有一组工具、一种提示、一个执行循环。当任务复杂度上升时（如"帮我分析这家公司的财报，然后写一份投资建议书"），单个 Agent 要么因为工具太多导致 LLM 选择困难，要么因为系统提示过长影响推理质量。多 Agent 系统通过分工协作解决这个问题：不同的 Agent 负责不同的能力域，通过明确的通信机制协同完成复杂任务。

LangGraph 的底层基础设施——StateGraph、Send、Command、子图——天然支持多 Agent 模式的构建。图的节点可以是子图（即嵌套的 Agent），条件边可以实现动态路由，Send 支持一对多的任务分发，Command 支持跨图的状态更新和导航。

本章将系统介绍基于 LangGraph 实现的四种典型多 Agent 架构：Supervisor（监督者）、Swarm（蜂群）、分层 Agent 和协作 Agent。每种模式都有其适用场景和权衡，我们会结合源码和实战代码深入分析它们的设计原理。

17.2 多 Agent 的核心抽象

17.2.1 Agent 作为子图

在 LangGraph 中，一个 Agent 本质上就是一个 CompiledStateGraph。多 Agent 系统就是一个图中嵌套了多个子图：

# 每个 Agent 是一个独立的图
research_agent = create_react_agent(model, research_tools, name="researcher")
writer_agent = create_react_agent(model, writing_tools, name="writer")

# 多 Agent 系统是一个包含子图的父图
builder = StateGraph(TeamState)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)

17.2.2 通信机制的三种模式

graph TB
    subgraph "模式 1：共享状态"
        A1[Agent A] -->|写入状态| S1[共享 State]
        S1 -->|读取状态| B1[Agent B]
    end
    subgraph "模式 2：Command 导航"
        A2[Agent A] -->|"Command(goto='B')"| B2[Agent B]
    end
    subgraph "模式 3：Send 分发"
        Router[路由器] -->|"Send('A', task1)"| A3[Agent A]
        Router -->|"Send('B', task2)"| B3[Agent B]
    end

17.2.3 重要迁移：create_react_agent 和 AgentState 自 v1.0 起已 deprecated

在深入具体模式前必须告知的一件事——本章所有示例里的 from langgraph.prebuilt import create_react_agent 和 AgentState 在 LangGraph v1.0 已经被标记 deprecated。打开 libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py：

# chat_agent_executor.py:274
@deprecated(
    "create_react_agent has been moved to `langchain.agents`. "
    "Please update your import to `from langchain.agents import create_agent`.",
    category=LangGraphDeprecatedSinceV10,
)
def create_react_agent(...):
    ...

对应的 AgentState / AgentStatePydantic / AgentStateWithStructuredResponse / AgentStateWithStructuredResponsePydantic 四个 state 类都挂了同样的 @deprecated 装饰器（libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py:53-107），迁移指引统一是"移到 langchain.agents"。

迁移之后的分工是：预设的 agent 形态由 langchain.agents 提供，LangGraph 这一侧保留 StateGraph、Pregel、checkpointer 等图执行层的原语（以及 ToolNode 这类底层节点）。

用户现在的迁移路径：

# 旧（本章示例，仍能用但会警告）
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(model, tools)

# 新（v1.0+ 推荐）
from langchain.agents import create_agent
agent = create_agent(model, tools)

两者短期内都存在并工作——create_react_agent 仍是 langgraph-prebuilt 里一份完整的独立实现（chat_agent_executor.py，1015 行），并没有转调 langchain.agents.create_agent；它只是挂了 @deprecated 装饰器：每次调用都会在运行时发出 LangGraphDeprecatedSinceV10 警告（实跑），IDE/类型检查器也能据此标出弃用。这个警告类别让用户可以 warnings.filterwarnings("ignore", category=LangGraphDeprecatedSinceV10) 暂时屏蔽——但长期建议尽快迁移。

本章接下来的示例代码为了讲清楚核心概念、继续使用 create_react_agent 的传统写法——因为多数生产代码库也还在用、读者会在真实环境里遇到。看到 deprecation warning 请按上面指引迁移、核心架构模式（Supervisor/Swarm/分层）本身不变。

这种 "预设组件从执行引擎迁到应用框架" 的分离，在本专栏第 16 章 §16.1.1 有同一份对照表（ToolNode 留下、create_react_agent 迁走），用 @deprecated + 迁移指引做平滑过渡。

17.2.4 源码核对：Send 类的真实定义与 map-reduce 动机

Send 和 Command 是多 Agent 编排的两个最重要原语。打开 libs/langgraph/langgraph/types.py:574-646，Send 的真实定义极其简单：

# types.py:574 起（省略 docstring 其余部分与 __repr__）
class Send:
    """A message or packet to send to a specific node in the graph."""

    __slots__ = ("node", "arg")

    node: str
    arg: Any

    def __init__(self, /, node: str, arg: Any) -> None:
        self.node = node
        self.arg = arg

    def __hash__(self) -> int:
        return hash((self.node, self.arg))

    def __eq__(self, value: object) -> bool:
        return (
            isinstance(value, Send)
            and self.node == value.node
            and self.arg == value.arg
        )

四个关键设计：

1、__slots__ 限制属性为两个。Send 只承载"往哪个节点 + 什么 payload"两件事——有了 slots，实例没有 __dict__，也不能再挂别的属性；载荷本身（arg，通常是字典）的开销另算。

2、它不是 dataclass，而是手写的 __slots__ 类。Send 只声明了 node: str / arg: Any 两个类型注解，__init__、__hash__、__eq__、__repr__ 全部手写——没有 @dataclass 装饰器，也就不是 frozen 的（send.node = "x" 能改，实跑）。

3、__hash__ 基于 (node, arg) —— 两个 Send 只有节点名和参数都相同才相等。但只有 arg 可哈希时 Send 才能进集合：最常见的字典载荷 hash(Send("a", {"x": 1})) 直接抛 TypeError: unhashable type: 'dict'（实跑）；框架也不靠哈希给 Send 去重——TASKS 是列表，两个相同的 Send 会派生两个任务（第 12 章 §12.2.3）。

4、/ 的位置在 self 之后：def __init__(self, /, node: str, arg: Any) ——Python 3.8+（PEP 570）的 positional-only 标记。按语言规范，/ 之前的形参只能按位置传，所以这里只有 self 是 positional-only，node 和 arg 仍然是 POSITIONAL_OR_KEYWORD——Send(node="x", arg=y) 是合法写法。由于签名里没有 **kwargs，这个 / 对调用方几乎没有可观察的影响：传 self=... 关键字照样报错，只是错误信息从 got multiple values for argument 'self' 变成 got some positional-only arguments passed as keyword arguments: 'self'（实跑；第 12 章 §12.9.6 对同一处代码有一致的说明）。

docstring 里的 map-reduce 例子（第 591-616 行）点出了 Send 最重要的用途："the sent state can differ from the core graph's state"——你可以给每个 Send 传不同的 arg，每个 target 节点看到的是自己的那份独立 state。Supervisor 给 researcher 和 writer 分别派任务时，不是让两个 Agent 看同一份全局 state——而是各自拿到定制的 task description。

17.2.5 源码核对：Command 的四个字段和它们之间的组合语义

Command（types.py:652-702）的定义比 Send 复杂得多——它是 LangGraph 多 Agent 通信的瑞士军刀：

@dataclass(**_DC_KWARGS)
class Command(Generic[N], ToolOutputMixin):
    graph: str | None = None
    update: Any | None = None
    resume: dict[str, Any] | Any | None = None
    goto: Send | Sequence[Send | N] | N = ()

    PARENT: ClassVar[Literal["__parent__"]] = "__parent__"

四个字段各司其职：

• graph：目标图。None=当前图；Command.PARENT（即字符串 "__parent__"）=上一级父图
• update：要对 state 做的更新——dict / NamedTuple / pydantic instance 都可
• resume：配合 interrupt() 的恢复值
• goto：下一步跳到哪——单节点名、节点名序列、Send、Send 序列都可

这四个字段可以任意组合——Command 不是 "do one thing" 的简单原语，是"一次动作里表达多种意图"的复合命令。三种最常见的组合：

# 1. 纯导航（Swarm 的 handoff 工具）
Command(goto="billing_agent")

# 2. 导航 + state 更新（Supervisor 决策）
Command(goto="researcher", update={"messages": [decision_msg]})

# 3. 跨图导航（子图内 handoff 到父图的别的 agent）
Command(goto="writer_agent", graph=Command.PARENT)

组合 1/2 分别对应 §17.4 / §17.3 的模式；组合 3 是 Swarm 与分层模式的关键——子图内部的节点能直接跳到父图里的兄弟节点（Command.PARENT 指直接外层那一个图，§17.3.5）。

注意 goto 类型是 Send | Sequence[Send | N] | N ——节点名和 Send 可以混合列表：

Command(goto=[
    "supervisor",                     # 激活 supervisor
    Send("async_task", {...}),        # 同时给 async_task 发任务
])

这种"一条 Command 同时触发多条路径"的写法里，列表中的节点和 Send 任务会在下一个超步一起执行——实跑时 supervisor 与 async_task 同在第 2 步，列表顺序不代表先后。

17.2.6 源码核对：Command._update_as_tuples 的五种 update 类型归一

Command.update 接受多种类型——dict、list[tuple]、NamedTuple、Pydantic、普通值。真实的 _update_as_tuples（types.py:687-700）把它们归一到 list[tuple[str, value]]：

def _update_as_tuples(self) -> Sequence[tuple[str, Any]]:
    if isinstance(self.update, dict):
        return list(self.update.items())
    elif isinstance(self.update, (list, tuple)) and all(
        isinstance(t, tuple) and len(t) == 2 and isinstance(t[0], str)
        for t in self.update
    ):
        return self.update
    elif keys := get_cached_annotated_keys(type(self.update)):
        # （本行注释为本专栏所加）NamedTuple / Pydantic / dataclass 等带类注解的对象
        return get_update_as_tuples(self.update, keys)
    elif self.update is not None:
        return [("__root__", self.update)]
    else:
        return []

五个分支分别处理：

1. dict → items()
2. list[tuple[str, Any]] → 原样使用
3. 带 __annotations__ 的类的实例（NamedTuple/Pydantic/dataclass 等；TypedDict 的实例本身就是 dict，走第 1 支）→ 用缓存的 annotated keys 提取字段
4. 任何其他非 None 值 → 包成 [("__root__", self.update)]——供只有单一字段的 state schema 使用
5. None → 空列表

第 3 种特别值得看——get_cached_annotated_keys(type(self.update))。它根据类型的 __annotations__ 缓存每个类对应的字段 keys——同一个 state 类型只做一次 annotation 扫描，后续都命中缓存（_internal/_fields.py 的 ANNOTATED_KEYS_CACHE）——频繁创建 Command 时，key 提取对每个类型只做一次。

第 4 种 "__root__" 是个特殊约定——某些 state schema 是 "只有一个字段、或者这个字段就是 state 本身" 的简化形式（比如 StateGraph(list[str]) 直接用 list 作 state），这时 Command.update 传一个 list 而不是 dict——框架自动包成 ("__root__", list) 写进 __root__ 通道。

17.3 Supervisor 模式

17.3.1 架构概述

Supervisor 模式是最直观的多 Agent 架构：一个"主管"Agent 接收用户请求，分析任务需求，然后将子任务分配给专业 Agent。每个专业 Agent 完成后将结果返回给 Supervisor，Supervisor 决定是否需要更多工作。

graph TB
    User[用户] --> Sup[Supervisor Agent]
    Sup -->|"分配研究任务"| RA[Research Agent]
    Sup -->|"分配写作任务"| WA[Writer Agent]
    Sup -->|"分配代码任务"| CA[Coder Agent]
    RA -->|"研究结果"| Sup
    WA -->|"文章草稿"| Sup
    CA -->|"代码片段"| Sup
    Sup -->|"最终回复"| User
    style Sup fill:#ffe6e6,stroke:#333,stroke-width:2px

17.3.2 实现方式

from typing import Annotated, Literal, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from langgraph.types import Command

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_agent: str

# 创建专业 Agent
research_agent = create_react_agent(model, research_tools, name="researcher")
writer_agent = create_react_agent(model, writing_tools, name="writer")

def supervisor_node(state: SupervisorState) -> Command:
    """Supervisor 决定下一步交给哪个 Agent"""
    response = supervisor_llm.invoke([
        SystemMessage(content="""你是一个任务分配主管。
        分析用户请求，决定交给哪个 Agent：
        - researcher: 负责信息检索和数据分析
        - writer: 负责内容创作和文档编写
        - FINISH: 任务完成，返回最终结果"""),
        *state["messages"]
    ])

    # 解析 LLM 的决策
    next_agent = parse_decision(response.content)

    if next_agent == "FINISH":
        return Command(goto=END, update={"messages": [response]})
    else:
        return Command(
            goto=next_agent,
            update={"messages": [response]}
        )

# 构建 Supervisor 图
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)

builder.add_edge(START, "supervisor")
builder.add_edge("researcher", "supervisor")  # 完成后返回 Supervisor
builder.add_edge("writer", "supervisor")

graph = builder.compile()

17.3.3 Supervisor 的关键设计

1. 中心化决策：Supervisor 是唯一的决策者，专业 Agent 只负责执行
2. 循环结构：Agent 完成后总是返回 Supervisor，由 Supervisor 决定下一步
3. Command 导航：使用 Command(goto=next_agent) 实现动态路由

17.3.4 适用场景与局限

适用场景：

• 子任务之间有明确的先后依赖
• 需要中央协调来决定任务顺序
• Agent 数量较少（3-5 个）

局限：

• Supervisor 是单点瓶颈
• 每次决策都需要额外的 LLM 调用
• 子任务之间无法直接通信

17.3.5 源码核对：ParentCommand 如何穿越图边界——pregel/_retry.py 的异常冒泡机制

§17.2.5 讲了 Command(graph=Command.PARENT) 可以"导航到父图的兄弟节点"。这个能力背后是一套完整的异常冒泡机制，值得完整拆开。

真实实现在 libs/langgraph/langgraph/pregel/_retry.py:122-140（节选，省略 task.writes.clear() 等行）：

try:
    return task.proc.invoke(task.input, config)
except ParentCommand as exc:
    ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
    cmd = exc.args[0]
    # strip task_ids from namespace for comparison
    if cmd.graph in (ns, recast_checkpoint_ns(ns), task.name):
        # this command is for the current graph, handle it
        for w in task.writers:
            w.invoke(cmd, config)
        break
    elif cmd.graph == Command.PARENT:
        # this command is for the parent graph, assign it to the parent.
        exc.args = (replace(cmd, graph=_checkpoint_ns_for_parent_command(ns)),)
    # bubble up
    raise

分三步理解：

1、子图里 return Command(goto="foo", graph=Command.PARENT) 时，Pregel 把它包装成 ParentCommand 异常向上 raise——和第 16 章 §16.3.5 讲过的 GraphBubbleUp 是同一个异常家族。

2、在每一层捕获时，Pregel 比较 cmd.graph 和当前图的 ns。如果相等，说明这条 Command 已经冒泡到了目标图——执行它；如果是 Command.PARENT，把 graph 字段改写成父图的真实 ns，继续 raise 让它向上冒。

3、_checkpoint_ns_for_parent_command（_retry.py:57-83）负责把子图的 ns 字符串转换成父图的 ns：

# _retry.py:57-83（docstring 与行内注释有删节）
def _checkpoint_ns_for_parent_command(ns: str) -> str:
    """Return the checkpoint namespace for the parent graph.
    The checkpoint namespace is a `|`-separated path. Each segment is usually
    of the form `name:task_id` (e.g. `parent_first:<uuid>|node:<uuid>`), but the
    runtime may also insert a purely-numeric segment (e.g. `|1`) to disambiguate
    concurrent tasks.
    Numeric segments are not real path levels, so we drop them before computing
    the parent namespace.
    """
    parts = ns.split(NS_SEP)

    # Drop any trailing numeric selectors
    while parts and parts[-1].isdigit():
        parts.pop()

    # Drop the current frame segment itself
    if parts:
        parts.pop()

    # Drop any trailing numeric selectors for the parent frame
    while parts and parts[-1].isdigit():
        parts.pop()

    return NS_SEP.join(parts)

这个函数体现了 checkpoint ns 结构的一个微妙细节——ns 里可能有纯数字段（比如 parent:<uuid>|1|node:<uuid> 里的 |1|）——docstring 称之为并发任务的消歧段；实际追加它的是 subgraph_counter：同一个任务先后启动第二、第三个子图时，命名空间后面多出 |1、|2（第 11 章 §11.3.2 实跑）。计算父 ns 时要跳过数字段，只看 name:task_id 形式的真实层级段。

这条设计让你在 Swarm 或分层模式里写 Command(goto="sibling_agent", graph=Command.PARENT) 时不用自己算 ns——Pregel 把 Command 沿调用栈冒泡到直接外层那一个图执行。注意 PARENT 只上跳一层：实跑一个三层嵌套图，最内层节点 goto="supervisor" 能跳到中间层的 supervisor，goto 顶层节点名则只在中间层留下一条 wrote to unknown channel branch:to:ceo, ignoring it 警告、跳转被静默忽略。

这条机制和第 9 章讲的 GraphInterrupt 是同一思路——用 exception 实现"脱离当前上下文、跳到更外层处理器"：_retry.py 里 except ParentCommand 和 except GraphBubbleUp 两个分支紧挨着，都走"不属于我就原样 raise"的路径。用异常栈做控制流转移，是最简洁的"单向向上传递、由最近的匹配者处理"的语义。

17.4 Swarm 模式

17.4.1 架构概述

Swarm 模式（蜂群模式）没有中央调度者。每个 Agent 在完成自己的工作后，可以直接将控制权交给另一个 Agent。这类似于客服系统中的"转接"——当前 Agent 判断用户的需求超出自己的能力范围时，主动转接给更合适的 Agent。

graph LR
    User[用户] --> A1[接待 Agent]
    A1 -->|"转接"| A2[技术支持 Agent]
    A2 -->|"转接"| A3[退款处理 Agent]
    A3 -->|"完成"| User
    A1 -.->|"也可以直接完成"| User
    A2 -.->|"也可以转回"| A1
    style A1 fill:#e6f3ff,stroke:#333
    style A2 fill:#f3ffe6,stroke:#333
    style A3 fill:#ffe6f3,stroke:#333

17.4.2 实现方式

Swarm 的核心是"handoff"工具——每个 Agent 有一个特殊工具用于将控制权转交给其他 Agent：

from typing import Annotated
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.types import Command

# 创建 handoff 工具
def create_handoff_tool(target_agent: str, description: str):
    @tool(f"transfer_to_{target_agent}", description=f"将对话转交给 {target_agent}。{description}")
    def handoff(
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_msg = ToolMessage(f"已转交给 {target_agent}", tool_call_id=tool_call_id)
        return Command(
            goto=target_agent,
            # 工具运行在 agent 子图里，目标是父图里的兄弟节点
            graph=Command.PARENT,
            # 把本 agent 的消息（含发起调用的 AIMessage）连同配对的 ToolMessage 带回父图
            update={"messages": state["messages"] + [tool_msg]},
        )
    return handoff

# 定义 Agent 和它们的 handoff 能力
greeting_tools = [
    create_handoff_tool("tech_support", "当用户有技术问题时转交"),
    create_handoff_tool("billing", "当用户有账单问题时转交"),
]

tech_tools = [
    search_docs_tool,
    create_handoff_tool("greeting", "当问题解决后转回"),
]

billing_tools = [
    check_balance_tool,
    process_refund_tool,
    create_handoff_tool("greeting", "当问题解决后转回"),
]

# 创建各 Agent
greeting_agent = create_react_agent(model, greeting_tools, name="greeting")
tech_agent = create_react_agent(model, tech_tools, name="tech_support")
billing_agent = create_react_agent(model, billing_tools, name="billing")

# 构建 Swarm 图
class SwarmState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

builder = StateGraph(SwarmState)
builder.add_node("greeting", greeting_agent)
builder.add_node("tech_support", tech_agent)
builder.add_node("billing", billing_agent)
builder.add_edge(START, "greeting")  # 入口始终是接待 Agent

# 每个 Agent 可以通过 handoff 工具跳转到其他 Agent；
# 没有出边的 Agent 一旦不再转交就直接结束
graph = builder.compile()

handoff 工具必须带 graph=Command.PARENT：create_react_agent 返回的 agent 是一个子图，工具在它的 tools 节点里执行，裸的 Command(goto="tech_support") 目标是这个子图本身，而子图里没有这个节点——实跑只得到一条 wrote to unknown channel branch:to:tech_support, ignoring it 警告，跳转被静默忽略，随后子图因发起调用的 AIMessage 缺少配对的 ToolMessage 而抛 ValueError。上面的写法用假模型实跑：接待 Agent 调用 transfer_to_tech_support 后，控制权转到 tech_support，父图最终消息为 Human → AI（tool_call）→ Tool → AI。这与官方独立包 langgraph-swarm（PyPI 上为 0.1.0，2026-09-25 查；不在本专栏 pin 树内，以下据其官方仓库 langchain-ai/langgraph-swarm-py 主分支）里 create_handoff_tool 的做法一致：Command(goto=agent_name, graph=Command.PARENT, update={"messages": [...state 中的消息, tool_message], "active_agent": agent_name})。

17.4.3 Swarm 的关键设计

1. 去中心化：没有 Supervisor，Agent 之间是对等关系
2. Handoff 工具：通过返回 Command(goto=..., graph=Command.PARENT) 实现控制权转移
3. 工具即路由：LLM 通过选择"转交"工具来决定路由

17.4.4 适用场景与局限

适用场景：

• 客服系统、对话路由
• Agent 之间是专业领域的分工
• 同一时间只有一个 Agent 活跃

局限：

• 难以协调并行任务
• 可能出现循环转交
• 需要每个 Agent 都了解其他 Agent 的能力

17.4.5 源码核对：Command.goto 被 StateGraph 转换成什么——_CHANNEL_BRANCH_TO

一个 node 返回 Command(goto="billing") 之后，StateGraph 怎么把它变成实际的跳转？打开 libs/langgraph/langgraph/graph/state.py:1537-1563：

def _control_branch(value: Any) -> Sequence[tuple[str, Any]]:
    if isinstance(value, Send):
        return ((TASKS, value),)
    commands: list[Command] = []
    if isinstance(value, Command):
        commands.append(value)
    elif isinstance(value, (list, tuple)):
        for cmd in value:
            if isinstance(cmd, Command):
                commands.append(cmd)
    rtn: list[tuple[str, Any]] = []
    for command in commands:
        if command.graph == Command.PARENT:
            raise ParentCommand(command)

        goto_targets = (
            [command.goto] if isinstance(command.goto, (Send, str)) else command.goto
        )

        for go in goto_targets:
            if isinstance(go, Send):
                rtn.append((TASKS, go))
            elif isinstance(go, str) and go != END:
                # END is a special case, it's not actually a node in a practical sense
                # but rather a special terminal node that we don't need to branch to
                rtn.append((_CHANNEL_BRANCH_TO.format(go), None))
    return rtn

四条关键处理：

1、Command.PARENT → 立刻 raise ParentCommand。这条代码是 §17.3.5 讲过的异常冒泡的起点——在子图内部一检测到 graph=Command.PARENT，就把 Command 包成 ParentCommand 异常抛出。这一层不处理、向上冒。

2、Send 对象 → 写入 TASKS 通道。Pregel 有一个特殊通道叫 TASKS——凡是 Send 对象最终都写进这个通道。Pregel 下一个超步启动时会从 TASKS 通道读取所有待执行的 Send、为每个 Send 创建一个 task 实例。这是 Send 机制能实现"并行分发"的底层。

3、节点名字符串 → 写入 _CHANNEL_BRANCH_TO.format(go) 通道。每个可跳转节点有一个专属的"branch to" channel。当这个 channel 被写入时，Pregel 知道"下一步要激活这个节点"。这条机制让条件分支和直接跳转复用同一套 channel 触发逻辑——跳转是 channel 写入、被触发是 channel 订阅。

4、END 是特殊字符串，不走 branch。源码注释 "END is a special case, it's not actually a node in a practical sense"——END 是个哨兵节点，表示"结束"。不需要跳转到它，只需要让 Pregel 知道"没有下一个活跃节点"就等同于结束。

这套设计让 Command 从用户视角的"导航+更新"语义，翻译到 Pregel 视角的"channel 写入"——Command 是前端 API、channel writes 是执行引擎的单一抽象。所有外部语义最终都塌缩到 channel 写入——这是 LangGraph 架构统一性的核心。

17.5 分层 Agent

17.5.1 架构概述

分层 Agent 是 Supervisor 模式的扩展：多个 Supervisor 形成树形结构，每个 Supervisor 管理一组专业 Agent 或下级 Supervisor。这种模式适用于大规模复杂任务的分解。

graph TB
    CEO[CEO Agent<br/>总调度] --> PM1[项目经理 Agent<br/>研究方向]
    CEO --> PM2[项目经理 Agent<br/>创作方向]
    PM1 --> R1[数据分析 Agent]
    PM1 --> R2[文献检索 Agent]
    PM2 --> W1[文案撰写 Agent]
    PM2 --> W2[校对编辑 Agent]
    style CEO fill:#ffe6e6,stroke:#333,stroke-width:2px
    style PM1 fill:#fff3e6,stroke:#333
    style PM2 fill:#fff3e6,stroke:#333

17.5.2 实现方式

# 第一层：专业 Agent
data_agent = create_react_agent(model, data_tools, name="data_analyst")
search_agent = create_react_agent(model, search_tools, name="searcher")
writing_agent = create_react_agent(model, writing_tools, name="writer")
editing_agent = create_react_agent(model, editing_tools, name="editor")

# 第二层：子 Supervisor
def research_supervisor(state):
    """研究方向的中层主管"""
    response = llm.invoke([
        SystemMessage(content="你管理数据分析和文献检索。决定分配给谁。"),
        *state["messages"]
    ])
    next_agent = parse_decision(response.content)
    return Command(goto=next_agent, update={"messages": [response]})

research_team = StateGraph(TeamState)
research_team.add_node("supervisor", research_supervisor)
research_team.add_node("data_analyst", data_agent)
research_team.add_node("searcher", search_agent)
research_team.add_edge(START, "supervisor")
research_team.add_edge("data_analyst", "supervisor")
research_team.add_edge("searcher", "supervisor")
research_graph = research_team.compile(name="research_team")

# 类似地创建 writing_team...

# 第三层：顶层 Supervisor
top_builder = StateGraph(TopState)
top_builder.add_node("ceo", ceo_supervisor)
top_builder.add_node("research_team", research_graph)  # 子图作为节点
top_builder.add_node("writing_team", writing_graph)
top_builder.add_edge(START, "ceo")
top_builder.add_edge("research_team", "ceo")
top_builder.add_edge("writing_team", "ceo")

top_graph = top_builder.compile()

17.5.3 子图作为节点的运行时行为

当 Pregel 执行到子图节点时：

1. 子图的输入从父图状态中提取
2. 子图内部按自己的超步循环运行
3. 子图的 checkpoint_ns 自动嵌套（如 research_team:<task_id>|supervisor:<task_id>）
4. 子图完成后，输出合并回父图状态

sequenceDiagram
    participant Parent as 父图 (CEO)
    participant Sub as 子图 (Research Team)
    participant Agent as 子 Agent (Data Analyst)
    Parent->>Sub: 输入子任务
    Note over Sub: checkpoint_ns = "research_team:<task_id>"
    Sub->>Agent: Supervisor 分配任务
    Note over Agent: checkpoint_ns = "research_team:<task_id>|data_analyst:<task_id>"
    Agent-->>Sub: 返回结果
    Sub-->>Parent: 合并输出到父图状态

17.5.4 源码核对：子图的 checkpoint_ns 嵌套格式与数字段

§17.5.3 提到了子图 checkpoint_ns 自动嵌套。ns 的结构写在 libs/langgraph/langgraph/pregel/_retry.py:57-83 对 _checkpoint_ns_for_parent_command 的注释：

The checkpoint namespace is a |-separated path. Each segment is usually of the form name:task_id (e.g. parent_first:<uuid>|node:<uuid>), but the runtime may also insert a purely-numeric segment (e.g. |1) to disambiguate concurrent tasks (e.g. parent_first:<uuid>|1|node:<uuid>).

三条从这段注释能读出的细节：

1、层级分隔符是 | 不是 :。注释明确用 "|-separated path"——ns 形如 research_team:<uuid>|supervisor:<uuid>——| 分层、: 分隔段内的 name 和 task_id。实跑 §17.5.2 那样的三层图，最内层节点拿到的是 research_team:<id>|data_analyst:<id>|agent:<id>。

2、: 后面是 task_id（UUID）。每个节点执行都产生一个独立的 task_id——即使同一个节点被多次调用，每次 task_id 都不同。这让 Pregel 能精确追踪"某个特定执行"的 checkpoint——比如"这个 supervisor 的这次特定 dispatch"。

3、中间可能插入纯数字段（|1|）。docstring 说它用来区分并发任务，但追加它的是每个任务自己的 subgraph_counter：同一个任务里先后启动第二、第三个子图时，命名空间后面多出 |1、|2（如 caller:<uuid>|1），两个子图的检查点才不会混在一起。用 Send 给同一节点分发多个任务则用不到它——每个 Send 本身是独立任务、task_id 各不相同（第 11 章 §11.3.2 实跑两个 Send("worker", ...) 得到两个不同的 worker:<task_id>，没有数字段）。

这条设计让 checkpoint 能精确到"顶层图的这次运行的这个子图的这个节点的这次执行"——路径长得像文件系统路径。你可以通过 ns 定位一条执行轨迹，做日志聚合、错误归因。

17.5.5 源码核对：_get_root 对 Command.PARENT 的特殊处理

libs/langgraph/langgraph/graph/state.py:1580-1600 里的 _get_root 函数在处理节点返回值时对 Command.PARENT 做了一个不起眼的分支（_get_root 只用于根类型 state，即 output_keys == ["__root__"] 的图；常见的 dict 型 state 走 CompiledStateGraph.attach_node 里的 _get_updates，state.py:1248 起，对 Command.PARENT 有同样的跳过逻辑）：

def _get_root(input: Any) -> Sequence[tuple[str, Any]] | None:
    if isinstance(input, Command):
        if input.graph == Command.PARENT:
            return ()
        return input._update_as_tuples()
    elif (
        isinstance(input, (list, tuple))
        and input
        and any(isinstance(i, Command) for i in input)
    ):
        updates: list[tuple[str, Any]] = []
        for i in input:
            if isinstance(i, Command):
                if i.graph == Command.PARENT:
                    continue
                updates.extend(i._update_as_tuples())
            else:
                updates.append(("__root__", i))
        return updates
    elif input is not None:
        return [("__root__", input)]

为什么 Command.PARENT 返回空或跳过？ 因为 Command.PARENT 的语义是"这个 Command 不属于当前图"——它即将被 raise 成 ParentCommand 异常向上冒泡。当前图不应该把它的 update 应用到自己的 state——update 会在父图处理时被应用到父图的 state。

它和 _control_branch 配合：同一次 ChannelWrite.do_write 里，_assemble_writes 先依次跑完所有 mapper、最后才统一提交（pregel/_write.py:123-126），而 _control_branch 一遇到 Command.PARENT 就抛 ParentCommand，所以当前图这一次一条也不会写入——即便同一返回列表里还有发给本图的 Command，它的写入也随之整批丢弃（第 10 章 §10.8.2 实跑）。跳过分支保证 PARENT 的 update 不会被当成本图的更新去转换。

17.6 协作 Agent

17.6.1 架构概述

协作 Agent 模式中，多个 Agent 共享同一个状态空间，按照预定义的顺序或条件轮流执行。每个 Agent 都能读取和修改共享状态，形成一种"接力赛"式的协作。

graph LR
    Start[START] --> P[规划 Agent]
    P --> R[研究 Agent]
    R --> W[写作 Agent]
    W --> Rev[审核 Agent]
    Rev -->|需要修改| R
    Rev -->|通过| End[END]
    style P fill:#e6f3ff
    style R fill:#f3ffe6
    style W fill:#ffe6f3
    style Rev fill:#fff3e6

17.6.2 实现方式

class CollaborativeState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    plan: str
    research_results: list[str]
    draft: str
    review_feedback: str
    iteration: int

def planner(state: CollaborativeState) -> dict:
    """规划 Agent：分析需求，制定计划"""
    response = planner_llm.invoke([
        SystemMessage(content="你是一个项目规划者。分析用户需求，制定详细的执行计划。"),
        *state["messages"]
    ])
    return {"plan": response.content, "messages": [response]}

def researcher(state: CollaborativeState) -> dict:
    """研究 Agent：根据计划收集信息"""
    response = research_llm.invoke([
        SystemMessage(content=f"执行以下研究计划：{state['plan']}"),
        *state["messages"]
    ])
    return {"research_results": [response.content], "messages": [response]}

def writer(state: CollaborativeState) -> dict:
    """写作 Agent：基于研究结果撰写"""
    context = "\n".join(state["research_results"])
    response = writer_llm.invoke([
        SystemMessage(content=f"基于以下研究结果撰写：\n{context}"),
        *state["messages"]
    ])
    return {"draft": response.content, "messages": [response]}

def reviewer(state: CollaborativeState) -> dict:
    """审核 Agent：评审草稿质量"""
    response = reviewer_llm.invoke([
        SystemMessage(content=f"审核以下草稿的质量：\n{state['draft']}"),
        *state["messages"]
    ])
    return {
        "review_feedback": response.content,
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1
    }

def review_decision(state: CollaborativeState) -> str:
    """决定是否需要修改"""
    if "approved" in state["review_feedback"].lower():
        return END
    if state.get("iteration", 0) >= 3:
        return END  # 最多迭代 3 次
    return "researcher"

builder = StateGraph(CollaborativeState)
builder.add_node("planner", planner)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "planner")
builder.add_edge("planner", "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_conditional_edges("reviewer", review_decision)

graph = builder.compile()

17.6.3 协作的关键设计

1. 共享状态：所有 Agent 读写同一个 CollaborativeState
2. 专用字段：每个 Agent 有自己的输出字段（plan、research_results、draft、review_feedback）
3. 迭代循环：reviewer 可以触发新一轮的 research -> write -> review

17.6.4 源码核对：Send 如何在 Pregel 内部变成一个独立 task

§17.2.4 讲了 Send 的数据结构、§17.4.5 讲了它被写入 TASKS 通道。真正把 Send 变成一个可执行 task 的代码在 libs/langgraph/langgraph/pregel/_algo.py:878-937（节选；这部分在 Pregel 准备每一个超步的任务时运行）：

def prepare_push_task_send(...) -> PregelTask | PregelExecutableTask | None:
    if len(task_path) == 2:
        # SEND tasks, executed in superstep n+1
        # (PUSH, idx of pending send)
        idx = cast(int, task_path[1])
        if not channels[TASKS].is_available():
            return
        sends: Sequence[Send] = channels[TASKS].get()
        if idx < 0 or idx >= len(sends):
            return
        packet = sends[idx]
        if not isinstance(packet, Send):
            logger.warning(f"Ignoring invalid packet type {type(packet)} in pending sends")
            return

        if packet.node not in processes:
            logger.warning(f"Ignoring unknown node name {packet.node} in pending sends")
            return

        # find process
        proc = processes[packet.node]
        proc_node = proc.node
        if proc_node is None:
            return

        # create task id
        triggers = PUSH_TRIGGER
        checkpoint_ns = (
            f"{parent_ns}{NS_SEP}{packet.node}" if parent_ns else packet.node
        )
        task_id = task_id_func(
            checkpoint_id_bytes,
            checkpoint_ns,
            str(step),
            packet.node,
            PUSH,
            str(idx),
        )

五条从这段代码能读出的工程细节：

1、Send 在上一超步写入 TASKS 通道，下一超步才执行（注释 "executed in superstep n+1"）。这意味着用户写 return Send("worker", {...}) 产出的任务不会立刻执行——需要等当前超步完成、所有 channel 更新生效、Pregel 开始下一超步。Send 是跨超步的异步分发，不是同步调用。

2、四道防御性校验：

• not channels[TASKS].is_available() → TASKS 通道没有待处理 Send，直接返回
• idx < 0 or idx >= len(sends) → 索引越界（可能因为 checkpoint 恢复时 Send 列表已变），忽略
• not isinstance(packet, Send) → 类型错误，log warning 并忽略
• packet.node not in processes → 未知节点名，log warning 并忽略

第四条特别值得注意——如果你写 Send("nonexistent_node", ...)，Pregel 不会 throw——只会 log 一条 Ignoring unknown node name nonexistent_node in pending sends 并静默跳过（实跑，条件边返回和 Command(goto=Send(...)) 两条路径都如此）。代价是用户可能不知道自己拼错了节点名——所以开发时要留意 warning。

3、每个 Send 生成一个独立 task_id。task_id 的输入包含 idx——Send 在列表中的位置——两个 Send 给同一节点但不同 idx 会生成不同 task_id。这让 Pregel 能精确追踪每个 Send 对应的执行。

4、checkpoint_ns = f"{parent_ns}{NS_SEP}{packet.node}"——Send 产生的 task 的 ns 追加到父 ns 后面，形成层级路径。这保证了 Send 调度在嵌套图里也有正确的 checkpoint 定位（§17.5.4 讲过）。

5、triggers = PUSH_TRIGGER——Send 产生的 task 用一个特殊的 trigger 标记。Pregel 区分 "通过 channel 订阅被触发" 的普通节点（任务路径以 PULL 开头）和 "通过 Send PUSH 被触发" 的任务（路径为 (PUSH, idx)），prepare_next_tasks 分两段分别准备它们。

这几十行代码让条件边里 return [Send("a", ...), Send("b", ...)] 这种简单的 Python 返回值，变成了 Pregel 并发调度的基础——一个 Send = 一个独立任务、带自己的 task_id 与 ns（注意 arg 不按目标节点的 schema 做类型转换，第 12 章 §12.8.3）。用户不用自己写调度代码，这是 LangGraph 作为编排框架的核心价值。

17.7 模式对比与选择

17.7.1 四种模式对比

graph TB
    subgraph Supervisor
        S_Hub[中央调度者]
        S_A[Agent A]
        S_B[Agent B]
        S_Hub --> S_A
        S_Hub --> S_B
        S_A --> S_Hub
        S_B --> S_Hub
    end
    subgraph Swarm
        SW_A[Agent A]
        SW_B[Agent B]
        SW_C[Agent C]
        SW_A <-->|handoff| SW_B
        SW_B <-->|handoff| SW_C
    end
    subgraph 分层
        H_Top[顶层 Sup]
        H_Mid1[中层 Sup 1]
        H_Mid2[中层 Sup 2]
        H_A[Agent A]
        H_B[Agent B]
        H_Top --> H_Mid1
        H_Top --> H_Mid2
        H_Mid1 --> H_A
        H_Mid2 --> H_B
    end
    subgraph 协作
        C_A[Agent A] --> C_B[Agent B]
        C_B --> C_C[Agent C]
        C_C -.->|可选循环| C_A
    end

维度	Supervisor	Swarm	分层	协作
决策方式	中央决策	自主决策	层级决策	预定义顺序
通信模式	星形（hub-spoke）	点对点	树形	链式/环形
并行能力	可通过 Send	单一活跃	层内可并行	可通过 Send
适用规模	小型（3-5 Agent）	中型	大型	任意
复杂度	低	中	高	低
灵活性	中	高	高	低

17.7.2 选择建议

flowchart TB
    Q1{任务可以分解为<br/>独立子任务？}
    Q1 -->|是| Q2{需要动态决策<br/>任务分配？}
    Q1 -->|否| Q3{Agent 之间需要<br/>直接交接？}
    Q2 -->|是| Sup[Supervisor 模式]
    Q2 -->|否| Collab[协作 Agent 模式]
    Q3 -->|是| Swarm[Swarm 模式]
    Q3 -->|否| Q4{Agent 数量 > 5？}
    Q4 -->|是| Hier[分层 Agent 模式]
    Q4 -->|否| Sup

17.8 高级技巧

17.8.1 并行 Agent 执行

使用 Send API 让 Supervisor 同时分配多个任务：

def supervisor_with_parallel(state):
    """Supervisor 同时分配多个任务"""
    tasks = analyze_subtasks(state["messages"])
    # 作为节点时须把 Send 列表放进 Command.goto；
    # 节点直接返回 list[Send] 会抛 InvalidUpdateError: Expected dict（实跑）
    return Command(goto=[
        Send(task["agent"], {"messages": state["messages"], "task": task["description"]})
        for task in tasks
    ])

同样的 Send 列表也可以由条件边函数直接返回（add_conditional_edges("supervisor", fn)），那里不需要包 Command。

17.8.2 Agent 间消息传递的状态设计

class MultiAgentState(TypedDict):
    # 公共消息历史
    messages: Annotated[list[BaseMessage], add_messages]
    # 各 Agent 的中间结果（使用 reducer 合并；merge_dicts 需自行定义，如 lambda a, b: {**a, **b}）
    agent_outputs: Annotated[dict[str, Any], merge_dicts]
    # 当前活跃的 Agent
    active_agent: str
    # 全局上下文
    shared_context: dict[str, Any]

17.8.3 跨 Agent 的 Store 共享

def research_agent_node(state, runtime: Runtime):
    """研究 Agent：结果保存到 Store"""
    results = do_research(state["messages"])
    runtime.store.put(
        ("shared", "research"),
        state["task_id"],
        {"results": results}
    )
    return {"messages": [AIMessage(content="研究完成")]}

def writer_agent_node(state, runtime: Runtime):
    """写作 Agent：从 Store 读取研究结果"""
    research = runtime.store.search(
        ("shared", "research"),
        limit=10
    )
    context = "\n".join(r.value["results"] for r in research)
    draft = write_with_context(context, state["messages"])
    return {"messages": [AIMessage(content=draft)]}

17.9 常见陷阱与最佳实践

17.9.1 避免无限循环

多 Agent 系统最常见的问题是循环转交——Agent A 认为应该交给 Agent B，Agent B 又认为应该交回 Agent A。解决方案：

class SafeState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    handoff_count: int
    max_handoffs: int

def safe_handoff(state: SafeState, target: str) -> Command:
    """带有次数限制的安全转交"""
    if state.get("handoff_count", 0) >= state.get("max_handoffs", 10):
        return Command(
            goto=END,
            update={"messages": [AIMessage(content="已达到最大转交次数，结束对话。")]}
        )
    return Command(
        goto=target,
        update={"handoff_count": state.get("handoff_count", 0) + 1}
    )

17.9.2 状态 Schema 设计原则

多 Agent 系统的状态设计需要平衡共享与隔离：

class WellDesignedState(TypedDict):
    # 公共字段：所有 Agent 共享
    messages: Annotated[list[BaseMessage], add_messages]
    task_description: str

    # Agent 特有字段：使用 NotRequired 避免其他 Agent 被强制提供
    research_notes: NotRequired[str]
    draft_content: NotRequired[str]
    review_score: NotRequired[float]

    # 元数据字段：追踪执行过程
    current_agent: str
    iteration_count: int

17.9.3 消息历史管理

多 Agent 系统中，消息历史会快速膨胀。建议使用 pre_model_hook 或专门的消息管理节点：

from langchain_core.messages import RemoveMessage, SystemMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES

def trim_messages_hook(state):
    """保留最近 20 条消息，加上系统消息"""
    messages = state["messages"]
    if len(messages) > 20:
        # 保留系统消息 + 最近 20 条
        system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
        # 最近 20 条里若也有系统消息，去掉以免与 system_msgs 重复
        recent = [m for m in messages[-20:] if not isinstance(m, SystemMessage)]
        return {
            "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)] + system_msgs + recent
        }
    return {"messages": messages}

17.9.4 源码核对：interrupt() 的 resume 机制与多 Agent 人审批场景

多 Agent 系统最需要的能力之一是"某个 Agent 执行到敏感操作时等待人审批"——比如写作 Agent 写完后等审核员 +OK 再发出。libs/langgraph/langgraph/types.py:705 的 interrupt() 函数是这个机制的实现：

def interrupt(value: Any) -> Any:
    """Interrupt the graph with a resumable exception from within a node.

    The `interrupt` function enables human-in-the-loop workflows by pausing graph
    execution and surfacing a value to the client. ...

    In a given node, the first invocation of this function raises a `GraphInterrupt`
    exception, halting execution. ...

    A client resuming the graph must use the `Command` primitive to specify a value
    for the interrupt and continue execution. The graph resumes from the start of
    the node, **re-executing** all logic.
    """

三条容易被忽略的语义：

1、interrupt 发生时，当前节点抛 GraphInterrupt——这个节点本次的写入不提交；同一超步里的其他并行任务不会被取消（pregel/_runner.py 的 _should_stop_others 不把 GraphBubbleUp 当失败），照常跑完、写入存进 checkpoint，图在这一超步结束后暂停。这条异常通过 §17.3.5 讲过的 GraphBubbleUp 家族冒泡。第 16 章 §16.3.5 讲过 ToolNode 默认让 GraphBubbleUp 穿透，保证 interrupt 能正确冒泡——例外是配了 wrap_tool_call 且 handle_tool_errors=True 时，interrupt 会被吞成一条错误 ToolMessage、图不暂停。

2、恢复时用 Command(resume=value)：客户端用 graph.invoke(Command(resume="approved"), config=...) 告诉 Pregel "从中断处恢复、这个值作为 interrupt 的返回"——graph 从当前节点重新开始执行，但 interrupt() 调用会直接返回 resume 值（不再抛异常）。

3、节点从头重跑：官方注释明确写了 "The graph resumes from the start of the node, re-executing all logic."。这意味着节点里 interrupt 之前的代码会被跑第二次——任何副作用（数据库写、API 调用）会被重复执行。这条陷阱让用户必须把 interrupt 前的操作做成幂等的、或者把 interrupt 尽可能放在节点最开始。

多 Agent 系统下，interrupt 配合 Swarm/Supervisor 能实现复杂的人审批工作流——比如：

def sensitive_agent_node(state):
    # ... 收集信息 ...
    decision = interrupt({
        "action": "approve_refund",
        "amount": state["refund_amount"],
        "user": state["user_id"],
    })
    if decision == "approved":
        return Command(goto="refund_processor", update={"approved": True})
    else:
        return Command(goto="denial_handler", update={"reason": decision})

客户端收到 interrupt 值（包含申请详情），人工审核后调 graph.invoke(Command(resume="approved"), ...)——Agent 恢复、走批准路径。

17.9.5 源码核对：RetryPolicy 与多 Agent 的失败恢复

多 Agent 系统涉及多次 LLM 调用、多次 API 调用——任何一次都可能失败。libs/langgraph/langgraph/types.py:404-423 的 RetryPolicy 给每个节点配置精细的重试策略（docstring 标注 "Added in version 0.2.24"）：

class RetryPolicy(NamedTuple):
    """Configuration for retrying nodes.

    !!! version-added "Added in version 0.2.24"
    """

    initial_interval: float = 0.5
    """Amount of time that must elapse before the first retry occurs. In seconds."""
    backoff_factor: float = 2.0
    """Multiplier by which the interval increases after each retry."""
    max_interval: float = 128.0
    """Maximum amount of time that may elapse between retries. In seconds."""
    max_attempts: int = 3
    """Maximum number of attempts to make before giving up, including the first."""
    jitter: bool = True
    """Whether to add random jitter to the interval between retries."""
    retry_on: (
        type[Exception] | Sequence[type[Exception]] | Callable[[Exception], bool]
    ) = default_retry_on
    """List of exception classes that should trigger a retry, or a callable that returns `True` for exceptions that should trigger a retry."""

五个字段表达了一套指数退避 + jitter的成熟重试策略：

• 第 1 次重试等 0.5s
• 第 2 次等 1s（0.5 × 2）
• 第 3 次等 2s（1 × 2）
• 最多等 128s（max_interval 封顶）
• jitter=True 加随机抖动避免惊群效应

retry_on 接受 Exception 类型、序列或 callable——默认值 default_retry_on（libs/langgraph/langgraph/_internal/_retry.py）用的是黑名单而不是白名单：ConnectionError 一律重试，httpx.HTTPStatusError / requests.HTTPError 只在 5xx 时重试，而 ValueError、TypeError、ArithmeticError、ImportError、LookupError（KeyError/IndexError 的父类）、NameError、SyntaxError、RuntimeError、ReferenceError、StopIteration、StopAsyncIteration、OSError 这一串（多为编程错误）一律不重试——这类错误重试多少次通常都是同样结果；名单之外的异常默认返回 True（重试）。

多 Agent 场景下的用法：

builder.add_node(
    "llm_heavy_node",
    llm_node,
    retry_policy=RetryPolicy(max_attempts=5, retry_on=(httpx.TimeoutException,)),
)

只对 httpx.TimeoutException 重试，最多尝试 5 次（max_attempts 含首次，即最多重试 4 次）——LLM API 调用偶尔超时，重试能救；应用层错误不重试，避免浪费 token。

这条设计和第 16 章 §16.3.7 讲过的 _infer_handled_types（ToolNode 的错误分类）是同一哲学——精确控制哪些错误被吞、哪些向上抛。多 Agent 系统的可靠性建立在每一层都有明确的错误处理策略上。

17.10 设计决策

17.10.1 为什么用子图而不是函数调用？

将 Agent 实现为子图（而非简单的函数调用）的优势：

1. 独立 Checkpoint：每个子图有自己的 checkpoint 命名空间，支持独立的中断恢复
2. 可视化：子图在图的可视化中自然展示层级关系
3. 复用：同一个 Agent 子图可以在不同的多 Agent 系统中复用
4. 流式输出：子图的流式事件自动携带命名空间前缀，便于追踪

17.10.2 为什么 Handoff 用 Command 而不是状态更新？

Swarm 模式中使用 Command(goto=target) 而不是 return {"next_agent": target} 的原因：

1. 原子性：Command 的 goto 和 update 在同一步中执行
2. 可静态声明：配合 Command[Literal["a", "b"]] 返回注解或 add_node(..., destinations=...)，目标节点能在 compile() 的 validate() 里被检查、也能画进图；而 {"next_agent": target} 这种字符串状态字段对编译期完全不可见
3. 跨图能力：Command 可以通过 graph=Command.PARENT 导航到父图

17.10.3 状态隔离 vs 状态共享

不同的多 Agent 模式对状态共享有不同的需求：

• Supervisor：子 Agent 的状态可以通过子图隔离
• Swarm：所有 Agent 共享同一个状态（消息历史）
• 协作：共享状态是核心机制，每个 Agent 写入不同的字段
• 分层：每层有自己的状态，通过子图接口传递数据

17.11 小结

本章系统介绍了基于 LangGraph 实现的四种多 Agent 模式。每种模式都是底层原语（StateGraph、Send、Command、子图）的不同组合方式：

• Supervisor 通过 Command(goto=agent) 实现中央调度
• Swarm 通过 handoff 工具返回 Command 实现点对点交接
• 分层 Agent 通过子图嵌套实现多级管理
• 协作 Agent 通过共享 State 的不同字段实现接力式协作

选择哪种模式取决于具体需求：任务是否可分解、是否需要动态路由、Agent 数量和并行需求。在实践中，这些模式也可以混合使用——例如，顶层使用 Supervisor 模式，底层使用协作模式。

LangGraph 的优势在于它不绑定任何特定的 Agent 框架模式。无论选择哪种多 Agent 架构，底层都是相同的 StateGraph + Pregel 调度器，享有相同的 Checkpoint 持久化、中断恢复、流式输出和类型安全能力。这使得开发者可以根据业务需求自由组合，而不是被框架限制在某个特定的模式中。

下一章是本专栏的最后一章，我们将从更高的视角审视 LangGraph 的设计模式与架构决策——Pregel 计算模型的选择、Channel 版本追踪的巧妙、Checkpoint 时间旅行的实现、中断/恢复机制的全貌，以及如何基于这些思想构建你自己的工作流引擎。

17.11.1 源码核对：multi_agent supervisor 的 Send 并行派发如何和 messages reducer 配合

§17.8.1 给了 Supervisor 用 Send 并行派发多个任务的示例。实现这种"并行+结果聚合"的完整模式需要 state 设计和 reducer 配合：

from typing import Annotated
from operator import add

class ParallelSupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    task_results: Annotated[list[str], add]   # ← 关键：用 add reducer
    completed_tasks: Annotated[int, add]

def supervisor(state) -> Command:
    """一次派发多个并行任务；结果收齐后结束"""
    if state.get("task_results"):
        return Command(goto=END)
    tasks = analyze_and_split(state["messages"])
    return Command(goto=[Send("worker", {"messages": state["messages"], "task": t}) for t in tasks])

def worker(state) -> dict:
    result = do_task(state["task"])
    return {"task_results": [result], "completed_tasks": 1}

三条关键点：

1、reducer 必须是累积/合并型的。Annotated[list, add] 用 operator.add——两个 list 合并；Annotated[int, operator.add] 让并发 worker 的 completed_tasks 累加。如果用 Annotated[list, lambda old, new: new]（即后写覆盖前写），三个并发 worker 的 result 只会保留最后一个。

2、Send 发给同一个 node 名是允许的：上面代码里 3 个 Send 都指向 "worker"——Pregel 会为每个 Send 独立实例化一个 worker 任务（task_id 不同，§17.6.4 讲过）。3 个 worker 并发执行、各自产出 task_results 更新、都被 add reducer 合并到总 state。

3、后续节点能读到完整聚合结果：supervisor 派发完返回装着 Send 列表的 Command，当前超步结束；下一超步 worker 们并发执行；再下一超步回到 supervisor（假设 worker 都有边连回 supervisor，即 add_edge("worker", "supervisor")），此时 state["task_results"] 是完整的合并列表。三次超步完成"分发-并行执行-聚合"的 map-reduce——按上面的写法接上 START → supervisor 与 worker → supervisor 两条边、用把任务名转大写的桩函数作 do_task 实跑，三个任务得到 task_results == ['T1', 'T2', 'T3']、completed_tasks == 3。

这条 pattern 是 LangGraph 里并行派发多个 worker 的常见写法——Pregel 的并发 + Send 的分发 + reducer 的聚合三件套。

17.11.2 源码核对：handoff 工具里 return Command(goto=...) 在 ToolNode 的执行路径

§17.4 讲 Swarm 的 handoff 工具。真实的 handoff 能生效，依赖 ToolNode 识别工具返回值类型的能力。第 16 章 §16.3.8 讲过工具可以返回 Command；识别发生在 ToolNode._execute_tool_sync（tool_node.py:988-997；异步版 _execute_tool_async 在 1141-1149 有同样一段）的结尾：

# Process successful response
if isinstance(response, Command):
    # Validate Command before returning to handler
    return self._validate_tool_command(response, request.tool_call, input_type)
if isinstance(response, ToolMessage):
    response.content = cast("str | list", msg_content_output(response.content))
    return response

msg = f"Tool {call['name']} returned unexpected type: {type(response)}"
raise TypeError(msg)

到这里时，响应不是 ToolMessage 就必须是 Command——否则 ToolNode 抛 TypeError。注意普通工具返回的字符串、dict 等在 langchain_core 的 BaseTool 里已被 _format_output 包成 ToolMessage（§17.11.3），所以走到 TypeError 的只有既不是 ToolMessage 也不是 Command 的其他 ToolOutputMixin 对象。这条规则是 Swarm 能工作的基础：handoff 工具返回 Command，ToolNode 识别它是 Command 类型、经过 _validate_tool_command（tool_node.py:1398）校验后原样往上传到 Pregel。注意这个校验不碰 goto——它只管 update：一是 update 的形态要和 ToolNode 的输入形态匹配（dict 输入才能给 dict update、list 输入才能给 list update），二是 graph is None（发给当前图）时 update 里必须有一条 tool_call_id 对得上本次调用的 ToolMessage，否则抛 ValueError 并在错误信息里直接给出修法示例。只写 goto 不写 update 的 Command 由第 1432 行的 else: return command 原样放行；§17.4.2 的 handoff 工具带 dict update、ToolNode 输入也是 dict，形态检查通过，又因为 graph=Command.PARENT 不是 None，配对 ToolMessage 并非强制——那里带上它是为了让父图的消息历史里每个 tool call 都有回应。

这条"update 形态必须匹配输入形态"的校验（1404-1432 行）让错误在 ToolNode 这一层就暴露、不要等到 Pregel commit update 时才发现不兼容；而"必须配一条同 id 的 ToolMessage"那条（1452-1467 行）则是第 16 章 §16.2.9 讲过的 _validate_chat_history 在工具侧的前置防线——少了它，下一轮调模型时才会炸。

这条链路一环扣一环：

1. 用户写 handoff 工具，返回 Command(goto="billing", graph=Command.PARENT, update=...)
2. LLM 决定调用 handoff 工具
3. ToolNode._run_one 执行 handoff
4. 返回 Command 被识别、经 _validate_tool_command 校验
5. 上传到 Pregel、在 agent 子图里走 §17.4.5 讲过的 _control_branch，因 graph=Command.PARENT 抛 ParentCommand，按 §17.3.5 冒泡到父图
6. 父图把它转换成 channel 写入、下一超步激活 billing 节点

6 步转换让用户视角的"工具返回跳转意图"变成执行引擎的"channel 状态变化"——一条完整的抽象漏斗。

17.11.3 源码核对：ToolOutputMixin 的双重角色——Command 既是 graph 原语也是 tool 返回值

§17.2.5 讲 Command 时提到它是 Generic[N], ToolOutputMixin——为什么要继承 ToolOutputMixin？打开 langchain_core（langgraph 依赖）的 ToolOutputMixin 定义：

# langchain_core/messages/tool.py:16
class ToolOutputMixin:
    """Mixin for objects that tools can return directly.

    If a custom BaseTool is invoked with a `ToolCall` and the output of custom code is
    not an instance of `ToolOutputMixin`, the output will automatically be coerced to
    a string and wrapped in a `ToolMessage`.
    """

它是一个marker class——类体只有 docstring，什么都不做，只作为一个"类型信号"存在。langchain_core 的 _format_output（libs/core/langchain_core/tools/base.py）第一行就是 if isinstance(content, ToolOutputMixin) or tool_call_id is None: return content——是 ToolOutputMixin 就跳过默认的"把返回值 stringify 成 ToolMessage content"逻辑、直接原样上传。

这条设计让 Command 能同时扮演两个角色：

1. 作为 graph 节点返回值：走 §17.4.5 的 _control_branch
2. 作为工具函数返回值：绕过工具层（BaseTool._format_output）的字符串转换、被 ToolNode 识别为"跳转意图"

一个 class、两条识别路径——这是 Python 多重继承 + marker class 模式的典型应用。对比一下如果不用 ToolOutputMixin——_format_output 会把 Command 对象字符串化为 "Command(goto='billing')" 字符串、塞进 ToolMessage.content——LLM 看到的是一段毫无意义的字符串、handoff 彻底失效。

marker class 是 Python 里一条"不需要额外代码、只需要类型信息"的扩展机制。你在自己的工具系统里如果也需要"某些工具返回值有特殊含义"——用 marker class 比用字典或枚举类型检查更干净。

17.11.4 读完本章能回答的 10 个具体问题

作为本章掌握度的自测表：

1. Send 和 Command 的数据结构差异？（§17.2.4/§17.2.5——Send 2 字段、Command 4 字段复合）
2. Command.PARENT 如何跨图跳转？（§17.3.5——raise ParentCommand 异常冒泡 + 计算父图 ns）
3. checkpoint_ns 里为什么会有数字段？（§17.5.4——同一任务先后启动多个子图时由 subgraph_counter 追加，用来区分它们；docstring 说是并发消歧）
4. Send("unknown_node", ...) 会怎样？（§17.6.4——Pregel log warning 并静默忽略）
5. Send 的任务什么时候执行？（§17.6.4——下一超步，不是同步）
6. 工具返回非 ToolMessage/Command 会怎样？（§17.11.2——普通返回值先被包成 ToolMessage；其他 ToolOutputMixin 对象才让 ToolNode 抛 TypeError）
7. Command 的 update 字段接受哪几种类型？（§17.2.6——dict / list of tuples / NamedTuple / Pydantic / 单一值）
8. interrupt() 后节点恢复时会重跑吗？（§17.9.4——从头重跑节点，副作用可能重复）
9. RetryPolicy 默认对什么异常重试？（§17.9.5——default_retry_on 对 ConnectionError 和 5xx 重试，对 ValueError 等一串异常不重试，名单外的异常默认重试）
10. ToolOutputMixin 的作用？（§17.11.3——marker class 让 Command 绕过工具层的字符串序列化）

能回答 8 条以上说明源码细节已经被吃透；不清楚的条目直接回到对应小节按行号核对源码。

17.11.5 常见陷阱与工程经验总结

多 Agent 系统在生产环境里会遇到单 Agent 不会有的一类独特问题。把本章所有讲过的源码细节+工程经验汇总成6 条生产级 checklist：

1、 handoff 工具必须返回 Command 而不是字符串：return "transfer to billing" 对 Pregel 毫无意义——Pregel 看不懂自然语言路由意图。必须 return Command(goto="billing", ...)——Pregel 才能真的跳转；工具跑在 agent 子图里时还要加 graph=Command.PARENT（§17.4.2）。

2、子 Agent 和父图的 state schema 类型要对齐：子图的 input_schema 定义了它能接受什么 state，父图传入不匹配的字段会静默丢失。推荐用 TypedDict 继承或共享 schema。注意 LangGraph 运行时并不按 total=True 强制字段齐全（实跑：两个必填字段的 TypedDict state，invoke({}) 照常运行）；total=False / NotRequired[...] 影响的是类型检查器。

3、messages 历史会爆炸：§17.9.3 讲过 trim_messages_hook。实际生产系统必须在关键节点加消息修剪——多个 Agent 共用一条消息历史时，上下文随轮次增长得很快。可以考虑用 summary reducer 压缩历史、或者只传当前任务相关的上下文切片。

4、循环转交必须有 circuit breaker：§17.9.1 讲过 handoff_count。真实生产里建议三层防护：handoff_count 上限、total_step 上限（Pregel 自带 recursion_limit）、单步超时（编译后图的 step_timeout 属性，pregel/main.py:614）。任何一层触发就返回"任务超限，人工介入"。

5、并发 Send 时要给共享字段配合并型 reducer：Supervisor 用 Send 同时派发多个子任务、每个子任务都往同一字段写——字段没有 reducer 时，同一超步收到多个写入直接抛 InvalidUpdateError: At key '...': Can receive only one value per step（实跑）；reducer 若是"后写覆盖"型，则只留下一个结果。要用 Annotated[list, add_messages]、Annotated[list, operator.add] 或自定义的字典合并函数这类累积/合并型 reducer。

6、ParentCommand 穿透子图时要处理 state 合并语义：当子 Agent 用 Command(goto="parent_sibling", graph=Command.PARENT, update={...}) 跳到父图时，update 是应用到父图 state 的，不是子图的。用户常误以为 update 会传到两侧——实际只作用于目标图。这条语义在 §17.5.5 的 _get_root 里体现得最清楚。

17.12 本章与全专栏体系的呼应

多 Agent 模式不是空中楼阁——它建立在本专栏前面所有章节讲过的原语之上。梳理一下本章和全专栏的关联能让你看到 LangGraph 架构的整体性：

与第 3-4 章（StateGraph、Channel）的基础：Supervisor 的中央调度、Swarm 的 handoff、协作 Agent 的共享状态——它们全部是 StateGraph 的节点 + 条件边 + reducer 的不同组合。分层 Agent 用子图作为节点，子图也是 StateGraph——没有任何一种多 Agent 模式引入了 StateGraph 之外的新原语。

与第 6-7 章（Pregel 执行引擎、任务调度）的支撑：Send 的并行分发、Command 的跨图导航、子图的 ns 嵌套——全部通过 Pregel 的超步模型实现。第 6 章讲过的 "每超步从 channel 读、节点执行、写回 channel、再次触发" 循环是多 Agent 协作的底层物理。

与第 8 章（Checkpoint）的结合：多 Agent 的长流程天然需要持久化。Supervisor 每次决策、Swarm 每次 handoff、协作 Agent 每次迭代——都是独立的 checkpoint。checkpoint_ns 的嵌套（§17.5.4）让你能在复杂多 Agent 系统里精确定位到"这次运行里 research_team 子图中某个 worker 任务的状态"。

与第 15 章（Store）的共享记忆：§17.8.3 讲过跨 Agent 的 Store 共享。Research Agent 把研究结果写 Store、Writer Agent 从 Store 读——这是多 Agent 系统里"共享知识"的一种自然实现——不占消息历史，也比通过 state 传递解耦。

与第 16 章（ToolNode + create_react_agent）的衔接：§17.2.1 讲的 "Agent = CompiledStateGraph" 实际上引用了第 16 章 §16.2 的 create_react_agent——里面的 agent 节点、tools 节点、should_continue 条件都是 §16 细讲过的。多 Agent 系统里每个单 Agent 都是这个 pattern 的实例。

这种"每一章的原语在后一章组合出新的模式"的递进结构是 LangGraph 教学最有效的路径——你不需要学习 N 个独立概念，真正的概念只有 4 个（StateGraph、Pregel、Checkpoint、Store），剩下的全是组合。从简单 ReAct Agent（§16）到四种多 Agent 模式（本章）再到完整生产工作流（§18），其实用的都是同一组积木——只是搭成了不同形状。

这是 LangGraph 作为编排引擎的特质——少量原语 + 自由组合。LangChain 官方也把两种模式打包成了独立包：langgraph-supervisor（create_supervisor，PyPI 上为 0.0.31）和 langgraph-swarm（create_swarm、create_handoff_tool，PyPI 上为 0.1.0；均为 2026-09-25 查，不在本专栏 pin 树内）——但它们同样只是用 StateGraph + Command 搭出来的预设（§17.4.2 引过 langgraph-swarm 的 handoff 工具实现）。学会这几个原语的组合，你就能自己实现这些包之外的多 Agent 模式。

17.13 源码定位索引

为便于读者按图索骥核实本章说法：

小节	源文件	关键行号
§17.2.4 Send	libs/langgraph/langgraph/types.py	class Send:574-646
§17.2.5-17.2.6 Command	同上	class Command:652-702 / _update_as_tuples:687-700
§17.3.5 ParentCommand	libs/langgraph/langgraph/pregel/_retry.py	捕获:122-140 / _checkpoint_ns_for_parent_command:57-83
§17.4.5 _control_branch	libs/langgraph/langgraph/graph/state.py	1537-1563
§17.5.4 ns 格式	libs/langgraph/langgraph/pregel/_retry.py	注释:58-67
§17.5.5 _get_root	libs/langgraph/langgraph/graph/state.py	1580-1600
§17.6.4 Send → task	libs/langgraph/langgraph/pregel/_algo.py	prepare_push_task_send:878-937
§17.9.4 interrupt	libs/langgraph/langgraph/types.py	interrupt:705+
§17.9.5 RetryPolicy	同上	class RetryPolicy:404-423

源码版本：LangGraph 1.1.6。

下一章（第 18 章）将从更高的视角对 LangGraph 做架构复盘——Pregel 计算模型、Channel 版本追踪、Checkpoint 时间旅行、中断恢复机制如何汇聚成一套一致的设计哲学。