18-大模型智能体开发工程师：多智能体系统（Multi-Agent System）
·
系列文章导航：AI系列文章导航目录-持续更新中
第18课：多智能体系统（Multi-Agent System）
📝 本文摘要：本文从零讲解多智能体系统——单个Agent能力有限，多个Agent协作实现"1+1>2"。内容包括：①为什么需要多Agent（单Agent的五大瓶颈深度分析、多Agent的核心价值）；②多Agent的核心概念（角色、通信、协调、编排的定义与关系）；③多Agent在Agent技术栈中的定位（与MCP/Skill/记忆系统的关系）；④五大架构模式详解（管理者/流水线/对话/群体/层级，每种的原理、适用场景、优缺点、真实案例）；⑤通信机制（共享状态/消息传递/黑板系统/事件驱动，每种的实现与对比）；⑥协调机制（中央调度/共识协议/市场机制/角色协商）；⑦A2A协议（Google 2025发布，现为Linux Foundation项目，Agent间通信的标准化）；⑧主流框架深度对比（CrewAI/AutoGen/LangGraph/OpenAI Agents SDK/Google ADK，架构差异、代码示例、选型指南）；⑨完整实战：从零构建一个多Agent协作系统；⑩工程最佳实践与常见陷阱。适合AI小白从零理解多智能体系统的本质、架构和实践。
单个Agent就像一个全能但平庸的"独行侠"——什么都能做一点，但什么都做不精。多Agent系统就是组建一支专业团队，每个人负责自己最擅长的部分，通过协作完成单个人无法胜任的复杂任务。
一、为什么需要多Agent
1.1 一句话理解
多Agent系统 = 多个专业Agent协作完成复杂任务
类比: 多Agent之于单Agent ≈ 公司团队之于个人 ≈ 交响乐团之于独奏
1.2 单Agent的五大瓶颈（深度分析）
一句话理解：就像一个人不可能同时是程序员、设计师、测试员、产品经理，单个Agent也很难同时做好所有事情。当任务复杂度超过一定阈值，单Agent就会"力不从心"。
瓶颈1: 上下文窗口的物理限制
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
问题本质:
LLM的上下文窗口是有限的（128K tokens ≈ 10万字）
一个复杂Agent需要装入:
- System Prompt (2-5K tokens)
- 工具定义 (每个工具约500 tokens)
- 对话历史 (随对话增长)
- 记忆检索结果 (2-3K tokens)
- 工具返回结果 (可能很大)
当Agent需要30+个工具时:
30个工具 × 500 tokens/工具 = 15K tokens 仅工具定义就占了
→ 留给对话和推理的空间大幅缩减
→ 工具选择准确率随工具数量增加而下降
实测数据 (来自多个研究):
工具数量 选择准确率
5个 95%+
10个 90%
20个 80%
30个 65%
50个 <50%
多Agent如何解决:
把30个工具分给5个Agent，每个Agent只有6个工具
→ 每个Agent的工具选择准确率回到95%+
→ 总体系统能力 = 5个95%准确率的Agent协作
瓶颈2: 角色混淆（Role Confusion）
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
问题本质:
让一个Agent同时扮演多个角色，它会"人格分裂"。
典型场景:
让同一个Agent既写代码又审查代码
→ 它会"自己放水"——写的时候不严谨，审查的时候不严格
→ 因为它"知道"代码是自己写的，潜意识里会宽容
类比:
让同一个人写作业又批作业 → 肯定不会严格
让同一个人当原告又当法官 → 不可能公正
实际表现:
单Agent写代码+审查: 发现bug率 ~30%
两个Agent分别写和审查: 发现bug率 ~70%
（数据来自AutoGen论文的实验）
多Agent如何解决:
Agent A: 专注写代码（角色=开发者）
Agent B: 专注审查代码（角色=审查者）
→ 角色清晰，互相制约，质量提升
瓶颈3: 串行瓶颈（Sequential Bottleneck）
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
问题本质:
单Agent只能一步步做，不能并行处理多个子任务。
典型场景:
任务: "分析这个系统的前端、后端、数据库三个层面的性能问题"
单Agent: 先分析前端(30s) → 再分析后端(30s) → 再分析数据库(30s) = 90s
多Agent: 三个Agent同时分析三个层面 = 30s
更复杂的场景:
任务: "调研5个竞品，每个竞品分析功能、定价、用户评价"
单Agent: 5个竞品 × 3个维度 = 15步串行 → 可能需要5分钟
多Agent: 5个Agent同时调研5个竞品 → 1分钟
多Agent如何解决:
把可并行的子任务分配给不同Agent同时执行
→ 总耗时 = 最慢的那个子任务的耗时（而非所有子任务之和）
瓶颈4: 专业性不足（Lack of Specialization）
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
问题本质:
一个Agent的System Prompt只能定义一种"人设"。
如果你想让它既是"安全专家"又是"性能专家"又是"架构专家"，
它每个方面都只能做到60分，而不是每个方面90分。
原因:
System Prompt的指令越多越杂，LLM的遵循度越低。
"你是一个安全专家" → LLM会深入思考安全问题
"你是一个安全+性能+架构+测试+运维专家" → LLM每个方面都浅尝辄止
类比:
一个人开公司，从研发到销售到财务都自己做 → 效率低、质量差
vs 雇专业的研发、销售、财务 → 每个岗位都是专业水平
多Agent如何解决:
每个Agent只有一个专精角色:
- 安全Agent: System Prompt专注安全，只配安全相关工具
- 性能Agent: System Prompt专注性能，只配性能分析工具
- 架构Agent: System Prompt专注架构，只配架构评审工具
→ 每个Agent在自己的领域都是"专家级"
瓶颈5: 缺乏自我纠错能力
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
问题本质:
单Agent很难发现自己的错误——"当局者迷"。
它生成了一段有bug的代码，自己很难意识到有问题。
原因:
LLM的输出是基于概率的，它不会"怀疑"自己的输出。
除非你明确要求它"检查一下"，否则它倾向于认为自己是对的。
多Agent如何解决:
引入"对抗性"角色:
- Agent A生成方案
- Agent B专门找Agent A方案的漏洞
- Agent A根据反馈修改
→ 类似"红蓝对抗"或"同行评审"
实际效果:
单Agent生成代码: 一次通过率 ~60%
双Agent(写+审): 一次通过率 ~85%
三Agent(写+审+测试): 一次通过率 ~92%
1.3 多Agent的核心价值
┌─────────────────────────────────────────────────────────────────────────┐
│ 多Agent的四大核心价值 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ 1. 专业分工 (Specialization) │
│ 每个Agent专精一个领域 → 整体质量提升 │
│ 类比: 医院有内科、外科、眼科...每个科室都是专家 │
│ │
│ 2. 并行加速 (Parallelism) │
│ 多个Agent同时工作 → 总耗时大幅缩短 │
│ 类比: 10个人同时搬砖 vs 1个人搬10次 │
│ │
│ 3. 相互纠错 (Cross-checking) │
│ 不同Agent互相审查 → 错误率降低 │
│ 类比: 论文的同行评审、代码的Code Review │
│ │
│ 4. 涌现能力 (Emergence) │
│ 多个Agent协作产生单个Agent不具备的能力 │
│ 类比: 蚂蚁个体很笨，但蚁群能建造复杂巢穴 │
│ │
└─────────────────────────────────────────────────────────────────────────┘
1.4 什么时候该用多Agent（决策框架）
⚠️ 重要原则: 能用单Agent解决的，不要用多Agent！
多Agent的复杂度远大于单Agent。
判断是否需要多Agent的决策树:
任务能被一个Agent在5步内完成吗？
├── 是 → 用单Agent（不需要多Agent）
└── 否 ↓
任务能明确拆分为独立的子任务吗？
├── 否 → 用单Agent + 更好的规划（Plan-and-Execute）
└── 是 ↓
子任务之间需要不同的专业能力吗？
├── 否 → 用单Agent + 循环（同一个Agent做多次）
└── 是 ↓
子任务可以并行执行吗？
├── 是 → 多Agent并行模式
└── 否 ↓
子任务之间需要相互审查/纠错吗？
├── 是 → 多Agent对话模式
└── 否 → 多Agent流水线模式
✅ 适合多Agent的场景:
- 软件开发（需求分析 + 编码 + 测试 + 部署）
- 内容创作（研究 + 写作 + 审校 + 排版）
- 数据分析（数据获取 + 清洗 + 分析 + 可视化）
- 故障排查（日志分析 + 指标分析 + 代码审查 + 根因定位）
- 竞品调研（多个竞品同时调研 + 汇总对比）
❌ 不适合多Agent的场景:
- 简单问答（"Python怎么读文件"）
- 单一工具调用（"帮我查一下天气"）
- 线性简单任务（"帮我写一封邮件"）
- 通信开销 > 计算开销的场景
二、多Agent的核心概念
2.1 四个核心概念
在深入架构模式之前，必须先理解多Agent系统的四个核心概念：
┌─────────────────────────────────────────────────────────────────────────┐
│ 多Agent系统的四个核心概念 │
└─────────────────────────────────────────────────────────────────────────┘
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 角色 │ │ 通信 │ │ 协调 │ │ 编排 │
│ Role │ │Communication │ │ Coordination │ │Orchestration │
├──────────────┤ ├──────────────┤ ├──────────────┤ ├──────────────┤
│ 每个Agent │ │ Agent之间 │ │ 谁先谁后 │ │ 整体流程 │
│ 是什么角色 │ │ 如何交换信息 │ │ 如何达成一致 │ │ 如何组织 │
├──────────────┤ ├──────────────┤ ├──────────────┤ ├──────────────┤
│ 类比: │ │ 类比: │ │ 类比: │ │ 类比: │
│ 公司的岗位 │ │ 开会/邮件 │ │ 投票/审批 │ │ 项目管理 │
└──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘
关系:
角色 → 定义每个Agent"是谁"
通信 → 定义Agent之间"怎么说话"
协调 → 定义Agent之间"怎么达成一致"
编排 → 定义整个系统"怎么运转"
2.2 角色（Role）
定义: 每个Agent的身份、职责和能力边界。
为什么角色很重要:
没有明确角色 → Agent之间职责重叠 → 互相推诿或重复劳动
有明确角色 → 每个Agent知道自己该做什么、不该做什么
角色定义的三要素:
1. 身份描述（你是谁）: "你是一个资深Python开发者"
2. 职责范围（你做什么）: "负责编写和优化Python代码"
3. 能力边界（你不做什么）: "不负责测试和部署"
好的角色设计原则:
✅ 互斥: 每个Agent的职责不重叠
✅ 完备: 所有Agent的职责加起来覆盖整个任务
✅ 专精: 每个Agent只做一件事，做到极致
❌ 反例: "你是一个全栈开发者，负责前端、后端、测试、部署"
→ 这和单Agent没区别，失去了多Agent的意义
2.3 通信（Communication）
定义: Agent之间交换信息的方式和格式。
为什么通信方式很重要:
通信方式决定了:
- Agent之间能传递什么信息
- 信息传递的效率
- 系统的耦合度
四种通信方式（详见第五章）:
1. 共享状态: 所有Agent读写同一个状态对象
2. 消息传递: Agent之间发送结构化消息
3. 黑板系统: Agent往"黑板"上写信息，其他Agent读
4. 事件驱动: Agent发布/订阅事件
2.4 协调（Coordination）
定义: 多个Agent之间如何达成一致、解决冲突、确定执行顺序。
为什么需要协调:
场景: Agent A说"应该用方案1"，Agent B说"应该用方案2"
→ 谁说了算？怎么决定？
协调机制（详见第六章）:
1. 中央调度: 一个"管理者"Agent做所有决策
2. 投票机制: 多个Agent投票，少数服从多数
3. 辩论机制: Agent之间辩论，最终由裁判Agent决定
4. 市场机制: Agent"竞标"任务，最合适的Agent获得执行权
2.5 编排（Orchestration）
定义: 整个多Agent系统的执行流程如何组织和管理。
编排 vs 协调的区别:
协调: 解决"Agent之间的冲突"（微观）
编排: 管理"整个系统的流程"（宏观）
编排的核心问题:
1. 任务如何拆分？（一个大任务拆成哪些子任务）
2. 子任务如何分配？（哪个Agent做哪个子任务）
3. 执行顺序如何确定？（哪些先做、哪些后做、哪些并行）
4. 结果如何汇总？（子任务的结果如何合并为最终结果）
5. 异常如何处理？（某个Agent失败了怎么办）
2.6 多Agent在Agent技术栈中的定位
┌──────────────────────────────────────────────────────────────────┐
│ 第1层: LLM模型本体（GPT/Claude等） │
│ │
│ 每个Agent内部都有一个LLM实例 │
│ Agent A用GPT-4o，Agent B可以用Claude，Agent C可以用本地模型 │
│ → 多Agent系统可以混合使用不同模型！ │
│ │
└──────────────────────────────────┬───────────────────────────────┘
│
┌──────────────────────────────────▼───────────────────────────────┐
│ 第2层: Agent编排层 │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 多Agent编排器（Multi-Agent Orchestrator） │ │
│ │ │ │
│ │ 职责: │ │
│ │ 1. 管理所有Agent的生命周期 │ │
│ │ 2. 路由任务到合适的Agent │ │
│ │ 3. 管理Agent之间的通信 │ │
│ │ 4. 协调冲突、汇总结果 │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Agent A │ │ Agent B │ │ Agent C │ │ Agent D │ │ │
│ │ │ 研究员 │ │ 开发者 │ │ 测试员 │ │ 审查员 │ │ │
│ │ │ │ │ │ │ │ │ │ │ │
│ │ │ Prompt │ │ Prompt │ │ Prompt │ │ Prompt │ │ │
│ │ │ Tools │ │ Tools │ │ Tools │ │ Tools │ │ │
│ │ │ Memory │ │ Memory │ │ Memory │ │ Memory │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ 与其他系统的关系: │
│ - MCP: 每个Agent可以连接不同的MCP Server获取工具 │
│ - Skill: 每个Agent可以加载不同的Skill获取能力 │
│ - 记忆: Agent之间可以共享记忆，也可以各自独立记忆 │
│ │
└──────────────────────────────────────────────────────────────────┘
三、五大架构模式详解
3.1 架构模式全景
┌─────────────────────────────────────────────────────────────────────────┐
│ 五大多Agent架构模式 │
├──────────────┬──────────────┬──────────────┬──────────────┬─────────────┤
│ 管理者模式 │ 流水线模式 │ 对话模式 │ 群体模式 │ 层级模式 │
│ Orchestrator │ Pipeline │ Conversation │ Swarm │ Hierarchical│
├──────────────┼──────────────┼──────────────┼──────────────┼─────────────┤
│ 一个管理者 │ 任务按顺序 │ Agent之间 │ Agent自治 │ 多层管理 │
│ 分配任务 │ 流转 │ 对话协商 │ 自由转交 │ 逐级委派 │
├──────────────┼──────────────┼──────────────┼──────────────┼─────────────┤
│ 类比: │ 类比: │ 类比: │ 类比: │ 类比: │
│ 项目经理 │ 工厂流水线 │ 圆桌会议 │ 自由市场 │ 公司组织架构 │
│ 分配工作 │ 一道道工序 │ 讨论决策 │ 自由交易 │ 总监→经理→员工│
├──────────────┼──────────────┼──────────────┼──────────────┼─────────────┤
│ 适合: │ 适合: │ 适合: │ 适合: │ 适合: │
│ 任务可拆分 │ 流程固定 │ 需要纠错 │ 任务不确定 │ 大规模复杂 │
│ 子任务独立 │ 上下游明确 │ 需要讨论 │ 需要灵活 │ 任务 │
└──────────────┴──────────────┴──────────────┴──────────────┴─────────────┘
3.2 管理者模式（Orchestrator Pattern）
是什么
定义: 一个"管理者"Agent负责理解任务、拆分子任务、分配给子Agent执行、
收集结果并汇总为最终输出。
结构:
┌──────────────────┐
│ Orchestrator │ ← 管理者Agent
│ (管理者) │ 理解任务、拆分、分配、汇总
└────────┬─────────┘
│
┌───────────┼───────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Agent A │ │ Agent B │ │ Agent C │ ← 工作Agent
│ (研究) │ │ (编码) │ │ (测试) │ 各自执行子任务
└──────────┘ └──────────┘ └──────────┘
执行流程:
1. 用户请求 → Orchestrator
2. Orchestrator分析任务，拆分为子任务
3. Orchestrator把子任务分配给合适的Agent
4. 子Agent执行，结果返回Orchestrator
5. Orchestrator汇总所有结果，生成最终回答
6. 最终回答 → 用户
类比:
项目经理收到需求 → 拆分为前端/后端/测试任务 →
分配给对应开发者 → 收集各自的产出 → 整合交付
为什么需要这种模式
核心优势:
1. 集中控制: 管理者有全局视角，能做出最优分配
2. 流程清晰: 用户只和管理者交互，不需要知道内部有多少Agent
3. 容错简单: 某个Agent失败了，管理者可以重新分配或换Agent
4. 易于扩展: 新增Agent只需注册到管理者，不影响其他Agent
核心劣势:
1. 单点瓶颈: 管理者Agent如果出错，整个系统瘫痪
2. 管理者负担重: 需要理解所有子Agent的能力，做出正确分配
3. 通信开销: 所有信息都要经过管理者中转
4. 不适合需要Agent间直接协作的场景
适用场景
✅ 最适合:
- 任务可以明确拆分为独立子任务
- 子任务之间没有强依赖
- 需要并行执行多个子任务
- 用户不需要参与中间过程
实际案例:
1. 技术调研系统: 管理者拆分为"搜索资料"+"分析趋势"+"写报告"
2. 客服系统: 管理者根据问题类型路由到"退款Agent"/"技术支持Agent"/"投诉Agent"
3. 代码生成系统: 管理者拆分为"设计架构"+"写代码"+"写测试"+"写文档"
真实项目案例
案例1: OpenAI的Swarm框架中的Triage Agent模式
- 一个Triage Agent（分诊Agent）作为管理者
- 根据用户问题类型，转交给专业Agent处理
- 用于客服场景: 分诊 → 退款/技术/投诉
案例2: Microsoft AutoGen的GroupChat Manager
- GroupChatManager作为管理者
- 管理多个Agent的发言顺序
- 决定每轮谁来发言
案例3: LangGraph的Supervisor模式
- 一个Supervisor节点作为管理者
- 根据当前状态决定下一步由哪个Agent执行
- 官方示例: multi-agent-supervisor
实现代码
"""
管理者模式实现（基于OpenAI API）
依赖: pip install openai
"""
from openai import OpenAI
import json
client = OpenAI()
# ═══════════════════════════════════════════
# 定义工作Agent
# ═══════════════════════════════════════════
class WorkerAgent:
"""工作Agent: 执行具体子任务"""
def __init__(self, name: str, role: str, system_prompt: str, tools: list = None):
self.name = name
self.role = role
self.system_prompt = system_prompt
self.tools = tools or []
def execute(self, task: str) -> str:
"""执行子任务"""
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": task}
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.3
)
return response.choices[0].message.content
# ═══════════════════════════════════════════
# 管理者Agent
# ═══════════════════════════════════════════
class OrchestratorAgent:
"""管理者Agent: 拆分任务、分配、汇总"""
def __init__(self, workers: list[WorkerAgent]):
self.workers = {w.name: w for w in workers}
def run(self, user_request: str) -> str:
"""完整的管理者流程"""
# Step 1: 拆分任务
print("📋 Step 1: 拆分任务...")
subtasks = self._decompose_task(user_request)
print(f" 拆分为 {len(subtasks)} 个子任务")
for st in subtasks:
print(f" - [{st['agent']}] {st['task']}")
# Step 2: 分配并执行
print("\n🔨 Step 2: 分配并执行...")
results = {}
for subtask in subtasks:
agent_name = subtask["agent"]
task_desc = subtask["task"]
if agent_name in self.workers:
print(f" → {agent_name} 正在执行...")
result = self.workers[agent_name].execute(task_desc)
results[agent_name] = result
print(f" ✅ {agent_name} 完成")
else:
results[agent_name] = f"[错误] 未找到Agent: {agent_name}"
# Step 3: 汇总结果
print("\n📊 Step 3: 汇总结果...")
final_output = self._synthesize(user_request, results)
return final_output
def _decompose_task(self, user_request: str) -> list[dict]:
"""用LLM拆分任务"""
worker_descriptions = "\n".join([
f"- {name}: {w.role}" for name, w in self.workers.items()
])
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""你是一个任务管理者。请将用户的请求拆分为子任务，并分配给合适的Agent。
可用的Agent:
{worker_descriptions}
用户请求: {user_request}
输出JSON格式:
{{"subtasks": [{{"agent": "agent名称", "task": "具体的子任务描述"}}]}}
规则:
1. 每个子任务只分配给一个Agent
2. 子任务描述要具体、可执行
3. 如果某个任务不需要某个Agent，就不要分配"""
}],
response_format={"type": "json_object"},
temperature=0.0
)
result = json.loads(response.choices[0].message.content)
return result.get("subtasks", [])
def _synthesize(self, original_request: str, results: dict) -> str:
"""汇总所有Agent的结果"""
results_text = "\n\n".join([
f"=== {name} 的结果 ===\n{result}"
for name, result in results.items()
])
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""请将以下各Agent的执行结果汇总为一份完整的回答。
原始请求: {original_request}
各Agent结果:
{results_text}
要求: 整合所有信息，生成结构清晰、内容完整的最终回答。"""
}],
temperature=0.3
)
return response.choices[0].message.content
# ═══════════════════════════════════════════
# 使用示例
# ═══════════════════════════════════════════
# 创建工作Agent
researcher = WorkerAgent(
name="researcher",
role="技术研究员，负责搜索和整理技术信息",
system_prompt="你是一个技术研究员。请根据任务要求，提供详细的技术调研信息。"
)
developer = WorkerAgent(
name="developer",
role="高级开发者，负责编写代码和技术方案",
system_prompt="你是一个高级开发者。请根据任务要求，提供代码实现或技术方案。"
)
reviewer = WorkerAgent(
name="reviewer",
role="代码审查专家，负责审查代码质量和安全性",
system_prompt="你是一个代码审查专家。请审查代码的质量、安全性和可维护性。"
)
# 创建管理者
orchestrator = OrchestratorAgent(workers=[researcher, developer, reviewer])
# 执行
result = orchestrator.run("帮我用Go实现一个带限流功能的HTTP中间件")
print("\n" + "="*60)
print("最终结果:")
print(result)
3.3 流水线模式（Pipeline Pattern）
是什么
定义: 任务按固定的顺序在多个Agent之间流转，
每个Agent处理完后把结果传给下一个Agent。
类似工厂的流水线——每道工序由专人负责。
结构:
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ Agent A │───→│ Agent B │───→│ Agent C │───→│ Agent D │
│ (需求分析)│ │ (架构设计)│ │ (代码实现)│ │ (测试验证)│
└──────────┘ └──────────┘ └──────────┘ └──────────┘
输入 ↓ ↓ ↓
用户需求 需求文档 设计文档 代码实现
↓
测试报告
执行流程:
1. 用户输入 → Agent A（需求分析）
2. Agent A的输出 → Agent B的输入（架构设计）
3. Agent B的输出 → Agent C的输入（代码实现）
4. Agent C的输出 → Agent D的输入（测试验证）
5. Agent D的输出 → 最终结果
类比:
汽车工厂: 冲压 → 焊接 → 涂装 → 总装 → 检测
每道工序专人负责，上一道的产出是下一道的输入
为什么需要这种模式
核心优势:
1. 流程标准化: 每个环节的输入输出格式固定，质量可控
2. 专业深度: 每个Agent只需要精通一个环节
3. 可追溯: 出了问题可以定位到具体哪个环节
4. 易于优化: 可以单独优化某个环节而不影响其他
核心劣势:
1. 串行瓶颈: 总耗时 = 所有环节耗时之和
2. 错误传播: 上游Agent出错，错误会传播到所有下游
3. 灵活性差: 流程固定，不适合需要反复迭代的场景
4. 反馈困难: 下游发现上游的问题，很难回头修改
适合场景:
✅ 流程固定、步骤明确的任务
✅ 每个步骤的输入输出格式清晰
✅ 不需要频繁回退和迭代
❌ 不适合需要反复修改的创意性任务
❌ 不适合步骤之间有复杂依赖的任务
真实项目案例
案例1: MetaGPT的软件开发流水线
产品经理Agent → 架构师Agent → 开发者Agent → 测试Agent
- 产品经理: 生成PRD文档
- 架构师: 根据PRD设计系统架构
- 开发者: 根据架构写代码
- 测试: 根据代码写测试用例并执行
案例2: 内容创作流水线
研究Agent → 大纲Agent → 写作Agent → 编辑Agent
- 研究: 收集素材和数据
- 大纲: 根据素材规划文章结构
- 写作: 根据大纲撰写正文
- 编辑: 校对、润色、排版
案例3: 数据处理流水线
采集Agent → 清洗Agent → 分析Agent → 可视化Agent
- 采集: 从多个数据源获取原始数据
- 清洗: 去重、填充缺失值、格式统一
- 分析: 统计分析、趋势发现
- 可视化: 生成图表和报告
实现代码
"""
流水线模式实现
"""
from openai import OpenAI
import json
client = OpenAI()
class PipelineAgent:
"""流水线中的一个Agent节点"""
def __init__(self, name: str, system_prompt: str,
output_format: str = None):
self.name = name
self.system_prompt = system_prompt
self.output_format = output_format # 期望的输出格式说明
def process(self, input_data: str) -> str:
"""处理输入，产生输出"""
prompt = self.system_prompt
if self.output_format:
prompt += f"\n\n输出格式要求: {self.output_format}"
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": prompt},
{"role": "user", "content": input_data}
],
temperature=0.3
)
return response.choices[0].message.content
class Pipeline:
"""流水线编排器"""
def __init__(self, stages: list[PipelineAgent]):
self.stages = stages
def run(self, initial_input: str, verbose: bool = True) -> dict:
"""执行整条流水线"""
current_input = initial_input
stage_outputs = {}
for i, stage in enumerate(self.stages):
if verbose:
print(f"\n{'='*60}")
print(f"📍 Stage {i+1}/{len(self.stages)}: {stage.name}")
print(f" 输入: {current_input[:100]}...")
output = stage.process(current_input)
stage_outputs[stage.name] = output
if verbose:
print(f" 输出: {output[:100]}...")
print(f" ✅ 完成")
# 当前输出成为下一阶段的输入
current_input = output
return {
"final_output": current_input,
"stage_outputs": stage_outputs
}
# ═══════════════════════════════════════════
# 使用示例: 技术博客写作流水线
# ═══════════════════════════════════════════
# Stage 1: 研究
researcher = PipelineAgent(
name="研究员",
system_prompt="""你是一个技术研究员。根据给定的主题，提供详细的技术调研信息。
包括: 核心概念、关键技术点、优缺点、适用场景、相关工具/框架。""",
output_format="结构化的研究笔记，包含5-8个关键发现"
)
# Stage 2: 大纲
outliner = PipelineAgent(
name="大纲规划师",
system_prompt="""你是一个内容规划专家。根据研究笔记，规划一篇技术博客的大纲。
要求: 逻辑清晰、层次分明、覆盖所有关键点。""",
output_format="Markdown格式的文章大纲，包含标题、各章节标题和要点"
)
# Stage 3: 写作
writer = PipelineAgent(
name="技术作者",
system_prompt="""你是一个技术写作专家。根据大纲撰写完整的技术博客文章。
要求: 通俗易懂、有代码示例、有实际案例。""",
output_format="完整的Markdown格式技术博客文章"
)
# Stage 4: 审校
editor = PipelineAgent(
name="编辑审校",
system_prompt="""你是一个技术编辑。审校文章的准确性、可读性和完整性。
如果发现问题，直接修改并标注修改原因。如果没有问题，输出优化后的最终版本。""",
output_format="审校后的最终版本文章"
)
# 组装流水线
pipeline = Pipeline(stages=[researcher, outliner, writer, editor])
# 执行
result = pipeline.run("写一篇关于'LLM Agent记忆系统'的技术博客")
print("\n" + "="*60)
print("最终文章:")
print(result["final_output"][:500])
3.4 对话模式（Conversation Pattern）
是什么
定义: 多个Agent之间通过对话来协作，互相讨论、质疑、改进，
直到达成共识或满足终止条件。
结构:
┌──────────┐ ┌──────────┐
│ Agent A │ ←──→ │ Agent B │
│ (写代码) │ │ (审查) │
└──────────┘ └──────────┘
│ │
└────── 对话 ──────┘
轮次1: A写代码 → B审查，发现3个问题
轮次2: A修改代码 → B再审查，还有1个问题
轮次3: A再修改 → B通过 ✅
更复杂的多方对话:
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Agent A │ ←──→ │ Agent B │ ←──→ │ Agent C │
│ (提出方案)│ │ (质疑) │ │ (裁判) │
└──────────┘ └──────────┘ └──────────┘
A提出方案 → B质疑漏洞 → A修改 → B再质疑 → C裁判决定
类比:
- 代码审查: 开发者提交PR → 审查者提意见 → 开发者修改 → 审查者通过
- 学术辩论: 正方陈述 → 反方质疑 → 正方回应 → 评委判定
- 头脑风暴: 多人讨论 → 互相补充 → 达成共识
为什么需要这种模式
核心优势:
1. 质量提升: 互相审查纠错，输出质量远高于单Agent
2. 多角度思考: 不同角色从不同角度分析同一问题
3. 自我改进: 通过多轮迭代，方案越来越完善
4. 模拟人类协作: 最接近人类团队讨论的模式
核心劣势:
1. 耗时长: 多轮对话意味着多次LLM调用
2. 成本高: 每轮对话都消耗Token
3. 可能不收敛: Agent之间可能"吵"不出结果
4. 需要收敛机制: 必须设置最大轮数或终止条件
关键设计:
必须有收敛机制，否则Agent可能无限对话下去！
收敛方式:
1. 最大轮数: 最多对话N轮，强制结束
2. 共识检测: 当所有Agent都同意时结束
3. 裁判Agent: 由第三方Agent判断是否可以结束
4. 质量阈值: 当输出达到某个质量标准时结束
适用场景
✅ 最适合:
- 代码写作+审查（最经典的场景）
- 方案设计+评审
- 论文写作+同行评审
- 任何需要"质量把关"的场景
实际案例:
1. AutoGen的双Agent对话
- Coder Agent写代码
- Critic Agent审查代码
- 多轮迭代直到代码通过审查
2. ChatDev的角色扮演对话
- CEO和CTO讨论技术方案
- 程序员和测试员讨论代码质量
- 通过角色扮演的对话推进项目
3. 辩论式推理（Debate）
- 多个Agent对同一问题给出不同答案
- 互相质疑对方的推理过程
- 最终由裁判Agent选出最佳答案
- 研究表明: 辩论后的答案准确率高于单Agent
实现代码
"""
对话模式实现: 代码写作 + 审查
"""
from openai import OpenAI
client = OpenAI()
class ConversationalAgent:
"""对话Agent"""
def __init__(self, name: str, system_prompt: str):
self.name = name
self.system_prompt = system_prompt
def respond(self, conversation_history: list[dict]) -> str:
"""根据对话历史生成回复"""
messages = [
{"role": "system", "content": self.system_prompt}
] + conversation_history
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.3
)
return response.choices[0].message.content
class ConversationOrchestrator:
"""对话编排器: 管理多Agent之间的对话"""
def __init__(self, agents: list[ConversationalAgent],
max_rounds: int = 5,
termination_keyword: str = "APPROVED"):
self.agents = agents
self.max_rounds = max_rounds
self.termination_keyword = termination_keyword
def run(self, initial_task: str) -> dict:
"""运行多轮对话"""
conversation = [{"role": "user", "content": initial_task}]
for round_num in range(self.max_rounds):
print(f"\n{'─'*60}")
print(f"🔄 Round {round_num + 1}/{self.max_rounds}")
for agent in self.agents:
# 每个Agent看到完整的对话历史
response = agent.respond(conversation)
print(f"\n 💬 [{agent.name}]:")
print(f" {response[:200]}...")
# 添加到对话历史
conversation.append({
"role": "assistant",
"content": f"[{agent.name}]: {response}"
})
# 检查终止条件
if self.termination_keyword in response:
print(f"\n ✅ {agent.name} 发出终止信号: {self.termination_keyword}")
return {
"rounds": round_num + 1,
"conversation": conversation,
"final_output": response,
"terminated_by": agent.name
}
print(f"\n ⚠️ 达到最大轮数 {self.max_rounds}，强制结束")
return {
"rounds": self.max_rounds,
"conversation": conversation,
"final_output": conversation[-1]["content"],
"terminated_by": "max_rounds"
}
# ═══════════════════════════════════════════
# 使用示例: 代码写作 + 审查
# ═══════════════════════════════════════════
coder = ConversationalAgent(
name="开发者",
system_prompt="""你是一个高级Python开发者。
你的职责是根据需求编写高质量的代码。
当审查者提出修改意见时，你需要认真修改代码并解释修改原因。
输出格式: 先说明思路，然后给出完整代码。"""
)
reviewer = ConversationalAgent(
name="审查者",
system_prompt="""你是一个严格的代码审查专家。
你的职责是审查代码的质量、安全性、可维护性和性能。
审查要点: 错误处理、边界条件、命名规范、代码复杂度、安全漏洞。
如果代码有问题，指出具体问题和修改建议。
如果代码质量达标（没有严重问题），回复中包含"APPROVED"表示通过。
不要轻易通过——至少要审查一轮。"""
)
# 运行对话
orchestrator = ConversationOrchestrator(
agents=[coder, reviewer],
max_rounds=4,
termination_keyword="APPROVED"
)
result = orchestrator.run("请实现一个线程安全的LRU缓存，支持过期时间")
print(f"\n{'='*60}")
print(f"对话轮数: {result['rounds']}")
print(f"终止原因: {result['terminated_by']}")
3.5 群体模式（Swarm Pattern）
是什么
定义: 没有固定的管理者或流程，Agent之间可以自由地将任务转交给
"更合适"的Agent处理。每个Agent自己判断是否能处理当前任务，
如果不能就转交。
结构:
┌──────────┐
│ Agent A │ ←──┐
│ (客服) │ │
└────┬─────┘ │
│ │
▼ │
┌──────────┐ │
│ Agent B │────┤ ← 任意Agent可以转交任务给任意其他Agent
│ (技术) │ │
└────┬─────┘ │
│ │
▼ │
┌──────────┐ │
│ Agent C │ ───┘
│ (退款) │
└──────────┘
执行流程:
1. 用户消息到达Agent A（入口Agent）
2. Agent A判断: "这个问题我能处理吗？"
- 能 → 直接处理并回复
- 不能 → 转交给更合适的Agent
3. Agent B收到任务，同样判断
4. 直到某个Agent能处理并给出最终回答
关键特征:
- 没有中央管理者
- Agent自治（自己决定是否转交）
- 转交是"带上下文"的（不是从零开始）
- 灵活但可能"踢皮球"
类比:
打客服电话:
"您好，这个问题我处理不了，帮您转接技术部门"
→ "您好，这个涉及退款，帮您转接财务部门"
→ "您好，我来处理您的退款..."
为什么需要这种模式
核心优势:
1. 极度灵活: 不需要预定义固定流程
2. 去中心化: 没有单点故障
3. 自适应: Agent自己判断最佳处理者
4. 易于扩展: 新增Agent不影响现有流程
核心劣势:
1. 可能"踢皮球": Agent之间互相转交，没人处理
2. 不可预测: 执行路径不确定，难以调试
3. 需要收敛机制: 必须限制最大转交次数
4. Agent需要"自知之明": 要能准确判断自己能否处理
关键设计:
1. 每个Agent必须有清晰的"能力边界"描述
2. 必须设置最大转交次数（防止无限循环）
3. 转交时必须携带完整上下文
4. 需要一个"兜底Agent"处理没人接的任务
真实项目案例
案例1: OpenAI Swarm → OpenAI Agents SDK
- OpenAI于2024年开源Swarm实验框架，后演变为正式的Agents SDK
- 核心概念: Agent + Handoff（转交）
- 每个Agent定义instructions和functions
- Agent可以通过返回另一个Agent来实现转交
- 现已发展为功能完整的Agents SDK: https://github.com/openai/openai-agents-python
案例2: 智能客服系统
- 入口Agent: 理解用户意图
- 技术支持Agent: 处理技术问题
- 退款Agent: 处理退款请求
- 投诉Agent: 处理投诉升级
- 用户的问题自动路由到最合适的Agent
案例3: Anthropic的Tool Use + Handoff模式
- Claude的multi-tool-use中
- Agent可以"委托"其他Agent处理子任务
实现代码
"""
群体模式实现（参考OpenAI Swarm的设计思想）
"""
from openai import OpenAI
import json
client = OpenAI()
class SwarmAgent:
"""Swarm中的Agent: 可以自主决定是否转交"""
def __init__(self, name: str, instructions: str,
functions: list = None, handoff_targets: list = None):
self.name = name
self.instructions = instructions
self.functions = functions or []
self.handoff_targets = handoff_targets or [] # 可以转交给哪些Agent
def get_system_prompt(self, available_agents: dict) -> str:
"""生成包含转交能力的System Prompt"""
prompt = self.instructions
if self.handoff_targets:
prompt += "\n\n你可以将任务转交给以下Agent（当你认为他们更适合处理时）:\n"
for target_name in self.handoff_targets:
if target_name in available_agents:
target = available_agents[target_name]
prompt += f"- {target_name}: {target.instructions[:100]}\n"
prompt += "\n如果需要转交，请在回复中包含: [HANDOFF: agent名称]"
prompt += "\n如果你能处理，直接回答用户问题。"
return prompt
class Swarm:
"""Swarm编排器: 管理Agent之间的转交"""
def __init__(self, agents: list[SwarmAgent], entry_agent: str,
max_handoffs: int = 5):
self.agents = {a.name: a for a in agents}
self.entry_agent = entry_agent
self.max_handoffs = max_handoffs
def run(self, user_message: str) -> dict:
"""运行Swarm"""
current_agent_name = self.entry_agent
messages = [{"role": "user", "content": user_message}]
handoff_chain = [current_agent_name]
for step in range(self.max_handoffs + 1):
current_agent = self.agents[current_agent_name]
system_prompt = current_agent.get_system_prompt(self.agents)
print(f"\n 🤖 [{current_agent_name}] 正在处理...")
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt}
] + messages,
temperature=0.3
)
reply = response.choices[0].message.content
# 检查是否需要转交
if "[HANDOFF:" in reply:
# 解析转交目标
import re
match = re.search(r'\[HANDOFF:\s*(\w+)\]', reply)
if match:
target_name = match.group(1)
if target_name in self.agents:
print(f" → 转交给 [{target_name}]")
current_agent_name = target_name
handoff_chain.append(target_name)
# 把转交原因加入上下文
messages.append({
"role": "assistant",
"content": f"[{handoff_chain[-2]}转交说明]: {reply}"
})
continue
# 没有转交，当前Agent直接回答
print(f" ✅ [{current_agent_name}] 给出最终回答")
return {
"final_agent": current_agent_name,
"response": reply,
"handoff_chain": handoff_chain,
"total_handoffs": len(handoff_chain) - 1
}
return {
"final_agent": current_agent_name,
"response": "达到最大转交次数，无法处理",
"handoff_chain": handoff_chain,
"total_handoffs": self.max_handoffs
}
# ═══════════════════════════════════════════
# 使用示例: 智能客服Swarm
# ═══════════════════════════════════════════
triage = SwarmAgent(
name="triage",
instructions="""你是客服入口Agent。你的职责是理解用户的问题，
如果是技术问题转交给tech，如果是退款问题转交给refund，
如果是简单的问候或通用问题，你直接回答。""",
handoff_targets=["tech", "refund"]
)
tech = SwarmAgent(
name="tech",
instructions="""你是技术支持Agent。你负责解答技术问题，
包括API使用、SDK集成、错误排查等。
如果用户的问题涉及退款，转交给refund。""",
handoff_targets=["refund"]
)
refund = SwarmAgent(
name="refund",
instructions="""你是退款处理Agent。你负责处理退款请求。
请确认订单信息，说明退款政策，并处理退款。
你不需要转交给其他Agent。"""
)
# 创建Swarm
swarm = Swarm(
agents=[triage, tech, refund],
entry_agent="triage",
max_handoffs=3
)
# 测试不同类型的问题
print("=== 测试1: 技术问题 ===")
result = swarm.run("你们的API怎么做认证？")
print(f"处理链: {' → '.join(result['handoff_chain'])}")
print(f"回答: {result['response'][:100]}")
print("\n=== 测试2: 退款问题 ===")
result = swarm.run("我想退款，订单号是ORD-12345")
print(f"处理链: {' → '.join(result['handoff_chain'])}")
print(f"回答: {result['response'][:100]}")
3.6 层级模式（Hierarchical Pattern）
是什么
定义: 多层管理结构，高层Agent管理中层Agent，中层Agent管理底层Agent。
类似公司的组织架构: 总监 → 经理 → 员工。
结构:
┌──────────────┐
│ 总管理者 │ ← Level 0: 战略决策
│ (CTO Agent) │
└──────┬───────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 前端主管 │ │ 后端主管 │ │ 测试主管 │ ← Level 1: 战术管理
│ Agent │ │ Agent │ │ Agent │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
┌───┼───┐ ┌───┼───┐ ┌───┼───┐
▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼
UI CSS JS API DB 缓存 单元 集成 E2E ← Level 2: 具体执行
Agent Agent Agent Agent Agent Agent Agent Agent Agent
执行流程:
1. 用户请求 → 总管理者
2. 总管理者拆分为"前端任务"+"后端任务"+"测试任务"
3. 前端主管收到前端任务，进一步拆分为"UI"+"CSS"+"JS"
4. 底层Agent执行具体工作
5. 结果逐级汇总: 底层→中层→顶层→用户
类比:
军队指挥: 将军→师长→团长→连长→士兵
公司管理: CEO→VP→Director→Manager→Engineer
为什么需要这种模式
核心优势:
1. 可扩展性强: 可以管理大量Agent（几十甚至上百个）
2. 职责清晰: 每一层有明确的管理范围
3. 复杂度可控: 每个管理者只需要管理3-5个下属
4. 适合大规模任务: 可以处理非常复杂的项目
核心劣势:
1. 层级越多，通信延迟越大
2. 中间层可能成为信息瓶颈
3. 实现复杂度高
4. 不适合简单任务（杀鸡用牛刀）
适合场景:
✅ 大型软件项目开发（前端+后端+移动端+测试+运维）
✅ 企业级复杂任务（涉及多个部门协作）
✅ Agent数量超过10个的场景
❌ 不适合简单任务
❌ 不适合Agent数量少于5个的场景
真实项目案例
案例1: MetaGPT
- 模拟软件公司的组织架构
- CEO → 产品经理 → 架构师 → 开发者 → 测试
- 每个角色有明确的输入输出格式（SOP）
- GitHub: https://github.com/FoundationAgents/MetaGPT (⭐ 68.5K)
案例2: ChatDev
- 模拟软件开发公司
- CEO + CTO + 程序员 + 测试 + 设计师
- 通过角色扮演对话推进项目
- GitHub: https://github.com/OpenBMB/ChatDev
案例3: CrewAI的Hierarchical Process
- 支持hierarchical模式
- 一个Manager Agent管理多个Worker Agent
- Manager负责任务分配和结果汇总
3.7 五种模式对比总结
┌──────────────┬──────────┬──────────┬──────────┬──────────┬──────────┐
│ │ 管理者 │ 流水线 │ 对话 │ 群体 │ 层级 │
├──────────────┼──────────┼──────────┼──────────┼──────────┼──────────┤
│ 控制方式 │ 集中 │ 固定流程 │ 协商 │ 自治 │ 分层 │
│ 灵活性 │ 中 │ 低 │ 高 │ 最高 │ 中 │
│ 可预测性 │ 高 │ 最高 │ 中 │ 低 │ 高 │
│ 并行能力 │ 高 │ 低(串行) │ 低 │ 中 │ 高 │
│ 实现复杂度 │ 中 │ 低 │ 中 │ 中 │ 高 │
│ 适合Agent数 │ 3-8 │ 3-6 │ 2-4 │ 3-10 │ 10+ │
│ 典型场景 │ 任务分配 │ 标准流程 │ 审查纠错 │ 客服路由 │ 大型项目 │
│ 代表框架 │ LangGraph │ MetaGPT │ AutoGen │ Swarm │ CrewAI │
│ │ Supervisor│ │ │ │ Hierarchy│
└──────────────┴──────────┴──────────┴──────────┴──────────┴──────────┘
选型建议:
- 不确定用哪个？→ 先试管理者模式（最通用）
- 流程固定？→ 流水线模式
- 需要质量把关？→ 对话模式
- 需要灵活路由？→ 群体模式
- 任务超级复杂？→ 层级模式
- 可以组合使用！→ 管理者 + 对话（管理者分配任务，子Agent之间对话协作）
四、通信机制详解
4.1 为什么通信机制很重要
多Agent系统的核心挑战不是"每个Agent怎么工作"，
而是"Agent之间怎么交换信息"。
通信机制决定了:
1. Agent能获取什么信息（信息范围）
2. 信息传递的效率（延迟和带宽）
3. 系统的耦合度（修改一个Agent是否影响其他）
4. 可扩展性（能否轻松添加新Agent）
选错通信机制的后果:
- 信息不足: Agent做出错误决策（因为不知道其他Agent的状态）
- 信息过载: Agent收到太多无关信息，干扰判断
- 耦合过紧: 修改一个Agent导致其他Agent全部出错
- 性能瓶颈: 通信开销大于计算开销
4.2 方式1: 共享状态（Shared State）
定义: 所有Agent读写同一个状态对象。
任何Agent的修改，其他Agent都能立即看到。
原理:
┌─────────────────────────────────────────┐
│ 共享状态 (State) │
│ │
│ messages: [...] ← 所有Agent共享 │
│ plan: [...] ← 所有Agent可读写 │
│ results: {...} ← 所有Agent可读写 │
│ current_agent: "A" ← 所有Agent可读写 │
│ │
└─────────────────────────────────────────┘
↑↓ ↑↓ ↑↓
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent C │
└─────────┘ └─────────┘ └─────────┘
优点:
✅ 实现简单（一个dict/TypedDict就行）
✅ 信息同步即时（不需要显式传递）
✅ 所有Agent都有全局视角
缺点:
❌ 状态可能很大（所有信息都在里面）
❌ 并发冲突（两个Agent同时修改同一字段）
❌ 耦合度高（Agent依赖状态的具体结构）
❌ 不适合分布式部署
代表框架: LangGraph（State就是共享状态）
# LangGraph的共享状态示例
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class SharedState(TypedDict):
"""所有Agent共享的状态"""
messages: Annotated[list, add_messages] # 对话历史
plan: list[str] # 执行计划
current_step: int # 当前步骤
research_result: str # 研究结果
code_result: str # 代码结果
review_result: str # 审查结果
final_output: str # 最终输出
def researcher_node(state: SharedState) -> dict:
"""研究Agent: 读取state中的messages，写入research_result"""
# 读取共享状态
messages = state["messages"]
# ... 执行研究 ...
# 写入共享状态
return {"research_result": "研究结果..."}
def coder_node(state: SharedState) -> dict:
"""开发Agent: 读取research_result，写入code_result"""
research = state["research_result"] # 读取其他Agent的产出
# ... 根据研究结果写代码 ...
return {"code_result": "代码..."}
def reviewer_node(state: SharedState) -> dict:
"""审查Agent: 读取code_result，写入review_result"""
code = state["code_result"] # 读取其他Agent的产出
# ... 审查代码 ...
return {"review_result": "审查通过"}
4.3 方式2: 消息传递（Message Passing）
定义: Agent之间通过发送结构化消息来通信。
每个Agent有自己的"收件箱"，只处理发给自己的消息。
原理:
┌─────────┐ 消息: {to: B, content: "请审查这段代码"} ┌─────────┐
│ Agent A │ ──────────────────────────────────────────→ │ Agent B │
└─────────┘ └─────────┘
│
┌─────────┐ 消息: {to: A, content: "发现3个问题..."} │
│ Agent A │ ←────────────────────────────────────────────────┘
└─────────┘
消息格式:
{
"from": "agent_a",
"to": "agent_b",
"type": "task_request", // task_request / task_result / feedback / handoff
"content": "具体内容",
"metadata": {"priority": "high", "deadline": "..."}
}
优点:
✅ 松耦合（Agent之间不共享内部状态）
✅ 适合分布式部署（消息可以通过网络传递）
✅ 可以异步通信（不需要同时在线）
✅ 信息精确（只传递需要的信息）
缺点:
❌ 实现复杂（需要消息路由、序列化等）
❌ 可能丢消息（需要确认机制）
❌ Agent没有全局视角（只知道发给自己的信息）
代表框架: AutoGen（Agent之间通过消息对话）
4.4 方式3: 黑板系统（Blackboard System）
定义: 一个共享的"黑板"，Agent往上面写信息，其他Agent可以读。
类似团队的共享文档或看板。
原理:
┌─────────────────────────────────────────────────────┐
│ 黑板 (Blackboard) │
│ │
│ [研究结果] Agent A写入: "MCP协议的核心是..." │
│ [代码草稿] Agent B写入: "def connect():..." │
│ [审查意见] Agent C写入: "第3行有安全漏洞" │
│ [任务状态] 系统写入: "Step 2/5 进行中" │
│ │
└─────────────────────────────────────────────────────┘
↑读/写 ↑读/写 ↑读/写
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ (研究) │ │ (开发) │ │ (审查) │
└─────────┘ └─────────┘ └─────────┘
与共享状态的区别:
共享状态: 所有Agent看到完全相同的状态（全量）
黑板系统: Agent可以选择性地读取黑板上的特定区域（按需）
优点:
✅ 信息持久化（写在黑板上不会丢）
✅ 选择性读取（不需要看所有信息）
✅ 异步协作（Agent不需要同时工作）
✅ 可追溯（谁写了什么一目了然）
缺点:
❌ 黑板可能变得很大
❌ 需要管理黑板的读写权限
❌ Agent需要知道去哪里找信息
适合场景:
- 多个Agent需要共享中间结果
- Agent工作节奏不同（有的快有的慢）
- 需要记录完整的协作过程
4.5 方式4: 事件驱动（Event-Driven）
定义: Agent通过发布/订阅事件来通信。
Agent发布事件，订阅了该事件的Agent会收到通知。
原理:
┌─────────────────────────────────────────────────────┐
│ 事件总线 (Event Bus) │
│ │
│ 事件类型: │
│ "research_complete" → 订阅者: [Agent B, Agent C] │
│ "code_ready" → 订阅者: [Agent C] │
│ "review_failed" → 订阅者: [Agent B] │
│ │
└─────────────────────────────────────────────────────┘
发布↑ 订阅↓ 发布↑ 订阅↓
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ 发布: │ │ 订阅: │ │ 订阅: │
│ research │ │ research │ │ research │
│ _complete│ │ _complete│ │ _complete│
│ │ │ 发布: │ │ code_ │
│ │ │ code_ │ │ ready │
│ │ │ ready │ │ │
└─────────┘ └─────────┘ └─────────┘
优点:
✅ 最松耦合（Agent之间完全不知道彼此的存在）
✅ 最适合分布式和大规模系统
✅ 易于扩展（新Agent只需订阅感兴趣的事件）
✅ 天然支持并行（多个Agent可以同时响应同一事件）
缺点:
❌ 调试困难（事件流难以追踪）
❌ 顺序不确定（不知道谁先收到事件）
❌ 实现复杂（需要事件总线基础设施）
❌ 不适合需要严格顺序的场景
适合场景:
- 大规模多Agent系统（10+个Agent）
- 微服务架构的Agent系统
- Agent需要动态加入/退出的场景
4.6 四种通信方式对比
┌──────────────┬──────────────┬──────────────┬──────────────┬──────────────┐
│ │ 共享状态 │ 消息传递 │ 黑板系统 │ 事件驱动 │
├──────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│ 耦合度 │ 高 │ 中 │ 中 │ 低 │
│ 实现复杂度 │ 低 │ 中 │ 中 │ 高 │
│ 分布式支持 │ 差 │ 好 │ 中 │ 最好 │
│ 信息可见性 │ 全局可见 │ 点对点 │ 选择性可见 │ 订阅可见 │
│ 适合规模 │ 小(3-5) │ 中(5-10) │ 中(5-10) │ 大(10+) │
│ 代表框架 │ LangGraph │ AutoGen │ 自定义 │ 自定义 │
│ 最佳场景 │ 简单协作 │ 对话协作 │ 异步协作 │ 大规模系统 │
└──────────────┴──────────────┴──────────────┴──────────────┴──────────────┘
选型建议:
Agent数量 < 5 → 共享状态（简单够用）
Agent需要对话 → 消息传递
Agent异步工作 → 黑板系统
Agent数量 > 10 → 事件驱动
五、协调机制
5.1 为什么需要协调
问题: 多个Agent可能产生冲突
场景1: 任务冲突
Agent A: "我来处理这个任务"
Agent B: "不，我来处理"
→ 谁来做？
场景2: 意见冲突
Agent A: "应该用方案1"
Agent B: "应该用方案2"
→ 听谁的？
场景3: 资源冲突
Agent A: "我需要调用数据库"
Agent B: "我也需要调用数据库"（但数据库有并发限制）
→ 谁先用？
没有协调机制 → 混乱、死锁、重复劳动
有协调机制 → 有序、高效、结果一致
5.2 中央调度（Central Scheduling）
原理: 一个"调度者"做所有决策，其他Agent只执行。
调度者决定:
- 谁做什么任务
- 什么顺序执行
- 冲突时听谁的
优点: 简单、可预测、不会死锁
缺点: 调度者是单点故障、可能成为瓶颈
实现: 就是"管理者模式"中的Orchestrator
5.3 投票/共识机制（Voting/Consensus）
原理: 多个Agent对同一问题给出答案，通过投票决定最终结果。
Agent A: "答案是X"
Agent B: "答案是Y"
Agent C: "答案是X"
→ 投票结果: X (2票) > Y (1票) → 最终答案是X
变体:
1. 简单多数: 超过50%即通过
2. 加权投票: 不同Agent的票权重不同（专家的票更重）
3. 一致性要求: 必须所有Agent都同意
适合场景:
- 需要高可靠性的决策（多个Agent交叉验证）
- 答案有明确对错的场景（如数学计算、事实判断）
实际应用:
- LLM-as-Judge: 多个LLM评估同一个输出，投票决定质量
- 多模型集成: 同一问题问3个不同模型，取多数答案
"""
投票机制实现
"""
from openai import OpenAI
from collections import Counter
client = OpenAI()
def vote_based_decision(question: str, num_voters: int = 3,
models: list = None) -> dict:
"""多Agent投票决策"""
if models is None:
models = ["gpt-4o-mini"] * num_voters
votes = []
for i, model in enumerate(models):
response = client.chat.completions.create(
model=model,
messages=[{
"role": "user",
"content": f"""{question}
请给出你的答案。只回答最终结论，不要解释过程。"""
}],
temperature=0.7 # 稍高温度增加多样性
)
answer = response.choices[0].message.content.strip()
votes.append(answer)
print(f" Voter {i+1}: {answer}")
# 统计投票结果
counter = Counter(votes)
winner = counter.most_common(1)[0]
return {
"votes": votes,
"winner": winner[0],
"confidence": winner[1] / len(votes),
"unanimous": len(counter) == 1
}
# 使用示例
result = vote_based_decision("Python中 list.sort() 和 sorted() 的区别是什么？最核心的一点是？")
print(f"\n投票结果: {result['winner']}")
print(f"一致性: {'全票通过' if result['unanimous'] else f'置信度{result[\"confidence\"]:.0%}'}")
5.4 辩论机制（Debate）
原理: Agent之间就某个问题进行辩论，互相质疑对方的推理过程，
最终由裁判Agent或人类决定采纳哪个方案。
流程:
1. 多个Agent各自给出答案和推理过程
2. Agent互相阅读对方的推理，指出漏洞
3. Agent根据对方的质疑修改自己的答案
4. 重复2-3轮
5. 裁判Agent选出最终答案
为什么辩论比投票好:
投票: 只看最终答案，不看推理过程
辩论: 通过质疑推理过程，能发现隐藏的错误
研究结果:
"Improving Factuality and Reasoning in Language Models through Multiagent Debate"
(Du et al., 2023)
→ 辩论后的答案准确率比单Agent高10-20%
→ 特别是在需要推理的复杂问题上效果显著
5.5 市场机制（Market Mechanism）
原理: Agent通过"竞标"来获取任务。
每个Agent评估自己完成任务的能力和成本，
最合适的Agent"赢得"任务执行权。
流程:
1. 管理者发布任务: "需要分析一段Go代码的性能问题"
2. Agent A竞标: "我是Go专家，置信度0.9，预计耗时30s"
3. Agent B竞标: "我是通用开发者，置信度0.6，预计耗时60s"
4. Agent C竞标: "我是Python专家，置信度0.3，不太适合"
5. 管理者选择Agent A（最高置信度）
优点:
✅ 自动选择最合适的Agent
✅ Agent可以动态加入/退出
✅ 天然的负载均衡
缺点:
❌ 竞标过程本身有开销
❌ Agent可能"高估"自己的能力
❌ 实现复杂
适合场景:
- Agent池很大，需要动态选择
- 任务类型多样，不同任务需要不同专家
- 类似"人才市场"的场景
六、A2A协议（Agent-to-Agent Protocol）
6.1 为什么需要A2A
背景:
MCP解决了"Agent如何连接工具"的标准化问题。
但还有一个问题没解决: "Agent如何连接Agent"？
当前的多Agent系统:
- CrewAI的Agent只能和CrewAI的Agent协作
- AutoGen的Agent只能和AutoGen的Agent协作
- 不同框架的Agent无法互通
类比:
MCP之前: 每个AI应用自己写工具连接代码 → MCP统一了
A2A之前: 每个框架自己定义Agent通信方式 → A2A要统一
A2A的目标:
让不同团队、不同框架、甚至不同公司的Agent能够互相通信和协作。
你的Agent（用LangGraph写的）
↕ A2A协议
合作方的Agent（用AutoGen写的）
↕ A2A协议
第三方的Agent（用CrewAI写的）
6.2 A2A的核心概念
A2A协议 (Google, 2025年4月首次发布，现已捐赠给Linux Foundation)
GitHub: https://github.com/a2aproject/A2A (⭐ 24.1K+)
官方文档: https://a2aproject.github.io/A2A/
四个核心概念:
1. Agent Card（智能体名片）
每个Agent对外暴露一个"名片"，描述自己的能力
类似MCP Server的工具列表，但描述的是Agent级别的能力
{
"name": "code-reviewer",
"description": "专业的代码审查Agent，支持Python/Go/Java",
"capabilities": ["code_review", "security_audit", "performance_analysis"],
"input_format": "代码文件或PR链接",
"output_format": "结构化的审查报告"
}
2. Task（任务）
Agent间传递的工作单元
{
"task_id": "task_001",
"from_agent": "orchestrator",
"to_agent": "code-reviewer",
"description": "审查这段代码的安全性",
"payload": {"code": "...", "language": "python"},
"deadline": "2024-03-15T10:00:00Z"
}
3. Message（消息）
Agent间的通信消息（任务进度、中间结果、问题反馈等）
4. Artifact（产出物）
任务完成后的产出
{
"task_id": "task_001",
"artifact_type": "review_report",
"content": {"issues": [...], "score": 85, "suggestions": [...]}
}
关键技术特性:
- 通信协议: JSON-RPC 2.0 over HTTP(S)
- 交互模式: 同步请求/响应、SSE流式、异步推送通知
- 数据格式: 支持文本、文件和结构化JSON
- 企业级: 内置安全认证和可观测性设计
官方SDK（多语言支持）:
- Python: pip install a2a-sdk
- Go: go get github.com/a2aproject/a2a-go
- JavaScript: npm install @a2a-js/sdk
- Java: Maven依赖
- .NET: dotnet add package A2A
- Rust: cargo add a2a-lf
A2A vs MCP:
┌──────────────┬──────────────────────────────────────────┐
│ │ 连接对象 │
├──────────────┼──────────────────────────────────────────┤
│ MCP │ Agent ←→ Tool（Agent连接工具） │
│ A2A │ Agent ←→ Agent（Agent连接Agent） │
├──────────────┼──────────────────────────────────────────┤
│ │ 粒度 │
├──────────────┼──────────────────────────────────────────┤
│ MCP │ 函数级（一个工具 = 一个函数） │
│ A2A │ Agent级（一个Agent = 一组能力） │
├──────────────┼──────────────────────────────────────────┤
│ │ 通信模式 │
├──────────────┼──────────────────────────────────────────┤
│ MCP │ 请求-响应（同步） │
│ A2A │ 任务-产出（可异步，支持长时间运行的任务） │
├──────────────┼──────────────────────────────────────────┤
│ │ 治理 │
├──────────────┼──────────────────────────────────────────┤
│ MCP │ Anthropic主导 │
│ A2A │ Linux Foundation开源治理 │
└──────────────┴──────────────────────────────────────────┘
6.3 A2A的当前状态
截至2026年6月:
- Google于2025年4月首次发布A2A协议规范
- 2025年下半年捐赠给Linux Foundation，成为开放治理项目
- 项目已从google/A2A迁移到独立组织a2aproject/A2A
- GitHub: 24.1K+ Stars, 2.4K+ Forks, 144+贡献者
- 已有完整的多语言SDK（Python/Go/JS/Java/.NET/Rust）
- DeepLearning.AI已推出官方课程（与Google Cloud和IBM Research合作）
- 多个主流框架已集成A2A支持:
- Google ADK: 原生支持A2A Server/Client
- LangGraph: 可将Agent暴露为A2A-compliant Server
- BeeAI: 原生A2A支持
- CrewAI: 通过插件支持A2A
- 协议仍在快速演进中，路线图包括:
- Agent Discovery增强（AgentCard中的授权方案）
- QuerySkill()方法（动态查询Agent能力）
- 动态UX协商（任务中途切换交互模式）
- 流式可靠性和推送通知机制改进
对开发者的建议:
- A2A已进入可用阶段，建议在新项目中考虑A2A兼容性
- 如果你的Agent需要跨团队/跨组织协作，优先采用A2A
- 使用Google ADK或LangGraph可以最快实现A2A兼容
- 学习资源: DeepLearning.AI官方课程 "A2A: The Agent2Agent Protocol"
- 关注a2aproject/A2A仓库的更新和samples目录
七、主流多Agent框架深度对比
7.1 框架全景
┌──────────────────────────────────────────────────────────────────────────────────────┐
│ 主流多Agent框架 (2024-2026) │
├──────────────┬──────────────┬──────────────┬──────────────┬─────────────┬────────────┤
│ CrewAI │ AutoGen │ LangGraph │ OpenAI │ MetaGPT │ Google ADK │
│ │ (Microsoft) │ (LangChain) │ Agents SDK │(Foundation │ (Google) │
│ │ │ │ │ Agents) │ │
├──────────────┼──────────────┼──────────────┼──────────────┼─────────────┼────────────┤
│ 高层抽象 │ 对话驱动 │ 图编排 │ 轻量+全能 │ SOP驱动 │ A2A原生 │
│ 开箱即用 │ 灵活强大 │ 精确控制 │ 多模型支持 │ 软件开发 │ 云原生 │
├──────────────┼──────────────┼──────────────┼──────────────┼─────────────┼────────────┤
│ Python │ Python │ Python │ Python/JS │ Python │ Python │
│ 2023.10 │ 2023.9 │ 2024.1 │ 2025.3 │ 2023.8 │ 2025.4 │
├──────────────┼──────────────┼──────────────┼──────────────┼─────────────┼────────────┤
│ ⭐ 52.9K │ ⭐ 58.7K │ ⭐ 33.9K │ ⭐ 26.9K │ ⭐ 68.5K │ ⭐ 20K │
│ (GitHub Star)│ │ │ │ │ │
└──────────────┴──────────────┴──────────────┴──────────────┴─────────────┴────────────┘
注: Star数据截至2026年6月
7.2 CrewAI
定位: "最简单的多Agent框架"——用最少的代码实现多Agent协作
核心概念:
- Agent: 有角色、目标、背景故事的智能体
- Task: 分配给Agent的具体任务
- Crew: 一组Agent + 一组Task的组合
- Process: 执行模式（sequential顺序 / hierarchical层级）
设计哲学:
"像组建一个团队一样简单"
定义角色 → 分配任务 → 一键运行
优点:
✅ 上手极快（10行代码就能跑起来）
✅ 概念直观（Agent/Task/Crew对应人/任务/团队）
✅ 内置工具集成（搜索、文件读写等）
✅ 支持层级模式（Manager自动分配任务）
缺点:
❌ 灵活性有限（复杂流程难以表达）
❌ 调试困难（黑盒执行，中间过程不透明）
❌ 不支持复杂的条件分支和循环
❌ 对LLM调用的控制粒度粗
"""
CrewAI示例: 技术调研团队
依赖: pip install crewai
"""
from crewai import Agent, Task, Crew, Process
# ═══════════════════════════════════════════
# 定义Agent（角色）
# ═══════════════════════════════════════════
researcher = Agent(
role="技术研究员",
goal="深入研究给定的技术主题，收集关键信息和最新动态",
backstory="""你是一位资深技术研究员，有10年的技术调研经验。
你擅长从多个角度分析技术，能快速识别技术的核心价值和潜在风险。""",
verbose=True,
allow_delegation=False, # 不允许委派给其他Agent
)
analyst = Agent(
role="技术分析师",
goal="从研究数据中提取关键洞察，形成结构化的分析结论",
backstory="""你是一位技术分析师，擅长从大量信息中提炼关键洞察。
你能识别技术趋势、评估优劣势、给出客观的技术评价。""",
verbose=True,
allow_delegation=False,
)
writer = Agent(
role="技术作者",
goal="将分析结果撰写为清晰易懂的技术报告",
backstory="""你是一位技术写作专家，能把复杂技术概念写成通俗易懂的文章。
你的文章结构严谨、重点突出、有理有据。""",
verbose=True,
allow_delegation=False,
)
# ═══════════════════════════════════════════
# 定义Task（任务）
# ═══════════════════════════════════════════
research_task = Task(
description="""研究"Agent记忆系统"的技术现状，包括:
1. 主流的记忆架构方案
2. 关键论文和开源项目
3. 生产环境的最佳实践
4. 当前的技术挑战""",
expected_output="包含4个方面的详细研究笔记，每个方面至少3个关键发现",
agent=researcher,
)
analysis_task = Task(
description="""基于研究笔记，分析:
1. 最成熟的技术方案是什么
2. 最大的技术挑战是什么
3. 未来发展趋势
4. 对开发者的建议""",
expected_output="结构化的分析报告，包含4个维度的深度分析",
agent=analyst,
)
writing_task = Task(
description="""基于研究和分析，撰写一篇技术报告:
1. 标题: "Agent记忆系统技术调研报告"
2. 包含: 背景、现状、分析、趋势、建议
3. 语言简洁，重点突出，1000字左右""",
expected_output="一篇结构清晰的技术报告",
agent=writer,
)
# ═══════════════════════════════════════════
# 组建Crew并运行
# ═══════════════════════════════════════════
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential, # 顺序执行（流水线模式）
verbose=True,
)
# 运行
result = crew.kickoff()
print("\n" + "="*60)
print("最终报告:")
print(result)
7.3 AutoGen
定位: "最灵活的多Agent对话框架"——Agent之间通过对话协作
核心概念:
- Agent: 可以发送和接收消息的实体
- GroupChat: 多个Agent的群聊
- GroupChatManager: 管理群聊中的发言顺序
- ConversableAgent: 可对话的Agent基类
设计哲学:
"Agent之间的协作本质上就是对话"
让Agent像人一样聊天、讨论、达成共识
优点:
✅ 极度灵活（任何对话模式都能实现）
✅ 支持人类参与（Human-in-the-loop）
✅ 丰富的Agent类型（Assistant/UserProxy/GroupChat等）
✅ 微软背书，社区活跃
缺点:
❌ 学习曲线陡（概念多，API复杂）
❌ 对话可能不收敛（需要仔细设计终止条件）
❌ Token消耗大（多轮对话累积）
❌ 版本迭代快，API不稳定（v0.2 → v0.4变化大）
"""
AutoGen示例: 代码写作+审查对话
依赖: pip install autogen-agentchat autogen-ext
"""
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
# 创建模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
# ═══════════════════════════════════════════
# 定义Agent
# ═══════════════════════════════════════════
coder = AssistantAgent(
name="coder",
model_client=model_client,
system_message="""你是一个Python开发者。
根据需求写代码，代码要求:
1. 有完整的类型注解
2. 有docstring
3. 有错误处理
4. 代码简洁高效
当审查者说APPROVED时，任务完成。""",
)
reviewer = AssistantAgent(
name="reviewer",
model_client=model_client,
system_message="""你是一个代码审查专家。
审查代码的:
1. 正确性（逻辑是否正确）
2. 健壮性（错误处理是否完善）
3. 可读性（命名、注释是否清晰）
4. 性能（是否有明显的性能问题）
如果有问题，指出具体问题和修改建议。
如果代码质量达标，回复"APPROVED"。""",
)
# ═══════════════════════════════════════════
# 组建团队
# ═══════════════════════════════════════════
# 终止条件: 当消息中出现"APPROVED"时停止
termination = TextMentionTermination("APPROVED")
team = RoundRobinGroupChat(
participants=[coder, reviewer],
termination_condition=termination,
max_turns=6, # 最多6轮对话
)
# ═══════════════════════════════════════════
# 运行
# ═══════════════════════════════════════════
async def main():
result = await team.run(
task="实现一个Python装饰器，用于函数调用的重试机制，支持指定最大重试次数和重试间隔"
)
print("\n" + "="*60)
print("对话记录:")
for msg in result.messages:
print(f"\n[{msg.source}]: {msg.content[:200]}...")
asyncio.run(main())
7.4 LangGraph多Agent
定位: "最精确控制的多Agent框架"——用图来定义Agent之间的协作流程
核心概念:
- State: 共享状态（所有节点共享）
- Node: 图中的节点（每个节点可以是一个Agent）
- Edge: 节点之间的连接（定义流转逻辑）
- Conditional Edge: 条件边（根据状态决定下一步）
设计哲学:
"多Agent协作 = 一个状态机/图"
用图的节点和边精确定义Agent之间的协作流程
优点:
✅ 精确控制（每一步都可以精确定义）
✅ 可视化（图结构直观）
✅ 支持复杂流程（条件分支、循环、并行）
✅ 与LangChain生态无缝集成
✅ 内置持久化（checkpoint）
缺点:
❌ 代码量大（需要定义State、Node、Edge）
❌ 学习曲线（需要理解图编程范式）
❌ 对简单场景过于复杂
"""
LangGraph多Agent示例: Supervisor模式
依赖: pip install langgraph langchain-openai
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import json
# ═══════════════════════════════════════════
# 定义共享状态
# ═══════════════════════════════════════════
class MultiAgentState(TypedDict):
messages: Annotated[list, add_messages]
next_agent: str # 下一个要执行的Agent
research_output: str
code_output: str
final_output: str
# ═══════════════════════════════════════════
# 定义Agent节点
# ═══════════════════════════════════════════
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def supervisor_node(state: MultiAgentState) -> dict:
"""Supervisor: 决定下一步由谁执行"""
messages = state["messages"]
response = llm.invoke([
SystemMessage(content="""你是一个任务管理者。根据当前状态，决定下一步由谁执行。
可选的Agent:
- researcher: 负责研究和收集信息
- coder: 负责编写代码
- FINISH: 任务完成
根据对话历史和已有结果，决定下一步。
只回答一个词: researcher / coder / FINISH"""),
*messages
])
next_agent = response.content.strip().lower()
if next_agent not in ["researcher", "coder", "finish"]:
next_agent = "finish"
return {"next_agent": next_agent}
def researcher_node(state: MultiAgentState) -> dict:
"""研究Agent"""
messages = state["messages"]
response = llm.invoke([
SystemMessage(content="你是一个技术研究员。根据用户的需求，提供详细的技术调研信息。"),
*messages
])
return {
"messages": [response],
"research_output": response.content
}
def coder_node(state: MultiAgentState) -> dict:
"""开发Agent"""
messages = state["messages"]
research = state.get("research_output", "")
context = f"参考研究结果:\n{research}\n\n" if research else ""
response = llm.invoke([
SystemMessage(content=f"你是一个高级开发者。{context}根据需求编写高质量代码。"),
*messages
])
return {
"messages": [response],
"code_output": response.content
}
# ═══════════════════════════════════════════
# 构建图
# ═══════════════════════════════════════════
def route_next(state: MultiAgentState) -> Literal["researcher", "coder", "__end__"]:
"""路由函数: 根据supervisor的决策路由到下一个节点"""
next_agent = state.get("next_agent", "finish")
if next_agent == "finish":
return "__end__"
return next_agent
# 构建StateGraph
graph = StateGraph(MultiAgentState)
# 添加节点
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
# 添加边
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_next)
graph.add_edge("researcher", "supervisor") # 研究完回到supervisor
graph.add_edge("coder", "supervisor") # 编码完回到supervisor
# 编译
app = graph.compile()
# ═══════════════════════════════════════════
# 运行
# ═══════════════════════════════════════════
result = app.invoke({
"messages": [HumanMessage(content="帮我用Go实现一个简单的限流器（令牌桶算法）")],
"next_agent": "",
"research_output": "",
"code_output": "",
"final_output": ""
})
print("最终结果:")
for msg in result["messages"]:
if hasattr(msg, "content"):
print(f"\n{msg.content[:300]}...")
7.5 OpenAI Agents SDK
定位: "轻量但全能的多Agent框架"——OpenAI官方出品，已发展为功能丰富的生产级框架
GitHub: https://github.com/openai/openai-agents-python (⭐ 26.9K)
安装: pip install openai-agents
核心概念（已从最初的3个扩展到9个）:
- Agent: 有instructions和tools的智能体
- Sandbox Agents: 预配置容器环境的Agent（v0.14.0新增），支持长时间任务
- Agents as tools / Handoffs: Agent之间的委托和转交
- Tools: 函数工具、MCP工具、托管工具
- Guardrails: 输入输出安全验证
- Human in the loop: 内置人类介入机制
- Sessions: 自动管理跨轮次的对话历史
- Tracing: 内置运行追踪、调试和优化
- Realtime Agents: 基于gpt-realtime-2的语音Agent
设计哲学:
"少即是多，但也要够用"——从极简起步，逐步扩展为全能框架
现在已支持100+个LLM（通过LiteLLM集成），不再仅限OpenAI模型
重大更新（2025-2026）:
- Provider-agnostic: 支持OpenAI、Anthropic、Google等100+模型
- Sandbox Agents: 支持容器化长时间任务执行
- Sessions: 自动对话历史管理（支持Redis持久化）
- JS/TS版本: 已发布JavaScript/TypeScript SDK
- Realtime Agents: 语音Agent支持
优点:
✅ API设计优雅（几行代码就能跑）
✅ OpenAI官方维护，迭代快
✅ 原生支持Handoff（Swarm模式）
✅ 内置Tracing和调试工具
✅ 支持100+模型（不再仅限OpenAI）
✅ Sandbox Agents支持复杂长时间任务
✅ 有Python和JS/TS双版本
缺点:
❌ 不支持复杂的图编排（不如LangGraph灵活）
❌ 没有内置的对话模式（不如AutoGen）
❌ 版本迭代快，API可能变化
"""
OpenAI Agents SDK示例
依赖: pip install openai-agents
"""
from agents import Agent, Runner, function_tool, handoff
# ═══════════════════════════════════════════
# 定义工具
# ═══════════════════════════════════════════
@function_tool
def search_docs(query: str) -> str:
"""搜索技术文档"""
return f"搜索'{query}'的结果: [模拟的搜索结果...]"
@function_tool
def write_code(spec: str, language: str = "python") -> str:
"""根据规格写代码"""
return f"```{language}\n# 根据规格: {spec}\ndef implement():\n pass\n```"
@function_tool
def run_tests(code: str) -> str:
"""运行测试"""
return "所有测试通过 ✅"
# ═══════════════════════════════════════════
# 定义Agent
# ═══════════════════════════════════════════
researcher = Agent(
name="researcher",
instructions="你是技术研究员。使用search_docs工具收集信息，然后总结关键发现。",
tools=[search_docs],
)
developer = Agent(
name="developer",
instructions="你是开发者。根据需求使用write_code工具编写代码。",
tools=[write_code],
)
tester = Agent(
name="tester",
instructions="你是测试工程师。使用run_tests工具验证代码。",
tools=[run_tests],
)
# 入口Agent（Triage/Router）
triage = Agent(
name="triage",
instructions="""你是任务路由Agent。根据用户的请求:
- 如果需要调研 → 转交给researcher
- 如果需要写代码 → 转交给developer
- 如果需要测试 → 转交给tester""",
handoffs=[
handoff(agent=researcher, description="需要技术调研时转交"),
handoff(agent=developer, description="需要编写代码时转交"),
handoff(agent=tester, description="需要运行测试时转交"),
],
)
# ═══════════════════════════════════════════
# 运行
# ═══════════════════════════════════════════
async def main():
result = await Runner.run(triage, "帮我调研一下Go的限流库有哪些")
print(f"最终结果: {result.final_output}")
import asyncio
asyncio.run(main())
7.6 框架选型指南
┌──────────────────────────────────────────────────────────────────────────┐
│ 框架选型决策树 │
└──────────────────────────────────────────────────────────────────────────┘
你的需求是什么？
├── "我想快速搭建一个多Agent原型"
│ → CrewAI（最快上手，10分钟跑起来）
│
├── "我需要Agent之间深度对话和纠错"
│ → AutoGen（对话模式最强）
│
├── "我需要精确控制每一步的执行流程"
│ → LangGraph（图编排，精确控制）
│
├── "我想要轻量但功能全面的方案"
│ → OpenAI Agents SDK（官方出品，支持100+模型，功能丰富）
│
├── "我要模拟一个软件开发团队"
│ → MetaGPT 或 ChatDev（专为软件开发设计）
│
└── "我需要生产级的可靠性和可观测性"
→ LangGraph（内置持久化、重试、可观测性）
├── "我需要原生A2A协议支持"
│ → Google ADK（A2A协议的官方参考实现）
更详细的对比:
┌──────────────┬──────────┬──────────┬──────────┬──────────┬──────────┬──────────┐
│ 维度 │ CrewAI │ AutoGen │ LangGraph│ OAI SDK │ MetaGPT │Google ADK│
├──────────────┼──────────┼──────────┼──────────┼──────────┼──────────┼──────────┤
│ 上手难度 │ ⭐ │ ⭐⭐⭐ │ ⭐⭐⭐ │ ⭐ │ ⭐⭐ │ ⭐⭐ │
│ 灵活性 │ ⭐⭐ │ ⭐⭐⭐⭐ │ ⭐⭐⭐⭐⭐│ ⭐⭐⭐ │ ⭐⭐ │ ⭐⭐⭐⭐ │
│ 生产就绪 │ ⭐⭐⭐ │ ⭐⭐⭐ │ ⭐⭐⭐⭐ │ ⭐⭐⭐ │ ⭐⭐ │ ⭐⭐⭐ │
│ 模型支持 │ 多模型 │ 多模型 │ 多模型 │ 100+模型 │ 多模型 │ 多模型 │
│ A2A支持 │ 插件 │ 社区 │ 已集成 │ 无 │ 无 │ 原生 │
│ 社区活跃度 │ 高 │ 最高 │ 高 │ 高 │ 高 │ 高 │
│ 文档质量 │ 好 │ 中 │ 好 │ 好 │ 中 │ 好 │
│ 调试体验 │ 一般 │ 一般 │ 好 │ 好 │ 一般 │ 好 │
└──────────────┴──────────┴──────────┴──────────┴──────────┴──────────┴──────────┘
八、完整实战：从零构建多Agent协作系统
8.1 实战目标
构建一个"技术方案评审"多Agent系统:
用户输入: 一个技术方案描述
系统输出: 多角度的评审报告
Agent团队:
1. 架构师Agent: 评审架构设计的合理性
2. 安全专家Agent: 评审安全风险
3. 性能专家Agent: 评审性能瓶颈
4. 汇总Agent: 整合所有评审意见，生成最终报告
架构模式: 管理者模式（汇总Agent作为管理者）
通信方式: 共享状态
8.2 完整实现
"""
完整实战: 技术方案评审多Agent系统
依赖: pip install openai
"""
from openai import OpenAI
from dataclasses import dataclass, field
from datetime import datetime
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
client = OpenAI()
# ═══════════════════════════════════════════════════════════════════
# Part 1: Agent定义
# ═══════════════════════════════════════════════════════════════════
@dataclass
class ReviewAgent:
"""评审Agent"""
name: str
role: str
system_prompt: str
review_dimensions: list[str] # 评审维度
def review(self, proposal: str) -> dict:
"""对技术方案进行评审"""
dimensions_text = "\n".join([f" {i+1}. {d}" for i, d in enumerate(self.review_dimensions)])
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"""请从你的专业角度评审以下技术方案:
{proposal}
请从以下维度进行评审:
{dimensions_text}
输出JSON格式:
{{
"score": 1-10的评分,
"summary": "一句话总结评审结论",
"issues": [
{{"severity": "high/medium/low", "description": "问题描述", "suggestion": "改进建议"}}
],
"highlights": ["方案的亮点1", "方案的亮点2"]
}}"""}
],
response_format={"type": "json_object"},
temperature=0.3
)
try:
result = json.loads(response.choices[0].message.content)
result["reviewer"] = self.name
result["role"] = self.role
return result
except:
return {
"reviewer": self.name,
"role": self.role,
"score": 5,
"summary": "评审过程出错",
"issues": [],
"highlights": []
}
# ═══════════════════════════════════════════════════════════════════
# Part 2: 评审团队
# ═══════════════════════════════════════════════════════════════════
class ReviewTeam:
"""评审团队: 管理多个评审Agent"""
def __init__(self):
self.reviewers = self._create_reviewers()
def _create_reviewers(self) -> list[ReviewAgent]:
"""创建评审团队"""
return [
ReviewAgent(
name="架构师",
role="系统架构评审",
system_prompt="""你是一位资深系统架构师，有15年分布式系统设计经验。
你擅长评估系统架构的合理性、可扩展性和可维护性。
你的评审风格: 严谨但建设性，指出问题的同时给出改进方案。""",
review_dimensions=[
"架构分层是否合理",
"组件职责是否清晰",
"可扩展性设计",
"技术选型是否合适",
"是否存在过度设计或设计不足"
]
),
ReviewAgent(
name="安全专家",
role="安全风险评审",
system_prompt="""你是一位信息安全专家，专注于应用安全和基础设施安全。
你擅长发现潜在的安全漏洞、数据泄露风险和权限问题。
你的评审风格: 从攻击者视角思考，找出最薄弱的环节。""",
review_dimensions=[
"认证和授权机制",
"数据安全（传输和存储）",
"输入验证和注入防护",
"敏感信息保护",
"第三方依赖的安全性"
]
),
ReviewAgent(
name="性能专家",
role="性能瓶颈评审",
system_prompt="""你是一位性能工程专家，专注于高并发系统的性能优化。
你擅长识别性能瓶颈、评估系统容量和优化方案。
你的评审风格: 用数据说话，关注P99延迟和吞吐量。""",
review_dimensions=[
"是否存在性能瓶颈",
"数据库查询效率",
"缓存策略是否合理",
"并发处理能力",
"资源使用效率"
]
),
]
def review(self, proposal: str, parallel: bool = True) -> dict:
"""执行评审（支持并行）"""
print(f"📋 开始评审，共 {len(self.reviewers)} 位评审专家")
print(f" 并行模式: {'是' if parallel else '否'}")
reviews = []
if parallel:
# 并行执行所有评审
with ThreadPoolExecutor(max_workers=len(self.reviewers)) as executor:
futures = {
executor.submit(reviewer.review, proposal): reviewer
for reviewer in self.reviewers
}
for future in as_completed(futures):
reviewer = futures[future]
result = future.result()
reviews.append(result)
print(f" ✅ {reviewer.name} 评审完成 (评分: {result.get('score', '?')}/10)")
else:
# 串行执行
for reviewer in self.reviewers:
print(f" → {reviewer.name} 正在评审...")
result = reviewer.review(proposal)
reviews.append(result)
print(f" ✅ {reviewer.name} 评审完成 (评分: {result.get('score', '?')}/10)")
# 汇总评审结果
print(f"\n📊 汇总评审结果...")
summary = self._synthesize(proposal, reviews)
return {
"individual_reviews": reviews,
"summary": summary,
"timestamp": datetime.now().isoformat()
}
def _synthesize(self, proposal: str, reviews: list[dict]) -> dict:
"""汇总所有评审意见"""
reviews_text = ""
for r in reviews:
reviews_text += f"\n--- {r['reviewer']}({r['role']}) 评分:{r.get('score','?')}/10 ---\n"
reviews_text += f"总结: {r.get('summary', '')}\n"
if r.get('issues'):
reviews_text += "问题:\n"
for issue in r['issues']:
reviews_text += f" [{issue.get('severity','')}] {issue.get('description','')}\n"
if r.get('highlights'):
reviews_text += f"亮点: {', '.join(r.get('highlights', []))}\n"
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""请汇总以下多位专家的评审意见，生成最终评审报告:
技术方案:
{proposal[:500]}
各专家评审:
{reviews_text}
输出JSON格式:
{{
"overall_score": 综合评分(1-10),
"verdict": "通过/有条件通过/需要修改/不通过",
"key_risks": ["最关键的风险1", "风险2"],
"must_fix": ["必须修改的问题1", "问题2"],
"nice_to_have": ["建议优化的点1", "点2"],
"final_comment": "给方案作者的综合建议（2-3句话）"
}}"""
}],
response_format={"type": "json_object"},
temperature=0.3
)
try:
return json.loads(response.choices[0].message.content)
except:
return {"overall_score": 5, "verdict": "需要修改", "final_comment": "汇总过程出错"}
# ═══════════════════════════════════════════════════════════════════
# Part 3: 使用示例
# ═══════════════════════════════════════════════════════════════════
if __name__ == "__main__":
team = ReviewTeam()
proposal = """
技术方案: 订单服务重构
1. 架构: 从单体拆分为微服务，使用gRPC通信
2. 数据库: 从MySQL迁移到PostgreSQL，引入读写分离
3. 缓存: 使用Redis缓存热点订单数据，TTL=5分钟
4. 消息队列: 引入Kafka处理异步任务（发送通知、更新统计）
5. API网关: 使用Kong作为API网关，统一认证和限流
6. 部署: Kubernetes部署，HPA自动扩缩容
"""
result = team.review(proposal, parallel=True)
print("\n" + "="*60)
print("📝 最终评审报告")
print("="*60)
summary = result["summary"]
print(f"\n综合评分: {summary.get('overall_score', '?')}/10")
print(f"评审结论: {summary.get('verdict', '?')}")
print(f"\n关键风险:")
for risk in summary.get('key_risks', []):
print(f" ⚠️ {risk}")
print(f"\n必须修改:")
for fix in summary.get('must_fix', []):
print(f" 🔴 {fix}")
print(f"\n建议优化:")
for nice in summary.get('nice_to_have', []):
print(f" 🟡 {nice}")
print(f"\n综合建议: {summary.get('final_comment', '')}")
九、工程最佳实践
9.1 设计原则
原则1: 最小Agent数量
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ 用最少的Agent完成任务
❌ 为了"看起来高级"而增加Agent
经验法则:
- 3-5个Agent是最佳数量
- 超过7个Agent就要考虑是否过度设计
- 每增加一个Agent = 增加一倍的通信复杂度和Token成本
原则2: 角色互斥完备
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ 每个Agent的职责不重叠（互斥）
✅ 所有Agent的职责加起来覆盖整个任务（完备）
❌ 两个Agent都能做同一件事（导致冲突或重复）
❌ 某些任务没有Agent负责（导致遗漏）
原则3: 明确的输入输出契约
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ 每个Agent的输入格式和输出格式都明确定义
❌ Agent之间传递模糊的自然语言
好的做法:
Agent A输出: {"type": "research_report", "findings": [...], "confidence": 0.8}
Agent B期望输入: {"type": "research_report", ...}
→ 格式匹配，不会出错
坏的做法:
Agent A输出: "我研究了一下，发现..."
Agent B: "这是什么格式？我该怎么解析？"
原则4: 必须有收敛机制
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ 设置最大轮数/最大步数
✅ 定义明确的终止条件
❌ 让Agent无限对话/无限转交
必须设置:
- max_rounds: 对话模式的最大轮数（建议3-5轮）
- max_handoffs: 群体模式的最大转交次数（建议3-5次）
- timeout: 整体超时时间（建议30-120秒）
原则5: 可观测性
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ 记录每个Agent的输入、输出、耗时
✅ 记录Agent之间的通信内容
✅ 记录整体执行路径
❌ 黑盒执行，出了问题不知道哪里错了
必须记录:
- 每个Agent被调用了几次
- 每次调用的Token消耗
- 执行路径（A→B→C还是A→C→B）
- 失败原因和重试次数
原则6: 优雅降级
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ 某个Agent失败时，系统仍能给出部分结果
❌ 一个Agent失败导致整个系统崩溃
做法:
- 每个Agent的调用都有try-catch
- 失败时返回默认值或跳过该Agent
- 最终结果标注"哪些Agent未能完成"
9.2 常见陷阱
陷阱1: Agent数量膨胀
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
症状: "我们需要一个Agent做X，一个做Y，一个做Z..."
最终搞出10+个Agent，系统复杂度爆炸
原因: 把"功能"和"Agent"混淆了
不是每个功能都需要一个独立的Agent
解决:
- 一个Agent可以有多个工具（不需要每个工具一个Agent）
- 只有"需要不同角色/专业"时才拆分Agent
- 问自己: "这两个功能需要不同的System Prompt吗？"
如果不需要 → 合并为一个Agent
陷阱2: 通信爆炸
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
症状: Agent之间传递大量信息，Token消耗远超预期
5个Agent的系统，Token消耗是单Agent的10倍而非5倍
原因: 每个Agent都把完整的对话历史传给下一个Agent
信息在Agent之间"滚雪球"式膨胀
解决:
- Agent之间只传递"摘要"而非完整内容
- 使用结构化的输入输出格式（JSON而非自然语言）
- 设置每个Agent的上下文Token预算
陷阱3: 无限循环
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
症状: Agent A转交给B，B转交给C，C又转交回A...
或者: 对话模式中Agent永远达不成共识
原因: 缺乏收敛机制
解决:
- 设置max_rounds / max_handoffs
- 记录转交历史，禁止回到已经去过的Agent
- 设置全局timeout
- 有一个"兜底"机制（超时后强制结束并返回当前最佳结果）
陷阱4: 角色泄露
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
症状: Agent A的输出中包含了Agent B的System Prompt信息
或者Agent开始"扮演"其他Agent的角色
原因: 对话历史中包含了其他Agent的System Prompt
LLM被混淆了
解决:
- 每个Agent的调用都是独立的（不共享System Prompt）
- 对话历史中只包含"内容"，不包含其他Agent的角色设定
- 使用独立的messages数组，而非共享一个
陷阱5: 成本失控
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
症状: 多Agent系统的API调用费用远超预期
原因:
- 每个Agent都用最贵的模型
- 对话轮数没有限制
- 没有Token预算管理
解决:
- 分级使用模型: 管理者用GPT-4o，工作Agent用GPT-4o-mini
- 设置每个Agent的Token预算
- 监控和告警: 单次请求超过X tokens时告警
- 简单任务不要用多Agent
9.3 成本优化策略
策略1: 模型分级
管理者/决策Agent: GPT-4o（需要强推理能力）
工作Agent: GPT-4o-mini（执行具体任务，够用就行）
简单判断: GPT-4o-mini 或更小的模型
策略2: 缓存
相同的输入 → 直接返回缓存结果
特别是"角色判断"类的调用（判断该转交给谁）
策略3: 并行执行
能并行的Agent同时执行 → 减少总耗时（虽然不减少Token）
用户体验更好
策略4: 早期终止
如果某个Agent的输出已经足够好 → 跳过后续Agent
不是每次都需要走完整个流程
策略5: 批量处理
多个小任务合并为一个大任务 → 减少Agent调用次数
十、总结与知识图谱
10.1 本课核心知识点
┌─────────────────────────────────────────────────────────────────────────┐
│ 多智能体系统 - 知识图谱 │
└─────────────────────────────────────────────────────────────────────────┘
多智能体系统
│
┌───────────────────┼───────────────────┐
│ │ │
为什么需要 架构模式 工程实践
│ │ │
┌─────┼─────┐ ┌──────┼──────┐ ┌────┼────┐
│ │ │ │ │ │ │ │ │
上下文 角色 串行 管理者 流水线 对话 设计 常见 成本
窗口 混淆 瓶颈 模式 模式 模式 原则 陷阱 优化
限制 群体模式 层级模式
┌───────────────────┼───────────────────┐
│ │ │
通信机制 协调机制 主流框架
│ │ │
┌─────┼─────┐ ┌──────┼──────┐ ┌────┼────┐
│ │ │ │ │ │ │ │ │
共享 消息 事件 中央 投票 辩论 CrewAI AutoGen LangGraph
状态 传递 驱动 调度 共识 机制 OpenAI SDK
黑板系统 市场机制 MetaGPT
10.2 与前后课程的关系
第14课 ReAct推理行动框架
→ 单Agent的推理-行动循环
→ 本课: 多个Agent各自有ReAct循环，通过协作完成更复杂的任务
第15课 MCP协议
→ Agent如何连接工具（Agent ←→ Tool）
→ 本课: Agent如何连接Agent（Agent ←→ Agent）
→ A2A协议是MCP在Agent间通信的对应物
第16课 Skill系统
→ 单Agent的能力扩展
→ 本课: 多Agent的能力组合（每个Agent加载不同Skill）
第17课 Agent记忆系统
→ 单Agent的记忆管理
→ 本课: 多Agent之间的记忆共享和隔离
下一课: Agent开发框架
→ 如何用具体框架实现本课讲的各种模式
→ LangGraph/CrewAI/AutoGen的深度实战
📝 作业
作业1：实现一个管理者模式的多Agent系统
用本课的OrchestratorAgent代码，实现一个"竞品分析"系统：
- Agent A: 负责收集竞品A的信息
- Agent B: 负责收集竞品B的信息
- Agent C: 负责对比分析
- 管理者: 分配任务并汇总
验证方法：输入"对比分析Cursor和Windsurf这两个AI编程工具"，看系统是否能正确拆分、并行执行、汇总。
作业2：实现一个对话模式的代码审查系统
基于ConversationOrchestrator，实现：
- 开发者Agent写代码
- 审查者Agent审查
- 多轮迭代直到通过
验证方法：观察对话轮数是否在2-4轮内收敛，最终代码质量是否优于单Agent一次性生成。
作业3（进阶）：实现一个混合模式系统
结合管理者模式和对话模式：
- 管理者拆分任务
- 子任务中的"写代码"部分用对话模式（写+审查）
- 其他子任务用单Agent直接执行
下一篇文章见：AI系列文章导航目录-持续更新中
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
12
0- 0
已为社区贡献5条内容
所有评论(0)