AI Agent 开发(24):多 Agent 架构

QuibblerAgentQuibblerAgent 2026-10-25 约 28 分钟 3 次阅读

第 10 篇讲子代理时留了个尾巴:把 Agent 包成工具,主编就能把活派出去。那是多 Agent 的最小形态——控制权始终攥在主编手里,子代理跑完就交卷。到了真实的业务规模,这颗种子会长成完整的多人班底:有人专职调度、有人埋头干活、有人互相移交接力棒。2026 年的生产级 Agent 系统里,多 Agent 已经从炫技架构沉淀为几类稳定模式,本篇把它们一次讲透。

本篇是《AI Agent 开发》系列第 24 篇,代码基于 langgraph 1.2.x 与 langchain 1.x(2026-10 基线),环境沿用第 2 篇的 uv 项目。全文按"何时需要 → Supervisor 调度 → Swarm 移交 → 层级团队 → 通信风格 → 实战创作团队 → 观测排障 → 总结"推进——先泼冷水,再上架构。

1、何时需要多 Agent

先说结论:单 Agent 够用就别上多 Agent。多 Agent 的三项收益都真实存在,但每一项都伴生代价,先看收益。

第一是上下文隔离。单 Agent 的消息历史是单份的:检索环节烧掉的五万 token 会原封不动跟着进写作环节,长上下文里模型注意力被稀释,账单同步膨胀。多 Agent 把中间产物关在各自的窗口里,调度者只看结论——脏活留在干活的人手里。

第二是专业分工。不同职责配不同的提示词与工具集。第 10 篇讲过工具 schema 一多选择质量就下滑:十几个工具全塞进每次模型调用,模型开始选错工具、漏看参数。子代理把一组工具收拢成一项职责,决策面瞬间干净。

第三是并行加速。无依赖的子任务可以扇出并发执行,墙钟时间只取决于最慢的一支,对调研、翻译、批量审核这类任务收益直接。

代价同样有三条,姑且叫"三涨":成本涨——每个工人都要重读自己那份上下文,调度者每跳一次就是一次完整的模型调用;延迟涨——串行的跳数层层叠加;调试涨——答案错了,错在哪一环?根因可能在三个 agent 之前的一次工具幻觉里。

多 Agent 是用系统复杂度换能力边界。单 Agent 加一份干净的提示词与十几个工具能覆盖大多数业务场景;多 Agent 是任务规模逼出来的,不是架构审美选出来的。

判断清单如下,命中两条以上再考虑拆分:

  • 工具清单超过 15~20 个,且天然分簇(检索一簇、写库一簇)
  • 任务是阶段流水线(调研→写作→审校),各阶段提示词互相打架
  • 单 Agent 上下文被中间产物撑爆,压缩摘要开始丢关键信息
  • 子任务彼此独立,并行有真实收益

官方文档如今把多 Agent 形态归为五类:subagents(agent as tool)、handoffs、skills、router 与 custom workflow,详见多 Agent 总览。本篇展开前两类对应的两种主流拓扑:Supervisor 与 Swarm。

2、Supervisor 调度模式

Supervisor 的结构一句话:一个调度者加 N 个工人。调度者不干活,只看任务决定"下一个谁来";工人是完整 Agent(自己的提示词、工具、循环),干完回调度台汇报。拓扑上调度者是唯一枢纽,START 进调度者,每个工人的出边都指回调度者。

先交代库的现状,这是 2026 年与老教程差异最大的地方。langgraph-supervisor 库曾是官方推荐入口,如今 PyPI 页面停在 0.0.31(2025-11-19 发布),官方迁移指南明确标注不再维护。老教程的写法长这样:

# 老教程常见写法:langgraph-supervisor 0.0.x(已停更,识别即可)
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

researcher = create_react_agent(model, tools=[web_search],
                                prompt="你是检索专家。")
writer = create_react_agent(model, tools=[write_file],
                            prompt="你是撰稿人。")

workflow = create_supervisor(
    [researcher, writer], model=model,
    prompt="按任务类型派工,全部完成后汇总作答。")
