多 Agent 系统 - 多个 Agent 协作 / Multi-Agent Systems and Agent Collaboration
📅 创建时间：2026-05-08 🏷️ 标签：#MultiAgent #CrewAI #AutoGen #Agent协作 📚 前置知识：[[05-agent-workflow]]
📋 本章目标
- 理解为什么需要多 Agent 系统
- 掌握多 Agent 的角色定义方法
- 理解两种协作模式：层级 vs 对等
- 了解 Agent 间的通信协议
- 掌握 CrewAI 的使用方法
- 理解多 Agent 的常见问题和解决方案
第0部分：多 Agent 之间的"通信"到底是怎么发生的？
Function Calling 那篇已经讲清楚了：单个 Agent 就是程序 + LLM API 之间的一个 while 循环。那"多个 Agent 协作"是什么？是不是每个 Agent 跑在独立的服务器上，通过 HTTP 或 gRPC 互相通信？
不是。绝大多数多 Agent 框架中，"Agent"只是一个概念——它们共享同一个程序进程，通信就是程序把 Agent A 的输出作为 Agent B 的输入。
0.1 你已有的认知——单 Agent 循环
python
# 单 Agent：一个 while 循环，一个 LLM，一组 tools
messages = [system_prompt, user_question]
while True:
response = llm.chat(messages, tools)
if no tool_calls:
return response.content
execute_tools(response.tool_calls)
messages.append(tool_results)0.2 多 Agent 就是在这个循环外面再套一层编排
python
# 多 Agent（CrewAI 风格）：编排器依次调用不同的 Agent
def run_multi_agent(task):
# Agent A 做研究
research_result = run_single_agent(
role="研究员", goal="搜索相关信息",
input=task
)
# Agent B 根据研究结果写报告
write_result = run_single_agent(
role="写手", goal="撰写报告",
input=research_result ← Agent A 的输出就是 Agent B 的输入
)
# Agent C 审核报告
review_result = run_single_agent(
role="审核员", goal="检查报告质量",
input=write_result ← Agent B 的输出就是 Agent C 的输入
)
return review_result每个 Agent 底层都是一次独立的 LLM API 调用（或一组循环调用），Agent 之间的"通信"就是程序把字符串从一个变量传给下一个变量。 不存在 Agent 之间的网络协议——它们不互相"发消息"，是编排程序在它们之间传递数据。
0.3 那"并行"是怎么实现的？
串行（Sequential）：
Agent A 执行 → 等结果 → Agent B 用结果执行
本质：程序先 run(A)，拿到 output_A，再 run(B, input=output_A)
并行（Parallel）：
Agent A 执行 ─┐
├→ 等所有完成 → 汇总结果
Agent B 执行 ─┘
本质：程序用 Promise.all / asyncio.gather 同时调多个 LLM APIAgent A 和 Agent B 的输入之间没有依赖关系时，就可以并行调用。它们各自调自己的 LLM API，程序等待两边都返回后再汇总。不是 Agent 之间在通信——是程序同时开了两个 LLM API 连接。
0.4 所以"多 Agent"本质上是什么？
┌─────────────────────────────────────────────────────────────┐
│ 多 Agent 的本质 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 多 Agent = 编排程序 + 多个 LLM API 调用 │
│ │
│ "Agent A" 不是一个独立进程，而是： │
│ • 一段独立的 system prompt（定义角色和目标） │
│ • 一组独立的 tools（研究员有搜索工具，写手没有） │
│ • 一次独立的 LLM API 调用（或循环调用） │
│ │
│ 编排程序做的事： │
│ • 决定哪个 Agent 先执行 │
│ • 把上一个 Agent 的输出传给下一个 Agent │
│ • 决定是串行还是并行 │
│ • 汇总多个 Agent 的结果 │
│ │
│ 这就是 CrewAI / AutoGen / LangGraph 等框架做的事情—— │
│ 它们提供了一套 DSL/API 让你声明"谁先谁后、谁传数据给谁"， │
│ 然后框架负责按声明的顺序调 LLM API。 │
│ │
└─────────────────────────────────────────────────────────────┘带着这个认知——多 Agent 就是编排程序在多个 LLM 调用之间传递字符串——下面的细节就不会觉得神秘了。
1.1 单 Agent 的局限
┌─────────────────────────────────────────────────────────────┐
│ 单 Agent 的瓶颈 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 局限1：能力边界 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 一个 Agent 要同时扮演：研究员 + 写手 + 审核员 + ... │ │
│ │ → 什么都做，什么都不精 │ │
│ │ → System Prompt 越来越长，越来越难维护 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 局限2：串行执行效率低 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 任务A（5分钟）→ 任务B（5分钟）→ 任务C（5分钟） │ │
│ │ 总耗时：15分钟 │ │
│ │ 如果能并行：总耗时：5分钟 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 局限3：专业化不足 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 让一个 Agent 同时擅长代码和写作是困难的 │ │
│ │ 不同领域需要不同的知识、不同的 Prompt │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 多 Agent 的优势
┌─────────────────────────────────────────────────────────────┐
│ 多 Agent 的价值 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 优势1：专业化 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Researcher Agent：专注搜索和分析 │ │
│ │ Writer Agent：专注写作和表达 │ │
│ │ Reviewer Agent：专注检查和优化 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 优势2：并行化 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent A：搜索竞品A │ │
│ │ Agent B：搜索竞品B 同时执行 → 节省时间 │ │
│ │ Agent C：搜索竞品C │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 优势3：可组合性 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 新的团队 = 新的 Agent 组合 │ │
│ │ 可以灵活组建不同角色的团队 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘第2部分：Agent 角色定义
2.1 角色的三要素
┌─────────────────────────────────────────────────────────────┐
│ Agent 角色三要素 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. Role（角色） │
│ • 定义 Agent 的身份 │
│ • 如："资深数据分析师"、"专业律师" │
│ │
│ 2. Goal（目标） │
│ • 定义 Agent 要完成什么 │
│ • 如："提取关键数据洞察"、"检查合同风险" │
│ │
│ 3. Backstory（背景故事） │
│ • 补充 Agent 的专业背景 │
│ • 帮助 LLM 理解如何"扮演"这个角色 │
│ │
└─────────────────────────────────────────────────────────────┘2.2 角色定义示例
python
# 差角色定义
researcher = Agent(
role="研究员",
goal="做研究"
)
# 好角色定义
researcher = Agent(
role="资深 AI 市场研究员",
goal="收集并分析目标公司最近的 AI 技术动态、产品发布和市场份额变化，输出结构化的分析报告",
backstory="""你是一位有 10 年经验的技术市场分析师，
专注于 AI 和机器学习领域。
你擅长从公开信息中提取关键洞察，
并能用商业语言表达技术价值。
你注重数据来源的可靠性，
习惯引用一手资料（财报、官方发布、权威报道）。"""
)2.3 常用 Agent 角色模板
┌─────────────────────────────────────────────────────────────┐
│ 常用 Agent 角色库 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 协调类： │
│ • Manager Agent — 分配任务、协调流程 │
│ • Orchestrator — 编排多个子 Agent │
│ │
│ 执行类： │
│ • Researcher — 信息检索与分析 │
│ • Writer — 内容撰写与编辑 │
│ • Coder — 代码编写与调试 │
│ • Analyst — 数据分析与可视化 │
│ │
│ 审核类： │
│ • Reviewer — 结果审核与质量把控 │
│ • Critic — 提出批评意见 │
│ • Validator — 结果验证 │
│ │
│ 辅助类： │
│ • Planner — 制定执行计划 │
│ • Summarizer — 汇总整理信息 │
│ • Translator — 跨语言转换 │
│ │
└─────────────────────────────────────────────────────────────┘第3部分：协作模式
3.1 模式1：层级模式（Hierarchical）
特点：有一个 Manager 统一指挥
┌─────────────────────────────────────────────────────────────┐
│ 层级协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ │
│ │ Manager │ │
│ │ 管理者 │ │
│ └────┬─────┘ │
│ 分配任务 ↓ 汇总结果 │
│ ┌─────────────┼─────────────┐ │
│ ↓ ↓ ↓ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │Researcher│ │ Writer │ │ Reviewer │ │
│ │ 研究员 │ │ 写手 │ │ 审核员 │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ ↓ ↓ ↓ │
│ └─────────────┼─────────────┘ │
│ 上报结果 ↓ 返回修改 │
│ └──────────┘ │
│ │
│ 流程：Manager 分配 → 各自执行 → 上报 → Manager 决策 │
│ │
└─────────────────────────────────────────────────────────────┘适用场景：任务有明确的主次之分、需要统一协调
3.2 模式2：对等模式（Peer-to-Peer）
特点：Agent 之间直接协作，没有中央指挥
┌─────────────────────────────────────────────────────────────┐
│ 对等协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ │
│ │Researcher│ ←───────→ │ Writer │ │
│ └────┬─────┘ 信息交换 └────┬─────┘ │
│ │ │ │
│ └─────────────┬─────────────────┘ │
│ ↓ │
│ ┌──────────┐ │
│ │ Reviewer │ │
│ └──────────┘ │
│ │
│ 流程：直接协商 → 相互配合 → 共同完成任务 │
│ │
└─────────────────────────────────────────────────────────────┘适用场景：Agent 能力相近、任务需要深度协作
3.3 模式3：流水线模式（Pipeline）
特点：像工厂流水线，每个 Agent 做特定环节
┌─────────────────────────────────────────────────────────────┐
│ 流水线协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入 → [Researcher] → [Analyst] → [Writer] → [Reviewer] → 输出
│ ↓ ↓ ↓ ↓ │
│ 收集信息 分析数据 撰写内容 最终审核 │
│ │
│ 特点： │
│ • 每个 Agent 专做一个环节 │
│ • 输出是下个环节的输入 │
│ • 像流水线一样高效 │
│ │
└─────────────────────────────────────────────────────────────┘适用场景：任务有明确的前后依赖关系
3.4 协作模式对比
┌─────────────────────────────────────────────────────────────┐
│ 协作模式对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ │ 模式 │ 复杂度 │ 适用场景 │ 优点 │
│ ├───────────┼────────┼────────────────────┼───────────────┤
│ │ 层级模式 │ 中 │ 需要统一协调 │ 结构清晰 │
│ │ 对等模式 │ 高 │ 深度协作 │ 灵活 │
│ │ 流水线模式 │ 低 │ 固定流程 │ 高效 │
│ │
└─────────────────────────────────────────────────────────────┘第4部分：CrewAI 实战
4.1 CrewAI 核心概念
┌─────────────────────────────────────────────────────────────┐
│ CrewAI 三要素 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Agent（角色） │
│ • role：角色名称 │
│ • goal：目标 │
│ • backstory：背景 │
│ • tools：可用工具 │
│ │
│ Task（任务） │
│ • description：任务描述 │
│ • expected_output：期望输出格式 │
│ • agent：谁来执行（可选） │
│ │
│ Crew（团队） │
│ • agents：Agent 列表 │
│ • tasks：Task 列表 │
│ • process：执行流程（sequential / hierarchical） │
│ │
└─────────────────────────────────────────────────────────────┘4.2 CrewAI 使用示例
python
from crewai import Agent, Task, Crew
from langchain.tools import Tool
# 定义工具
search_tool = Tool(name="search", func=search_web, description="搜索网络信息")
analyze_tool = Tool(name="analyze", func=analyze_data, description="分析数据")
# 定义 Agent
researcher = Agent(
role="市场研究员",
goal="收集目标公司的最新动态和行业趋势",
backstory="你是一位资深市场分析师，擅长从公开信息中提取关键洞察",
tools=[search_tool]
)
analyst = Agent(
role="数据分析师",
goal="分析收集到的信息，提取关键数据洞察",
backstory="你是一位数据驱动的分析师，擅长用数据讲故事",
tools=[analyze_tool]
)
writer = Agent(
role="商业报告撰写人",
goal="将分析结果撰写成专业的商业报告",
backstory="你是一位经验丰富的商业作家，文章结构清晰、逻辑严谨"
)
reviewer = Agent(
role="质量审核员",
goal="审核报告质量，确保准确性和可读性",
backstory="你是一位严格的编辑，对细节有敏锐的洞察力"
)
# 定义任务
task1 = Task(
description="搜索并整理目标公司最近三个月的 AI 相关动态",
expected_output="结构化的动态列表，包含时间、事件、影响",
agent=researcher
)
task2 = Task(
description="分析研究员收集的信息，提取关键洞察",
expected_output="3-5个关键数据洞察，带数据支撑",
agent=analyst
)
task3 = Task(
description="基于分析师的洞察撰写一份500字商业报告",
expected_output="结构清晰的专业商业报告",
agent=writer
)
task4 = Task(
description="审核报告，检查事实准确性、逻辑清晰度",
expected_output="修改建议列表，或确认通过",
agent=reviewer
)
# 创建团队并执行
crew = Crew(
agents=[researcher, analyst, writer, reviewer],
tasks=[task1, task2, task3, task4],
process="sequential" # 顺序执行
)
result = crew.kickoff()
print(result)4.3 层级模式 CrewAI
python
from crewai import Crew
from crewai.process import HierarchicalProcess
# 层级模式：自动创建 Manager
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[task1, task2, task3],
process=HierarchicalProcess(
manager_llm=ChatOpenAI(model="gpt-4o"), # 指定 Manager LLM
number_of_agents=3
)
)
# Manager 自动分配任务、协调流程
result = crew.kickoff()第5部分：通信协议
5.1 消息传递方式
┌─────────────────────────────────────────────────────────────┐
│ Agent 间通信方式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 方式1：共享消息队列 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent A → [消息队列] → Agent B │ │
│ │ 消息格式：{"from": "A", "content": "...", │ │
│ │ "type": "result"} │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 方式2：共享状态 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ ┌─────────────┐ │ │
│ │ │ 共享状态 │ │ │
│ │ │ state = { │ │ │
│ │ │ resultA, │ │ │
│ │ │ resultB │ │ │
│ │ │ } │ │ │
│ │ └──────┬──────┘ │ │
│ │ ┌───────────┼───────────┐ │ │
│ │ ↓ ↓ ↓ │ │
│ │ Agent A Agent B Agent C │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 方式3：链式传递 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent A → 结果 → Agent B → 结果 → Agent C → 最终结果 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘5.2 通信消息模板
python
from typing import TypedDict
from enum import Enum
class MessageType(Enum):
RESULT = "result" # 任务结果
REQUEST = "request" # 请求协作
FEEDBACK = "feedback" # 反馈意见
STATUS = "status" # 状态更新
class AgentMessage(TypedDict):
from_agent: str
to_agent: str # "all" 表示广播
type: MessageType
content: str
metadata: dict
# 示例消息
message = AgentMessage(
from_agent="researcher",
to_agent="analyst",
type=MessageType.RESULT,
content="已完成市场调研，发现3个关键趋势",
metadata={"source": "web_search", "confidence": 0.9}
)第6部分：多 Agent 的挑战与解决方案
6.1 常见问题
┌─────────────────────────────────────────────────────────────┐
│ 多 Agent 常见问题 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题1：循环依赖 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent A 需要 Agent B 的结果 │ │
│ │ Agent B 需要 Agent A 的结果 │ │
│ │ → 死锁！ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 问题2：状态不一致 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 多个 Agent 各自维护状态 │ │
│ │ Agent A 看到的状态 ≠ Agent B 看到的状态 │ │
│ │ → 结果冲突！ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 问题3：通信开销 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent 之间频繁传递大量数据 │ │
│ │ → 延迟增加、成本上升 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 问题4：缺乏全局视角 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 每个 Agent 只知道自己的任务 │ │
│ │ 缺乏整体优化能力 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘6.2 解决方案
┌─────────────────────────────────────────────────────────────┐
│ 问题解决方案 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题1：循环依赖 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 解决：显式声明依赖关系，框架自动处理拓扑排序 │ │
│ │ 或：引入第三方 Agent 作为协调者 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 问题2：状态不一致 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 解决：使用单一真相来源（Single Source of Truth） │ │
│ │ 所有 Agent 读写同一份共享状态 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 问题3：通信开销 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 解决：批量传递 + 压缩 + 按需同步 │ │
│ │ 避免传递原始数据，传递引用或摘要 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 问题4：缺乏全局视角 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 解决：保留 Manager/Orchestrator 角色 │ │
│ │ 或：定期让所有 Agent 共享状态，全局优化 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘核心总结
总结1：多 Agent 的价值
专业化 → 各自专注擅长领域
并行化 → 效率提升
可组合 → 灵活应对不同任务总结2：三种协作模式
| 模式 | 特点 | 适用场景 |
|---|---|---|
| 层级模式 | Manager 统筹 | 需要协调 |
| 对等模式 | 直接协作 | 深度配合 |
| 流水线模式 | 顺序执行 | 固定流程 |
总结3：CrewAI 三要素
Agent：角色 + 目标 + 背景 + 工具
Task：描述 + 期望输出 + 执行者
Crew：Agent 列表 + Task 列表 + 执行流程章节测试
测试1：单 Agent 局限
单 Agent 的三个主要局限是什么？
测试2：角色定义
Agent 角色的三个核心要素是什么？
测试3：协作模式
什么场景下适合用层级模式？什么场景下适合用流水线模式？
测试4：CrewAI
CrewAI 的三个核心概念是什么？
测试5：多 Agent 问题
Agent 之间的循环依赖应该怎么解决？
参考答案
测试1答案
答案：
- 能力边界：一个 Agent 什么都做，什么都不精
- 串行效率低：任务必须顺序执行
- 专业化不足：不同领域需要不同的知识和 Prompt
测试2答案
答案：Role（角色身份）、Goal（要完成的目标）、Backstory（背景故事，补充专业背景）
测试3答案
答案：
- 层级模式：任务有明确的主次之分、需要统一协调时使用
- 流水线模式：任务有明确的前后依赖关系、每个环节固定时使用
测试4答案
答案：Agent（角色定义）、Task（任务定义）、Crew（团队编排）
测试5答案
答案：显式声明依赖关系让框架自动处理拓扑排序，或引入第三方 Agent 作为协调者打破循环。
相关笔记
- [[05-agent-workflow]] - 单 Agent 工作流是理解多 Agent 的基础
- [[02-framework-evolution]] - CrewAI、AutoGen 等多 Agent 框架
- [[08-real-world-applications]] - 多 Agent 的实际应用场景
下一步学习
- [ ] 阅读 07 - RAG 进阶
学习状态：🟡 开始学习