1. 从单兵作战到团队协同多 Agent 系统的架构分水岭过去一年我经手过四个 Agent 相关的项目从最早的单体 Agent 一把梭到后来拆成多 Agent 协作中间踩的坑足够写一本小册子。最深的体会是多 Agent 系统不是把单 Agent 复制几份就完事了它本质上是一次架构范式的迁移。你面对的不再是一个模型加几个工具而是一群有各自职责、需要通信、需要协调、还会互相甩锅的数字员工。这篇文章想聊的核心就是标题里那个词——架构取舍。从超级 Agent到能力平台中间隔着的不是技术栈的堆叠而是一系列非此即彼的选择什么时候该用一个全能 Agent什么时候必须拆成多 Agent拆了之后靠什么通信是共享内存、消息队列还是 MCP 协议编排逻辑放在代码里还是交给模型自己决策这些问题没有标准答案但每一个都有明确的适用边界和代价。我假设读这篇文章的你已经写过至少一个能跑通的 Agent demo知道什么是 tool calling、什么是 ReAct 循环也大概听说过 MCP、Workflow 编排这些词。如果你还在纠结Agent 到底是什么那这篇可能稍微超前了一点。但如果你正卡在我的 Agent 越来越臃肿加一个功能就要改一堆 prompt这个阶段那接下来的内容应该能帮你少走几个月的弯路。整篇会围绕四个层面展开先讲清楚架构选型的底层逻辑再拆解多 Agent 的核心通信与编排机制然后给出一套可复现的落地实现路径最后把我踩过的坑和排查经验整理成速查表。全程不堆概念尽量用我实际项目里的场景说话。2. 架构选型什么时候该拆什么时候该忍2.1 超级 Agent 的天花板在哪里先说结论超级 Agent 的崩溃点通常出现在工具数量超过 15 到 20 个或者单次任务需要跨越 3 个以上领域的时候。这不是玄学是有具体原因的。我做过一个内部测试同一个模型工具集从 5 个逐步加到 30 个任务成功率的变化非常明显。5 到 10 个工具时成功率稳定在 90% 以上到 15 个左右开始出现选错工具的情况超过 20 个之后模型经常在几个功能相近的工具之间反复横跳甚至出现幻觉调用——调用一个根本不存在的工具名。背后的原理其实不复杂。工具描述是要占 context 的每个工具的 name、description、参数 schema 加起来少说 100 到 200 token。20 个工具就是 3000 到 4000 token 的固定开销还没算对话历史。context 一长模型的注意力就被稀释它要在理解用户意图和从一堆工具里挑对的之间分配算力两头都做不好。另一个隐性成本是 prompt 维护。超级 Agent 的系统提示词往往要写几百上千字把每个工具的使用场景、边界、注意事项都塞进去。改一个工具的描述可能影响另一个工具的被选中概率。这种耦合让迭代变得极其痛苦我管它叫prompt 泥球——越滚越大越滚越不敢动。提示如果你现在的 Agent 工具数在 10 个以内任务领域也比较单一别急着拆。过早引入多 Agent 带来的协调开销很可能比它解决的问题还多。2.2 多 Agent 拆分的三个触发信号那什么时候该动手拆我总结了三个比较可靠的信号只要命中两个基本就可以考虑多 Agent 架构了。信号一工具集出现明显的领域聚类。比如你的 Agent 同时管着数据库查询文件处理外部 API 调用报表生成四类工具每类下面又有好几个。这种天然的聚类就是拆分的边界一个领域一个 Agent各管各的工具。信号二任务流程存在明确的阶段划分。像先调研、再分析、后生成报告这种线性流程每个阶段的输入输出都很清晰拆成流水线上的多个 Agent 反而比一个 Agent 从头做到尾更稳。因为每个 Agent 的 context 更干净职责更聚焦。信号三需要不同的模型或不同的权限。有些任务用便宜的小模型就够了有些必须上大模型有些操作需要写权限有些只读就行。这种差异用单 Agent 很难优雅处理拆开之后每个 Agent 可以独立配置模型和权限成本和安全性都好控制。反过来说如果你的任务高度依赖全局上下文比如根据整段对话历史做一个综合判断那拆开反而会让信息在 Agent 之间传递时丢失。这种情况我建议还是保持单 Agent或者用主 Agent 持有全局状态、子 Agent 只做局部处理的混合模式。2.3 一张表看清三种架构的取舍为了让你更直观地做决策我把三种典型架构放在一起对比。这张表是我从实际项目里总结出来的不是理论推演。维度超级 Agent单体多 Agent 协作能力平台Agent as Service适用工具数5-15 个每 Agent 5-10 个无上限按需注册上下文隔离无全局共享按 Agent 隔离完全隔离协调开销低中高调试难度低单点中链路追踪高分布式扩展性差改一处动全身中新增 Agent好插件式典型场景单一领域助手流程化任务企业级能力中台失败模式工具选错、context 溢出通信丢失、死循环服务发现、版本兼容看这张表你会发现架构选择本质是在协调开销和扩展性之间找平衡点。超级 Agent 协调开销最低但扩展性最差能力平台扩展性最好但协调开销最高。多 Agent 协作是中间态也是大多数团队应该停留的位置——除非你确实有平台化的需求否则别一步跨到能力平台那个复杂度不是一般团队扛得住的。3. 多 Agent 的核心机制通信、编排与状态管理3.1 Agent 之间到底怎么说话拆成多 Agent 之后第一个要解决的问题就是通信。我见过太多人一上来就搞消息队列、搞 RPC结果发现根本没必要。通信机制的选择取决于你的 Agent 是紧耦合还是松耦合。紧耦合场景下Agent 之间是流水线关系A 的输出直接是 B 的输入。这种情况最简单的方式就是函数调用式的直接传递——A 返回一个结构化对象编排层把它塞给 B。不需要任何中间件一个 Python 函数调用就搞定。我早期项目里就是这么干的简单直接调试也方便。松耦合场景下Agent 之间是发布订阅关系一个 Agent 的产出可能被多个下游消费或者需要异步处理。这时候才需要引入消息机制。但即便如此我也不建议一上来就上 Kafka、RabbitMQ 这种重型中间件。一个内存队列或者干脆用数据库表做任务队列对大多数项目来说足够了。这里要重点说一下MCPModel Context Protocol。很多人把 MCP 理解成Agent 通信协议其实不太准确。MCP 解决的是Agent 和外部能力工具、数据源之间的标准化接口问题它让工具的描述和调用有了统一格式这样不同的 Agent 可以共享同一套工具定义不用每个 Agent 都重新写一遍。在我的架构里MCP 主要用在能力层——把数据库、API、文件系统这些封装成 MCP Server然后各个 Agent 通过 MCP Client 去调用。Agent 之间的通信我还是用更轻量的方式。注意MCP 的价值在于标准化和复用不在于性能。如果你的工具就三五个自己写函数调用比搭 MCP Server 快得多。别为了用而用。3.2 编排逻辑代码编排 vs 模型编排这是多 Agent 架构里最容易被低估的一个决策点。编排逻辑到底写在代码里还是交给模型自己决策代码编排Workflow 式就是把流程写死先调 AA 的结果给 BB 的结果给 C中间加几个条件判断。LangChain 的 Workflow、各种 DAG 编排工具都是这个思路。优点是可控、可预测、好调试缺点是灵活性差遇到预设流程之外的输入就抓瞎。模型编排Agent 自主决策是给一个协调者 Agent让它根据任务动态决定调用哪些子 Agent、以什么顺序调用。优点是灵活能处理开放式任务缺点是行为不可预测容易陷入死循环调试起来像开盲盒。我的实践经验是主干流程用代码编排局部决策交给模型。比如一个报告生成系统整体流程调研→分析→撰写→审核是固定的用代码串起来但调研这一步内部具体查哪些数据源、查几轮交给调研 Agent 自己决定。这样既有流程的稳定性又有局部的灵活性。具体实现上我通常用一个编排器类来管理流程每个步骤是一个 Agent 调用步骤之间的数据传递用显式的数据结构定义。下面是一个简化后的骨架class Orchestrator: def __init__(self, agents: dict): self.agents agents # {researcher: ..., analyst: ..., writer: ...} def run(self, task: str) - str: # 阶段一调研内部由 researcher agent 自主决策 research_result self.agents[researcher].run(task) # 阶段二分析输入是调研结果 analysis_result self.agents[analyst].run(research_result) # 阶段三撰写带条件重试 for attempt in range(3): draft self.agents[writer].run(analysis_result) if self._quality_check(draft): return draft return draft # 兜底返回最后一版这个骨架看起来简单但里面有几个关键设计每个 Agent 的输入输出都是明确的编排器不关心 Agent 内部怎么实现重试逻辑放在编排层而不是 Agent 内部这样 Agent 保持纯粹质量检查是独立的函数可以随时替换。3.3 状态管理共享还是隔离多 Agent 系统里状态管理是个绕不开的难题。共享状态意味着所有 Agent 读写同一份数据好处是信息不丢失坏处是并发写会冲突而且一个 Agent 的脏数据会污染全局。隔离状态意味着每个 Agent 有自己的 context好处是干净、可并行坏处是信息传递要靠显式的序列化容易丢细节。我的做法是分层状态全局有一个任务状态对象只存关键的结构化信息任务 ID、当前阶段、已完成步骤的摘要每个 Agent 内部有自己的工作状态可以随意折腾但对外只暴露结构化的输出。这样既保证了 Agent 之间的信息传递是可控的又给了每个 Agent 内部足够的自由度。举个具体例子。一个客服多 Agent 系统全局状态里存的是用户 ID、问题分类、已尝试的解决方案列表而处理具体问题的 Agent内部可能维护着完整的对话历史、检索到的知识库片段、推理过程。当它把结果交回编排层时只返回解决方案 置信度 依据摘要而不是把整个内部状态都倒出来。这样下游 Agent 拿到的输入是干净的不会被无关信息干扰。4. 从零搭一套多 Agent 系统可复现的落地路径4.1 环境与依赖准备这一节我尽量给到能直接抄的程度。技术栈我选的是 Python 一个轻量 Agent 框架 MCP 做工具层这套组合是我试过上手最快、坑最少的。基础依赖就几个pip install fastapi uvicorn pydantic httpx pip install mcp # MCP 官方 SDK框架选择上我不绑定具体某个因为核心逻辑是通用的。你可以用 LangChain、可以用 LlamaIndex甚至自己手写一个 ReAct 循环都行。关键是理解架构而不是记 API。目录结构我建议这样组织后面扩展起来不会乱project/ ├── agents/ # 各个 Agent 的实现 │ ├── base.py # Agent 基类 │ ├── researcher.py │ ├── analyst.py │ └── writer.py ├── tools/ # 工具定义 │ ├── mcp_servers/ # MCP Server 实现 │ └── registry.py # 工具注册中心 ├── orchestrator/ # 编排逻辑 │ └── pipeline.py ├── state/ # 状态管理 │ └── task_state.py └── main.py # 入口这个结构的好处是职责清晰Agent 只管自己的推理逻辑工具只管能力封装编排只管流程状态只管数据。任何一块要改不会牵动其他部分。4.2 定义 Agent 基类统一接口是关键多 Agent 系统能不能维护很大程度上取决于 Agent 的接口是否统一。我见过有人每个 Agent 写一套自己的调用方式结果编排层里全是 if-else 判断类型改一个 Agent 要动编排层。这是典型的架构债。我的做法是定义一个极简的基类所有 Agent 都实现同一个run方法from abc import ABC, abstractmethod from pydantic import BaseModel class AgentInput(BaseModel): task: str context: dict {} class AgentOutput(BaseModel): result: str metadata: dict {} confidence: float 1.0 class BaseAgent(ABC): def __init__(self, name: str, tools: list, model: str): self.name name self.tools tools self.model model abstractmethod def run(self, input: AgentInput) - AgentOutput: 所有 Agent 的统一入口 pass这个基类只有三个字段result是给下游用的核心产出metadata放辅助信息比如用了哪些工具、耗时多少confidence是置信度编排层可以据此决定要不要重试或转人工。实操心得confidence这个字段看起来可有可无但在实际系统里非常有用。我让每个 Agent 在输出时自评一个 0 到 1 的置信度编排层设置阈值低于阈值的自动触发重试或者降级处理。这一招把线上事故率降了不少。4.3 用 MCP 封装工具层工具层的核心诉求是一次定义多处复用。MCP 正好解决这个问题。我把数据库查询、外部 API、文件操作都封装成 MCP Server每个 Server 暴露一组标准化的工具。一个最简单的 MCP Server 长这样from mcp.server import Server from mcp.types import Tool, TextContent server Server(database-tools) server.list_tools() async def list_tools(): return [ Tool( namequery_user, description根据用户ID查询用户基本信息, inputSchema{ type: object, properties: {user_id: {type: string}}, required: [user_id] } ) ] server.call_tool() async def call_tool(name: str, arguments: dict): if name query_user: result db.query(arguments[user_id]) return [TextContent(typetext, textstr(result))]封装好之后任何 Agent 只要连上这个 MCP Server就能用这些工具不用重复写。这就是 MCP 的核心价值——把工具从 Agent 里解耦出来变成独立的能力单元。这里有个细节要注意工具的描述要写得像给新人看的说明书。模型选工具全靠 description写得含糊它就会选错。我踩过的坑是查询用户和搜索用户两个工具描述没写清楚区别模型经常混用。后来改成根据精确 ID 查询和根据姓名模糊搜索问题就解决了。4.4 编排层实现把流程串起来编排层是整个系统的大脑。我的实现思路是显式流程 隐式决策主干流程写死在代码里每个节点的内部决策交给 Agent。class Pipeline: def __init__(self, agents: dict, state_store): self.agents agents self.state_store state_store def execute(self, task_id: str, task: str): state self.state_store.create(task_id, task) # 阶段一调研 research self.agents[researcher].run( AgentInput(tasktask, contextstate.summary()) ) state.record(research, research) # 阶段二分析带置信度检查 analysis self.agents[analyst].run( AgentInput(tasktask, contextstate.summary()) ) if analysis.confidence 0.6: # 置信度不足让调研 Agent 补充信息后重试 research self.agents[researcher].run( AgentInput(taskf补充调研{analysis.metadata.get(gap)}) ) state.record(research_supplement, research) analysis self.agents[analyst].run( AgentInput(tasktask, contextstate.summary()) ) state.record(analysis, analysis) # 阶段三撰写 draft self.agents[writer].run( AgentInput(tasktask, contextstate.summary()) ) state.record(draft, draft) return draft.result这段代码里有几个设计点值得说。第一状态通过state.summary()传递而不是把整个 state 对象丢给 Agent这样 Agent 拿到的永远是精简后的关键信息。第二置信度检查触发的补充调研是一个典型的反馈回路让系统有自我修正的能力。第三每个阶段的产出都记录到 state方便事后追溯和调试。4.5 状态存储别小看这一层状态存储我一开始用的是内存字典跑单机 demo 没问题一上生产就出问题——服务重启状态全丢多实例部署状态不共享。后来换成了 Redis问题解决但引入了新的复杂度。我的建议是分阶段开发和测试用内存生产环境用 Redis 或数据库。抽象一个StateStore接口底层实现可替换class StateStore(ABC): abstractmethod def create(self, task_id: str, task: str) - TaskState: ... abstractmethod def load(self, task_id: str) - TaskState: ... class MemoryStateStore(StateStore): def __init__(self): self._store {} def create(self, task_id, task): state TaskState(task_id, task) self._store[task_id] state return state这样切换存储后端只需要换一个实现类业务代码不用动。这个抽象看起来多此一举但等你真的要上生产的时候会感谢自己当初留了这个口子。5. 踩坑实录多 Agent 系统的高频故障与排查5.1 死循环Agent 之间互相踢皮球这是多 Agent 系统最经典的故障。A 把任务交给 BB 觉得不该自己处理又交回 AA 再交给 B无限循环。我遇到过一次两个 Agent 来回传了 40 多轮token 烧了一大笔。根因通常是职责边界不清。每个 Agent 都觉得这不是我的活但又没有明确的拒绝并上报机制只能往下游推。解决办法有两个一是在编排层设置最大轮次超过就强制中断并上报二是给每个 Agent 明确的职责描述和无法处理时的返回格式让它知道该往哪退。MAX_HANDOFFS 10 def execute_with_guard(self, task): handoff_count 0 current_agent self.entry_agent while handoff_count MAX_HANDOFFS: result current_agent.run(task) if result.metadata.get(handoff_to): current_agent self.agents[result.metadata[handoff_to]] handoff_count 1 else: return result raise RuntimeError(f任务在 {MAX_HANDOFFS} 轮内未完成疑似死循环)注意最大轮次这个保护一定要加而且要在编排层加不要指望 Agent 自己控制。Agent 是模型驱动的它没有全局视角不知道自己在循环里。5.2 信息在传递中丢失多 Agent 系统里信息每经过一次传递就可能衰减一次。A 的输出经过序列化、截断、摘要到 C 手里可能已经面目全非。我遇到过一个案例调研 Agent 找到了关键数据但摘要时被压缩掉了导致分析 Agent 得出完全错误的结论。对策是关键信息用结构化字段传递不要靠自然语言摘要。比如调研结果里的数据来源关键数字时间范围这些用明确的字段存而不是塞在一段文字里让下游去解析。自然语言适合表达推理过程结构化字段适合传递事实。信息类型传递方式原因关键事实、数字结构化字段不能被摘要丢失推理过程自然语言需要保留上下文工具调用记录结构化日志用于追溯和调试中间结论结构化 置信度下游据此决策5.3 常见问题速查表下面这张表是我从多个项目里攒出来的基本覆盖了 80% 的故障场景。现象可能原因排查方向解决手段Agent 反复调用同一工具工具描述不清或结果不符合预期看工具调用日志优化 description加结果校验任务中途卡住无响应某个 Agent 超时未返回加超时日志设置单 Agent 超时超时降级输出质量忽高忽低上下文污染或模型不稳定对比不同输入的 context隔离 context加输出校验成本异常飙升死循环或重复调用统计 token 消耗加轮次限制加缓存多实例状态不一致状态存储未共享检查存储后端换 Redis 或数据库工具调用报权限错误Agent 权限配置错误检查工具权限矩阵按 Agent 分配最小权限5.4 几个不那么显然的经验最后分享几个我在文档里看不到、只能靠踩坑攒出来的经验。第一给每个 Agent 起个人名。听起来很傻但当你调试一个五 Agent 系统时researcher 说 analyst 给的数据有问题比Agent A 说 Agent B 的输出异常好理解太多。人名的心理暗示还能帮你更自然地设计职责边界。第二日志要记录完整的调用链。每个 Agent 的输入、输出、耗时、token 消耗、调用的工具全部记下来用同一个 task_id 串起来。出问题的时候你能一眼看出是哪个环节掉的链子。我用的是一张简单的日志表字段就 task_id、agent_name、input、output、duration、tokens够用了。第三先跑通两个 Agent 再扩展。别一上来就设计五个 Agent 的架构先用两个 Agent 把通信、编排、状态这套机制跑通验证没问题了再往上加。我见过太多人设计了完美的五 Agent 架构结果卡在两个 Agent 的通信上整个项目停滞。第四给模型编排留后路。如果你用了模型自主决策的编排方式一定要有一个降级到固定流程的开关。模型抽风的时候切到写死的流程至少能保证系统可用。这个开关平时不用但关键时刻能救命。从超级 Agent 到能力平台这条路我走了差不多一年中间推翻重来过两次。现在回头看最大的收获不是学会了某个框架或协议而是建立了一套判断什么时候该拆、怎么拆、拆完怎么管的思维框架。技术会过时但这套判断逻辑不会。如果你正在这个路口犹豫我的建议是先用单 Agent 把业务跑通等它真的撑不住了再拆拆的时候从两个 Agent 起步把通信和状态这两件事做扎实剩下的都是水到渠成的事。