第 19 章多智能体系统：Supervisor / Router / Handoffs / Swarm

单一 Agent（Monolithic Agent）在面对复杂、跨领域任务时存在三类结构性瓶颈：上下文窗口限制——所有历史对话、工具输出、检索片段、反思草稿堆积在同一个 messages 列表里，超长上下文导致注意力稀释、关键信息被淹没、成本按轮次线性飙升；技能边界限制——一个 LLM 难以同时精通代码编写、SQL 查询、法律审查、市场分析、数学推理等异构技能，角色专用 prompt 与专用工具集才能稳定产出高质量结果；任务复杂度限制——某些任务天然需要多角色协作循环（研究员调研→作者撰写→审稿人评审→作者修订→审稿人复审），单 Agent 很难稳定完成这种多轮、跨角色的迭代。多智能体系统（Multi-Agent System，MAS）将任务拆分给多个拥有独立角色、独立上下文、独立工具集的 Agent，由显式协调机制组织其协作。LangGraph 官方将多智能体协作归纳为四种主流模式：Supervisor（主管-工人）、Router（路由）、Handoffs（交接）、Swarm（蜂群），四种模式均可由 StateGraph + Command + 子图直接表达，不需要引入额外框架。

19.1 四种协作模式与通信方式

模式	中央协调器	控制权流转	适用场景	可控性	复杂度
Supervisor	有（Supervisor LLM）	Worker 完成后返回 Supervisor，由 Supervisor 做下一轮决策	通用复杂任务（研究→写作→审稿循环），最常用、最可控	高	中
Router	有（入口 Router，一次决策）	Router 一次路由到目标 Agent，Agent 执行完即 END	单轮任务分发、意图分类（客服/技术/账单分流）	高	低
Handoffs	无固定中心	Agent 之间通过 transfer 工具主动把控制权交给对方	客服电话转接式场景、跨域支持	中	中
Swarm	无中心协调器	Agent 通过共享黑板/主题队列自由订阅/发布消息	开放研究、群体智慧、涌现行为实验	低	高

Supervisor（主管-工人模式）。一张图内包含一个 Supervisor 节点与若干 Worker 节点（researcher/writer/critic 等）；Supervisor 是一个 LLM 节点，接收任务描述与所有 Worker 产出的结构化摘要，决策下一步由哪个 Worker 执行（或判定任务完成路由到 END）；Worker 执行完毕通过 Command(goto="supervisor") 回到 Supervisor 做下一轮决策。该模式控制流清晰、可解释性强、终止条件容易保证，Supervisor 的 prompt 可以显式约束决策空间，是生产场景下的首选模式。

Router（路由模式）。入口只有一个 Router LLM（或轻量分类器），根据用户输入一次性选择最合适的专业 Agent，被选中 Agent 执行完直接到 END，不再回到 Router。该模式简单高效、轮次最少（Router 1 次 + Worker 1 次），但不支持多 Agent 协作，适合单轮任务分发（如"写代码走 code_agent、查天气走 weather_agent、查订单走 order_agent"）。Router 可以是 LLM（处理自然语言分类），也可以是规则/小模型（高确定性场景）。

Handoffs（交接模式）。不设中心 Supervisor，每个 Agent 的工具集中包含一组 transfer_to_xxx 工具；当 Agent 判断当前问题不属于自己领域（或已完成自己部分需要交接），通过调用 transfer 工具把控制权交给对应 Agent——工具函数返回 Command(goto=xxx) 实现跳转。这种模式类似电话转接，Agent 之间可以灵活转移对话、不需要中央决策点，但容易出现 A→B→A→B 的无限循环，必须加轮次上限、显式终止工具和转移检测。

Swarm（蜂群模式）。无中央协调器，Agent 之间通过 Topic Channel（发布-订阅）或共享 BaseStore（黑板模式）自由发布/订阅消息，每个 Agent 读取黑板上自己感兴趣的消息、产出新消息发布出去，直到某个终止条件满足（如某个 Agent 发布了 FINAL_ANSWER 类型的消息、或达到最大轮次）。该模式最灵活、最有可能产生涌现行为，但也最不可控——调试困难、token 消耗大、终止条件难保证，主要适合研究场景，不推荐直接用于对外生产服务。

多智能体通信方式有五种典型形态，按耦合度从高到低：

