多 Agent 系统的难点不在于“让多个 Agent 跑起来”,而在于让它们稳定地协作:谁负责决策,谁持有上下文,什么时候交接控制权,交接失败后怎么恢复,状态冲突怎么处理。
如果只是在一个任务里调用多个工具,很多时候单 Agent 就够了。多 Agent 真正解决的是异构协作问题,也就是不同任务阶段需要不同角色、不同提示词、不同工具权限,甚至不同模型能力。
一个典型的软件研发 Agent 系统可能是这样:
flowchart LR
U[用户需求] --> P[产品分析 Agent]
P --> C[代码生成 Agent]
C --> R[代码审查 Agent]
R --> T[测试 Agent]
T --> O[最终交付结果]
这里的几个 Agent 不只是“分工不同”,而是能力边界不同:
| Agent | 核心职责 | 需要的工具 | 输出形式 |
|---|---|---|---|
| 产品分析 Agent | 澄清需求、拆任务 | 需求模板、历史 PRD | 需求规格 |
| 代码生成 Agent | 编写代码 | 代码仓库、编辑器、依赖管理 | 源码变更 |
| 代码审查 Agent | 检查质量和风险 | 静态分析、规范检查 | Review 结论 |
| 测试 Agent | 生成并执行测试 | 测试框架、CI 环境 | 测试报告 |
如果这些职责用同一个 System Prompt、同一批工具、同一种输出格式就能完成,那就没有必要拆成多个 Agent。拆分会带来额外成本:调用链变长、上下文传递更复杂、错误更容易级联。
什么时候应该使用多 Agent
多 Agent 适合解决“角色、工具、模型、上下文”明显不同的问题,而不是单纯解决“步骤多”的问题。
| 场景 | 是否适合多 Agent | 原因 |
|---|---|---|
| 一个 Agent 调用搜索、数据库、计算器等多个工具 | 不一定适合 | 只是工具多,单 Agent + Function Calling 通常足够 |
| 先检索资料,再写报告,再做事实核查 | 适合 | 检索、写作、核查的提示词和输出标准不同 |
| 代码生成、代码审查、测试执行协作 | 适合 | 角色目标不同,工具权限也不同 |
| 辩论、红蓝对抗、多角度评估 | 适合 | Agent 需要持有不同立场 |
| 固定表单填写流程 | 不一定适合 | 状态机或普通工作流可能更可靠 |
| 企业审批、风控、审计流程 | 适合但要谨慎 | 需要角色隔离、日志追踪和权限控制 |
判断是否该拆 Agent,可以看四个信号:
-
提示词是否明显不同
如果每个阶段都要用完全不同的角色设定和约束,就有拆分价值。 -
工具权限是否不同
例如代码 Agent 可以写仓库,审查 Agent 只能读仓库,测试 Agent 可以访问 CI(持续集成)环境。 -
输出契约是否不同
一个 Agent 输出 Markdown 报告,另一个 Agent 输出 JSON 测试结果,下游消费方式完全不同。 -
模型选择是否不同
有的阶段需要强推理模型,有的阶段只需要低成本模型做格式整理。
多 Agent 系统在特定研究和基准中出现过较高失败率,范围可达到 41% 到 86.7%。这个数字不应该被理解成“多 Agent 一定不能用”,而是说明:只要拆成多个 Agent,就必须认真处理交接、状态、失败恢复和可观测性。
Supervisor 与 Swarm:两种常见协作模式
多 Agent 系统最常见的两种编排方式是 Supervisor 和 Swarm。
Supervisor:集中式调度
Supervisor 模式里有一个中心 Agent,负责理解用户请求、拆分任务、分配工作、收集结果和生成最终回复。其他 Agent 通常不直接面对用户。
flowchart TD
U[用户] --> S[Supervisor Agent]
S --> A1[检索 Agent]
S --> A2[分析 Agent]
S --> A3[写作 Agent]
A1 --> S
A2 --> S
A3 --> S
S --> U
这种模式的优势是控制清晰。所有关键决策都经过 Supervisor,因此容易做日志、审计、权限控制和失败恢复。
但它也有明显代价:Supervisor 每次路由都要消耗一次模型调用或规则判断,还要在不同 Agent 之间转换上下文。对于强交互场景,这部分开销可能占据较大比例。
Supervisor 适合这些场景:
- 企业级流程,需要审计和追踪;
- 工作流结构比较明确;
- 需要统一权限控制;
- 下游 Agent 不应该直接暴露给用户;
- 每一步都需要中心节点判断是否继续。
Swarm:去中心化交接
Swarm 模式没有固定的中心调度者。当前活跃 Agent 可以根据任务需要,把控制权直接交给另一个 Agent。
flowchart LR
U[用户] --> A[接待 Agent]
A --> B[技术支持 Agent]
B --> C[账单 Agent]
C --> B
B --> U
Swarm 的核心是 Agent 之间可以直接 Handoff。它更像一个专家网络:当前 Agent 发现自己不适合继续处理,就把上下文和控制权交给更合适的 Agent。
这种方式的优势是交互灵活、延迟较低。不必每一步都回到中心节点,适合动态对话和低延迟任务。
代价是全局控制变弱。Agent 之间如果没有清晰的交接协议,容易出现循环交接、上下文丢失、责任不清等问题。
Swarm 适合这些场景:
- 对话式任务,用户问题会频繁变化;
- Agent 数量可能动态增加或减少;
- 更关注响应速度,而不是严格流程;
- 每个 Agent 都能独立判断下一步应该交给谁。
混合架构:上层集中,下层灵活
复杂系统经常采用混合架构:顶层用 Supervisor 控制主流程,局部任务内部用 Swarm 做灵活协作。
flowchart TD
U[用户] --> S[顶层 Supervisor]
S --> G1[研发子系统]
S --> G2[数据分析子系统]
subgraph 研发子系统 Swarm
C[代码 Agent] --> R[Review Agent]
R --> T[测试 Agent]
T --> C
end
subgraph 数据分析子系统 Swarm
Q[查询 Agent] --> A[分析 Agent]
A --> V[校验 Agent]
end
G1 --> S
G2 --> S
S --> U
选型可以按下面这张表判断:
| 需求 | 更适合的模式 |
|---|---|
| 需要审计、追踪、权限控制 | Supervisor |
| 流程固定,步骤清晰 | Supervisor |
| 对话动态变化,路由不固定 | Swarm |
| 希望减少中心调度开销 | Swarm |
| 既要控制主流程,又要局部灵活 | 混合架构 |
Handoff 不是传消息,而是控制权、上下文和状态的交接
Handoff 是多 Agent 系统里最容易被低估的机制。它不是简单地把一句话传给下一个 Agent,而是包含三件事:
- 控制权转移:下一个由谁继续处理?
- 上下文迁移:应该传哪些历史信息和中间结果?
- 状态同步:共享状态如何更新,冲突如何处理?
完整的 Handoff 过程可以表示成这样:
sequenceDiagram
participant A as 当前 Agent
participant S as 状态存储
participant B as 目标 Agent
participant L as 日志系统
A->>A: 判断无法继续处理
A->>S: 写入阶段结果和交接原因
A->>B: 发送精简上下文
B->>S: 读取必要状态
B->>B: 校验输入契约
B-->>A: 接收控制权
A->>L: 记录 Handoff 事件
一个可靠的 Handoff 至少要回答这些问题:
| 问题 | 如果不处理会怎样 |
|---|---|
| 为什么要交接? | 下游 Agent 不知道任务背景 |
| 交给谁? | 可能路由到错误角色 |
| 传什么上下文? | 传太少会丢信息,传太多会干扰判断 |
| 当前状态是否已经写入? | 下游读到旧状态或半成品 |
| 交接失败怎么办? | 任务卡住或重复执行 |
| 是否允许交回? | 可能出现无限循环 |
LangGraph 中的两种 Handoff 写法
LangGraph 适合构建多 Agent 工作流,因为它把 Agent 调用建模成图,把共享数据建模成状态。节点是执行单元,边决定下一步去哪里。
条件边:适合固定流程
如果流程比较固定,可以用条件边根据状态选择下一个节点。
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END
class State(TypedDict):
task: str
route: Literal["researcher", "writer", "finish"]
research_notes: str
final_answer: str
def router(state: State) -> State:
# 实际项目里可以由规则或 LLM 决定 route
if not state.get("research_notes"):
return {**state, "route": "researcher"}
if not state.get("final_answer"):
return {**state, "route": "writer"}
return {**state, "route": "finish"}
def choose_next(state: State):
return state["route"]
builder = StateGraph(State)
builder.add_node("router", router)
builder.add_node("researcher", researcher_agent)
builder.add_node("writer", writer_agent)
builder.set_entry_point("router")
builder.add_conditional_edges(
"router",
choose_next,
{
"researcher": "researcher",
"writer": "writer",
"finish": END,
},
)
builder.add_edge("researcher", "router")
builder.add_edge("writer", "router")
graph = builder.compile()
条件边的好处是可控、可视化、容易测试。缺点是灵活性有限,流程变化多时,边会越来越复杂。
Command:适合动态交接
如果希望 Agent 自己决定下一站,可以返回 Command,同时更新状态并指定目标节点。
from typing import TypedDict
from langgraph.types import Command
class State(TypedDict):
task: str
research_notes: str
draft: str
def researcher_agent(state: State) -> Command:
notes = "整理后的检索结论..."
return Command(
update={
"research_notes": notes,
},
goto="writer",
)
def writer_agent(state: State) -> Command:
draft = f"基于资料生成回答:{state['research_notes']}"
return Command(
update={
"draft": draft,
},
goto="reviewer",
)
Command 的优势是交接逻辑可以靠近 Agent 本身,Agent 执行完后直接声明“我已经完成了什么,应该交给谁”。在 Swarm 风格的系统中,这种写法更自然。
ToolMessage 与 tool_call_id:容易踩的工程细节
当大语言模型通过工具调用触发 Handoff 时,对话历史里通常会出现一条 tool call。这个 tool call 必须有对应的 ToolMessage,而且 tool_call_id 要匹配。
from langchain_core.messages import ToolMessage
def handoff_to_writer(tool_call):
# 执行控制权转移
target_agent = "writer"
return ToolMessage(
content=f"Handoff accepted. Next agent: {target_agent}",
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
如果模型消息里声明了工具调用,但历史记录里没有对应的 ToolMessage,后续模型调用可能会认为对话结构损坏。表现出来的问题可能是:
- 模型重复调用同一个 Handoff 工具;
- 下一个 Agent 无法理解上一轮工具调用结果;
- 框架校验消息历史时报错;
- 日志里出现“工具调用未闭合”的异常。
多 Agent 调试时,不能只看最终输出,还要检查消息历史是否成对出现:AIMessage.tool_calls 和 ToolMessage.tool_call_id 必须对齐。
上下文迁移:不要把所有历史都塞给下一个 Agent
Handoff 时常见错误是把完整聊天记录、Supervisor 的路由过程、所有工具返回结果一起传给下游 Agent。这样做看似保险,实际会引入噪声。
下游 Agent 需要的是“完成自己任务所需的信息”,不是系统内部所有历史。
一个更稳妥的做法是构造标准化交接包:
{
"task_id": "task-20260607-001",
"goal": "根据检索资料生成一份技术方案",
"handoff_from": "researcher",
"handoff_to": "writer",
"handoff_reason": "资料收集完成,需要生成结构化方案",
"constraints": [
"输出必须是 Markdown",
"不要包含未验证的数据"
],
"artifacts": {
"research_notes": "核心资料摘要...",
"references": [
"https://example.com/a",
"https://example.com/b"
]
},
"decisions": [
"采用 Supervisor + 局部 Swarm 架构"
],
"open_questions": [
"是否需要加入人工审批节点"
]
}
上下文迁移要遵守三个原则:
| 原则 | 说明 |
|---|---|
| 只传业务信息 | 路由日志、内部推理草稿、无关聊天历史不应该传给下游 |
| 保留关键决策 | 下游需要知道已经做过哪些判断,避免重复推理 |
| 明确输出契约 | 交接包里要写清楚下游应该产出什么格式 |
如果两个 Agent 的输入格式不同,中间还需要做格式转换。例如检索 Agent 输出的是引用列表,写作 Agent 需要的是按主题聚合后的资料摘要,不能直接把原始搜索结果扔过去。
状态同步:共享状态不是共享草稿纸
多 Agent 系统往往会维护一个共享状态,例如任务目标、消息历史、中间结果、文件路径、执行状态等。共享状态如果没有约束,很容易出现冲突。
常见状态问题有三类:
| 问题 | 例子 | 后果 |
|---|---|---|
| 写冲突 | 两个 Agent 同时修改 final_answer | 后写入覆盖先写入 |
| 脏读 | 一个 Agent 读取到另一个 Agent 未完成的中间结果 | 下游基于半成品继续执行 |
| 状态污染 | 审查 Agent 把自己的评论写进代码生成字段 | 后续 Agent 误判数据含义 |
更安全的做法是做状态分区:每个 Agent 只能写自己负责的字段,共享字段只能通过 Reducer 合并。
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
# 消息历史用专门的消息合并器
messages: Annotated[list, add_messages]
# 每个 Agent 写自己的字段
research_notes: Annotated[list[str], add]
code_changes: Annotated[list[str], add]
review_comments: Annotated[list[str], add]
test_reports: Annotated[list[dict], add]
# 最终输出最好由固定节点生成,避免多个 Agent 同时写
final_answer: str
这里的 add 表示列表追加,而不是覆盖。多个 Agent 同时返回更新时,Reducer 会把结果合并起来。
状态设计可以按所有权划分:
| 状态字段 | 写入者 | 读取者 |
|---|---|---|
research_notes | 检索 Agent | 写作 Agent、审查 Agent |
code_changes | 代码 Agent | 审查 Agent、测试 Agent |
review_comments | 审查 Agent | 代码 Agent、Supervisor |
test_reports | 测试 Agent | Supervisor、写作 Agent |
final_answer | 汇总 Agent | 用户 |
最危险的是让所有 Agent 都能读写同一个大对象,例如 context 或 memory。短期看起来方便,长期会变成不可调试的黑盒。
AutoGen、CrewAI、LangGraph 怎么选
多 Agent 框架不能只看热度,要看它解决的是哪类问题。
| 框架 | 核心思路 | 适合场景 | 主要代价 |
|---|---|---|---|
| LangGraph | 图 + 状态机 | 复杂工作流、可控编排、需要状态管理 | 学习曲线较陡 |
| AutoGen | Agent 之间通过对话协作 | 企业应用、对话式协作、可靠性要求高 | 配置和抽象较重 |
| CrewAI | 角色、任务、流程编排 | 快速原型、MVP、任务型协作 | 复杂控制能力有限 |
LangGraph:适合复杂流程和精细控制
LangGraph 的优势是可以明确描述节点、边、状态和循环。它适合这些场景:
- RAG(检索增强生成)和多工具混合;
- 有条件分支、循环修正、人工审批;
- 需要 checkpoint 和恢复;
- 需要清晰地控制每一步状态更新。
如果任务里存在“生成 → 审查 → 修改 → 再审查”的循环,LangGraph 会比简单链式调用更合适。
AutoGen:适合对话式协作和企业场景
AutoGen 把多 Agent 协作建模成对话。它适合多个角色围绕一个问题来回讨论,例如规划、执行、审查、反馈。
它的优势在于企业级能力相对完整,包括日志、可观测性、错误处理等方面的支持。代价是抽象较重,配置也更复杂。
CrewAI:适合快速验证想法
CrewAI 的上手门槛低,角色、任务、工具的概念直观。它适合快速验证一个多 Agent 想法,例如:
- 自动生成市场分析报告;
- 多角色内容生产;
- 简单研发辅助流程;
- MVP 阶段的任务编排。
当流程开始出现复杂分支、循环、并发和强状态管理时,就需要评估是否迁移到 LangGraph 这类更可控的框架。
多 Agent 稳定性的核心:先约束,再重试
多 Agent 系统失败并不完全随机,常见问题有明确模式。公开研究对大量执行轨迹的分析显示,规格问题和协调失败占比较高,其中规格问题可达 41.77%,协调失败可达 36.94%。
规格问题:Agent 对输入输出理解不一致
规格问题通常表现为:
- 上游输出 Markdown,下游却按 JSON 解析;
- 字段名不一致,例如
summary和abstract混用; - Agent 自由发挥,输出了额外解释文本;
- 下游要求数组,上游返回字符串。
解决方式是强制输入输出结构化。
from pydantic import BaseModel, Field
from typing import Literal
class ReviewResult(BaseModel):
status: Literal["approved", "need_changes", "rejected"]
risk_level: Literal["low", "medium", "high"]
comments: list[str] = Field(default_factory=list)
required_changes: list[str] = Field(default_factory=list)
每个 Agent 都应该有明确契约:
def parse_review_result(raw_output: str) -> ReviewResult:
return ReviewResult.model_validate_json(raw_output)
配套的契约测试也很重要:
def test_review_agent_output_contract():
raw = run_review_agent(
code_diff="修改了登录接口的鉴权逻辑",
requirement="检查安全风险"
)
result = ReviewResult.model_validate_json(raw)
assert result.status in {"approved", "need_changes", "rejected"}
assert result.risk_level in {"low", "medium", "high"}
assert isinstance(result.comments, list)
没有 Schema 的多 Agent 系统,很容易把错误推迟到下游才暴露。等到最后一步才发现格式不对,排查成本会很高。
协调失败:一个错误被下游不断放大
协调失败通常不是单点错误,而是错误传播链。
flowchart LR
A[检索 Agent 返回错误资料] --> B[分析 Agent 基于错误资料推理]
B --> C[写作 Agent 生成错误结论]
C --> D[审查 Agent 未发现问题]
D --> E[最终输出错误结果]
防护手段包括:
| 手段 | 作用 |
|---|---|
| 熔断 | 某个 Agent 连续失败后停止调用,避免浪费成本 |
| 降级 | 非关键 Agent 失败时使用备用流程 |
| 检查点 | 每个关键阶段保存状态,失败后从最近检查点恢复 |
| 幂等设计 | 重试不会造成重复写入或重复执行危险操作 |
| 超时控制 | 防止某个 Agent 长时间占用流程 |
| 最大交接次数 | 防止 Agent 之间无限 Handoff |
例如可以给 Handoff 设置最大次数:
MAX_HANDOFFS = 5
def guard_handoff(state, target_agent: str):
handoff_count = state.get("handoff_count", 0)
if handoff_count >= MAX_HANDOFFS:
return {
**state,
"status": "failed",
"error": "handoff limit exceeded",
}
return {
**state,
"next_agent": target_agent,
"handoff_count": handoff_count + 1,
}
Swarm 模式尤其需要这个保护,否则两个 Agent 可能来回交接:
客服 Agent -> 技术 Agent -> 账单 Agent -> 客服 Agent -> 技术 Agent ...
运行时问题:工具、网络和外部系统失败
多 Agent 系统通常依赖外部工具,例如搜索 API、数据库、代码仓库、CI 系统。LLM(大语言模型)本身的不确定性之外,外部系统也会失败。
常见防护包括:
- 工具调用设置超时;
- 网络错误使用指数退避重试;
- 写操作使用幂等键;
- 高风险操作加入人工确认;
- 每个工具返回结构化错误,而不是直接抛出裸异常。
import time
def call_tool_with_retry(fn, *, retries=3, base_delay=0.5):
last_error = None
for attempt in range(retries):
try:
return fn()
except TimeoutError as e:
last_error = e
time.sleep(base_delay * (2 ** attempt))
return {
"ok": False,
"error_type": "timeout",
"message": str(last_error),
}
返回结构化错误后,Agent 才能判断是重试、降级,还是交给人工处理。
可观测性:没有 Trace,就没有调试能力
多 Agent 系统的错误经常跨越多个节点。如果没有完整 Trace,只看最终回答,很难定位问题发生在哪一步。
每次 Agent 调用至少要记录这些信息:
| 字段 | 说明 |
|---|---|
task_id | 一次用户任务的唯一 ID |
agent_name | 当前执行的 Agent |
input_schema_version | 输入契约版本 |
output_schema_version | 输出契约版本 |
handoff_from | 来源 Agent |
handoff_to | 目标 Agent |
handoff_reason | 交接原因 |
token_usage | Token 消耗 |
latency_ms | 调用耗时 |
tool_calls | 调用过哪些工具 |
status | 成功、失败、降级、熔断 |
Trace 的价值不是“方便看日志”,而是能回答生产问题:
- 哪个 Agent 最容易失败?
- 哪个 Handoff 最常丢上下文?
- 哪个工具调用耗时最高?
- 哪类输入最容易触发格式错误?
- 成本主要消耗在哪些节点?
没有这些数据,稳定性优化只能靠猜。
一个可落地的多 Agent 设计清单
设计多 Agent 系统时,可以按这份清单检查:
| 模块 | 必须明确的问题 |
|---|---|
| Agent 边界 | 每个 Agent 负责什么,不负责什么 |
| 输入输出契约 | 每个 Agent 的输入和输出 Schema |
| 协作模式 | Supervisor、Swarm 还是混合架构 |
| Handoff 协议 | 交接原因、目标 Agent、上下文包、失败处理 |
| 状态管理 | 哪些字段共享,哪些字段只能由特定 Agent 写入 |
| 错误处理 | 重试、熔断、降级、超时、人工介入 |
| 可观测性 | Trace、日志、指标、成本统计 |
| 测试策略 | 单 Agent 契约测试、端到端流程测试、失败注入测试 |
多 Agent 的关键不是把 Agent 数量堆起来,而是让每个 Agent 有清晰职责,并且让它们之间的交接可控、可验证、可恢复。
能用单 Agent + 多工具解决的问题,不必强行拆成多 Agent。确实需要多角色协作时,重点要放在 Handoff、状态同步、结构化输出和失败防护上。框架只是实现手段,系统边界和协作协议才决定稳定性。