Appearance
65. 多智能体模式（Multi-Agent Pattern）
难度 P1 高频 · 岗位 应用 · 频率 ★★★★☆ · 预计阅读 10 min 公司 OpenAI · Anthropic 相关题 Q60 设计模式 · Q61 反射模式
本题阅读地图
面试场景还原
👔面试官：假设你要做一个 AI 助手，能自动完成「帮我调研 AI 编程工具市场，写一份 50 页的行业分析报告，包含竞品对比、市场规模、技术趋势、投资建议」。单 Agent 能做吗？
🙋♂️我：可能做不完，任务太复杂了，涉及搜索、数据分析、文案写作多个环节。
👔面试官：对，这种跨领域、多步骤、长周期的超复杂任务，单 Agent 很难搞定。这时候需要多智能体模式。你能讲讲多智能体的架构是怎么设计的吗？比如 Manus 是怎么实现的？
🙋♂️我：呃……多个 Agent 分工合作？
👔面试官：可以这么理解。但具体有哪些协作方式？主从式、对等式、流水线式分别适用于什么场景？Agent 之间怎么通信、怎么协调？这些都是多智能体架构的关键问题。
TL;DR 速记
- 多智能体核心价值：处理单 Agent 无法完成的超复杂、跨领域、长周期任务
- 三种协作拓扑：主从式（最常见）、对等式（灵活复杂）、流水线式（顺序执行）
- 角色专业化：每个 Agent 有专属领域（搜索、分析、写作、审核）
- 通信机制：共享状态（黑板）、消息传递、函数调用
- 关键挑战：协调开销、错误传播、调试困难、成本高昂
- 典型产品：Manus（主从式）、MetaGPT（角色分工）、AutoGen（通用框架）
- 选型原则：任务复杂度 > 单 Agent 能力时，考虑多智能体
图解
图 1：多智能体系统架构
图 2：三种协作拓扑对比
详细解析
为什么需要多智能体？
单 Agent 的局限性：
| 局限 | 说明 | 示例 |
|---|---|---|
| 上下文限制 | LLM 上下文窗口有限 | 无法同时处理 50 页报告的全部内容 |
| 能力单一 | 单个 Agent 难以精通所有领域 | 搜索、分析、写作、设计都擅长 |
| 稳定性差 | 复杂任务容易出错 | 长链路任务一步错步步错 |
| 效率低下 | 串行处理慢 | 搜索和写作可以并行 |
| 成本高 | 全用大模型不划算 | 简单子任务可以用小模型 |
多智能体的优势：
- 专业化分工：每个 Agent 精通特定领域
- 并行处理：独立子任务同时执行
- 容错性强：单个 Agent 失败不影响整体
- 可扩展性：新增 Agent 扩展新能力
- 成本优化：不同 Agent 可用不同模型
多智能体的核心架构
三层架构模型：
┌─────────────────────────────────────────────────────────────┐
│ 协调层（Orchestrator） │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 任务理解 │→│ 任务拆解 │→│ 资源分配 │→│ 结果整合 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
↓ 分发任务
┌─────────────────────────────────────────────────────────────┐
│ 执行层（Execution Layer） │
│ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │搜索Agent│ │分析Agent│ │写作Agent│ │设计Agent│ │审核Agent│ │
│ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ │
└─────────────────────────────────────────────────────────────┘
↓ 读写数据
┌─────────────────────────────────────────────────────────────┐
│ 共享层（Shared Layer） │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 任务上下文 │ │ 中间结果 │ │ 执行状态 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘角色设计原则：
| 角色 | 职责 | 常用模式 | 工具 |
|---|---|---|---|
| 规划 Agent | 理解任务、拆解子任务、分配资源 | 规划模式 | 任务管理工具 |
| 搜索 Agent | 信息检索、数据收集 | ReAct + 工具 | 搜索引擎、数据库 |
| 分析 Agent | 数据处理、模式识别、洞察提取 | ReAct | 分析工具、Python |
| 写作 Agent | 内容生成、文案撰写 | 反射模式 | 文档工具 |
| 设计 Agent | 可视化、UI/UX 设计 | 反射模式 | 设计工具 |
| 审核 Agent | 质量检查、合规审查、事实核查 | 反射模式 | 规则引擎 |
| 整合 Agent | 汇总结果、生成最终输出 | 规划模式 | 文档生成 |
协作拓扑：主从式
架构特点：
- 一个 Supervisor（主 Agent）负责协调
- 多个 Worker（子 Agent）负责执行
- Supervisor 决定任务分配和结果整合
- Worker 之间不直接通信，都通过 Supervisor
工作流程：
python
class SupervisorWorkerSystem:
"""主从式多智能体系统"""
def __init__(self):
self.supervisor = SupervisorAgent()
self.workers = {
"researcher": ResearchAgent(),
"analyst": AnalysisAgent(),
"writer": WritingAgent(),
}
async def execute(self, task: str) -> str:
# Supervisor 理解并拆解任务
subtasks = await self.supervisor.decompose(task)
# 分配任务给 Workers
results = {}
for subtask in subtasks:
worker = self.workers[subtask.assigned_role]
result = await worker.execute(subtask)
results[subtask.id] = result
# Supervisor 检查中间结果
if await self.supervisor.need_revision(subtask, result):
revised = await worker.revise(result)
results[subtask.id] = revised
# Supervisor 整合最终结果
final = await self.supervisor.synthesize(task, results)
return final适用场景：
| 场景 | 说明 |
|---|---|
| 任务拆解清晰 | 可以明确分配子任务 |
| 需要中心协调 | 子任务之间有依赖关系 |
| 质量要求高 | Supervisor 可以审核和修正 |
| 团队规模中等 | 5-10 个 Worker 比较合适 |
典型产品：Manus、MetaGPT
协作拓扑：对等式
架构特点：
- 没有中心协调者
- Agent 之间直接通信
- 更灵活，适合动态协作
- 复杂度高，难以调试
工作流程：
python
class PeerToPeerSystem:
"""对等式多智能体系统"""
def __init__(self):
self.agents = {
"agent_a": AgentA(),
"agent_b": AgentB(),
"agent_c": AgentC(),
}
# 注册通信接口
for name, agent in self.agents.items():
agent.register_peers(self.agents)
async def execute(self, task: str) -> str:
# 启动一个 Agent 作为入口
entry_agent = self.agents["agent_a"]
# Agent 自主决定调用哪个其他 Agent
result = await entry_agent.handle(task)
# 结果在 Agent 之间传递，直到完成
return result通信协议示例：
python
class AgentMessage:
"""Agent 间消息格式"""
def __init__(self, sender: str, receiver: str, content: dict, msg_type: str):
self.sender = sender
self.receiver = receiver
self.content = content
self.type = msg_type # REQUEST, RESPONSE, BROADCAST
self.timestamp = datetime.now()适用场景：
| 场景 | 说明 |
|---|---|
| 任务边界模糊 | 难以预先定义子任务 |
| 需要动态协作 | Agent 根据情况自主决定调用谁 |
| 探索性任务 | 不确定最佳路径 |
| 容错要求高 | 单个 Agent 失败可以切换 |
典型产品：AutoGen、CAMEL
协作拓扑：流水线式
架构特点：
- Agent 按顺序排列
- 输出直接作为下一个 Agent 的输入
- 类似工厂流水线
- 适合有明确步骤的任务
工作流程：
python
class PipelineSystem:
"""流水线式多智能体系统"""
def __init__(self):
self.stages = [
DataCollectionAgent(),
DataCleaningAgent(),
AnalysisAgent(),
VisualizationAgent(),
ReportGenerationAgent(),
]
async def execute(self, input_data: str) -> str:
data = input_data
for i, stage in enumerate(self.stages):
print(f"Stage {i+1}: {stage.__class__.__name__}")
data = await stage.process(data)
# 可选：中间结果检查点
if i < len(self.stages) - 1:
await self.checkpoint(data, stage=i)
return data适用场景：
| 场景 | 流水线示例 |
|---|---|
| 数据处理 | 采集 → 清洗 → 转换 → 存储 |
| 内容生产 | 选题 → 资料 → 写作 → 编辑 → 发布 |
| 软件开发 | 需求 → 设计 → 编码 → 测试 → 部署 |
| 报告生成 | 搜索 → 分析 → 撰写 → 排版 → 校对 |
错误处理策略：
python
async def execute_with_recovery(self, input_data: str) -> str:
data = input_data
for i, stage in enumerate(self.stages):
try:
data = await stage.process(data)
except Exception as e:
# 策略 1：重试当前阶段
if await self.can_retry(stage, e):
data = await stage.retry(data)
# 策略 2：回退到上一阶段
elif i > 0:
data = await self.rollback(i-1)
data = await self.stages[i-1].reprocess(data)
# 策略 3：人工介入
else:
raise PipelineError(f"Stage {i} failed: {e}")
return data通信与协调机制
三种通信模式：
| 模式 | 说明 | 优点 | 缺点 |
|---|---|---|---|
| 直接调用 | Agent A 直接调用 Agent B | 简单、延迟低 | 紧耦合、难扩展 |
| 消息队列 | 通过消息中间件通信 | 解耦、异步、可扩展 | 延迟高、复杂 |
| 共享状态 | 通过共享存储通信 | 状态可见、可恢复 | 需要同步机制 |
黑板模式（Blackboard）：
python
class BlackboardSystem:
"""黑板模式：共享状态通信"""
def __init__(self):
self.blackboard = {
"task": None,
"hypotheses": [], # 各 Agent 的假设/结果
"solutions": [],
"status": "idle"
}
self.agents = []
async def execute(self, task: str):
self.blackboard["task"] = task
self.blackboard["status"] = "running"
# 各 Agent 轮询黑板，贡献知识
while not self.is_complete():
for agent in self.agents:
if agent.can_contribute(self.blackboard):
contribution = await agent.contribute(self.blackboard)
self.blackboard["hypotheses"].append(contribution)
return self.blackboard["solutions"]消息总线模式：
python
class MessageBus:
"""消息总线：发布-订阅通信"""
def __init__(self):
self.subscribers = defaultdict(list)
def subscribe(self, event_type: str, agent):
self.subscribers[event_type].append(agent)
async def publish(self, event_type: str, message: dict):
for agent in self.subscribers[event_type]:
await agent.receive(event_type, message)五种设计模式完整对比：
| 模式 | 核心机制 | 复杂度 | 适用场景 | 典型代表 |
|---|---|---|---|---|
| 反射模式 | 生成→检查→优化 | ⭐⭐ | 高精度输出 | 代码生成助手 |
| 工具使用模式 | LLM + API 调用 | ⭐⭐⭐ | 需要实时数据 | 天气查询 Bot |
| ReAct 模式 | 思考-行动循环 | ⭐⭐⭐⭐ | 复杂单任务 | 数据分析助手 |
| 规划模式 | 任务拆解+执行 | ⭐⭐⭐⭐⭐ | 战略级长任务 | Deep Research |
| 多智能体模式 | 多 Agent 协作 | ⭐⭐⭐⭐⭐ | 跨领域超复杂任务 | Manus |
常见踩坑与反例
踩坑 1：过早引入多智能体
错误做法： 单 Agent 能搞定的任务硬上多智能体，增加不必要的复杂度。
正确做法： 从简单模式开始，只有当任务复杂度超过单 Agent 能力时才升级。
踩坑 2：Agent 划分过细
错误做法： 一个简单报告生成任务拆成 10 个 Agent，协调开销超过收益。
正确做法： Agent 数量控制在合理范围（通常 3-7 个），避免过度拆分。
踩坑 3：通信协议不统一
错误做法： 不同 Agent 用不同的消息格式，集成困难。
正确做法： 定义统一的通信协议和接口标准。
踩坑 4：忽视错误传播
错误做法： 一个 Agent 失败导致整个系统崩溃。
正确做法： 实现错误隔离和降级策略（跳过失败 Agent、使用默认值、人工介入）。
踩坑 5：缺少可观测性
错误做法： 多 Agent 协作过程黑盒，出问题无法定位。
正确做法： 完善的日志、追踪、状态可视化（如 LangSmith、Agent 执行看板）。
面试官可能继续追问
追问 1：Manus 的多智能体架构具体是怎么设计的？ 答题要点：Manus 采用主从式架构，有一个中央规划 Agent 负责任务拆解和资源分配，多个专业 Agent（搜索、编码、分析等）并行执行，通过共享状态（黑板模式）通信，最后由整合 Agent 汇总结果。
追问 2：多智能体系统的性能瓶颈在哪里？ 答题要点：协调开销（Supervisor 成为瓶颈）、通信延迟（Agent 间频繁通信）、上下文传递（大量数据在各 Agent 间传递）、LLM 调用成本（多 Agent 意味着多倍成本）。
追问 3：如何处理 Agent 之间的冲突？ 答题要点：中央仲裁（Supervisor 裁决）、投票机制（多数决）、优先级机制（高优先级 Agent 覆盖）、人工介入（复杂冲突人工判断）。
追问 4：多智能体系统如何保证一致性？ 答题要点：共享状态版本控制、事务机制（ACID）、最终一致性设计、冲突检测与解决、定期同步检查点。
面试总结
多智能体模式是 AI Agent 设计模式中最复杂的层级。面试时强调三点：
- 三种协作拓扑：主从式（最常见，适合明确分工）、对等式（灵活，适合探索性任务）、流水线式（顺序执行，适合流程明确任务）
- 核心价值：通过专业化分工和并行处理，解决单 Agent 无法完成的超复杂任务
- 关键挑战：协调开销、通信设计、错误隔离、可观测性——这些是多智能体系统的工程难点
记住：多智能体不是万能药。只有当任务复杂度确实超过单 Agent 能力时，才值得引入多智能体的复杂度。面试时展示你能根据任务特征选择合适的协作拓扑，并意识到多智能体带来的工程挑战。