通信方式	耦合度	实现手段	适用场景
共享 State	高	所有 Agent 读写同一份 State 字段	简单 Router、小规模 Supervisor
Command/goto	中	节点返回 Command(goto=..., update=...)	Supervisor/Handoffs 显式控制流
消息传递（Send/Topic）	低	Send 派发任务、Topic 发布-订阅	Swarm/Map-Reduce/广播事件
共享 Store（黑板）	低	BaseStore put/get/search（跨线程持久）	跨会话协作、Swarm、异步 Agent
直接调用（子图嵌套）	强耦合	CompiledStateGraph 作为节点 add_node	强隔离的 Agent 封装（推荐）

多 Agent Memory 的核心问题是上下文隔离。如果所有 Agent 共用同一个 messages 字段，协作几轮后 messages 会膨胀到数万 token，旧 Agent 的推理痕迹、工具调用、错误尝试都会干扰新 Agent 的判断，成本也快速攀升。推荐架构是把每个 Agent 封装为独立子图，各自维护私有 messages 字段（只保留自己角色相关的历史、自己的工具调用记录），Supervisor 层只维护三个轻量字段：task（用户原始任务）、worker_results（各 Worker 产出的结构化摘要字典，而非完整 messages）、next_agent（调度目标）或 turn（轮次计数）。Worker 子图被调度时通过输入转换器拿到必要上下文（任务描述 + 前序结果摘要），不需要读取其他 Worker 的完整历史。

Token 成本控制策略：(1) Supervisor prompt 保持精简，只看结构化摘要（每个 Worker 一段总结+状态），不塞原始 messages；(2) Worker 返回给 Supervisor 的结果做摘要压缩（Worker 内部可以产出长文，但回给 Supervisor 时用专门的 summarize 节点压到 200 字以内）；(3) 限制 Supervisor 最大轮次（如 10 轮），超限兜底终止或转人工；(4) 子图 messages 字段内部用滑动窗口（保留最近 K 轮）或 Summary Memory 定期压缩；(5) 监控每次任务的 token 消耗与轮次数，对异常高消耗任务打告警。

终止条件必须显式设计，这是多 Agent 系统最常见的生产 Bug 来源：Supervisor 必须能输出 FINISH（或调用 finish 工具，或输出 Literal["FINISH"]），不能靠"自然聊完"；Handoffs 必须有 max_turns 上限与显式 finish 工具；Swarm 必须有 Agent 能产出终止消息，且必须设置 max_turns/recursion_limit 兜底。任何缺省终止条件的多 Agent 图都会在达到 recursion_limit 时抛 GraphRecursionError，且会反复调用 LLM 大量消耗 token 才停止。

19.2 Supervisor 模式完整示例

python

# langgraph==1.2.x
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class SupState(TypedDict):
    task: str
    research_note: str
    draft: str
    review: str
    next_agent: Literal["researcher", "writer", "critic", "FINISH"]
    turn: int

def researcher(s: SupState) -> Command:
    prompt = f"请对以下主题做 3-5 条要点研究（每条不超过 50 字）：{s['task']}"
    note = llm.invoke([("user", prompt)]).content
    return Command(goto="supervisor",
                   update={"research_note": note, "turn": s.get("turn",0)+1})

def writer(s: SupState) -> Command:
    prompt = f"基于以下研究要点写一篇 300 字短文：\n{s['research_note']}"
    draft = llm.invoke([("user", prompt)]).content
    return Command(goto="supervisor",
                   update={"draft": draft, "turn": s.get("turn",0)+1})

def critic(s: SupState) -> Command:
    prompt = (f"评审以下短文，严格以 APPROVED 或 REVISE: <理由> 开头回复：\n"
              f"[草稿]\n{s['draft']}\n[研究要点]\n{s['research_note']}")
    review = llm.invoke([("user", prompt)]).content
    goto = "FINISH" if review.startswith("APPROVED") else "writer"
    return Command(goto=goto, update={"review": review, "turn": s.get("turn",0)+1})

SUP_PROMPT = """你是任务主管，根据当前状态严格只输出一个词决定下一步：
researcher（需要研究要点）/ writer（需要写草稿）/ critic（草稿已写待评审）/ FINISH（任务完成）。
当前状态：research_note={'已完成' if s.get('research_note') else '未完成'}，
draft={'已完成' if s.get('draft') else '未完成'}，turn={s.get('turn',0)}。"""

def supervisor(s: SupState) -> Command:
    if s.get("turn", 0) >= 8:
        return Command(goto=END, update={"next_agent": "FINISH"})
    if not s.get("research_note"):
        nxt = "researcher"
    elif not s.get("draft"):
        nxt = "writer"
    else:
        # 让 LLM 判断是否需要 critic 再一轮
        r = llm.invoke([("user", SUP_PROMPT)]).content.strip().split()[0]
        nxt = r if r in {"researcher","writer","critic","FINISH"} else "critic"
    goto = END if nxt == "FINISH" else nxt
    return Command(goto=goto, update={"next_agent": nxt})