app = workflow.compile()

解读:老代码能跑就别动,新项目别再引入这个依赖。官方给的迁移方向正是第 10 篇讲过的 agent as tool——调度者本身是一个 create_agent,每个工人被 @tool 包一层塞进它的工具列表。于是 Supervisor 在现代有两种实现路线,核心差异在控制权:

维度 工具式(agent as tool) 图式(Command 路由)
控制权 始终在调度者手里 工人执行期间接管
产出回流 工具返回值,即一条 ToolMessage 写进共享 State,回边交还
面向用户 产出必须经调度者转述 工人可直接续接对话
适合场景 多源产出要汇总加工 流水线、需要回环打回

工具式第 10 篇已有完整代码,不再重复。图式是本篇主角:工人是 StateGraph 上的节点,调度节点返回 Command(goto=...) 指派下一站,第 6 节完整实现。选型一句话:产出要"合"——多份材料汇总成一份报告,用工具式;流程要"流"——阶段推进、打回重做,用图式。

3、Swarm 与 Handoff 模式

Swarm 把调度者撤了:没有中心节点,每个 Agent 手里握着移交(handoff)工具,判断"这活不归我"就直接转给同事。客服是教科书场景:售前顾问正聊报价,客户问出报错细节,售前调用 handoff 移交技术支持,技术支持答完又把物流问题转给物流专员。用户视角始终是一个连续对话,后台换了三个人。

库这边是好消息:langgraph-swarm 仍在活跃维护,2026-10 的版本是 0.1.0(2025-12-04 发布),依赖 langchain 与 langgraph 1.x。安装后照客服场景搭一个双员 swarm:

# uv add langgraph-swarm
from langchain.agents import create_agent
from langgraph_swarm import create_handoff_tool, create_swarm

presales = create_agent(
    model,
    tools=[query_price, create_handoff_tool(
        agent_name="tech_support",
        description="涉及安装、报错、配置的技术问题,移交技术支持")],
    system_prompt="你是售前顾问,只回答产品与报价。",
    name="presales",
)
tech_support = create_agent(
    model,
    tools=[search_docs, create_handoff_tool(
        agent_name="presales", description="报价与选型问题,移交售前")],
    system_prompt="你是技术支持工程师,依据文档原文回答。",
    name="tech_support",
)

swarm = create_swarm([presales, tech_support],
                     default_active_agent="presales")
app = swarm.compile(checkpointer=checkpointer)  # 不挂会忘掉谁在岗

解读三个要点。其一,create_handoff_tool 生成的是一个"转接"工具,description 就是教模型何时转接的说明书,移交条件写得越具体,swarm 的路由质量越高。其二,create_swarm 返回的是 StateGraph 而非编译好的图,要自己 compile;共享状态里有个 active_agent 字段记录当前谁在岗,多轮对话自动从最后活跃的 Agent 续接——但前提是挂上 checkpointer,否则下一轮就忘了谁在岗。其三,移交是双向的:每个 Agent 都得持有指向同事的 handoff 工具,不然移交过去就回不来。

create_handoff_tool 生成的工具,掀开盖子本质是返回一个 Command。官方 handoffs 文档给了手写版,这是理解移交机制的最佳教材:

from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

@tool
def transfer_to_tech(runtime: ToolRuntime) -> Command:
    """技术问题移交技术支持。"""
    last_reply = next(
        msg for msg in reversed(runtime.state["messages"])
        if isinstance(msg, AIMessage))
    receipt = ToolMessage(content="已移交技术支持",
                          tool_call_id=runtime.tool_call_id)
    return Command(
        goto="tech_support",       # 父图里的兄弟节点
        graph=Command.PARENT,      # 目的地不在当前子图内
        update={"active_agent": "tech_support",
                "messages": [last_reply, receipt]},
    )

