多智能体协作概述
多智能体协作是构建复杂 AI 系统的基础架构，通过多个 Agent 的分工协作，实现超越单一 Agent 能力边界的任务处理。理解协作模式、设计原则和核心概念是掌握多智能体系统的第一步。
一、核心原理
1.1 多智能体系统的定义
多智能体系统（Multi-Agent System, MAS）由多个自治 Agent 组成，每个 Agent 具有以下特征：
| 特征 | 描述 | 示例 |
|---|---|---|
| 自治性 | Agent 能独立决策和行动 | 自主选择执行哪个任务 |
| 社会性 | Agent 能与其他 Agent 交互 | 通过消息传递协作 |
| 反应性 | Agent 能感知环境并响应 | 根据任务状态调整策略 |
| 主动性 | Agent 能主动追求目标 | 主动请求协作或资源 |
┌─────────────────────────────────────────────────────────────┐
│ 多智能体系统架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ │
│ │ 环境 │ │
│ │ Environment │ │
│ └────────┬────────┘ │
│ │ │
│ ┌───────────────────┼───────────────────┐ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ Agent A │ ←───→ │ Agent B │ ←───→ │ Agent C │ │
│ │ │ │ │ │ │ │
│ │ • 自治 │ │ • 自治 │ │ • 自治 │ │
│ │ • 感知 │ │ • 感知 │ │ • 感知 │ │
│ │ • 决策 │ │ • 决策 │ │ • 决策 │ │
│ │ • 行动 │ │ • 行动 │ │ • 行动 │ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ │ │ │ │
│ └───────────────────┴───────────────────┘ │
│ │ │
│ ┌────────┴────────┐ │
│ │ 通信网络 │ │
│ │ Communication │ │
│ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 协作的本质
协作的本质是信息共享与行动协调：
┌─────────────────────────────────────────────────────────────┐
│ 协作的三个维度 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 信息维度：知识共享 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent A 拥有知识 K1 ──→ 共享 ──→ Agent B 获知 K1 │ │
│ │ │ │
│ │ 示例： │ │
│ │ 搜索Agent 找到数据 → 传递给 → 分析Agent 处理数据 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 任务维度：工作分配 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 总任务 T ──→ 分解 ──→ 子任务 {t1, t2, t3} │ │
│ │ ↓ │ │
│ │ 分配给不同Agent执行 │ │
│ │ │ │
│ │ 示例： │ │
│ │ "开发网站" → 需求分析 + 设计 + 前端 + 后端 + 测试 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. 目标维度：统一方向 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent A 目标 GA ──┐ │ │
│ │ Agent B 目标 GB ──┼──→ 协调 ──→ 共同目标 G │ │
│ │ Agent C 目标 GC ──┘ │ │
│ │ │ │
│ │ 示例： │ │
│ │ 各Agent优化各自目标 → 协调 → 整体最优解 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘二、协作模式详解
2.1 顺序协作（Sequential）
定义：Agent 按照预设顺序依次执行任务，前一个 Agent 的输出作为后一个 Agent 的输入。
┌─────────────────────────────────────────────────────────────┐
│ 顺序协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入 ──→ [Agent A] ──→ [Agent B] ──→ [Agent C] ──→ 输出 │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ 处理1 处理2 处理3 │
│ │
│ 特点： │
│ • 流程固定，易于理解和调试 │
│ • 每个步骤有明确的输入输出 │
│ • 适合有依赖关系的任务 │
│ │
│ 示例：新闻写作流水线 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 选题 │ ─→ │ 撰写 │ ─→ │ 编辑 │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │
│ 适用场景： │
│ • 内容生产流水线 │
│ • 数据处理管道 │
│ • 审批流程 │
│ │
└─────────────────────────────────────────────────────────────┘代码示例：
from crewai import Agent, Task, Crew, Process
# 定义 Agent
researcher = Agent(
role="研究员",
goal="收集和分析信息",
backstory="专业的研究员，擅长信息搜集",
allow_delegation=False
)
writer = Agent(
role="撰稿人",
goal="撰写高质量文章",
backstory="资深作家，文笔流畅",
allow_delegation=False
)
editor = Agent(
role="编辑",
goal="审校和优化文章",
backstory="专业编辑，追求完美",
allow_delegation=False
)
# 定义任务
research_task = Task(
description="研究主题：{topic}",
agent=researcher
)
write_task = Task(
description="根据研究结果撰写文章",
agent=writer
)
edit_task = Task(
description="审校文章",
agent=editor
)
# 创建顺序协作的 Crew
crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, write_task, edit_task],
process=Process.sequential # 顺序执行
)
# 执行
result = crew.run()2.2 并行协作（Parallel）
定义：多个 Agent 同时执行独立任务，最后汇总结果。
┌─────────────────────────────────────────────────────────────┐
│ 并行协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ ┌──→│ Agent A │──┐ │
│ │ └─────────┘ │ │
│ 输入 ──→ [分发器] │ [汇总器] ──→ 输出 │
│ │ ┌─────────┐ │ │
│ └──→│ Agent B │──┘ │
│ └─────────┘ │
│ │
│ 特点： │
│ • 执行效率高，多任务同时进行 │
│ • 任务之间无依赖关系 │
│ • 需要结果汇总机制 │
│ │
│ 示例：多源数据收集 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 天气API │ │ 新闻API │ │ 股票API │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │
│ └──────────────┼──────────────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 汇总Agent │ │
│ └─────────────┘ │
│ │
│ 适用场景： │
│ • 多源数据收集 │
│ • 批量任务处理 │
│ • 多方案对比分析 │
│ │
└─────────────────────────────────────────────────────────────┘2.3 层次协作（Hierarchical）
定义：高层 Agent 负责规划和协调，底层 Agent 负责具体执行。
┌─────────────────────────────────────────────────────────────┐
│ 层次协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Manager │ ← 规划、协调、监督 │
│ │ Agent │ │
│ └──────┬──────┘ │
│ │ │
│ ┌─────────────┼─────────────┐ │
│ ↓ ↓ ↓ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Worker │ │ Worker │ │ Worker │ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ │ (执行者) │ │ (执行者) │ │ (执行者) │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ └─────────────┼─────────────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 结果汇总 │ │
│ └─────────────┘ │
│ │
│ 特点： │
│ • 清晰的职责分层 │
│ • 便于管理和监控 │
│ • 支持动态任务分配 │
│ │
│ 示例：项目管理 │
│ ┌───────────────────────────────────────────────┐ │
│ │ 项目经理Agent │ │
│ │ (规划、分配、监督) │ │
│ └─────────────────────┬─────────────────────────┘ │
│ ┌──────────┼──────────┐ │
│ ↓ ↓ ↓ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │开发Agent│ │测试Agent│ │运维Agent│ │
│ └────────┘ └────────┘ └────────┘ │
│ │
│ 适用场景： │
│ • 大型项目开发 │
│ • 复杂任务分解 │
│ • 企业级应用 │
│ │
└─────────────────────────────────────────────────────────────┘2.4 混合协作（Hybrid）
定义：根据任务特点，灵活组合多种协作模式。
┌─────────────────────────────────────────────────────────────┐
│ 混合协作模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Orchestrator│ ← 总调度 │
│ │ Agent │ │
│ └──────┬──────┘ │
│ │ │
│ ┌─────────────────┼─────────────────┐ │
│ ↓ ↓ ↓ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 顺序链 │ │ 并行组 │ │ 层次结构 │ │
│ │ A → B → C │ │ D ║ E ║ F │ │ G │ │
│ │ │ │ │ │ / \ │ │
│ │ (流水线) │ │ (并行) │ │H I │ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ │ │ │ │
│ └─────────────────┼─────────────────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 结果整合 │ │
│ └─────────────┘ │
│ │
│ 特点： │
│ • 最大灵活性 │
│ • 适应复杂场景 │
│ • 需要精细设计 │
│ │
│ 示例：智能客服系统 │
│ ┌───────────────────────────────────────────────┐ │
│ │ 调度Agent │ │
│ │ (分析问题、路由分发) │ │
│ └─────────────────────┬─────────────────────────┘ │
│ ┌──────────┼──────────┐ │
│ ↓ ↓ ↓ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │FAQ Agent│ │人工客服│ │技术支持│ │
│ │(自动应答)│ │Agent │ │Agent │ │
│ └────────┘ │(人机协作)│ │(专家链) │ │
│ └────────┘ └────────┘ │
│ │
│ 适用场景： │
│ • 企业级应用 │
│ • 复杂业务流程 │
│ • 多场景覆盖 │
│ │
└─────────────────────────────────────────────────────────────┘三、核心组件设计
3.1 Agent 角色设计
每个 Agent 需要明确定义其角色：
| 设计要素 | 说明 | 示例 |
|---|---|---|
| 角色名称 | 标识 Agent 身份 | "数据分析师" |
| 目标 | Agent 追求的结果 | "从数据中提取洞察" |
| 能力 | Agent 具备的技能 | "数据清洗、统计分析、可视化" |
| 工具 | Agent 可用的工具 | "Python解释器、数据库查询" |
| 约束 | Agent 的行为边界 | "不执行删除操作" |
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class AgentRole:
"""Agent 角色定义"""
name: str # 角色名称
goal: str # 目标
backstory: str # 背景故事
skills: List[str] # 技能列表
tools: List[str] # 可用工具
constraints: List[str] # 约束条件
def to_prompt(self) -> str:
"""转换为系统提示词"""
return f"""
你是一个{self.name}。
背景：{self.backstory}
目标：{self.goal}
技能：{', '.join(self.skills)}
可用工具：{', '.join(self.tools)}
约束条件：
{chr(10).join(f'- {c}' for c in self.constraints)}
"""
# 定义角色
data_analyst = AgentRole(
name="数据分析师",
goal="从数据中提取有价值的洞察",
backstory="拥有10年数据分析经验，擅长从复杂数据中发现规律",
skills=["数据清洗", "统计分析", "可视化", "机器学习"],
tools=["Python解释器", "数据库查询", "图表生成器"],
constraints=["不修改原始数据", "不执行删除操作"]
)3.2 任务分配机制
┌─────────────────────────────────────────────────────────────┐
│ 任务分配流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 总任务 │ │
│ └──────┬──────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 任务分解 │ │
│ │ │ │
│ │ 总任务 ──→ 子任务列表 ──→ 依赖关系图 │ │
│ │ │ │
│ │ 示例： │ │
│ │ "分析销售数据" ──→ { │ │
│ │ t1: 数据收集 (无依赖), │ │
│ │ t2: 数据清洗 (依赖t1), │ │
│ │ t3: 统计分析 (依赖t2), │ │
│ │ t4: 生成报告 (依赖t3) │ │
│ │ } │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 能力匹配 │ │
│ │ │ │
│ │ 子任务 ──→ 匹配Agent ──→ 检查负载 │ │
│ │ │ │
│ │ 匹配算法： │ │
│ │ score = skill_match * 0.5 + │ │
│ │ availability * 0.3 + │ │
│ │ success_rate * 0.2 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 执行调度 │ │
│ │ │ │
│ │ 按依赖顺序 ──→ 分配任务 ──→ 监控执行 │ │
│ │ │ │
│ │ 调度策略： │ │
│ │ • 拓扑排序确定执行顺序 │ │
│ │ • 无依赖任务并行执行 │ │
│ │ • 失败任务重新分配或重试 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.3 通信基础设施
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class Message:
"""标准消息格式"""
sender: str # 发送者ID
receiver: str # 接收者ID (broadcast表示广播)
message_type: str # 消息类型
content: Any # 消息内容
timestamp: datetime # 时间戳
metadata: Dict[str, Any] # 元数据
def to_json(self) -> str:
return json.dumps({
"sender": self.sender,
"receiver": self.receiver,
"message_type": self.message_type,
"content": self.content,
"timestamp": self.timestamp.isoformat(),
"metadata": self.metadata
})
@classmethod
def from_json(cls, json_str: str) -> "Message":
data = json.loads(json_str)
return cls(
sender=data["sender"],
receiver=data["receiver"],
message_type=data["message_type"],
content=data["content"],
timestamp=datetime.fromisoformat(data["timestamp"]),
metadata=data["metadata"]
)
class CommunicationBus(ABC):
"""通信总线抽象"""
@abstractmethod
def send(self, message: Message) -> bool:
"""发送消息"""
pass
@abstractmethod
def receive(self, agent_id: str) -> Optional[Message]:
"""接收消息"""
pass
@abstractmethod
def broadcast(self, message: Message) -> bool:
"""广播消息"""
pass
class InMemoryBus(CommunicationBus):
"""内存通信总线（用于单机）"""
def __init__(self):
self.mailboxes: Dict[str, list] = {} # agent_id -> messages
def register(self, agent_id: str):
"""注册 Agent"""
self.mailboxes[agent_id] = []
def send(self, message: Message) -> bool:
if message.receiver not in self.mailboxes:
return False
self.mailboxes[message.receiver].append(message)
return True
def receive(self, agent_id: str) -> Optional[Message]:
if agent_id not in self.mailboxes or not self.mailboxes[agent_id]:
return None
return self.mailboxes[agent_id].pop(0)
def broadcast(self, message: Message) -> bool:
for agent_id in self.mailboxes:
if agent_id != message.sender:
self.mailboxes[agent_id].append(
Message(
sender=message.sender,
receiver=agent_id,
message_type=message.message_type,
content=message.content,
timestamp=message.timestamp,
metadata=message.metadata
)
)
return True四、设计原则
4.1 单一职责原则
每个 Agent 应该只负责一个特定领域：
┌─────────────────────────────────────────────────────────────┐
│ 单一职责原则示例 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ❌ 反模式：全能 Agent │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 超级 Agent │ │
│ │ │ │
│ │ • 搜索信息 │ │
│ │ • 分析数据 │ │
│ │ • 撰写文章 │ │
│ │ • 代码开发 │ │
│ │ • 测试验证 │ │
│ │ • 部署发布 │ │
│ │ │ │
│ │ 问题：职责不清、难以优化、出错难定位 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ✅ 正确模式：职责分离 │
│ ┌───────┐ ┌───────┐ ┌───────┐ ┌───────┐ │
│ │搜索 │ │分析 │ │撰写 │ │发布 │ │
│ │Agent │ → │Agent │ → │Agent │ → │Agent │ │
│ │ │ │ │ │ │ │ │ │
│ │专注 │ │专注 │ │专注 │ │专注 │ │
│ │信息 │ │数据 │ │写作 │ │发布 │ │
│ │检索 │ │处理 │ │创作 │ │运维 │ │
│ └───────┘ └───────┘ └───────┘ └───────┘ │
│ │
│ 优势：职责清晰、易于优化、出错易定位 │
│ │
└─────────────────────────────────────────────────────────────┘4.2 松耦合原则
Agent 之间应该通过接口通信，而不是直接依赖：
| 耦合程度 | 描述 | 问题 |
|---|---|---|
| 紧耦合 | Agent 直接调用另一个 Agent 的方法 | 难以替换、测试困难 |
| 松耦合 | 通过消息队列或共享状态通信 | 易于替换、测试简单 |
4.3 容错原则
系统应该能够处理单个 Agent 的失败：
class FaultTolerantOrchestrator:
"""容错编排器"""
def __init__(self, agents: List[Agent], max_retries: int = 3):
self.agents = {agent.id: agent for agent in agents}
self.max_retries = max_retries
self.fallback_agents = {} # 备用 Agent
def register_fallback(self, primary_id: str, fallback_agent: Agent):
"""注册备用 Agent"""
self.fallback_agents[primary_id] = fallback_agent
async def execute_task(self, task: Task) -> Result:
"""执行任务（带容错）"""
agent_id = self._select_agent(task)
for attempt in range(self.max_retries):
try:
agent = self.agents[agent_id]
result = await agent.execute(task)
return result
except Exception as e:
print(f"Agent {agent_id} 失败 (尝试 {attempt + 1}): {e}")
# 尝试备用 Agent
if agent_id in self.fallback_agents:
try:
fallback = self.fallback_agents[agent_id]
result = await fallback.execute(task)
return result
except Exception as e2:
print(f"备用 Agent 也失败: {e2}")
raise RuntimeError(f"任务执行失败，已尝试 {self.max_retries} 次")五、适用场景
5.1 最佳适用场景
| 场景类型 | 描述 | 推荐模式 |
|---|---|---|
| 软件开发 | 需求、设计、开发、测试协作 | 层次协作（MetaGPT） |
| 内容创作 | 研究者、写作者、编辑协作 | 顺序协作（CrewAI） |
| 数据分析 | 多源数据收集和处理 | 并行协作 |
| 智能客服 | 多场景问题处理 | 混合协作 |
| 项目规划 | 规划、执行、监控 | 层次协作 |
5.2 不适用场景
| 场景 | 原因 | 推荐替代方案 |
|---|---|---|
| 简单任务 | 协作开销大于收益 | 单 Agent + ReAct |
| 实时响应 | 通信延迟难以接受 | 单 Agent 一步执行 |
| 资源受限 | 多 Agent 内存消耗大 | 单 Agent |
| 高度耦合任务 | 难以分解和并行 | 单 Agent + Plan-Execute |
六、面试问答
Q1: 什么时候应该使用多 Agent 而不是单 Agent？
回答要点：
- 任务复杂度超出单 Agent 能力边界
- 需要多种专业能力组合
- 任务可并行化分解
- 需要冗余容错机制
- 任务有明确的子任务分解
Q2: 如何设计 Agent 间的通信协议？
回答要点：
- 定义标准消息格式（发送者、接收者、类型、内容）
- 确定通信模式（同步/异步、点对点/广播）
- 设计消息压缩和过滤机制
- 处理消息丢失和重复
Q3: 多 Agent 系统的可扩展性如何保证？
回答要点：
- 模块化设计，Agent 可独立部署
- 松耦合架构，通过消息通信
- 动态注册机制，支持 Agent 动态加入
- 负载均衡，避免单点瓶颈
Q4: 如何评估多 Agent 协作效果？
回答要点：
- 任务完成率
- 协作效率（通信次数 / 任务完成时间）
- 资源利用率
- 冲突发生频率
- 容错能力
七、小结
| 概念 | 一句话总结 | 面试关键词 |
|---|---|---|
| 协作模式 | 四种模式应对不同场景 | 顺序、并行、层次、混合 |
| 角色设计 | 明确定义 Agent 身份和职责 | 目标、能力、工具、约束 |
| 任务分配 | 能力匹配和负载均衡 | 分解、匹配、调度 |
| 设计原则 | 单一职责、松耦合、容错 | SOLID、可扩展 |
一句话总结：多智能体协作通过明确的角色分工和协调机制，实现了复杂任务的分解和并行处理，是构建企业级 AI 系统的核心架构。
最后更新：2026年3月19日