g = StateGraph(SupState)
g.add_node("supervisor", supervisor)
g.add_node("researcher", researcher)
g.add_node("writer", writer)
g.add_node("critic", critic)
g.add_edge(START, "supervisor")
for w in ["researcher","writer","critic"]:
    g.add_edge(w, "supervisor")
graph = g.compile()

r = graph.invoke({"task": "为什么流式输出对 LLM 应用用户体验至关重要", "turn": 0})
print("FINAL DRAFT:\n", r.get("draft","")[:300])
print("(review)", r.get("review","")[:80])

19.3 Handoffs 模式完整示例

python

# langgraph==1.2.x
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class CSState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    current_agent: str
    transfer_count: int

@tool
def transfer_to_technical():
    """将对话转给技术支持团队。"""
    return Command(goto="technical")

@tool
def transfer_to_billing():
    """将对话转给账单/支付团队。"""
    return Command(goto="billing")

@tool
def transfer_to_cs():
    """将对话转回一线客服。"""
    return Command(goto="cs")

@tool
def finish():
    """问题解决，结束对话，向用户道别。"""
    return Command(goto=END)

def make_agent(sys_prompt: str, tools: list):
    bound = llm.bind_tools(tools)
    def node(s: CSState):
        # 轮次熔断
        if s.get("transfer_count", 0) >= 6:
            return {"messages": [{"role":"assistant",
                                  "content":"为您转接人工客服，请稍候。"}],
                    "current_agent": "human"}
        resp = bound.invoke([("system", sys_prompt)] + s["messages"])
        if resp.tool_calls:
            tc = resp.tool_calls[0]
            tmap = {t.name: t for t in tools}
            if tc["name"] in tmap:
                ret = tmap[tc["name"]].invoke(tc["args"])
                if isinstance(ret, Command):
                    goto = ret.goto
                    new_agent = goto if isinstance(goto, str) and goto != END else s["current_agent"]
                    return Command(
                        goto=goto,
                        update={"messages": [resp],
                                "current_agent": new_agent,
                                "transfer_count": s.get("transfer_count",0)+1
                                if new_agent != s.get("current_agent")
                                else s.get("transfer_count",0)})
        return {"messages": [resp], "current_agent": s.get("current_agent","cs")}
    return node

cs_node = make_agent(
    "你是一线客服，负责通用咨询。技术问题(报错/功能异常)调用 transfer_to_technical，"
    "账单支付问题调用 transfer_to_billing，问题解决调用 finish。不要自己回答技术/账单细节。",
    [transfer_to_technical, transfer_to_billing, finish],
)
tech_node = make_agent(
    "你是技术支持，负责处理技术报错与功能异常。非技术问题调用 transfer_to_cs，"
    "解决后调用 finish。",
    [transfer_to_cs, finish],
)
bill_node = make_agent(
    "你是账单支持，负责支付/发票/退款问题。非账单问题调用 transfer_to_cs，"
    "解决后调用 finish。",
    [transfer_to_cs, finish],
)

g = StateGraph(CSState)
g.add_node("cs", cs_node).add_node("technical", tech_node).add_node("billing", bill_node)
g.add_edge(START, "cs")
graph = g.compile()

r = graph.invoke({
    "messages": [("user", "我支付的时候报错了，提示扣款成功但订单未支付，订单号 12345")],
    "current_agent": "cs", "transfer_count": 0,
}, config={"recursion_limit": 20})
print(r["messages"][-1].content[:200])
print("最后所在团队：", r.get("current_agent"))

本章小结

多智能体系统通过角色分工解决单 Agent 的上下文窗口限制、技能边界限制与任务复杂度限制。LangGraph 官方归纳四种协作模式：Supervisor（中央 LLM 调度 Worker 节点，Worker 执行完回到 Supervisor 再决策，最可控，生产首选）、Router（入口一次性路由到专业 Agent，执行完即 END，简单高效）、Handoffs（Agent 通过 transfer_to_xxx 工具主动交接控制权，灵活但易循环）、Swarm（无中心黑板/主题队列协作，最灵活但最难调试，仅适研究）。通信方式包括共享 State、Command/goto、Send/Topic、共享 Store、子图嵌套五种。多 Agent Memory 推荐每个 Agent 封装为子图独立维护 messages（私有 State），Supervisor 层只维护任务摘要、Worker 结果与调度字段，避免上下文无限膨胀。Token 成本通过 Supervisor prompt 精简、Worker 结果摘要化、轮次上限、消息裁剪多重手段控制。所有模式必须设计显式终止条件（FINISH/finish 工具 + max_turns + recursion_limit 兜底），防止 GraphRecursionError 与 token 雪崩。

