多 Agent 系统
为什么需要多 Agent?
Section titled “为什么需要多 Agent?”单个 Agent 面对复杂任务时会遇到瓶颈:
- 上下文窗口不够:一个 Agent 难以同时处理所有信息
- 角色冲突:一个人同时当产品经理、程序员和测试员,效果不好
- 错误放大:单点推理链越长,错误累积越严重
多 Agent 系统的理念:让专业的 Agent 做专业的事,通过协作完成复杂任务。
类比:一个人独自装修整套房子 vs 请一支施工队(水电工、泥瓦匠、木工各司其职)。
分析 Agent / 编码 Agent / 测试 Agent
各自专注自己的领域
1. 中心化(Orchestrator)
Section titled “1. 中心化(Orchestrator)”一个”总管” Agent 协调所有工作 Agent。
flowchart TD
Orch["Orchestrator\n(总管 Agent)"] --> W1["Worker 1"]
Orch --> W2["Worker 2"]
Orch --> W3["Worker 3"]
优点: 控制流清晰,易于调试 缺点: 总管是单点瓶颈,扩展性有限 适用: 大多数生产场景
2. 去中心化(Peer-to-Peer)
Section titled “2. 去中心化(Peer-to-Peer)”Agent 之间平等通信,无中心节点。
flowchart TD
A["Agent A"] <--> B["Agent B"]
A <--> C["Agent C"]
A <--> D["Agent D"]
B <--> C
B <--> D
C <--> D
优点: 无单点故障,可扩展 缺点: 通信复杂,难以追踪,可能死循环 适用: 需要高可用的分布式系统
3. 层级式(Hierarchical)
Section titled “3. 层级式(Hierarchical)”多级管理结构,类似公司组织架构。
flowchart TD
CEO["CEO Agent"] --> EM["工程经理 Agent"]
CEO --> PM["产品经理 Agent"]
EM --> FE["前端 Agent"]
EM --> BE["后端 Agent"]
PM --> Design["设计 Agent"]
优点: 分层授权,任务分解自然 缺点: 层级过多增加延迟 适用: 大型复杂项目
共享状态(Shared State)
Section titled “共享状态(Shared State)”所有 Agent 读写同一个状态存储。
# 共享状态示例shared_state = { "task": "构建一个 TODO 应用", "requirements": [...], # 产品 Agent 写入 "code": {...}, # 开发 Agent 写入 "test_results": [...], # 测试 Agent 写入 "status": "in_progress",}
# 每个 Agent 读取并更新共享状态def developer_agent(state): requirements = state["requirements"] code = generate_code(requirements) state["code"] = code state["status"] = "ready_for_testing" return state消息传递(Message Passing)
Section titled “消息传递(Message Passing)”Agent 之间通过消息直接通信。
# 消息传递示例class Message: sender: str receiver: str content: str msg_type: str # "request", "response", "broadcast"
# Agent A 发送消息给 Agent Bmsg = Message( sender="planner", receiver="coder", content="请实现用户认证模块,使用 JWT", msg_type="request",)agent_b.receive(msg)黑板模式(Blackboard)
Section titled “黑板模式(Blackboard)”所有 Agent 共享一个”黑板”,各自在上面读写信息。
flowchart TD
Arch["架构 Agent"] <--> BB["黑板 Blackboard\n需求 / 架构方案 / 代码 / 测试报告"]
Dev["开发 Agent"] <--> BB
Test["测试 Agent"] <--> BB
Ops["运维 Agent"] <--> BB
A2A 协议简介
Section titled “A2A 协议简介”A2A(Agent-to-Agent Protocol) 是 Google 在 2025 年提出的开放协议,旨在标准化不同 Agent 之间的通信。
flowchart LR
Discover["发现"] --> Negotiate["协商"] --> Delegate["委托"] --> Execute["执行"] --> Return["返回结果"]
A2A 解决的问题:
- 不同框架构建的 Agent 如何互相调用?
- 如何发现哪些 Agent 可以帮忙完成某个任务?
- 如何标准化任务委托和结果返回?
A2A 与 MCP(Model Context Protocol)是互补关系:MCP 解决 Agent 与工具/数据的连接,A2A 解决 Agent 与 Agent 的连接。
主流框架中的多 Agent 实现
Section titled “主流框架中的多 Agent 实现”LangGraph
Section titled “LangGraph”# LangGraph 多 Agent 示例(简化版)from langgraph.graph import StateGraph
def researcher(state): """研究员 Agent:搜索信息""" result = search_tool(state["query"]) return {"research": result}
def writer(state): """写手 Agent:撰写内容""" content = llm.chat(f"基于以下研究写一篇文章:\n{state['research']}") return {"draft": content}
def reviewer(state): """审稿 Agent:审核内容""" feedback = llm.chat(f"请审核这篇文章:\n{state['draft']}") return {"feedback": feedback}
# 构建工作流graph = StateGraph()graph.add_node("researcher", researcher)graph.add_node("writer", writer)graph.add_node("reviewer", reviewer)graph.add_edge("researcher", "writer")graph.add_edge("writer", "reviewer")CrewAI
Section titled “CrewAI”# CrewAI 示例from crewai import Agent, Task, Crew
researcher = Agent( role="研究员", goal="深入研究给定主题", backstory="你是一名资深研究员...",)
writer = Agent( role="技术写手", goal="将研究成果写成易懂的文章", backstory="你是一名技术博客作者...",)
research_task = Task(description="研究 AI Agent 最新进展", agent=researcher)write_task = Task(description="撰写研究报告", agent=writer)
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])result = crew.kickoff()Autogen
Section titled “Autogen”微软的多 Agent 框架,强调对话驱动的协作。Agent 之间通过对话交流,支持人类参与讨论。
何时不该使用多 Agent
Section titled “何时不该使用多 Agent”多 Agent 系统的最大陷阱是过度工程。如果一个任务可以用单 Agent + 好的 Prompt 完成,引入多 Agent 只会增加通信开销、调试难度和成本。经验法则:先用单 Agent 原型验证可行性,只有当你明确遇到以下问题时才考虑拆分为多 Agent——上下文窗口不够用、单次推理无法覆盖所有专业知识、或者需要并行执行独立子任务以提升速度。