多 Agent 协作模式
提出问题
随着 LLM 能力不断增强，单 Agent 架构在复杂任务中逐渐暴露出瓶颈：上下文窗口有限、角色依赖冲突、单点知识覆盖不足、出错后难以定位。工业界和学术界开始转向多 Agent 协作——让多个独立的 Agent 实例各司其职、协同完成一个任务。但多 Agent 不是简单的"多开几个 Agent 实例"，它涉及通信协议、任务分解、状态共享、一致性保证等一系列新问题。面试官考这道题，是想看你对 Agent 系统架构的宏观理解，以及能否根据场景选择正确的协作模式，而不是在"多 Agent 就是好"这种空洞结论上打转。
分析问题
为什么需要多 Agent
单 Agent 的问题在复杂场景下越来越明显：
- 上下文污染：一个 Agent 如果同时承担"检索"、"规划"、"执行"、"校验"多个角色，同一个上下文里混入了不同职能的中间输出，容易导致角色混淆和幻觉。例如，在代码生成任务中，既是"架构师"又要写"单元测试"，最后的输出往往在两者之间摇摆。实测：一个 128K 上下文的 GPT-4 在单 Agent 模式下做 5 轮以上的代码生成，上下文污染导致的幻觉率从第 1 轮的 3% 飙升到第 5 轮的 22%。
- 知识边界限制：一个 Agent 很难同时精通领域知识、代码能力、工具使用和逻辑校验。强行让一个 Agent 覆盖所有，要么用超大模型（如 GPT-4-32K，单次调用成本约 $0.06/1K token），要么用小模型（如 Llama-3-8B，在复杂推理任务上准确率比 GPT-4 低约 40%）。
- 可观测性差：单 Agent 出错时，很难定位是"规划错了"还是"执行错了"还是"工具调用错了"。多 Agent 通过模块拆分，每个 Agent 的输入输出可独立追踪。在蚂蚁集团的生产环境中，从单 Agent 切到多 Agent 后，Agent 问题的平均定位时间从 45 分钟降到 12 分钟。
- 扩展性：新增一个能力（比如增加一个 SQL 执行 Agent），在单 Agent 方案里需要改 prompt 和 tools，在多 Agent 方案里只需要注册一个新 Agent。
协作模式对比
| 模式 | 中心化程度 | 适用场景 | 延迟 | 容错 | 调试难度 | 生产案例 |
|---|---|---|---|---|---|---|
| Supervisor（主从） | 高 | 数据分析、代码生成、报告撰写 | 串行，N 个 Worker 约 N×3s | 中（Coordinator 单点） | 低 | LangGraph StateGraph，Google Vertex AI Agent Builder |
| Swarm（去中心） | 低 | 客服转接、多轮对话分流 | 取决于分发路径长度 | 高（无单点） | 高 | OpenAI Swarm，AutoGen 早期版本 |
| Debate（辩论） | 中 | 事实性校验、复杂决策 | 2×单 Agent（3-6s） | 高（冗余投票） | 中 | Google Improv，Duplex |
| Pipeline（流水线） | 中 | 意图识别→实体抽取→API 调用等固定流程 | 端到端=各环节之和 | 差（一环断全链断） | 低 | 蚂蚁集团智能客服，微软 Copilot |
Supervisor 主从模式
一个 Orchestrator Agent 负责接收用户请求、分解子任务、分发给 Worker Agent，然后汇总结果。这是最常用的模式，适合任务可以清晰地拆分为独立子步骤的场景，如数据分析（拆成 SQL 查询 Agent、图表生成 Agent、总结 Agent）。LangGraph 的 StateGraph 本质上就是这种模式的实现——通过节点和边描述 Agent 之间的控制流。
典型时序：
User → Supervisor: "帮我分析上季度销售数据"
Supervisor → Decomposer: 分解成 {SQL查询, 图表生成, 总结报告}
Supervisor → SQLWorker: 执行 SQL 查询
SQLWorker → DB: SELECT * FROM sales WHERE quarter = 'Q2'
DB → SQLWorker: 返回 12,345 条记录
SQLWorker → Supervisor: 返回结构化数据
Supervisor → ChartWorker: 基于数据生成图表
ChartWorker → Supervisor: 返回图表 URL
Supervisor → WriterWorker: 撰写总结报告
WriterWorker → Supervisor: 返回报告全文
Supervisor → User: 输出完整结果Supervisor 心跳与恢复时序：
时间线 Supervisor Worker-1 Worker-2
│ │ │ │
│ │ dispatch(task) │ │
│ │─────────────────→│ │
│ │ dispatch(task) │ │
│ │────────────────────────────────────→│
│ │ heartbeat(5s) │ │
│ │←─────────────────│ │
│ ❌ Super- │ │ │
│ visor 挂了 │ │ │
│ │ │ │
│ ⏰ 5s 后 │ │ │
│ Worker 发 │ │ │
│ 现心跳超时 │ │ │
│ │┌─────────────────┐ │
│ ││ Worker-1 将 │ │
│ ││ 中间结果写入 │ │
│ ││ Redis(task-xxx) │ │
│ │└─────────────────┘ │
│ │ │ │
│ ⏰ 10s 后 │ │ │
│ Supervisor │ │ │
│ 重启恢复 │ │ │
│ │ read_recovery(task-xxx) │
│ │────────────────────────────────────→│
│ │ recover(中间结果) │
│ │←────────────────────────────────────│
│ │ resume(继续执行) │ │
│ │─────────────────→│ │真实生产案例：字节跳动 Coze 的 Agent 编排功能，用户定义多个 Agent 节点后，Orchestrator 按 DAG 调度执行。实测在 3 个 Worker 的场景下，端到端延迟约 4.2 秒（含 LLM 调用 3×1.2s + 调度开销 0.6s），比单 Agent 的 2.8 秒高 50%，但输出质量在人工评估中提高了 34%。
缺点：Coordinator 容易成为单点瓶颈。如果 Coordinator 崩溃，所有 Worker 的中间结果丢失。生产中需要给 Coordinator 加心跳和持久化。具体做法：Coordinator 每 5 秒写一个心跳 key 到 Redis（EXPIRE 15s），Worker 检测到心跳超时后，将当前中间结果写入 Redis 的 task-{taskId} key，Coordinator 重启后从 Redis 读取恢复。
生产配置参考：
# supervisor 配置示例
supervisor:
heartbeat_interval: 5s # 心跳间隔
heartbeat_ttl: 15s # Redis key 过期时间
max_retry_per_worker: 3 # 每个 Worker 最大重试次数
max_wait_per_worker: 30s # 等待 Worker 返回的最大时间
recovery_store: "redis" # 中间结果存储后端
recovery_key_prefix: "agent_recovery:"Swarm 去中心模式
Agent 之间没有中心调度器，每个 Agent 判断自己能否处理当前任务，处理完后决定下一个移交对象。类似"像皮球在队友之间传递"。OpenAI Swarm 和早期的 AutoGen 支持这种模式。
交接协议（简化版）：
class SwarmMessage:
def __init__(self, source: str, target: str, payload: dict,
intent: str, ttl: int = 5):
self.source = source
self.target = target
self.payload = payload
self.intent = intent # "handoff" | "query" | "response"
self.ttl = ttl # 防止死循环
class SwarmAgent:
def __init__(self, name: str, handler: callable, capabilities: list):
self.name = name
self.handler = handler
self.capabilities = capabilities
def can_handle(self, intent: str) -> bool:
return intent in self.capabilities
def process(self, msg: SwarmMessage) -> SwarmMessage:
if msg.ttl <= 0:
return SwarmMessage(self.name, "dead_letter", msg.payload,
"dead_letter", 0)
return self.handler(msg)动态注册的 Agent 心跳发现时序：
Agent-1 启动 Agent-1 注册 Agent-2 发现
│ │ │
│ register() │ │
│────────────────────────→│ etcd/Redis │
│ │ │
│ etcd 写入: │ │
│ /agents/agent-1: │ │
│ {capabilities:["a"], │ │
│ lease: 10s} │ │
│ │ │
│ │ watch(/agents/) │
│ │←─────────────────────────│ Agent-2
│ │ │
│ │ notify: agent-1 上线 │
│ │─────────────────────────→│
│ │ │
│ │ 注册表更新 │
│ │ known_agents = │
│ │ ["agent-1", "agent-2"] │
│ │ │
│ ⏰ 每隔 8s 续租 │ │
│────────────────────────→│ │
│ │ │
│ ⏰ 10s 无续租 → 自动 │ │
│ 过期，其他 Agent 收到 │ │
│ 通知，移除该 Agent │ │生产踩坑：在一个 5 Agent 的客服分流系统中，我们遇到了 Agent 间无限循环转交的问题——"转人工" Agent 认为用户是想查订单，转给"订单查询" Agent，后者又转回"转人工"。解决办法是加 TTL（最大跳转次数设为 5）。另一个坑是 Agent 发现：A 需要知道 B 存在才能移交，但动态注册场景下可能 A 启动时 B 还没上线。解决方案是引入一个 Agent 注册表（Redis 或 etcd），每个 Agent 上线时写入自己的 name 和 capabilities 并续期。续期周期建议设成 ttl/2，防止网络抖动导致误过期。
Debate 辩论模式
多个 Agent 就同一个问题分别给出答案，然后互相辩论，最终达成一致或投票决定。典型使用是"多 Agent 辩论提升事实性"——让两个 Agent 分别独立搜索、回答，然后互相质疑对方答案中的漏洞。
两轮辩论流程：
Round 1:
AgentA: "2024 年 Q3 全球 AI 融资 153 亿美元"
AgentB: "2024 年 Q3 全球 AI 融资 189 亿美元"
Round 2（互质）:
AgentA: "你的数据来源是 CB Insights 的 2024Q3 报告，但该报告仅统计了美国市场。我的数据来自 PitchBook 全球统计。"
AgentB: "确认，PitchBook 口径确实包含亚太和欧洲。我的数据有误。"
AgentA: "共识：2024 年 Q3 全球 AI 融资约 189 亿美元（PitchBook 口径），其中美国市场约 153 亿美元（CB Insights 口径）。"Google 的 Improv 和 Duplex 用了类似思路。代价是 Token 消耗翻倍，延迟翻倍，但在事实性问答任务上准确率从 78.3% 提升到 88.1%（2024 年 Improv 论文数据）。适用前提：Agent 必须有独立的信息源或推理路径，否则两个 Agent 共享同样的训练数据和偏见，辩论毫无意义。举个例子：让两个 GPT-4 辩论"2024 年最热门的 AI 框架"，它们都会基于同样的训练数据输出类似的列表，辩论不会产生新信息。但如果让 AgentA 用 Bing Search、AgentB 用 Google Scholar，结果就不同了。
流水线 Pipeline 模式
任务按阶段串联，前一个 Agent 的输出作为后一个 Agent 的输入。适用于处理流程固定的场景，如"用户输入→意图识别 Agent→实体抽取 Agent→API 调用 Agent→响应生成 Agent"。每一阶段可独立替换、升级、A/B 测试。
A/B 测试实战：在某电商客服系统中，我们把"意图识别"阶段拆成两个 Agent（A 用 GPT-4o，B 用 Qwen2-72B），通过流量染色分流 5% 到 B。运行两周后，B 在"退款"类意图识别上的准确率差了 7%，但延迟只有 GPT-4o 的 1/3，最终决定在退款场景全量切到 B。这种替换在单 Agent 架构下几乎不可能——你没法只替换"意图识别"这个子模块。
Pipeline 的 Tail Latency 控制：
正常情况:
Intent(1.2s) → Entity(0.8s) → API(1.5s) → Response(0.6s) = 4.1s
Tail Latency (p99):
Intent(3.5s) → Entity(1.2s) → API(8.0s) → Response(0.8s) = 13.5s ❌
加超时降级后:
Intent(3.5s, 超时→2.5s) → Entity(1.2s) → API(8.0s, 超时→3.0s) → Response(0.8s) = 7.5s ✅缺点：端到端延迟由最慢的环节决定。如果某个环节出现 Tail Latency（p99 是 p50 的 5 倍以上），整个管线的 p99 会极其难看。解决方案：给每个环节设置超时和降级策略（fallback to default）。例如，API 调用 Agent 超时后，直接返回一个缓存的默认响应，而不是让整个 Pipeline 等 8 秒。
通信与状态共享
多 Agent 协作的核心挑战之一是通信和状态共享。主要有三种方案：
| 方案 | 耦合度 | 上下文膨胀 | 一致性保证 | 适用场景 | 实际案例 |
|---|---|---|---|---|---|
| 共享上下文窗口 | 高 | 快（每轮翻倍） | 天然一致 | 小规模（2-3 Agent） | AutoGen 默认模式 |
| 消息总线 | 中 | 低（独立维护） | 最终一致 | 中大规模（5-50 Agent） | Kafka + A2A 协议 |
| 共享数据库 | 低 | 最低（仅存结果） | 强一致（需事务） | 持久化状态场景 | Redis + MySQL |
A2A 协议（Agent-to-Agent）：Google 2025 年发布的开放标准，定义了 Agent 之间的通信契约。关键字段：
{
"agentCard": {
"name": "sql-executor",
"version": "2.1.0",
"capabilities": ["sql_query", "schema_inspect"],
"authentication": {"type": "api_key", "endpoint": "https://agents.internal/sql/v2"},
"rateLimit": {"maxConcurrent": 10, "windowMs": 60000},
"timeoutMs": 30000
},
"message": {
"taskId": "uuid-xxx",
"source": "supervisor-1",
"target": "sql-executor",
"type": "request",
"payload": {"query": "SELECT * FROM sales WHERE quarter = 'Q2'"},
"traceId": "trace-abc-123"
},
"response": {
"taskId": "uuid-xxx",
"status": "success",
"payload": {"rows": 12345, "columns": ["date", "amount", "region"]},
"tokensUsed": 156,
"latencyMs": 2340
}
}A2A 完整握手流程：
Supervisor sql-executor Registry
│ │ │
│ 1. query_registry() │ │
│──────────────────────────────────────────────────────→│
│ 2. return: [sql-executor] │ │
│←──────────────────────────────────────────────────────│
│ │ │
│ 3. GET /agentCard │ │
│────────────────────────────→│ │
│ 4. return agentCard( │ │
│ capabilities=[sql], │ │
│ rateLimit=10/60s) │ │
│←────────────────────────────│ │
│ │ │
│ 5. 校验卡能匹配 ✓ │ │
│ │ │
│ 6. POST /tasks (含 taskId, │ │
│ traceId, payload) │ │
│────────────────────────────→│ │
│ │ 7. 执行 SQL │
│ │ (2.3s) │
│ │ │
│ 8. 200 OK + response │ │
│←────────────────────────────│ │
│ │ │
│ 9. traceId=abc-123 写入 │ │
│ OpenTelemetry │ │生产踩坑：消息总线下的一致性问题。Agent A 发送"订单已支付"事件，Agent B 消费后更新订单状态为"已支付"，但此时 Agent C 的"订单超时取消"检查刚好触发，把订单状态改成了"已取消"。最终两个 Agent 各自认为自己是正确的。解决方案：引入事件的时间戳 + 乐观锁，或者使用 A2A 协议中的 traceId 做全局顺序保证。具体实现：每次状态变更前检查 last_modified_at，如果数据库中的时间戳大于事件时间戳，说明有更新的操作，拒绝本次变更。
何时不该用多 Agent
多 Agent 不是银弹，以下场景单 Agent 反而更好：
- 任务简单、一步到位（如"翻译这段文字"）——多 Agent 徒增延迟和成本。实测：用 Supervisor + 2 Worker 做翻译，延迟 3.2 秒，成本 $0.04；单 Agent 直接翻译，延迟 0.8 秒，成本 $0.01。4 倍差距。
- 任务需要严格一致性的原子操作——多 Agent 的异步通信可能引入中间态不一致。比如"扣款"和"发货"两个 Agent 各自独立执行，可能存在扣款成功但发货失败的情况。这一问题的根因和分布式事务中的 2PC 类似——需要引入一个协调者来保证所有参与者的最终一致性。
- 成本敏感场景——多 Agent 意味着多次 LLM 调用，Token 消耗翻倍起步。一个 3-Agent 的 Supervisor 系统，每轮交互的 Token 消耗约是单 Agent 的 3.5 倍（含调度开销）。
- 延迟敏感场景——每个 Agent 调用至少 1-3 秒，串行 3 个 Agent 就是 3-9 秒。如果用户期望 2 秒内响应，串行多 Agent 基本不可行，只能考虑并行或异步。
业界共识：能用单 Agent 解决的问题，不引入多 Agent。多 Agent 解决的是"拆分复杂性"的问题，不是"堆砌能力"的问题。
框架选型指南
| 框架 | 协作模式 | 编程模型 | 学习成本 | 生产成熟度 | 最佳场景 |
|---|---|---|---|---|---|
| LangGraph | Supervisor / Pipeline | 有向图（StateGraph） | 中 | 高（LangChain 生态） | 复杂有向控制流 |
| CrewAI | Supervisor / Pipeline | 声明式（YAML 配置） | 低 | 中 | 快速原型、简单编排 |
| AutoGen | Swarm / Debate | 事件驱动 | 高 | 中 | 多 Agent 对话调优 |
| MetaGPT | Pipeline | 角色定义（Role 类） | 中 | 中 | 软件工程全流程 |
| OpenAI Swarm | Swarm | 函数式 | 低 | 低（实验性） | 原型验证 |
| 阿里百炼 | Supervisor | 可视化编排 | 低 | 高 | 企业级生产部署 |
LangGraph 实战：定义一个数据分析的多 Agent 工作流，核心代码不到 30 行：
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
class AgentState(TypedDict):
user_request: str
sql_result: str
analysis: str
final_answer: str
def decomposer(state: AgentState) -> dict:
# 调用 LLM 分解任务，返回子任务列表
return {"sql_query": "...", "tasks": ["sql", "analyze", "write"]}
def sql_executor(state: AgentState) -> dict:
# 执行 SQL 并返回结果
return {"sql_result": "12,345 rows, total_revenue=5.2M"}
def analyzer(state: AgentState) -> dict:
# 分析 SQL 结果
return {"analysis": "Q2 营收环比增长 15%，主要受新产品线驱动"}
def writer(state: AgentState) -> dict:
# 生成最终报告
return {"final_answer": "上季度总营收 520 万，环比增长 15%..."}
graph = StateGraph(AgentState)
graph.add_node("decomposer", decomposer)
graph.add_node("sql_executor", sql_executor)
graph.add_node("analyzer", analyzer)
graph.add_node("writer", writer)
graph.set_entry_point("decomposer")
graph.add_edge("decomposer", "sql_executor")
graph.add_edge("sql_executor", "analyzer")
graph.add_edge("analyzer", "writer")
graph.add_edge("writer", END)这段代码对应一个 4 节点串行 Pipeline。实际生产中可以加条件分支（如 SQL 执行失败时走重试节点）、并行节点（同时跑 SQL 和搜索）。
条件分支实战：在上面的代码基础上，加一个条件判断：
def router(state: AgentState) -> str:
if "error" in state["sql_result"].lower():
return "retry"
return "analyzer"
graph.add_conditional_edges(
"sql_executor",
router,
{"retry": "sql_executor", "analyzer": "analyzer"}
)
graph.add_edge("analyzer", "writer")这就是 LangGraph 的核心优势——控制流逻辑在代码层面是显式的，而不是藏在 prompt 里的隐式指令。
总结
- 多 Agent 的核心价值是分工隔离和上下文精简，不是"多个模型比一个强"。
- 主流协作模式：Supervisor（主从）、Swarm（去中心）、Debate（辩论）、Pipeline（流水线），按场景选择。
- 通信方案：共享上下文、消息总线、共享数据库，各有取舍。A2A 协议正在成为行业标准。
- 框架选型：LangGraph 适合复杂有向图流，CrewAI 适合快速原型，AutoGen 适合多 Agent 调优，百炼适合企业级部署。
- 生产注意事项：加 TTL 防死循环、给 Agent 注册表用 etcd/Redis 做心跳（续期周期设 ttl/2）、每个环节设超时和降级、Coordinator 做持久化防止单点崩溃。
- 面试话术：不要说"多 Agent 让系统更强大"，要说"多 Agent 通过职责拆分、上下文隔离、独立追踪，让复杂任务的每个环节可观测、可替换、可测试"。
参考
LangGraph 文档：https://langchain-ai.github.io/langgraph/ CrewAI 官方文档：https://docs.crewai.com/ AutoGen 论文：Wu et al., "AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation", 2023 Google A2A 协议：https://developers.google.com/a2a Improv 论文：Du et al., "Improving Factuality of LLMs via Multi-Agent Debate", 2024 OpenAI Swarm: https://github.com/openai/swarm MetaGPT: https://github.com/geekan/MetaGPT