常见错误

多 Agent 共用同一 messages 字段导致上下文无限膨胀。现象：对话 5 轮后 messages 累积上万 token、API 成本飙升、回答质量下降（注意力分散）、最终超过模型上下文窗口报错。原因：所有 Agent 把自己的全部思考、工具调用、草稿都追加到同一个列表，互相污染。修复：每个 Agent 封装为子图，私有 messages 在子图 State 内维护；Supervisor 层只存结构化摘要不存原文。
没有明确终止条件导致 GraphRecursionError。现象：任务跑到 recursion_limit 时报错结束，token 已经大量消耗。原因：Supervisor 永远不输出 FINISH、Handoffs 没有 max_turns 上限、Swarm 没有 Agent 发布终止消息。修复：Supervisor 硬约束 turn 计数（≥N 强制 END）；Handoffs 在 State 维护 transfer_count，超过阈值转到人工节点；Swarm 必须设置消息类型 FINAL_ANSWER 与最大轮次。
Handoffs 形成循环（A→B→A→B 反复转接）。现象：客服转技术、技术说这是账单问题转账单、账单说这是账号问题转客服，循环 3-4 次后熔断。原因：Agent 领域边界 prompt 不清，Agent 都觉得问题不属于自己；transfer 工具未强制记录理由。修复：transfer 工具加 reason 参数强制说明理由；State 维护最近 N 次 transfer 记录（from/to/reason），检测到 A→B→A 重复路径时强制路由到人工节点；system prompt 明确各自领域边界，要求先尝试回答再考虑转接。
Supervisor 频繁错误路由到错误 Agent。现象：研究还没做就派给 critic、草稿还没写就让 reviewer 审稿、简单问题在多个 Worker 之间来回调度浪费 token。原因：Supervisor prompt 不够清晰、让 LLM 完全自由决策、没有硬规则兜底。修复：Supervisor 用 with_structured_output 强制输出结构化决策（next_agent 枚举 + reason + confidence）；字段存在性硬规则优先——research_note 为空就必须先 researcher、draft 为空必须先 writer，LLM 判断只用于多轮后的评审循环；低置信度路由时 interrupt 让人工分派。
Worker 结果 Supervisor 没有正确消费。现象：Worker 返回了完整文本，Supervisor 下一轮像没看到一样重复派工到同一 Worker，或错误判断字段状态。原因：Worker 返回的消息格式与 Supervisor 预期不匹配（字段名错误、未 update 到 Supervisor 关注的字段、把结果只追加到 messages 没写入结构化字段）。修复：Worker 必须返回 Command(update={"结构化字段名": 值}) 而不是只追加 messages；Supervisor prompt 明确读哪些字段；测试时断言 Supervisor 收到的 State 中关键字段非空。

生产建议

生产场景首选 Supervisor 模式；Handoffs 用于客服/工单等天然转接场景；Router 用于大规模意图分类后的单轮分发；Swarm 仅用于内部研究试验，不作为对外服务模式；
每个 Agent 子图单独测试（单元测试覆盖其工具调用、输入输出 schema、transfer 条件、finish 条件），集成测试覆盖多轮协作路径与终止路径（正常完成/超限终止/转人工）；
Supervisor 使用 Pydantic with_structured_output 强制输出 next_agent/reason/confidence 结构化字段，禁止自由文本解析路由决策；
所有 Handoffs 强制记录 transfer 链（from/to/reason/timestamp）写入 State 与审计日志，便于事后分析循环原因与转接准确率；
在 Supervisor 层引入置信度阈值：低置信度派单前先 interrupt 让人工确认分派，避免错误路由浪费 token 与时间；
多 Agent 对话在 LangSmith 中通过 metadata.mas_run_id 串联所有子 trace，跨 Agent 追踪 token 消耗、延迟分布、错误率；
预先压测最坏路径（max_turns 触发、所有 Worker 都被派一遍、最长循环）的 token 成本与端到端延迟，以此设置成本报警与业务 SLA；
多 Agent 系统上线初期开启"影子模式"：Supervisor 决策先记录日志但不实际派单，人工对比决策准确率，达标后再切换为自动执行。

C 语言

语言点

深度

新标准

面试整理

第 19 章 多智能体系统：Supervisor / Router / Handoffs / Swarm ​

19.1 四种协作模式与通信方式 ​

19.2 Supervisor 模式完整示例 ​

19.3 Handoffs 模式完整示例 ​

本章小结 ​

常见错误 ​

生产建议 ​