为什么需要多 Agent 协作
单个 Agent 在简单任务上表现不错。但当你遇到以下场景，单 Agent 就力不从心了：
场景一：研究 + 写作流水线
你需要让一个 Agent 做深度调研（收集资料、筛选信源），另一个 Agent 基于调研结果写文章。单 Agent 的 context window 会很快被原始材料塞满，写作质量急剧下降。拆成两个 Agent，各司其职，效果更好。
场景二：代码审查工作流
一个 Agent 负责理解变更意图，另一个 Agent 检查安全漏洞，第三个 Agent 检查性能问题。每个审查维度需要不同的 system prompt 和工具集。单 Agent 很难同时戴好三顶帽子。
场景三：客户支持升级
一线 Agent 处理常见问题，遇到复杂技术问题自动升级到技术专家 Agent，涉及账单争议则转给退款 Agent。每个 Agent 有自己的知识库和权限边界，单 Agent 要搞定所有分支会导致 prompt 臃肿、幻觉增加。
核心原则：当单 Agent 的 prompt 复杂到你需要用注释来分区时，就该考虑拆分了。
五种核心协作模式
1. 委派模式（Delegation / Handoff）
一个主 Agent 接收请求，根据内容类型将工作委派给专业 Agent。专业 Agent 完成后，控制权回到主 Agent。
┌─────────────┐
│ 用户请求 │
└──────┬──────┘
│
┌──────▼──────┐
│ Router Agent │
└──┬───┬───┬──┘
│ │ │
┌────────┘ │ └────────┐
│ │ │
┌──────▼───┐ ┌─────▼────┐ ┌─────▼────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ (技术) │ │ (退款) │ │ (常规) │
└──────┬───┘ └─────┬────┘ └─────┬────┘
│ │ │
└────────────┴────────────┘
│
┌──────▼──────┐
│ Router Agent │
│ (汇总回复) │
└──────┬──────┘
│
┌──────▼──────┐
│ 返回用户 │
└─────────────┘
适用场景：客服路由、技术支持分级、内容分类分发
不适用场景：任务之间有复杂依赖关系、需要多轮协商
框架支持：OpenAI Agents SDK（handoffs 一等公民）、CrewAI（delegation）、LangGraph（conditional edges）
OpenAI Agents SDK 实现委派：
from agents import Agent, Runner
tech_agent = Agent(
name="Tech Support",
instructions="你处理技术问题。用中文回复。",
tools=[search_docs, run_diagnostic],
)
billing_agent = Agent(
name="Billing",
instructions="你处理账单和退款问题。用中文回复。",
tools=[check_invoice, process_refund],
)
router = Agent(
name="Router",
instructions="判断用户问题类型，转给对应专家。",
handoffs=[tech_agent, billing_agent],
)
result = await Runner.run(router, "我的订阅被多扣了一次")
LangGraph 实现委派：
from langgraph.graph import StateGraph, END
def route_query(state):
if "技术" in state["query"] or "bug" in state["query"]:
return "tech_agent"
elif "退款" in state["query"] or "账单" in state["query"]:
return "billing_agent"
return "general_agent"
graph = StateGraph(AgentState)
graph.add_node("router", router_node)
graph.add_node("tech_agent", tech_node)
graph.add_node("billing_agent", billing_node)
graph.add_node("general_agent", general_node)
graph.add_conditional_edges("router", route_query, {
"tech_agent": "tech_agent",
"billing_agent": "billing_agent",
"general_agent": "general_agent",
})
for agent in ["tech_agent", "billing_agent", "general_agent"]:
graph.add_edge(agent, END)
2. 流水线模式（Pipeline / Sequential）
Agent 按固定顺序排列，每个 Agent 处理上一个 Agent 的输出。像工厂流水线一样，原材料经过每一道工序变成成品。
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ 输入数据 │────▶│ Agent 1 │────▶│ Agent 2 │────▶│ Agent 3 │────▶ 输出
│ │ │ (调研) │ │ (写作) │ │ (审校) │
└──────────┘ └──────────┘ └──────────┘ └──────────┘
适用场景：内容生产（调研 -> 写作 -> 审校）、数据处理管道、文档翻译链
不适用场景：步骤之间需要条件分支、中间某步可能失败需要重试其他路径
框架支持：LangGraph（线性 graph）、CrewAI（sequential process）、Google ADK（Workflow）
CrewAI 实现流水线：
from crewai import Agent, Task, Crew, Process
researcher = Agent(
role="研究员",
goal="收集并整理指定主题的深度资料",
backstory="你是一名资深研究员，擅长信息检索和交叉验证。",
)
writer = Agent(
role="技术写作者",
goal="基于调研结果写出清晰的技术文章",
backstory="你擅长把复杂技术概念讲清楚。",
)
research_task = Task(
description="调研 {topic} 的最新进展",
agent=researcher,
expected_output="结构化的调研报告",
)
write_task = Task(
description="基于调研报告撰写文章",
agent=writer,
expected_output="2000字技术文章",
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential,
)
3. 监督者模式（Supervisor / Orchestrator）
一个中央监督者 Agent 负责理解整体目标，动态分配任务给工作 Agent，收集结果，决定下一步。这是最灵活但也最复杂的模式。
┌─────────────┐
│ Supervisor │
│ (中央调度) │
└──┬───┬───┬──┘
┌──────────┘ │ └──────────┐
│ │ │
┌──────▼─────┐ ┌─────▼────┐ ┌───────▼────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ (搜索) │ │ (分析) │ │ (代码) │
└──────┬─────┘ └─────┬────┘ └───────┬────┘
│ │ │
└──────────────┴──────────────┘
结果回传 Supervisor
适用场景：复杂分析任务（需要动态决定调用哪些 Agent）、项目规划、需要多轮迭代优化的任务
不适用场景：简单路由（用委派模式更高效）、步骤固定的流程（用流水线更可控）
框架支持：LangGraph（supervisor multi-agent）、Google ADK（AgentTree）、AG2（GroupChat with select_speaker）
LangGraph 监督者模式：
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command
def supervisor(state: MessagesState) -> Command:
response = llm.invoke(
f"根据以下对话决定下一步。可选 Agent: researcher, coder, reviewer。\n"
f"对话历史: {state['messages']}\n"
f"回复格式: Agent名称 或 'FINISH'"
)
goto = response.content.strip().lower()
if goto == "finish":
return Command(goto=END)
return Command(goto=goto)
graph = StateGraph(MessagesState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_node("reviewer", reviewer_node)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", lambda s: s.get("next"), {
"researcher": "researcher",
"coder": "coder",
"reviewer": "reviewer",
END: END,
})
for node in ["researcher", "coder", "reviewer"]:
graph.add_edge(node, "supervisor")
4. 对等协作模式（Peer-to-Peer / Discussion）
多个 Agent 在一个共享空间里讨论，通过多轮对话达成共识。没有中央调度者，Agent 之间平等发言。
┌───────────┐ ┌───────────┐
│ Agent A │◀───────▶│ Agent B │
│ (正方) │ │ (反方) │
└─────┬─────┘ └─────┬─────┘
│ │
│ ┌───────────┐ │
└───▶│ Agent C │◀───┘
│ (裁判) │
└───────────┘
适用场景：头脑风暴、辩论式推理、多视角代码审查、创意生成
不适用场景：需要确定性输出的生产流程、对延迟敏感的场景
框架支持：AG2（GroupChat）、CrewAI（hierarchical process with manager_llm）
AG2 对等讨论：
import ag2
critic = ag2.AssistantAgent(
"critic",
system_message="你是一个严格的技术评审。找出方案中的漏洞。",
llm_config=llm_config,
)
engineer = ag2.AssistantAgent(
"engineer",
system_message="你是一个资深工程师。提出具体的技术方案。",
llm_config=llm_config,
)
judge = ag2.AssistantAgent(
"judge",
system_message="你汇总双方观点，给出最终结论。",
llm_config=llm_config,
)
groupchat = ag2.GroupChat(
agents=[engineer, critic, judge],
messages=[],
max_round=6,
speaker_selection_method="round_robin",
)
manager = ag2.GroupChatManager(groupchat=groupchat, llm_config=llm_config)
manager.initiate_chat(
engineer,
message="讨论：微服务架构是否适合一个 5 人初创团队？",
)
5. 层级模式（Hierarchical）
嵌套的监督者结构。顶层 Supervisor 管理几个子团队 Leader，每个子团队 Leader 又管理自己的 Worker Agent。类似公司的组织架构。
┌─────────────────┐
│ Top Supervisor │
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌──────▼─────┐ ┌─────▼──────┐ ┌─────▼──────┐
│ Sub-Super │ │ Sub-Super │ │ Sub-Super │
│ (前端团队) │ │ (后端团队) │ │ (测试团队) │
└──┬─────┬──┘ └──┬─────┬──┘ └──┬─────┬──┘
│ │ │ │ │ │
┌──▼─┐┌──▼─┐ ┌──▼─┐┌──▼─┐ ┌──▼─┐┌──▼─┐
│ W1 ││ W2 │ │ W3 ││ W4 │ │ W5 ││ W6 │
└────┘└────┘ └────┘└────┘ └────┘└────┘
适用场景：大型项目（前端/后端/测试各一个子团队）、企业级复杂工作流、需要清晰职责边界的系统
不适用场景：中小规模任务（过度设计）、团队少于 5 个 Agent
框架支持：Google ADK（AgentTree 原生支持）、LangGraph（嵌套 graph）、CrewAI（嵌套 Crew）
Google ADK 层级结构：
from google.adk import Agent, Workflow
frontend_dev = Agent(name="frontend_dev", instructions="实现前端组件和样式。")
backend_dev = Agent(name="backend_dev", instructions="实现 API 和数据库逻辑。")
tester = Agent(name="tester", instructions="编写和运行测试用例。")
frontend_lead = Agent(
name="frontend_lead",
instructions="管理前端开发任务，分配给 frontend_dev。",
sub_agents=[frontend_dev],
)
backend_lead = Agent(
name="backend_lead",
instructions="管理后端开发任务，分配给 backend_dev。",
sub_agents=[backend_dev],
)
test_lead = Agent(
name="test_lead",
instructions="管理测试任务，分配给 tester。",
sub_agents=[tester],
)
project_manager = Agent(
name="pm",
instructions="协调前端、后端和测试团队，完成项目。",
sub_agents=[frontend_lead, backend_lead, test_lead],
)
框架对照表
| 特性 | LangGraph | OpenAI Agents SDK | CrewAI | AG2 | Google ADK |
|---|---|---|---|---|---|
| 委派/Handoff | conditional edges | handoffs (原生) | delegation | 无原生支持 | tool-based |
| 流水线 | 线性 graph | 串行调用 | sequential process | 无原生支持 | Workflow 类 |
| 监督者 | supervisor multi-agent | agents-as-tools | hierarchical process | GroupChat + manager | AgentTree |
| 对等讨论 | 可实现但非主流 | 无原生支持 | 无原生支持 | GroupChat (原生) | 无原生支持 |
| 层级嵌套 | 嵌套 graph | 无原生支持 | 嵌套 Crew | 嵌套 GroupChat | AgentTree (原生) |
| 状态管理 | StateGraph (强) | RunContext | Crew memory | 对话历史 | Session |
| 持久化 | Checkpointer | 无内建 | 文件/memory | 无内建 | 无内建 |
| 人机交互 | interrupt + resume | guardrails | human_input | 无原生支持 | 无内建 |
| 生产就绪度 | 高 | 高 | 中 | 中 | 新（评估中） |
| 学习曲线 | 陡 | 平缓 | 平缓 | 中 | 中 |
| A2A 支持 | 通过集成可用 | 无原生 | 无原生 | 无原生 | 实验性 |
表中"原生支持"指框架提供了直接的 API 或模式来处理该场景，不需要你自己组装底层逻辑。
A2A：跨框架互操作
当你需要不同系统构建的 Agent 互相通信时，A2A（Agent-to-Agent）协议提供了标准化的方式。
为什么需要 A2A
实际企业环境中，你可能遇到：
- 客服系统用 OpenAI Agents SDK 构建
- 内部知识库用 LangGraph 构建
- 财务系统用 Google ADK 构建
这些 Agent 需要互相协作，但它们的运行时、状态管理、工具注册完全不同。A2A 解决的是这个互操作性问题。
核心概念
Agent Card：类似名片，描述一个 Agent 的能力、接受的输入格式、返回的输出格式。其他 Agent 通过读取 Agent Card 来决定是否调用它。
Task 生命周期：A2A 把 Agent 间的交互建模为 Task。一个 Agent 创建 Task，另一个 Agent 接收并执行。Task 有明确的状态流转：
Submitted -> Working -> Completed
\-> Failed
\-> Canceled
消息与 Artifact：Task 中传递的信息分为两类——消息（Message，用于协商和沟通）和产出（Artifact，最终交付物）。
简单示意
# Agent A 发布自己的 Agent Card
agent_card = {
"name": "code_reviewer",
"description": "审查 Python 代码的安全性和规范性",
"capabilities": ["security_audit", "style_check", "performance_review"],
"input_schema": {"type": "string", "description": "Python 代码"},
"output_schema": {"type": "object", "properties": {
"issues": {"type": "array"},
"score": {"type": "number"},
}},
}
# Agent B 通过 A2A 协议调用 Agent A
task = await a2a_client.create_task(
target_agent="code_reviewer",
input={"code": "def login(user, pwd):\n ..."},
)
result = await a2a_client.wait_for_completion(task.id)
A2A 目前 v1.0，生态仍在建设中。如果你不需要跨框架互操作，暂时不需要关注。
实战选择指南
根据你的实际需求选择模式：
你的任务是什么？
│
├─ 单一职责，需要路由到不同专家
│ └─▶ 委派模式
│ 推荐：OpenAI Agents SDK（最简洁）
│
├─ 固定步骤，每步处理上一步的输出
│ └─▶ 流水线模式
│ 推荐：CrewAI（开箱即用）或 LangGraph（需要更多控制时）
│
├─ 复杂目标，需要动态决定调用谁、调用几次
│ └─▶ 监督者模式
│ 推荐：LangGraph（最灵活，支持持久化）
│
├─ 需要多视角讨论、头脑风暴
│ └─▶ 对等协作模式
│ 推荐：AG2 GroupChat
│
├─ 大型项目，多个子团队
│ └─▶ 层级模式
│ 推荐：Google ADK（原生 AgentTree）或 LangGraph（嵌套 graph）
│
└─ 需要跨系统/跨框架的 Agent 互操作
└─▶ A2A 协议
推荐：直接基于 A2A spec 实现
按规模选择：
| 规模 | 推荐 | 原因 |
|---|---|---|
| 2-3 个 Agent | OpenAI Agents SDK | Handoff 简洁，上手快 |
| 4-8 个 Agent | CrewAI 或 LangGraph | CrewAI 简单，LangGraph 灵活 |
| 8+ 个 Agent | LangGraph | 状态管理和持久化在大规模下不可替代 |
| 跨系统互操作 | A2A | 唯一标准化选项 |
常见陷阱
1. 过度工程化
这是最常见的错误。不是所有问题都需要多 Agent。
信号：你的"多 Agent 系统"中，每个 Agent 只是换了 system prompt，共享完全相同的工具和上下文。这时候你真正需要的是一个 Agent + 多个 prompt 模板。
修正：单 Agent 能搞定的不要拆。拆分的标准是：不同的 Agent 需要不同的工具集、不同的知识库、或不同的权限边界。
2. 状态管理灾难
多 Agent 之间传递状态是复杂度的主要来源。
信号：你发现自己在写越来越多的胶水代码来同步 Agent 之间的状态。
修正：优先选择有内建状态管理的框架（LangGraph 的 StateGraph）。避免把所有状态塞进一个巨大的 dict——用 TypedDict 明确每个字段的生命周期。
3. Token 成本爆炸
每个 Agent 都是一次 LLM 调用。5 个 Agent 各调用 3 轮 = 15 次 LLM 调用。
修正：
- 在 Agent 之间传递摘要而非完整上下文
- 用更小的模型处理简单步骤（分类、路由用 gpt-4o-mini，复杂推理用 gpt-4o）
- 设定 max_round 限制对话轮次
4. 调试困难
多 Agent 系统出错时，很难定位是哪个 Agent 的问题。
修正：
- 每个 Agent 的输入输出都做日志
- 使用框架的 tracing 功能（OpenAI Agents SDK 内建 tracing，LangGraph 配合 LangSmith）
- 开发阶段用 mock Agent 替换真实 LLM 调用，逐个验证
5. 忽略失败处理
Agent 调用可能超时、返回格式错误、或产生幻觉。
修正：
- 每个 Agent 调用都加 timeout
- 对 Agent 输出做 schema 校验
- 监督者模式中，supervisor 要能检测到 worker 失败并重试或降级
一句话总结
多 Agent 不是银弹——先问自己"单 Agent 能不能搞定"，如果能就不要拆；如果必须拆，从委派模式开始，按需升级到更复杂的模式。