当前位置：首页 > news >正文

05——多 Agent 架构

news 2026/5/12 3:19:32

多 Agent 架构：什么时候需要，以及如何避免过度设计

开篇：多 Agent 看起来很先进，为什么项目反而更难交付

很多团队在做 Agent 应用时，都会很快走到这个节点：

单 Agent 版本已经能跑
但复杂任务完成率不稳定
于是想通过“多 Agent 分工”提升质量

结果往往是：

架构图变漂亮了，问题没变少
Debug 难度暴涨
Token 成本与时延双双上涨
出错后很难定位到底谁错了

这说明一个现实：

多 Agent 不是能力加法，首先是复杂度乘法。

所以正确姿势不是“能拆就拆”，而是先回答：

业务问题是否真的需要多角色协作
复杂度上升是否有明确收益
你是否具备治理这套系统的能力

一、先做决策：什么时候用单 Agent，什么时候上多 Agent

1.1 优先单 Agent 的场景

满足以下条件时，建议坚持单 Agent：

任务链条短（2~4 步）
主要依赖固定工具集
路由逻辑可规则化
失败可通过简单重试与兜底处理

一句话：
如果流程能写成稳定 Workflow，就别急着引入多 Agent。

1.2 适合多 Agent 的场景

当你出现这些特征，才值得考虑多 Agent：

任务天然多角色（如分析、执行、审校）
子任务知识域差异明显（法务、财务、技术）
单 Agent 提示过长导致性能和稳定性下降
需要并行处理多个独立子问题

二、三种主流多 Agent 模式（从易到难）

2.1 模式 A：单 Agent + 工具集（最推荐起点）

本质：还是单 Agent，只是通过工具扩展能力边界。
优点：简单、稳定、可控。
缺点：当任务复杂到需要多角色审校时会吃力。

2.2 模式 B：Supervisor + Specialists（实战主流）

一个总控 Agent 负责拆解任务，多个专家 Agent 负责子任务。

优点：

分工明确，提示词长度可控
便于替换某个子 Agent 能力

缺点：

协调成本高
状态一致性要求更高

2.3 模式 C：去中心化协作（谨慎使用）

多个 Agent 彼此调用、协商、迭代。
理论上更灵活，实际上极难控制。

不建议初期采用，常见后果是：

执行路径不可预测
无限循环风险
成本失控

三、多 Agent 的核心不是“数量”，而是“协作协议”

你可以把多 Agent 看成“多服务协同系统”。
没有协议，协作必乱。

3.1 必须定义的协作契约

每个 Agent 至少要有：

输入契约（需要哪些上下文字段）
输出契约（结构化 JSON）
责任边界（做什么，不做什么）
失败契约（错误码、是否可重试）

示例（简化）：

{"agent":"review_agent","input_schema":{"draft_answer":"string","citations":"array"},"output_schema":{"pass":"boolean","risk_level":"low|medium|high","feedback":"string"}}

3.2 不要让 Agent 之间传“自然语言段子”

如果 A Agent 给 B Agent 的输入是大段自由文本，
你很难做自动校验与错误定位。

建议传“结构化工单”：

任务 ID
子任务类型
必填字段
上游证据引用

四、状态管理：多 Agent 失败最多的地方

多 Agent 一旦涉及共享状态，就会面临一致性问题。

4.1 三类状态要区分

会话状态：当前用户会话上下文（短期）
任务状态：工作流执行进度（中期）
业务状态：订单、审批、工单等真实业务数据（长期）

原则：

会话状态可缓存，必须有 TTL
任务状态要可恢复（checkpoint）
业务状态必须以源系统为准，不能依赖模型“记忆”

4.2 推荐状态机设计

这套状态机会让你在故障场景下仍可追踪与恢复。

五、上下文共享策略：共享太少不协同，共享太多会污染

多 Agent 不是每个都拿全量上下文。
应按职责最小化共享。

5.1 推荐“分层上下文”

全局层：用户身份、权限、目标任务
任务层：当前子任务必要资料
私有层：Agent 自身推理中间态（默认不外泄）

5.2 避免“上下文污染”

常见问题：

一个 Agent 的错误中间结论被全链路复用
过期信息覆盖最新业务状态

治理策略：

中间结果必须带时间戳和来源
高风险结论必须二次校验
过期上下文自动失效

六、并行与串行：不是越并行越快

并行能降时延，但会提高协调成本与冲突概率。

6.1 适合并行的任务

子任务独立，互不依赖
最终只需聚合结果

6.2 必须串行的任务

下游依赖上游输出
涉及写操作或状态变更

工程建议：

对可变更状态的任务默认串行，
对纯分析任务优先并行。

七、代码示例：Supervisor 模式最小可用实现（Python）

fromtypingimportDict,Any,ListclassSupervisor:def__init__(self,agents):self.agents=agentsdefrun(self,task:Dict[str,Any])->Dict[str,Any]:# 1) 拆解任务plan=self._plan(task)results:List[Dict[str,Any]]=[]# 2) 调度执行（示例为串行，可扩展并行）forstepinplan["steps"]:agent_name=step["agent"]payload=step["payload"]res=self.agents[agent_name].execute(payload)ifnotres.get("ok"):# 失败策略：重试/降级/中止fallback=self._handle_failure(step,res)ifnotfallback.get("ok"):return{"ok":False,"error":fallback}res=fallback results.append({"step":step["id"],"result":res})# 3) 聚合与审校final=self._merge_and_review(results)return{"ok":True,"data":final}def_plan(self,task):return{"steps":[{"id":"s1","agent":"research_agent","payload":task},{"id":"s2","agent":"execution_agent","payload":task},{"id":"s3","agent":"review_agent","payload":task},]}def_handle_failure(self,step,res):return{"ok":False,"code":"STEP_FAILED","step":step["id"],"detail":res}def_merge_and_review(self,results):return{"summary":"done","details":results}