本文承接长期记忆与 RAG,系统讲解 AI Agent 如何从单智能体进化为多智能体协作系统。文章分析单 Agent 的能力边界,介绍主从、流水线、辩论评审等典型架构,讲解 Agent 间的通信协议、任务分发、角色定义、冲突解决与结果汇总,并结合代码说明如何构建可控、可观测、可扩展的多 Agent 团队。
目录
- 一、单 Agent 的边界与多 Agent 的价值
- 二、多 Agent 的典型架构模式
- 三、Agent 间的通信与消息传递
- 四、任务分发与角色定义
- 五、冲突避免与结果汇总
- 六、可观测性:追踪一个团队的决策链
- 七、成本、安全与权限控制
- 八、上线前检查清单
- 结语
引言
在前四篇文章中,我们分别讨论了 AI Agent 的提示词传输与异常处理、Function Calling、任务规划与多工具编排,以及长期记忆与 RAG。这些能力让一个 Agent 能够独立完成从理解意图、调用工具到记住上下文的完整闭环。
然而,真实世界的任务往往比单个 Agent 能处理的更复杂。一份深度研究报告可能需要同时检索新闻、分析财报、整理竞品并统一文风;一个代码重构任务可能需要一个 Agent 写代码、另一个 Agent 写测试、第三个 Agent 做代码审查。当任务的专业深度、并行度或校验要求超出单个 Agent 的能力范围时,多智能体协作就成为必然选择。
本文将从工程实践出发,系统梳理多 Agent 系统的架构模式、通信机制、任务分发、冲突解决与可观测性设计,并给出可直接参考的代码示例。
一、单 Agent 的边界与多 Agent 的价值
1.1 单 Agent 能做什么,不能做什么
一个具备工具调用、任务规划和长期记忆的 Agent,已经能独立完成相当多的任务。但它存在几类固有边界:
- 专业深度边界:单个 Agent 的系统提示词不可能同时精通法律、财务、医学和工程。让一个通用 Agent 处理跨领域任务,结果往往是”样样通、样样松”。
- 上下文容量边界:即使有 RAG 和记忆压缩,单 Agent 在一次任务中能处理的信息量仍然有限。面对需要同时阅读几十份文档的任务,单 Agent 容易遗漏关键信息。
- 并行能力边界:单 Agent 的执行循环本质上是串行的——规划、执行、观察、再规划。即使工具可以并行调用,决策过程仍然是单点的。
- 自我校验边界:让同一个 Agent 既写代码又审查代码,容易陷入”自己的错误自己看不见”的盲区。独立的评审 Agent 能显著提升输出质量。
1.2 多 Agent 带来什么价值
| 维度 | 单 Agent | 多 Agent |
|---|---|---|
| 专业深度 | 通用能力,难以深耕 | 每个 Agent 专注一个领域 |
| 并行处理 | 决策串行,工具可有限并行 | 多 Agent 可真正并行执行 |
| 质量校验 | 自我校验,容易有盲区 | 独立评审,交叉验证 |
| 上下文利用 | 单次上下文容量有限 | 每个 Agent 独立上下文,信息分流 |
| 故障隔离 | 单点故障影响整个任务 | 单个 Agent 失败可降级或替换 |
| 系统复杂度 | 较低,易于调试 | 较高,需要通信与协调机制 |
多 Agent 的核心价值可以概括为三句话:让专业的人做专业的事,让独立的人互相校验,让并行的人同时推进。
但多 Agent 不是银弹。Agent 数量增加会带来通信开销、协调成本和结果冲突。一个设计良好的多 Agent 系统,应该在”单 Agent 不够用”的临界点引入,而不是为了多而多。
二、多 Agent 的典型架构模式
2.1 主从模式(Orchestrator-Worker)
主从模式是最常见的多 Agent 架构。一个”协调者 Agent”负责理解用户目标、拆解任务、分发给”工作者 Agent”,并汇总结果。
用户目标 → 协调者 Agent → 拆解任务
├→ 工作者 A(检索新闻)
├→ 工作者 B(分析财报)
└→ 工作者 C(整理竞品)
↓
协调者汇总 → 最终输出
主从模式的优点是结构清晰、易于控制,协调者掌握全局视野。缺点是协调者可能成为瓶颈,且工作者之间无法直接通信。
适用场景:研究报告、数据分析、内容生成等需要统一汇总的任务。
2.2 流水线模式(Pipeline)
流水线模式将任务分解为串行的多个阶段,每个阶段由一个专门的 Agent 负责,前一个 Agent 的输出是后一个 Agent 的输入。
用户输入 → 理解 Agent → 检索 Agent → 分析 Agent → 撰写 Agent → 审校 Agent → 最终输出
流水线模式的优点是每个 Agent 职责单一、可独立优化,且可以在阶段之间设置质量门。缺点是无法并行,且任何一个阶段的错误会传递到下游。
适用场景:内容生产流水线(选题→检索→撰写→审校)、数据处理管道等步骤明确的任务。
2.3 辩论与评审模式(Debate / Review)
辩论模式让多个 Agent 从不同角度分析同一问题,通过互相质疑和反驳来逼近更优解。评审模式则是一个 Agent 产出结果,另一个独立 Agent 进行审查并给出修改意见。
方案 A Agent ←→ 方案 B Agent
↓ 互相质疑
裁判 Agent 综合判断
↓
最终方案
辩论模式的优点是能显著减少单一视角的偏见,提高复杂决策的质量。缺点是耗时较长、成本较高,且需要一个中立的裁判来收敛结论。
适用场景:投资决策、方案选型、风险评估等高风险、需要多角度审视的任务。
2.4 平等协作模式(Peer-to-Peer)
平等协作模式中没有明确的协调者,所有 Agent 地位平等,通过共享状态(如黑板)或消息传递来协作。每个 Agent 根据自己的能力和当前状态主动认领任务。
平等协作模式的优点是灵活性高、可扩展性好,Agent 可以动态加入或退出。缺点是协调逻辑复杂,容易出现任务冲突或重复劳动,且难以预测整体行为。
适用场景:开放式探索、创意发散、需要高度灵活性的研究任务。
2.5 架构选择建议
| 架构模式 | 协调复杂度 | 并行度 | 质量保障 | 适用场景 |
|---|---|---|---|---|
| 主从模式 | 中 | 高 | 中 | 研究报告、数据分析 |
| 流水线模式 | 低 | 低 | 高(阶段门) | 内容生产、数据处理 |
| 辩论评审模式 | 高 | 中 | 高 | 决策、选型、风险评估 |
| 平等协作模式 | 高 | 高 | 低 | 开放式探索、创意发散 |
| 混合模式 | 高 | 高 | 高 | 复杂企业级任务 |
实践中最常用的是混合模式:顶层用主从模式协调,某个子任务内部用流水线或辩论模式。例如,一个投资分析系统可以由协调者分发任务,财务分析用流水线,投资决策用辩论模式。
三、Agent 间的通信与消息传递
3.1 消息是 Agent 间的通用语言
无论采用哪种架构,Agent 之间都需要通过消息来传递信息。一个设计良好的消息格式应该包含以下字段:
{
"message_id": "msg_20260908_001",
"from": "researcher",
"to": "analyst",
"type": "task_result",
"task_id": "task_001",
"content": {
"summary": "2026年Q3行业收入同比增长15%",
"data_source": "财报数据",
"confidence": 0.85
},
"metadata": {
"created_at": "2026-09-08T16:00:00Z",
"tokens_used": 1250,
"latency_ms": 3200
}
}
关键字段的作用:
from/to:明确消息的发送者和接收者,便于追踪和权限控制;type:消息类型(任务分配、结果返回、错误通知、状态同步等),接收者据此决定处理逻辑;task_id:关联到具体任务,让多个 Agent 的协作围绕同一个任务上下文展开;content:实际载荷,建议使用结构化 JSON 而非自由文本,便于解析和校验;metadata:追踪信息,包括时间戳、Token 消耗、延迟等,用于可观测性和成本核算。
3.2 三种通信范式
| 通信范式 | 特点 | 适用场景 |
|---|---|---|
| 请求-响应 | 发送者等待接收者回复 | 任务分配、结果查询 |
| 事件驱动 | 发布事件,感兴趣者订阅 | 状态变更通知、异步协作 |
| 共享状态(黑板) | 所有 Agent 读写同一个共享存储 | 平等协作、中间结果共享 |
主从模式通常使用请求-响应:协调者向工作者发送任务,工作者返回结果。流水线模式也是请求-响应的变体:上一阶段完成后触发下一阶段。
平等协作模式更适合事件驱动或共享状态:Agent 在黑板上发布自己的发现,其他 Agent 可以读取并基于此继续工作。
3.3 共享状态的设计
当多个 Agent 需要共享中间结果时,”黑板”(Blackboard)是一种经典模式。所有 Agent 都可以读写黑板上的内容,但需要遵循以下规则:
- 写入时标注来源和置信度:每个 Agent 写入的内容都应附带
agent_id和confidence,避免把不确定的推断当作事实; - 读取时校验时效性:黑板上的内容可能已被其他 Agent 更新,读取时应检查
updated_at; - 冲突时保留多版本:当两个 Agent 对同一事实给出不同结论时,不应直接覆盖,而应保留两个版本并标注冲突,由协调者或裁判来裁决。
一个简化的黑板数据结构:
{
"task_id": "research_001",
"entries": [
{
"key": "industry_growth_rate",
"value": "15%",
"source_agent": "researcher",
"confidence": 0.85,
"updated_at": "2026-09-08T16:05:00Z"
},
{
"key": "industry_growth_rate",
"value": "12%",
"source_agent": "analyst",
"confidence": 0.70,
"updated_at": "2026-09-08T16:10:00Z",
"conflict_with": "entry_001"
}
]
}
四、任务分发与角色定义
4.1 角色不是”人设”,而是能力边界
在多 Agent 系统中,每个 Agent 的角色定义决定了它能做什么、不能做什么、在什么情况下被调用。一个好的角色定义应该包含:
- 名称与描述:用动词加领域的方式命名,例如
financial_analyst、legal_reviewer,避免”助手”、”专家”这类模糊名称; - 能力清单:明确列出该 Agent 可以调用的工具和可以处理的任务类型;
- 输入输出格式:规定该 Agent 接收什么格式的输入、返回什么格式的输出;
- 权限边界:该 Agent 可以访问哪些数据、可以执行哪些操作;
- 失败处理:当该 Agent 无法完成任务时,应该返回什么错误、是否可以降级。
一个角色定义的示例:
{
"agent_id": "financial_analyst",
"name": "财务分析师",
"description": "分析上市公司财报,计算关键财务指标,识别异常变动",
"capabilities": [
"query_financial_statements",
"calculate_ratios",
"identify_anomalies"
],
"input_schema": {
"company_code": "string",
"period": "string",
"focus_areas": ["string"]
},
"output_schema": {
"summary": "string",
"key_metrics": ["object"],
"anomalies": ["object"],
"confidence": "number"
},
"permissions": {
"data_access": ["public_financial_data"],
"write_operations": false
},
"failure_policy": "return_structured_error_with_retry_suggestion"
}
4.2 任务分发的两种策略
任务分发有两种基本策略:
静态分发:协调者根据任务类型直接指定由哪个 Agent 处理。例如,涉及财务分析的任务自动分发给 financial_analyst。静态分发的优点是可预测、易于调试,缺点是不够灵活,当任务类型超出预设范围时可能无法处理。
动态分发:协调者先分析任务,然后根据各 Agent 的能力描述动态选择最合适的 Agent。动态分发的优点是灵活,能处理未预设的任务类型,缺点是选择可能不准确,且需要额外的推理开销。
实践中通常采用混合策略:常见任务类型用静态分发以保证稳定性,复杂或模糊的任务用动态分发以保证灵活性。
4.3 能力注册与发现
当 Agent 数量较多时,协调者不可能硬编码所有 Agent 的能力。可以引入”能力注册表”(Capability Registry),每个 Agent 启动时注册自己的能力,协调者通过查询注册表来发现可用的 Agent。
class CapabilityRegistry:
def __init__(self):
self._agents = {}
def register(self, agent_def):
self._agents[agent_def["agent_id"]] = agent_def
def find(self, task_description):
"""根据任务描述找到最合适的 Agent"""
scored = []
for agent in self._agents.values():
score = self._calculate_match(task_description, agent)
scored.append((score, agent))
scored.sort(key=lambda x: x[0], reverse=True)
return [agent for _, agent in scored[:3]]
def _calculate_match(self, task, agent):
"""简化的匹配评分:关键词重合度 + 能力描述相似度"""
task_words = set(task.lower().split())
desc_words = set(agent["description"].lower().split())
overlap = len(task_words & desc_words)
return overlap / max(len(task_words), 1)
五、冲突避免与结果汇总
5.1 多 Agent 一定会产生冲突
当多个 Agent 独立工作时,冲突是必然的,而不是例外。常见的冲突类型包括:
- 事实冲突:两个 Agent 对同一事实给出不同结论(例如,一个说收入增长 15%,另一个说 12%);
- 方案冲突:两个 Agent 提出不同的解决方案,各有优劣;
- 优先级冲突:两个 Agent 对任务的重要性排序不同;
- 资源冲突:两个 Agent 同时请求同一个有限资源(如同一个工具的调用配额)。
5.2 冲突解决策略
| 冲突类型 | 解决策略 | 说明 |
|---|---|---|
| 事实冲突 | 溯源 + 置信度加权 | 检查数据来源,优先采纳来源更权威、置信度更高的结论 |
| 方案冲突 | 多方案并行 + 评审 | 保留多个方案,由评审 Agent 或用户选择 |
| 优先级冲突 | 协调者裁决 | 由掌握全局的协调者根据任务目标决定 |
| 资源冲突 | 队列 + 限流 | 按优先级排队,或分配独立配额 |
事实冲突的解决尤其重要。一个好的做法是:不直接覆盖,而是保留冲突并标注。协调者在汇总时,如果发现同一事实有多个版本,应该:
- 检查每个版本的数据来源和置信度;
- 如果来源和置信度差异明显,采纳更可信的版本,并在输出中说明;
- 如果差异不明显,保留两个版本并向用户说明存在分歧,由用户决定。
5.3 结果汇总:从碎片到完整答案
当多个工作者 Agent 返回各自的结果后,协调者需要把这些碎片整合成一个完整、一致的答案。汇总过程应该遵循以下原则:
- 按任务结构组织:如果任务被拆解为”检索→分析→撰写”,汇总时也应按这个结构组织结果;
- 去重与合并:多个 Agent 可能返回重叠的信息,需要去重;互补的信息需要合并;
- 一致性检查:检查不同 Agent 的结论是否矛盾,如有矛盾按冲突策略处理;
- 统一文风:不同 Agent 的输出风格可能不同,汇总时应统一为一致的表达方式;
- 标注来源:最终输出中的每个关键结论都应标注来自哪个 Agent、基于什么数据,便于溯源。
一个简化的汇总器示例:
class ResultAggregator:
def __init__(self):
self.results = []
def add(self, agent_id, result):
self.results.append({"agent_id": agent_id, **result})
def aggregate(self, task_goal):
# 1. 按任务目标的结构组织结果
sections = self._organize_by_structure(task_goal)
# 2. 去重与合并
sections = self._deduplicate_and_merge(sections)
# 3. 一致性检查
conflicts = self._check_consistency(sections)
if conflicts:
sections = self._resolve_conflicts(sections, conflicts)
# 4. 统一文风并标注来源
final = self._unify_style(sections)
final = self._annotate_sources(final)
return final
六、可观测性:追踪一个团队的决策链
6.1 多 Agent 的可观测性比单 Agent 更重要
单 Agent 的调试已经不容易,多 Agent 的调试则复杂得多。当最终结果出错时,你需要知道:
- 是哪个 Agent 出了错?
- 是任务拆解错了,还是某个 Agent 的执行错了?
- 是 Agent 间的通信丢了消息,还是汇总时丢了信息?
- 是冲突没有被发现,还是冲突解决策略选错了?
因此,多 Agent 系统必须从第一天就设计可观测性,而不是等出了问题再补。
6.2 关键追踪指标
| 维度 | 指标 | 作用 |
|---|---|---|
| 任务级 | 总耗时、成功率、人工接管率 | 评估整体性能 |
| Agent 级 | 每个 Agent 的调用次数、成功率、平均耗时、Token 消耗 | 识别瓶颈和异常 Agent |
| 通信级 | 消息数量、丢失率、平均延迟 | 评估通信质量 |
| 冲突级 | 冲突发生次数、冲突类型分布、解决耗时 | 评估协作质量 |
| 成本级 | 总 Token 消耗、总费用、单任务平均成本 | 成本控制 |
6.3 全链路追踪的实现
每个任务应该有一个唯一的 trace_id,贯穿所有 Agent 的调用和通信。每个 Agent 的每次执行、每条消息都应该附带这个 trace_id,以及自己的 span_id 和父级 parent_span_id,形成完整的调用链。
{
"trace_id": "trace_abc123",
"span_id": "span_001",
"parent_span_id": null,
"agent_id": "orchestrator",
"action": "task_decomposition",
"input": "分析本月销售下降原因",
"output": ["task_001", "task_002", "task_003"],
"start_time": "2026-09-08T16:00:00Z",
"end_time": "2026-09-08T16:00:02Z",
"tokens_used": 350
}
通过这个调用链,你可以清晰地看到:协调者在什么时间拆解了任务,分发给了哪些 Agent,每个 Agent 花了多长时间、用了多少 Token、返回了什么结果,以及最终是如何汇总的。当结果出错时,可以快速定位到具体的 span。
6.4 回放与调试
可观测性的最终目标是可调试。一个好的多 Agent 系统应该支持”任务回放”:给定一个 trace_id,可以完整复现整个任务的执行过程,包括每个 Agent 的输入输出、每条消息的内容、每次冲突的解决过程。这对于排查问题、优化系统和向用户解释”为什么得到这个答案”都至关重要。
七、成本、安全与权限控制
7.1 多 Agent 的成本是单 Agent 的数倍
多 Agent 意味着更多的模型调用、更多的 Token 消耗、更高的成本。一个 3-Agent 的系统,成本可能是单 Agent 的 2-4 倍(因为协调者也需要调用模型,且 Agent 间通信会产生额外的上下文)。
成本控制的关键策略:
- 设置任务级预算:每个任务有最大 Token 消耗和最大费用上限,超过则终止或降级;
- 设置 Agent 级配额:每个 Agent 有独立的调用次数和 Token 配额,防止单个 Agent 失控;
- 缓存重复调用:相同输入的 Agent 调用结果可以缓存,避免重复计算;
- 按需启动 Agent:不是所有任务都需要全部 Agent,根据任务类型只启动必要的 Agent;
- 小模型做协调,大模型做执行:协调者的任务相对简单,可以用较小的模型;执行复杂分析的 Agent 用大模型。
7.2 权限隔离:每个 Agent 只能看到它该看到的
多 Agent 系统中,权限隔离比单 Agent 更重要。不同的 Agent 可能需要访问不同级别的数据,如果权限隔离不到位,一个低权限 Agent 可能通过协作间接获取高权限数据。
权限控制的基本原则:
- 最小权限:每个 Agent 只能访问完成其任务所必需的数据和工具;
- 数据范围绑定:Agent 的数据访问范围由服务端鉴权决定,不接受 Agent 自己声明的范围;
- 通信过滤:Agent 间传递的消息应经过敏感信息过滤,防止一个 Agent 把敏感数据传递给无权查看的另一个 Agent;
- 操作审计:每个 Agent 的每次数据访问和工具调用都应记录审计日志,包括操作人、操作对象、操作时间和结果。
7.3 高风险操作必须人工确认
与单 Agent 一样,多 Agent 系统中的高风险操作(转账、删除、对外发送、权限变更等)必须经过人工确认。不同的是,多 Agent 系统中可能出现”Agent A 建议操作,Agent B 审核通过,然后系统自动执行”的情况——这仍然不够,Agent 的审核不能替代人的确认。
高风险操作的确认流程应该是:
- Agent 提出操作建议,明确展示操作对象、参数和影响;
- 系统暂停执行,等待用户确认;
- 用户确认后,由服务端校验确认记录与当前请求完全匹配;
- 执行操作并记录审计日志。
八、上线前检查清单
- 是否明确了”什么时候用单 Agent、什么时候用多 Agent”的判断标准?
- 选择的架构模式(主从/流水线/辩论/平等/混合)是否匹配任务特点?
- 每个 Agent 的角色定义是否包含能力清单、输入输出格式和权限边界?
- Agent 间的消息格式是否结构化、可追踪、可校验?
- 是否有任务级和 Agent 级的预算控制,防止成本失控?
- 冲突检测和解决策略是否覆盖了事实冲突、方案冲突和资源冲突?
- 结果汇总是否包含去重、一致性检查、文风统一和来源标注?
- 是否实现了全链路追踪(trace_id + span_id),支持任务回放?
- 每个 Agent 的权限是否遵循最小权限原则,通信是否经过敏感信息过滤?
- 高风险操作是否必须经过人工确认,而非 Agent 间互相审核?
- 是否用真实任务集评估了多 Agent 相比单 Agent 的质量提升和成本增加?
结语
从单 Agent 到多 Agent,不是简单地”多开几个模型实例”,而是从”一个聪明的个体”进化为”一个协作的团队”。团队需要分工、需要通信、需要协调、需要解决冲突、需要有人对最终结果负责。
一个设计良好的多 Agent 系统,应该让每个 Agent 专注于自己最擅长的事,让协调者掌握全局但不陷入细节,让通信高效但不冗余,让冲突被发现而不是被掩盖,让成本可控、权限清晰、结果可追溯。
多 Agent 不是终点。当 Agent 团队能够稳定协作之后,下一个问题是:如何评估一个 Agent(或一个 Agent 团队)的质量?如何知道它的输出是”足够好”的?如何在持续迭代中量化改进?下一篇可以继续讨论 AI Agent 的评测与质量保障:从单轮回答的准确性到长任务的完成率,从人工评估到自动化基准,构建可量化、可对比、可持续改进的 Agent 质量体系。
本文来自投稿,不代表知派立场,如若转载,请注明出处:https://www.zinpai.com/news/4629.html