深色模式
多 Agent 协作与编排
摘要:单 Agent 适合线性任务,复杂任务需要多个 Agent 按角色协作。本文梳理三种主流拓扑(流水线、层级监督、群聊协商),讲清职责切分、上下文/状态传递、会合(handoff)与终止条件等编排难点,并给出生产落地的可靠性与成本要点。框架选型见
frameworks.md。
核心概念:什么时候需要多 Agent
当任务具备以下特征时,单 Agent 的上下文与单一 persona 会成为瓶颈:
- 子任务异质:需要不同的专业知识(研究 / 写作 / 校验)。
- 上下文过长:单 Agent 上下文被一种子任务占满,影响其他环节。
- 需要相互制衡:生成与审核分离,避免「自己写自己过」。
常见拓扑对比:
| 拓扑 | 结构 | 优点 | 风险 |
|---|---|---|---|
| 流水线 (Pipeline) | A→B→C 顺序 | 简单、可预测 | 前序错误向后传播 |
| 层级 (Hierarchical) | Manager 派发/汇总 | 可控、易回退 | Manager 成单点 |
| 群聊 (Group Chat) | 多 Agent 自由对话 | 思辨质量高 | 顺序不可控、易发散 |
能用单 Agent 就别上多 Agent
多 Agent 带来 N 倍的 token 消耗与 N 倍的失败面。很多「多 Agent」需求其实用单 Agent + 多工具 + 清晰 prompt 就能解决。先验证单 Agent 瓶颈,再引入协作。
架构与原理:三种拓扑
编排的关键不是「让 Agent 们聊天」,而是明确 handoff(交接)规则与终止条件:
- handoff:上一个 Agent 的输出以何种结构交给下一个;群聊中由「经理/仲裁」或规则决定谁发言。
- termination:达到最大轮次、出现
FINAL标记、或收敛判据(如审核通过)即停止,否则无限循环。
生产实践:层级编排示例(伪代码)
python
# multi_agent_hierarchical.py —— 层级编排示意 [未实测]
def manager(task):
plan = llm_plan(task) # 拆子任务
results = {}
for sub in plan:
worker = pick_worker(sub)
out = worker.run(sub, context=results) # 传递已有结果作上下文
results[sub.id] = out
if not reviewer.check(out): # 审核门
out = worker.run(sub, context=results, feedback=reviewer.last)
results[sub.id] = out
return llm_summarize(results)
# 终止保护
def run_with_guard(task, max_rounds=8):
for _ in range(max_rounds):
if manager.should_stop(task):
break
manager.step(task)
return manager.final()1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
群聊拓扑必须设终止条件
Group Chat 下「下一个谁发言」常由 LLM 决定,最容易出现无休止讨论。务必设置 max_rounds、强制 FINAL_ANSWER 标记、并对单轮 token 设上限,否则成本与延迟都会失控。
操作步骤 / 配置
以 CrewAI 风格表达角色与流程(示意):
python
# crew_example.py —— 角色/任务声明 [未实测,API 以 crewai 文档为准]
from crewai import Agent, Task, Crew
researcher = Agent(role="研究员", goal="收集事实", backstory="资深数据分析师", allow_delegation=False)
writer = Agent(role="写手", goal="基于研究成稿", backstory="技术作者", allow_delegation=False)
task1 = Task(description="调研 Agent 安全威胁", agent=researcher, expected_output="结构化要点")
task2 = Task(description="据此写 800 字综述", agent=writer, expected_output=" Markdown 文章")
crew = Crew(agents=[researcher, writer], tasks=[task1, task2], process="sequential")
result = crew.kickoff()1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
bash
pip install crewai langgraph autogen-agentchat # 多框架按需安装 [版本相关]1
验证
- 功能验证:用一条端到端任务跑通,检查每个 Agent 的输入/输出结构符合预期。
- 终止验证:故意构造会发散的任务,确认
max_rounds能拦停。 - 隔离验证:单独单测每个 Worker 工具与 prompt,避免多 Agent 放大单点故障。
回滚 / 清理
- 多 Agent 系统的回滚单元是「编排图/流程定义」,纳入版本管理;出问题切回上一版流程图。
- 临时调试会产生大量中间产物与日志,定期清理避免泄露任务数据。
- 下线时吊销各 Worker 持有的工具凭证。
故障排查
- 上下文丢失:Worker 间传递的是摘要而非原始上下文,关键事实被丢。对策:传递引用/ID 而非全文,或共享一个记忆层(见
memory.md)。 - 职责重叠/冲突:两个 Agent 抢同一子任务。对策:在 prompt 与 handoff 规则中显式切分边界。
- Manager 瓶颈:层级中所有流量过 Manager,成为延迟与成本中心。对策:扁平化、只让 Manager 做调度不做内容。
安全与合规
- 提示注入跨 Agent 传播:一个 Agent 被注入后,其输出会污染下游。对策:Agent 间传递内容视为不可信,关键决策点加校验。
- 越权工具调用:Worker 可能调用超出其角色的敏感工具。对策:按角色授予最小工具集。
- 数据泄露:多 Agent 共享上下文易累积 PII。对策:工具侧脱敏 + 最终输出扫描。
- 成本失控:N 个 Agent 并发放大调用与 token。对策:并发上限 + 总预算 + 终止条件。
成本 / 性能
多 Agent 成本 ≈ 各 Agent 调用次数之和,通常是单 Agent 的 2–5 倍。性能上,流水线可并行子任务缩短 wall-clock,但群聊的顺序不确定性会拉高 P95 延迟。建议:对可并行的子任务用层级 + 并行 Worker;对需要思辨质量的用群聊但严控轮次;所有路径都加 token 预算与并发上限。以单任务平均 5 万 token、单价 $10/MTok 估算,多 Agent 单任务约 $0.5,日 1 万任务即 $5000,必须靠预算熔断 [版本相关]。