解读:goto 指向父图里的兄弟节点,graph=Command.PARENT 声明目的地不在当前 Agent 子图内部(参数全集见 Command API 参考)。最值得盯的是 update 里那对消息:带工具调用的 AIMessage 与 tool_call_id 严格对上的 ToolMessage 必须成对写入,缺一个消息历史就是畸形的,后续模型调用会报工具调用无响应。官方还建议只递这对消息而非全量历史——接收方拿到摘要而非流水账,上下文干净。

另外提一句:handoffs 文档还给了第三条路——单个 create_agent 配 @wrap_model_call 中间件,按状态字段动态换系统提示词与工具集。物理上是单 Agent,逻辑上多人格切换,小规模场景比真拆图轻得多。

4、层级团队

Supervisor 的调度者自己也是 Agent,也会过载:工人列表超过二十个,路由决策质量开始下滑,调度者的上下文被一堆工具 schema 挤满。解法是再套一层——supervisor 的 supervisor,总调度只面对几个组调度,组调度面对自己的组员:

                    ┌──────────────┐
     用户请求 ─────► │ 总调度 Agent  │
                    └──────┬───────┘
               ┌───────────┴───────────┐
               ▼                       ▼
      ┌────────────────┐     ┌────────────────┐
      │ 研究组调度 Agent │     │ 写作组调度 Agent │
      └───────┬────────┘     └───────┬────────┘
          ┌───┴────┐           ┌────┴────┐
          ▼        ▼           ▼         ▼
        检索员   资料核对员    撰稿员     审校员

解读:每一层调度者的注意力只花在直接下属上,这与人科组织的分工逻辑一致。但层级不是免费的——层越深,trace 越难读、状态传递路径越长。官方迁移指南对层级用法的态度很务实:能拍平就拍平,每片叶子做成总调度的一个工具;真正需要分层的信号是某个小组的内部协调本身就是一个复杂任务,比如研究组内部的"检索→核对→汇总"自己就是条流水线。先数工人数量,再决定要不要分层。

5、黑板与消息传递

多个 Agent 之间怎么通信?两种风格。共享黑板:全员读写同一份 State,LangGraph 的默认风格——add_messages 之外的字段就是公共白板,任何节点缺什么直接从 State 里拿。消息传递:交互只携带必要消息,handoff 的那对消息、Command(update=...) 的增量,各 Agent 上下文互不可见。

维度 共享黑板 消息传递
载体 全员读写的同一份 State Command 携带的增量消息
可见性 任何节点都能读全量字段 只有接收方看到
token 成本 字段多了全员买单 只递必要消息,上下文干净
调试 打印一份 state 全局可见 数据流隐式,漏传即丢信息
适用 同一张图内的小团队 跨子图、跨层协作

量级参考:官方文档的性能对比里,无状态的 subagents 每次请求成本稳定但全价,带状态的 handoff 在重复请求上能省约 40%~50% 的模型调用。实践建议:一个团队内部(同一张图)用黑板,字段收敛并注释清楚每个字段的写者;跨子图、跨层协作用消息传递,显式携带、成对传递。第 6 节的实战就是黑板风格的收敛示范。

6、内容创作团队实战

需求:输入选题,planner 出大纲,writer 写初稿,reviewer 审稿——不合格带意见打回 writer 重写,合格收工。reviewer 用 interrupt 换成人工审,AI 起草、人把关。这张图融合第 8 篇的 Command 路由与第 11 篇的 interrupt,是 supervisor 图式的完整落地。

6.1、状态与工人

State 设计是黑板风格的收敛示范:五个字段各管一摊,feedback 是流程旗标(待审、打回、通过),comment 是打回通道,审稿意见经它流回 writer 的提示词。

from typing import Literal
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt

class TeamState(TypedDict):
    task: str      # 选题,invoke 时传入
    outline: str   # planner 产出的大纲
    draft: str     # writer 产出的初稿
    feedback: str  # 流程旗标:待审 / 打回 / 通过
    comment: str   # 审稿意见,打回时递给 writer

model = init_chat_model("openai:gpt-5")  # 第 2 篇的统一模型入口

