2874 字
14 分钟
单Agent vs 多Agent:何时使用哪种架构
T47.1 AI Agent 系列专题 - 第三篇(基础概念)✅
一、架构选择的重要性
1.1 背景
在构建 AI Agent 系统时,开发者面临的第一个关键决策是:应该使用单Agent还是多Agent架构?
这个选择会直接影响:
- 系统的复杂度和开发成本
- 任务处理的准确性和效率
- 系统的可扩展性和维护性
- 最终的用户体验
1.2 两种架构的直观对比
┌─────────────────────────────────┬─────────────────────────────────┐│ 单 Agent 架构 │ 多 Agent 架构 │├─────────────────────────────────┼─────────────────────────────────┤│ │ ││ ┌─────────────────┐ │ ┌─────────┐ ┌─────────┐ ││ │ │ │ │ Agent A │ │ Agent B │ ││ │ ┌───────────┐ │ │ └────┬────┘ └────┬────┘ ││ │ │ Single │ │ │ │ │ ││ │ │ Agent │ │ │ └─────┬──────┘ ││ │ └───────────┘ │ │ ┌─────┴─────┐ ││ │ │ │ │ Coordinator│ ││ └────────┬────────┘ │ └─────┬─────┘ ││ │ │ ┌──────┴──────┐ ││ │ │ ┌────┴────┐ ┌────┴────┐ ││ ▼ │ │SubAgent1│ │SubAgent2│ ││ ┌────────┐ │ └─────────┘ └─────────┘ ││ │ 输出 │ │ ││ └────────┘ │ │└─────────────────────────────────┴─────────────────────────────────┘二、单Agent架构详解
2.1 什么是单Agent架构
单Agent架构是指整个系统围绕一个核心Agent构建,所有任务都由这个Agent处理。Agent可能配备多个工具,但决策和执行都集中在同一实体。
2.2 核心特征
单Agent架构特征:├── 单一决策中心├── 共享上下文和记忆├── 工具集合(Tool Use)├── 顺序或并行任务执行└── 统一输出接口2.3 典型实现
class SingleAgent: """单Agent架构示例"""
def __init__(self, llm, tools: List[callable], memory: Memory): self.llm = llm self.tools = {t.name: t for t in tools} self.memory = memory
def process(self, task: str) -> str: """ 处理单一任务 """ # 1. 理解任务 understanding = self.llm.generate(f"理解任务: {task}")
# 2. 规划步骤 plan = self.plan(understanding)
# 3. 顺序执行 results = [] for step in plan: result = self.execute_step(step) results.append(result) self.memory.add_step(step, result)
# 4. 生成最终输出 return self.synthesize(results)
def plan(self, task: str) -> List[dict]: """规划任务步骤""" prompt = f""" 任务:{task} 可用工具:{list(self.tools.keys())}
将任务分解为步骤,并选择每步使用的工具。 """ response = self.llm.generate(prompt) return self._parse_plan(response)
def execute_step(self, step: dict) -> str: """执行单个步骤""" tool_name = step['tool'] params = step['parameters']
if tool_name in self.tools: return self.tools[tool_name](**params) else: return f"Error: Unknown tool {tool_name}"
def process_batch(self, tasks: List[str]) -> List[str]: """ 批量处理任务(并行) """ with ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(self.process, task) for task in tasks] return [f.result() for f in futures]2.4 适用场景
| 场景 | 说明 | 示例 |
|---|---|---|
| 简单问答 | 单一问题快速响应 | FAQ 机器人 |
| 单一工具调用 | 只需使用1-2个工具 | 天气查询、日程管理 |
| 线性流程 | 任务可以顺序执行 | 文章撰写、数据格式化 |
| 资源受限 | 计算资源有限 | 边缘设备部署 |
| 快速原型 | 需要快速验证概念 | POC 开发 |
2.5 优势与劣势
优势:
- 简单直观:架构清晰,易于理解和实现
- 低延迟:无需多Agent通信开销
- 成本可控:单一LLM调用,成本可预测
- 易于调试:问题定位简单,流程透明
劣势:
- 扩展性有限:任务复杂度增加时,Agent负担加重
- 容错性差:单点故障影响整体
- 并行能力弱:难以同时处理多个相关子任务
- 专业性受限:通用Agent在特定领域不如专家Agent
三、多Agent架构详解
3.1 什么是多Agent架构
多Agent架构由多个专业化Agent组成,每个Agent负责特定职责,通过协作完成复杂任务。
3.2 核心特征
多Agent架构特征:├── 专业化分工(Role-based)├── 分布式决策├── Agent间通信协议├── 协调器(Coordinator)├── 共享或私有记忆└── 协作决策机制3.3 协作模式
3.3.1 层级模式(Hierarchical)
┌─────────────────────────────────────┐│ Orchestrator ││ (协调器/编排器) │└───────────────┬─────────────────────┘ │ ┌───────────┼───────────┐ ▼ ▼ ▼┌───────┐ ┌───────┐ ┌───────┐│Agent A│ │Agent B│ │Agent C││(专家1)│ │(专家2)│ │(专家3)│└───────┘ └───────┘ └───────┘class HierarchicalMultiAgent: """层级式多Agent架构"""
def __init__(self, orchestrator, sub_agents: Dict[str, Agent]): self.orchestrator = orchestrator self.sub_agents = sub_agents
def process(self, task: str) -> str: """ 层级处理流程 """ # 1. 协调器分析任务 task_analysis = self.orchestrator.analyze(task)
# 2. 协调器决定分配给哪些子Agent assignments = self.orchestrator.assign(task_analysis)
# 3. 并行执行子任务 results = {} with ThreadPoolExecutor() as executor: futures = { agent_name: executor.submit( self.sub_agents[agent_name].process, assignment['task'] ) for agent_name, assignment in assignments.items() } results = {name: f.result() for name, f in futures.items()}
# 4. 协调器整合结果 return self.orchestrator.synthesize(results)3.3.2 对等模式(Peer-to-Peer)
┌─────────────────────────────────────┐│ Agent Network │├─────────────────────────────────────┤│ ││ ┌───────┐ ┌───────┐ ││ │Agent A│◄──►│Agent B│ ││ └───┬───┘ └───┬───┘ ││ │ │ ││ └─────┬─────┘ ││ ▼ ││ ┌───────┐ ││ │Agent C│ ││ └───────┘ ││ │└─────────────────────────────────────┘class PeerToPeerAgent: """对等式多Agent"""
def __init__(self, agents: List[Agent], communication_protocol): self.agents = {a.name: a for a in agents} self.protocol = communication_protocol
def broadcast(self, sender: str, message: dict): """广播消息给所有Agent""" for name, agent in self.agents.items(): if name != sender: agent.receive(message)
def send_message(self, from_agent: str, to_agent: str, message: dict): """向特定Agent发送消息""" self.protocol.send( from_agent, to_agent, message ) self.agents[to_agent].receive(message)
def consensus(self, topic: str) -> str: """达成共识""" # 收集各Agent的意见 opinions = {} for name, agent in self.agents.items(): opinions[name] = agent.opine(topic)
# 投票或加权平均 return self.protocol.resolve(opinions)3.3.3 雾件模式(Fan-out/Fan-in)
┌─────────────────────────────────────┐│ Fan-out / Fan-in │├─────────────────────────────────────┤│ ││ ┌───────┐ ││ │Router │ ││ └───┬───┘ ││ ┌───────┼───────┐ ││ ▼ ▼ ▼ ││ ┌───────┐┌───────┐┌───────┐ ││ │Agent A││Agent B││Agent C│ ││ └───┬───┘└───┬───┘└───┬───┘ ││ └───────┼───────┘ ││ ▼ ││ ┌───────┐ ││ │Reducer│ ││ └───────┘ ││ │└─────────────────────────────────────┘class FanOutFanInAgent: """雾件模式多Agent"""
def __init__(self, router, workers: List[Agent], reducer): self.router = router self.workers = workers self.reducer = reducer
def process(self, task: str) -> str: """ 雾件模式处理 """ # Fan-out: 路由器分发任务 sub_tasks = self.router.distribute(task, len(self.workers))
# 并行执行 results = [] for worker, sub_task in zip(self.workers, sub_tasks): result = worker.process(sub_task) results.append(result)
# Fan-in: 归约器整合结果 return self.reducer.combine(results)3.4 通信协议设计
from enum import Enumfrom dataclasses import dataclassfrom typing import Any
class MessageType(Enum): REQUEST = "request" RESPONSE = "response" QUERY = "query" RESULT = "result" ERROR = "error" HEARTBEAT = "heartbeat"
@dataclassclass AgentMessage: """Agent间消息格式""" sender: str receiver: str # "broadcast" for all type: MessageType content: Any metadata: dict = None timestamp: float = None
class AgentProtocol: """Agent通信协议"""
def __init__(self): self.inbox = {} # agent_name -> list of messages
def send(self, from_agent: str, to_agent: str, content: Any): """发送消息""" msg = AgentMessage( sender=from_agent, receiver=to_agent, type=MessageType.RESPONSE, content=content, timestamp=time.time() )
if to_agent not in self.inbox: self.inbox[to_agent] = [] self.inbox[to_agent].append(msg)
def broadcast(self, from_agent: str, content: Any, recipients: List[str]): """广播消息""" for recipient in recipients: self.send(from_agent, recipient, content)
def receive(self, agent_name: str) -> List[AgentMessage]: """接收消息""" messages = self.inbox.get(agent_name, []) self.inbox[agent_name] = [] # 清空 return messages3.5 适用场景
| 场景 | 说明 | 示例 |
|---|---|---|
| 复杂工作流 | 多步骤、多角色协作 | 软件开发团队Agent |
| 专业领域 | 需要领域专家 | 法律/医疗咨询 |
| 并行处理 | 独立子任务可并行 | 市场调研报告 |
| 容错要求 | 需要冗余设计 | 关键业务系统 |
| 多职能 | 需要多种能力 | 智能客服+销售+售后 |
3.6 优势与劣势
优势:
- 专业化:每个Agent可成为特定领域专家
- 并行性:独立任务可同时处理
- 容错性:单Agent故障不影响整体
- 可扩展性:新增Agent即可扩展功能
- 模块化:易于独立开发和测试
劣势:
- 复杂度高:架构设计和调试难度大
- 通信开销:Agent间通信增加延迟和成本
- 一致性挑战:多Agent决策可能冲突
- 资源消耗:多Agent需要更多计算资源
- 协调成本:需要有效的协调机制
四、架构选择决策框架
4.1 决策流程图
┌─────────────────────────────────────────────────────────────────┐│ Agent 架构选择决策树 │└─────────────────────────────────────────────────────────────────┘
┌─────────────┐ │ 任务复杂度 │ └──────┬──────┘ │ ┌─────────────────┼─────────────────┐ │ │ │ ▼ ▼ ▼ 简单 (<3步) 中等 (3-7步) 复杂 (>7步) │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │单Agent │ │评估专业性│ │多Agent │ │足够 │ │需求 │ │协作架构 │ └──────────┘ └────┬─────┘ └──────────────┘ │ ┌──────────┼──────────┐ │ │ │ ▼ ▼ ▼ 不需要 需要 复杂 专家 专家 协作 │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ 单Agent │ │多Agent │ │层级+对等混合 │ │+多工具 │ │角色分工 │ │架构 │ └──────────┘ └──────────┘ └──────────────┘4.2 决策检查清单
在选择架构时,请逐项检查:
任务特性
- 任务是否可以分解为独立子任务?
- 子任务是否需要不同领域的专业知识?
- 子任务之间是否有依赖关系?
- 是否需要并行处理提升效率?
系统要求
- 对延迟的容忍度如何?
- 对容错性的要求如何?
- 系统资源的限制(计算力、内存)?
开发考量
- 团队是否有能力维护多Agent系统?
- 是否需要快速原型验证?
4.3 选型建议表
| 因素 | 倾向单Agent | 倾向多Agent |
|---|---|---|
| 任务复杂度 | 低 | 高 |
| 领域多样性 | 单一 | 多领域 |
| 并行需求 | 低 | 高 |
| 容错要求 | 一般 | 高 |
| 响应延迟 | 敏感 | 可容忍 |
| 开发资源 | 有限 | 充足 |
| 扩展预期 | 低 | 高 |
五、实战案例分析
5.1 案例一:智能客服系统
场景:电商平台的智能客服,需要处理咨询、售后、推荐等多种需求。
分析:
- 任务类型多样(咨询、售后、推荐)
- 需要不同专业知识
- 用户期望快速响应
推荐架构:多Agent + 对等通信
┌────────────────────────────────────────┐│ Customer Service │├────────────────────────────────────────┤│ ││ ┌────────┐ ┌────────┐ ┌────────┐ ││ │ 咨询 │ │ 售后 │ │ 推荐 │ ││ │ Agent │ │ Agent │ │ Agent │ ││ └───┬────┘ └───┬────┘ └───┬────┘ ││ │ │ │ ││ └───────────┴───────────┘ ││ │ ││ ┌────┴────┐ ││ │ Router │ ││ └─────────┘ ││ │└────────────────────────────────────────┘5.2 案例二:代码审查助手
场景:帮助开发者审查代码、检测bug、提供优化建议。
分析:
- 任务相对单一
- 需要代码执行和静态分析
- 需要理解多种编程语言
推荐架构:单Agent + 多工具
class CodeReviewAgent: """ 单Agent架构的代码审查助手 """ def __init__(self, llm): self.llm = llm self.tools = [ StaticAnalyzer(), Linter(), SecurityScanner(), CodeFormatter() ]
def review(self, code: str, language: str) -> dict: # 使用单一Agent协调多个工具 # 无需多个专业Agent pass5.3 案例三:金融分析助手
场景:投资银行使用的金融分析助手,需要整合财报分析、市场数据、行业研究等。
分析:
- 多数据源整合
- 专业领域知识要求高
- 需要多维度交叉验证
- 容错要求高
推荐架构:多Agent层级架构
┌─────────────────────────────────────────┐│ 金融分析 Orchestrator │├─────────────────────────────────────────┤│ 任务分解 | 结果整合 | 质量控制 │└───────────────────┬─────────────────────┘ │ ┌───────────────┼───────────────┐ ▼ ▼ ▼┌────────┐ ┌────────┐ ┌────────┐│财报分析│ │市场数据│ │行业研究││ Agent │ │ Agent │ │ Agent │└───┬────┘ └───┬────┘ └───┬────┘ │ │ │ └────────────┴────────────────┘ │ ┌────┴────┐ │ 报告生成 │ │ Agent │ └─────────┘六、架构演进策略
6.1 从单Agent到多Agent的演进路径
阶段1: 单Agent MVP │ │ 发现瓶颈 ▼阶段2: 单Agent + 增强工具 │ │ 复杂度增加 ▼阶段3: 垂直拆分(领域专家) │ │ 协作需求增加 ▼阶段4: 多Agent协作 │ │ 规模扩大 ▼阶段5: 层级+对等混合架构6.2 演进信号
何时应该考虑多Agent:
- 单Agent处理时间超过预期
- 不同任务的准确率差异显著
- 需要集成多个外部API/数据源
- 系统需要支持多用户并发
- 维护单一Agent变得困难
七、总结
| 维度 | 单Agent | 多Agent |
|---|---|---|
| 架构复杂度 | 低 | 高 |
| 开发成本 | 低 | 高 |
| 任务复杂度支持 | 中 | 高 |
| 并行处理能力 | 弱 | 强 |
| 容错性 | 差 | 好 |
| 扩展性 | 差 | 好 |
| 延迟 | 低 | 中-高 |
| 成本 | 低 | 中-高 |
最终建议:
- 起步选单Agent:简单场景先用单Agent,快速验证
- 按需演进:遇到瓶颈再考虑多Agent
- 渐进式拆分:从垂直拆分开始,逐步演进
- 工具优先:优先考虑工具增强,再考虑架构变更
相关阅读:
单Agent vs 多Agent:何时使用哪种架构
https://shenhuanjie.github.io/posts/single-agent-vs-multi-agent-architecture-20260511/