def planner(state: TeamState) -> dict:
    outline = model.invoke([
        ("system", "你是主编,只输出三段式大纲。"),
        ("human", state["task"])]).content
    return {"outline": outline}

def writer(state: TeamState) -> dict:
    draft = model.invoke([
        ("system", "你是撰稿人,严格按大纲成文。"),
        ("human", f"大纲:\n{state['outline']}\n"
                  f"审稿意见:{state.get('comment', '首稿,无意见')}")]).content
    return {"draft": draft, "feedback": "待审"}  # 旗标复位,回环才收敛

解读两点。其一,planner 与 writer 是最朴素的一次模型调用,教学刻意为之;真实系统里把任一工人换成 create_agent 子图即可挂上检索、写库等工具,图结构一字不动。其二,writer 每次重写都把 feedback 拉回"待审"——这是回环能收敛的关键,旗标不复位,调度者会无限派活。

6.2、调度与人工审

reviewer 与 supervisor 是这张图的灵魂,前者用 interrupt 把终审权交给人,后者用 Command 驱动全流程:

def reviewer(state: TeamState) -> dict:
    decision = interrupt({   # 第 11 篇:人工审稿位,暂停并递出申请单
        "draft": state["draft"],
        "标准": "结构完整、论据充分、语言通顺",
    })
    if decision == "approve":            # resume 的值流回这里
        return {"feedback": "通过"}
    return {"feedback": "打回", "comment": decision}

def supervisor(state: TeamState) -> Command[
        Literal["planner", "writer", "reviewer", "__end__"]]:
    if not state.get("outline"):          # 第一站:出大纲
        return Command(goto="planner")
    if state.get("feedback") == "打回":    # 带意见回炉
        return Command(goto="writer")
    if not state.get("draft"):            # 有纲无稿:写初稿
        return Command(goto="writer")
    if state.get("feedback") == "通过":    # 审稿通过:收工
        return Command(goto=END)
    return Command(goto="reviewer")       # 其余情况:送审

解读两点。其一,reviewer 的 interrupt() 携带申请单(初稿加审稿标准)暂停,人工 resume 的值原样流回 decision:传 "approve" 通过,传其他文本就是驳回意见,落进 comment。其二,supervisor 这里是确定性规则而非模型调用,是教学取舍;生产环境把判据换成一次结构化输出的模型调用(第 4 篇),图结构不变。注意返回类型注解 Command[Literal[...]] 不是装饰,LangGraph 靠它登记合法目的地。最后组装成图:

from langgraph.checkpoint.memory import InMemorySaver

builder = StateGraph(TeamState)
for name, fn in [("supervisor", supervisor), ("planner", planner),
                 ("writer", writer), ("reviewer", reviewer)]:
    builder.add_node(name, fn)
builder.add_edge(START, "supervisor")
for name in ("planner", "writer", "reviewer"):
    builder.add_edge(name, "supervisor")   # 工人干完活,回调度台

graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "essay-1"}}

graph.invoke({"task": "写一篇 LangGraph 入门导览"}, cfg)  # 停在 reviewer
graph.invoke(Command(resume="开头太拖沓,直接进正题"), cfg)  # 驳回重写
graph.invoke(Command(resume="approve"), cfg)  # 通过,图走到 END

解读:工人的出边统一指回 supervisor,调度者是拓扑上唯一的枢纽——这正是第 2 节说的 Supervisor 骨架。compile 挂 checkpointer、invoke 带 thread_id,是 interrupt 的两个硬前提(第 11 篇)。运行时序:第一次 invoke 推进到 reviewer 的 interrupt 处暂停返回;第一次 resume 带驳回意见,图走打回回环(writer 重写、再送审、再次暂停);第二次 resume 传 approve,图走到 END,终稿在 result["draft"] 里。

打回是回环,recursion_limit 就是保险丝:默认 25 次节点执行够三四轮打回,真遇到审稿人死磕,要么调上限,要么给 comment 加轮次计数到点强制放行。

7、多 Agent 观测排障

单 Agent 的日志是一条线,多 Agent 的日志是一团毛线:一次用户请求在三层图里跑出上百次节点执行,答案出错了,根因可能埋在三个 agent 之前的一次工具幻觉里。观测不是锦上添花,是多 Agent 能否进生产的前提。三板斧:

  • 全员命名:name 参数与图节点名全部显式命名
  • 全链路 trace:一次 invoke 的全部子图执行串成一棵树
  • 观测先行:trace 打通之前别进生产

第一条解决定位:create_agent 传了 name、节点名不偷懒,trace 树就按 agent 分组,错误出在哪一层、哪一次调用,一眼可见。第二条是能力本体:LangSmith 把跨层嵌套的子图执行串成一棵完整的树,每次模型调用、工具调用、token 消耗全部展开——多 Agent 的观测噩梦,靠的是这棵树而不是逐节点打印。第三条是纪律:多 Agent 系统的第一次线上事故,事故报告大概率以"不知道谁干的"开头。

trace 是排障用的,评测是防回归用的:把典型任务做成数据集回放、用 LLM-as-judge 给各工人产出打分,这套在第 27 篇展开。经验法则:多 Agent 改一处提示词,全链路 trace 至少跑一轮再合入。

8、总结

回顾全文。多 Agent 的正当理由是上下文隔离、专业分工、并行加速,代价是成本、延迟、调试三涨——单 Agent 够用就别拆。Supervisor 有两条现代路线:工具式控制权不出调度者、适合汇总;图式用 Command 路由、适合流水线与打回,停更的 langgraph-supervisor 库不必再引入。Swarm 靠 handoff 直接移交,langgraph-swarm 0.1.0 仍活跃,create_handoff_tool 的本质是返回 Command(goto=..., graph=Command.PARENT),消息必须成对。层级团队能拍平就拍平;通信风格黑板对内、消息传递对外。实战把 planner、writer、reviewer 组成带打回回环的 supervisor 图,人工审用 interrupt 收口。

练习:给创作团队加一个 fact-checker——插在 writer 与 reviewer 之间,挂上第 15 篇的检索工具,对初稿的关键论断逐条核查并输出核查表。二选一扩展:核查不通过的论断直接打回 writer(把核查表写进 comment),或仅作为 reviewer 申请单的附件供人工参考;进阶再把 supervisor 的判据从确定性规则换成模型决策。

对于正准备把单 Agent 拆成团队的场景,建议先用本篇的四节点图跑通最小闭环:工人先用纯函数占位,拓扑稳定后再逐个换成 create_agent——拓扑先行,能力后补,每一步都在 trace 里看得见。

相关推荐

Hermes Agent 详解:自进化学习型 AI 代理
Agent

Hermes Agent 详解:自进化学习型 AI 代理

Hermes Agent 详解:自进化学习型 AI 代理Hermes Agent 是由 Nous Research 于 2026 年 2 月开源的自主 AI 代理框架,其核心定位是"与你共同成长的智能体"。与传统的 IDE 编程助手或单一 API 聊天机器人不同,Hermes 运行在你的服务器上,能够记住所学内容,运行时间越长能力越强。它采用 MIT 开源许可,支持 Linux、macOS 和 W

393
Hermes Agent:与你一起成长的开源 AI 代理
Agent

Hermes Agent:与你一起成长的开源 AI 代理

Hermes Agent:与你一起成长的开源 AI 代理大多数 AI 助手是"健忘的租客"——换个窗口就失忆、换个平台就重学、干过的活不留下任何沉淀。Nous Research 出品的 Hermes Agent 反其道而行,口号是"The Agent That Grows With You"(与你一起成长的代理):它是一个开源(MIT)、自托管的 AI 代理,具备持久记忆(学习你的项目、自动生成技

322
speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线
AI

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线语音 Agent 的"全开源"路线一直有个尴尬:框架开源了,模型却离不开云端 API。huggingface/speech-to-speech 把这条路走通了:VAD → STT → LLM → TTS 四段低延迟流水线,每个组件都可换成开源模型本地跑——Apple Silicon 一台 Mac 就能撑起

264