1. 从一次失败的单体 Agent 上线说起去年下半年我接手了一个内部知识库问答助手的改造项目。需求听起来不复杂用户用自然语言提问Agent 自主决定是查文档、调接口还是直接回答。团队一开始信心满满选了一个当时口碑不错的 Agent 框架用 ReAct 范式把工具列表一挂Prompt 一写Demo 跑得飞起。演示那天老板问了三个问题前两个答得漂亮第三个涉及跨部门数据对比Agent 开始反复调用同一个检索工具循环了七八次最后吐出一句“我无法完成这个任务”。那次之后我们复盘了很久。问题不在于模型不够强也不在于工具写得烂而在于我们把一个本质上需要多角色协作的复杂任务硬塞进了一个单体 Agent 的循环里。它既要理解意图又要规划步骤还要执行工具、校验结果、组织语言所有职责压在一个上下文窗口里一旦任务链条变长它就开始“精神分裂”——忘了自己刚才查过什么或者陷入无意义的重复。这篇文章想聊的就是这件事单体 Agent 的能力边界到底在哪里为什么复杂任务几乎必然走向 Multi-Agent 架构以及如果你现在要动手搭一套 Multi-Agent 系统哪些坑是绕不过去的。适合已经写过至少一个能跑通的 Agent Demo、正在被复杂任务折磨的开发者也适合还在观望要不要上多智能体的技术负责人。我会尽量把原理讲透把代码和配置给到能直接抄的程度同时把我在实际项目里踩过的坑摊开来说。2. 单体 Agent 的能力天花板究竟卡在哪2.1 上下文窗口不是“越大越好”的万能药很多人第一反应是上下文窗口不够那就换更大的模型呗。128K 不够上 200K200K 不够等 1M。这个思路在单体 Agent 场景下有个致命误区——上下文长度和有效注意力是两回事。我做过一个粗糙但很有说服力的测试让单体 Agent 处理一个需要调用 6 个不同工具、总共 15 步的任务。当我把历史对话完整保留时模型在第 10 步之后开始出现明显的“遗忘”它会重新调用第 3 步已经调用过的工具理由是“我需要确认一下”。而当我把历史压缩成摘要再喂进去它又会丢失一些关键的中间结果导致最终答案缺斤少两。这背后的机制其实不复杂。Transformer 的注意力机制在处理长序列时早期 token 的权重会被稀释。你塞进去的 100K token 里真正对当前决策有用的可能只有 2K但模型没法自动帮你做这个筛选。单体 Agent 的所有状态——目标、计划、工具返回、中间推理——全挤在同一个序列里互相干扰。提示判断你的任务是否超出单体 Agent 舒适区有个简单信号——如果任务步骤超过 8 步或者需要同时维护 3 个以上不同来源的状态就该考虑拆分了。2.2 角色混淆一个 Agent 演不了整台戏单体 Agent 最隐蔽的问题不是能力不足而是角色混淆。当你给一个 Agent 同时下达“你是严谨的数据分析师”和“你是富有创造力的文案”这两个指令时它会在两种人格之间摇摆输出的东西既不够严谨也不够有创意。我在做内容生成 Agent 时深有体会。需求是先分析用户提供的产品数据再写一段营销文案。单体 Agent 的做法是分析到一半突然开始写文案写文案写到一半又回头补数据分析。最后交付的东西逻辑断裂数据引用和文案风格完全不搭。这不是 Prompt 写得不好而是单一 Agent 缺乏角色隔离机制。它的系统提示词是一个大杂烩所有职责混在一起模型在每一步都要重新判断“我现在该以什么身份说话”。而 Multi-Agent 的核心价值之一就是让每个 Agent 只专注一个角色用独立的系统提示词和独立的上下文把职责边界划清楚。2.3 错误累积与循环陷阱单体 Agent 还有一个要命的问题错误会沿着推理链一路累积而且它自己很难跳出来。举个真实例子。我让 Agent 查某只股票近 30 天的收盘价然后计算波动率。第一步它调用了行情接口返回的数据格式和预期不符字段名变了。单体 Agent 的反应是重试同一个接口参数不变。重试三次失败后它开始“编造”数据用一些看起来合理但完全错误的数字继续往下算。最终输出的波动率毫无意义但 Agent 自己浑然不觉。为什么会这样因为单体 Agent 的“自我校验”能力受限于同一个上下文。它没有独立的“质检员”角色来审视自己的输出只能靠自己判断“我做得对不对”而模型对自己刚生成的内容往往过于自信。Multi-Agent 架构里你可以专门设一个 Verifier Agent它的唯一职责就是挑毛病而且它的上下文里没有“我努力过了”这种情绪包袱判断会更客观。3. Multi-Agent 到底解决了什么本质问题3.1 职责分离带来的上下文净化Multi-Agent 最直接的好处是每个 Agent 的上下文窗口只装自己关心的东西。还是拿那个知识库问答的例子。拆成 Multi-Agent 之后架构变成这样一个 Router Agent 负责判断用户意图决定走哪条路径一个 Retrieval Agent 只负责查文档它的上下文里只有查询语句和检索结果一个 Analysis Agent 只负责对比分析它拿到的是 Retrieval Agent 整理好的结构化数据最后还有一个 Writer Agent 负责组织语言。每个 Agent 的上下文都很“干净”没有无关信息干扰。Retrieval Agent 不需要知道最终答案要写成什么风格Writer Agent 也不需要知道检索时用了什么关键词。这种隔离带来的效果是立竿见影的——在我们的实测中同一个任务单体 Agent 的完成率是 47%拆成 4 个 Agent 后提升到 82%。3.2 并行化让该同时干的事同时干单体 Agent 是严格串行的一步做完才能做下一步。但很多复杂任务里有些子任务之间没有依赖关系完全可以并行。比如做竞品分析需要抓取 A、B、C 三家公司的公开信息然后对比。单体 Agent 会依次抓取耗时是三者之和。Multi-Agent 可以同时启动三个 Fetcher Agent各自抓一家最后汇总给 Analyzer。在我们的测试里这种并行化能把整体耗时压缩 60% 以上。这里有个实操细节并行 Agent 的调度不是简单开三个线程就完事。你需要一个 Orchestrator 来管理它们的生命周期处理超时、重试和结果聚合。Python 里可以用asyncio.gather配合超时控制但要注意 Agent 之间的状态隔离别让一个 Agent 的异常拖垮整个流程。3.3 专业化分工与工具集收敛单体 Agent 通常挂着一大堆工具检索、计算、API 调用、文件操作全在一起。工具越多模型选错工具的概率越大。我见过一个 Agent 挂了 20 多个工具结果它在该用计算器的时候去调了搜索引擎。Multi-Agent 的思路是每个 Agent 只挂自己真正需要的工具。Retrieval Agent 只有检索工具Calculator Agent 只有计算工具Code Agent 只有代码执行工具。工具集收敛之后选择准确率大幅提升。我们的数据是工具数量从 18 个降到每个 Agent 平均 3 个工具调用准确率从 71% 升到 94%。3.4 可观测性与可调试性单体 Agent 出问题时你面对的是一个巨大的日志流很难定位是哪一步的推理出了偏差。Multi-Agent 天然把流程切成了多个节点每个节点的输入输出都清晰可查。我们后来在系统里加了一个 Trace 模块记录每个 Agent 的每次调用输入是什么、输出是什么、耗时多少、是否触发重试。出问题时直接看哪个节点的输出异常定位时间从原来的平均 40 分钟缩短到 5 分钟以内。这个收益在系统上线后的运维阶段尤其明显。4. 拆解一个可落地的 Multi-Agent 架构4.1 角色划分别为了多而多Multi-Agent 最常见的误区是为了显得架构先进而硬拆。我见过一个团队把“查天气”这种任务拆成三个 Agent一个解析城市名一个调天气 API一个格式化输出。这纯属自找麻烦。角色划分的原则应该是当一个子任务需要独立的上下文、独立的工具集或者需要不同的“人格”时才拆成独立 Agent。具体来说我通常按这几个维度判断判断维度适合拆分的信号不适合拆分的信号上下文需求子任务需要大量专属信息子任务共享同一批信息工具集子任务有专属工具工具高度重叠角色定位需要不同专业视角同一视角的不同步骤并行可能子任务之间无依赖严格串行依赖校验需求需要独立质检自我校验足够按这个标准一个典型的复杂任务 Multi-Agent 架构通常包含这几类角色Orchestrator编排者负责拆解任务和调度Specialist Agents专家 Agent各自负责一个领域Verifier校验者负责质量把关Aggregator聚合者负责汇总输出。小规模场景下Orchestrator 和 Aggregator 可以合并Verifier 也可以由 Orchestrator 兼任但 Specialist 一定要拆开。4.2 通信机制消息传递还是共享状态Multi-Agent 的通信方式主要有两种消息传递和共享状态。消息传递是每个 Agent 完成工作后把结果作为消息发给下一个 Agent。这种方式解耦彻底每个 Agent 不需要知道全局状态适合流程固定的场景。缺点是如果流程需要回溯或跳转消息链会变得复杂。共享状态是维护一个全局的 State 对象所有 Agent 读写同一个状态。这种方式灵活适合需要动态调整流程的场景但要注意并发读写的问题。Python 里可以用dataclass定义 State配合锁机制保证一致性。我的建议是流程相对固定的用消息传递需要动态决策的用共享状态。实际项目里往往是混合使用——Orchestrator 维护一个轻量级的全局状态Agent 之间通过消息传递具体数据。from dataclasses import dataclass, field from typing import Any dataclass class AgentState: task: str intermediate_results: dict field(default_factorydict) current_step: str errors: list field(default_factorylist) final_output: Any None这个 State 结构很朴素但够用。关键是intermediate_results用字典存每个 Agent 往里写自己的产出键名用 Agent 名避免冲突。4.3 编排模式从顺序到动态路由编排模式决定了 Agent 之间怎么协作。常见的有四种顺序编排最简单Agent A 做完给 BB 做完给 C。适合流水线式的任务比如“抓数据 → 清洗 → 分析 → 写报告”。路由编排由一个 Router 根据输入决定走哪条分支。比如用户问技术问题走 Technical Agent问业务问题走 Business Agent。适合意图分类明确的场景。并行编排多个 Agent 同时工作最后汇总。适合子任务独立的场景比如多源数据抓取。动态编排最灵活Orchestrator 根据当前状态实时决定下一步调哪个 Agent。适合流程不确定、需要根据中间结果调整策略的复杂任务。实现上通常是一个循环每轮让 Orchestrator 输出下一步动作直到任务完成。async def dynamic_orchestrate(task: str, max_rounds: int 10): state AgentState(tasktask) for _ in range(max_rounds): next_action await orchestrator.decide(state) if next_action FINISH: break agent agent_registry[next_action.agent_name] result await agent.run(state, next_action.instruction) state.intermediate_results[next_action.agent_name] result return state.final_output这段代码是骨架实际用的时候要加超时、重试和错误处理。max_rounds是防止死循环的保险丝我一般设 10 到 15根据任务复杂度调整。4.4 工具与记忆的隔离策略每个 Agent 的工具集要独立配置别搞一个全局工具池让所有 Agent 共享。记忆也一样短期记忆当前任务的上下文每个 Agent 独立长期记忆跨任务的知识可以共享但要有命名空间隔离。我们用的是按 Agent 名分区的向量库每个 Agent 只能检索自己分区的内容。这样既避免了信息污染又保留了跨任务学习的可能。实现上就是在写入时给每条记忆打上agent_name标签检索时加过滤条件。5. 用 Python 搭一套最小可用的 Multi-Agent 系统5.1 环境准备与依赖选择先说环境。Python 3.10 以上推荐 3.11因为asyncio的改进对并发 Agent 很友好。核心依赖就几个pip install openai1.0.0 pip install pydantic2.0 pip install asyncio pip install tenacityopenai是模型调用pydantic用来定义 Agent 的输入输出结构强类型能省很多调试时间tenacity做重试。别一上来就上 LangChain 或 AutoGen 这类重框架先用原生 API 把核心逻辑跑通理解每一步在干什么之后再考虑要不要用框架提效。注意如果你用的是其他模型服务把openai换成对应的 SDK 即可接口逻辑大同小异。关键是保持 Agent 的抽象层和模型调用层解耦方便后续换模型。5.2 定义 Agent 基类与消息协议先定义一个 Agent 基类把通用逻辑抽出来from abc import ABC, abstractmethod from pydantic import BaseModel class AgentMessage(BaseModel): sender: str receiver: str content: str metadata: dict {} class BaseAgent(ABC): def __init__(self, name: str, system_prompt: str, tools: list None): self.name name self.system_prompt system_prompt self.tools tools or [] self.memory [] abstractmethod async def run(self, state: AgentState, instruction: str) - str: pass def _build_messages(self, instruction: str) - list: messages [{role: system, content: self.system_prompt}] messages.extend(self.memory[-5:]) # 只保留最近5轮防止上下文膨胀 messages.append({role: user, content: instruction}) return messages这里有个关键设计memory只保留最近 5 轮。为什么是 5我试过 3、5、105 是在效果和成本之间的平衡点。太少会丢失必要上下文太多会引入噪声。当然这个数字要根据你的任务调整但原则是每个 Agent 的记忆要精简别把整个对话历史都塞进去。5.3 实现一个检索 Agent 和一个分析 Agent检索 Agent 的职责很单一拿到查询语句调检索工具返回结构化结果。class RetrievalAgent(BaseAgent): async def run(self, state: AgentState, instruction: str) - str: query instruction results await self._search(query) formatted self._format_results(results) self.memory.append({role: assistant, content: formatted}) return formatted async def _search(self, query: str) - list: # 实际项目里替换成你的检索实现 # 这里用伪代码示意 return await vector_store.search(query, top_k5) def _format_results(self, results: list) - str: lines [] for i, r in enumerate(results, 1): lines.append(f[{i}] {r[title]}: {r[content][:200]}) return \n.join(lines)分析 Agent 拿到检索结果做对比分析class AnalysisAgent(BaseAgent): async def run(self, state: AgentState, instruction: str) - str: context state.intermediate_results.get(retrieval, ) prompt f基于以下资料进行分析\n{context}\n\n分析要求{instruction} response await self._call_llm(prompt) self.memory.append({role: assistant, content: response}) return response async def _call_llm(self, prompt: str) - str: messages self._build_messages(prompt) # 调用模型API return await llm_client.chat(messages)这两个 Agent 的实现都很朴素但组合起来已经能处理“查资料 分析”这类任务了。关键是每个 Agent 只做一件事上下文干净工具专一。5.4 编排器与状态流转的代码骨架编排器是整个系统的大脑它决定什么时候调哪个 Agentclass Orchestrator: def __init__(self, agents: dict): self.agents agents self.planner_prompt 你是一个任务编排器。根据当前状态决定下一步调用哪个Agent。 可用Agent{agent_list} 当前任务{task} 已完成步骤{completed} 请输出JSON格式{{next_agent: agent_name, instruction: 具体指令}} 如果任务已完成输出{{next_agent: FINISH}} async def decide(self, state: AgentState) - dict: prompt self.planner_prompt.format( agent_listlist(self.agents.keys()), taskstate.task, completedlist(state.intermediate_results.keys()) ) response await llm_client.chat([{role: user, content: prompt}]) return json.loads(response) async def run(self, task: str, max_rounds: int 12) - str: state AgentState(tasktask) for round_num in range(max_rounds): decision await self.decide(state) if decision[next_agent] FINISH: break agent self.agents[decision[next_agent]] try: result await asyncio.wait_for( agent.run(state, decision[instruction]), timeout30 ) state.intermediate_results[decision[next_agent]] result except asyncio.TimeoutError: state.errors.append(f{decision[next_agent]} 超时) continue return state.intermediate_results.get(writer, 任务未完成)这段代码有几个实操要点。asyncio.wait_for给每个 Agent 设了 30 秒超时防止某个 Agent 卡死拖垮全局。超时后记录错误并继续而不是直接崩溃。max_rounds设 12 是经验值大部分任务 6 到 8 轮能完成留点余量。6. 实测中那些文档不会告诉你的坑6.1 Agent 之间的“踢皮球”现象Multi-Agent 上线第一周我们遇到一个诡异问题任务在 Retrieval Agent 和 Analysis Agent 之间来回跳谁也不肯出最终结果。日志显示 Retrieval 说“资料已提供请分析”Analysis 说“资料不足请补充检索”循环了 9 轮。根因是两个 Agent 对“资料是否充足”的判断标准不一致。Retrieval 觉得返回了 5 条结果就算充足Analysis 觉得这 5 条里只有 2 条相关不算充足。而 Orchestrator 没有仲裁机制只能看着它们互相推诿。修复方案是加一个仲裁规则当同一个 Agent 被连续调用超过 2 次Orchestrator 强制进入下一阶段或者触发 Verifier 做裁决。具体实现是在 State 里加一个调用计数器state.call_counts[agent_name] state.call_counts.get(agent_name, 0) 1 if state.call_counts[agent_name] 2: # 强制推进或触发仲裁这个坑的教训是Multi-Agent 系统必须有防死循环机制而且不能只靠 max_rounds 这种粗粒度的限制要针对单个 Agent 的调用频率做监控。6.2 上下文传递中的信息损耗Agent A 的输出传给 Agent B 时如果直接传原始文本B 可能抓不住重点。我们试过让 A 输出结构化 JSONB 解析后使用效果好很多。但这里有个新问题JSON 的字段设计如果太死会限制 A 的表达。比如 A 发现了一个预期外的信息但 JSON schema 里没有对应字段它只能丢弃。后来我们改成“核心字段 扩展字段”的混合模式核心字段强类型扩展字段用extra: dict兜底。class RetrievalResult(BaseModel): query: str documents: list[dict] confidence: float extra: dict {} # 兜底放预期外的发现6.3 模型调用成本失控的三种典型场景Multi-Agent 的成本比单体 Agent 高这是事实。但失控往往不是因为 Agent 多而是因为这三个场景场景一Orchestrator 每轮都调模型。如果任务有 10 轮Orchestrator 就调了 10 次模型每次都要传完整状态。优化方法是让 Orchestrator 用更便宜的模型或者用规则引擎处理简单决策只在复杂分支才调模型。场景二Agent 重试没有上限。一个 Agent 失败后重试 5 次每次都是完整的模型调用。我们后来给重试加了指数退避和最大次数限制超过 3 次就上报错误不再重试。场景三记忆无限增长。前面提到每个 Agent 只保留最近 5 轮记忆但有些 Agent 的intermediate_results会越积越多。我们加了一个清理机制每个 Agent 完成后只保留它输出的摘要原始详细结果存到外部存储需要时再取。6.4 调试 Multi-Agent 的实用工具链调试 Multi-Agent 比调试单体 Agent 复杂因为你要追踪多个 Agent 的状态。我们最后搭了一套简单的 Trace 系统每个 Agent 的每次调用记录agent_name、input、output、duration、token_usage用trace_id串联同一次任务的所有调用输出成 JSON Lines 格式方便用jq或写脚本分析import time import json def trace(agent_name: str, trace_id: str): def decorator(func): async def wrapper(*args, **kwargs): start time.time() result await func(*args, **kwargs) log_entry { trace_id: trace_id, agent: agent_name, duration: time.time() - start, output_preview: str(result)[:200] } with open(agent_trace.jsonl, a) as f: f.write(json.dumps(log_entry) \n) return result return wrapper return decorator这个装饰器很简陋但足够定位大部分问题。上线后我们靠它发现了不少隐蔽的 bug比如某个 Agent 在特定输入下会返回空字符串导致下游 Agent 拿到空上下文后开始胡编。7. 什么时候不该上 Multi-Agent聊了这么多 Multi-Agent 的好处但必须说清楚它不是银弹很多场景下单体 Agent 更合适。如果你的任务满足以下条件老老实实用单体 Agent任务步骤少于 5 步不需要多种专业视角工具集不超过 5 个对延迟敏感Multi-Agent 的多次模型调用天然更慢团队还没有调试 Multi-Agent 的基础设施。我见过最离谱的案例是一个团队把“根据用户输入生成一句问候语”拆成了三个 Agent意图识别、风格选择、文本生成。结果延迟从 800ms 涨到 3 秒成本翻了 4 倍效果还不如直接让一个 Agent 干。判断标准其实很简单当你发现单体 Agent 的失败原因主要是“上下文太杂”或“角色冲突”时才考虑 Multi-Agent。如果失败原因是模型能力不足或工具本身有问题拆成再多 Agent 也没用。另外Multi-Agent 的引入应该是一个渐进过程。我的建议路径是先用单体 Agent 跑通核心流程记录失败案例当失败案例中超过 30% 是上下文或角色问题导致的再开始拆分拆分时先拆最痛的那个环节验证效果后再逐步扩展。别一上来就设计一个五六个 Agent 的复杂架构那样调试成本会让你怀疑人生。8. 从单体到多体的迁移路线图如果你已经决定要迁移这里给一个我实际用过的路线图分四个阶段。第一阶段埋点与诊断。在现有单体 Agent 里加详细的日志记录每次失败的具体原因。分类统计多少是上下文溢出多少是工具选错多少是角色混淆。这个阶段大概花一周但能帮你精准定位该拆哪里。第二阶段拆出第一个 Specialist。选一个最独立、边界最清晰的子任务拆出来。通常是检索或计算这类“输入输出明确”的任务。拆出来后让单体 Agent 把这块工作委托给 Specialist其他逻辑不变。对比拆分前后的成功率和延迟。第三阶段引入 Orchestrator。当 Specialist 超过两个时手动调度开始变得混乱这时候引入 Orchestrator。初期可以让 Orchestrator 用规则做决策稳定后再换成模型决策。第四阶段加 Verifier 和监控。系统稳定运行后加上 Verifier Agent 做质量把关同时完善 Trace 和告警。这个阶段的目标是让系统能自我发现和报告问题而不是等用户投诉。每个阶段之间留出至少一周的观察期别急着往下推。我见过太多团队一次性重构结果出了问题连是哪个环节导致的都说不清。最后分享一个我在迁移过程中总结的小技巧给每个 Agent 写一份“岗位说明书”内容包括它的职责、输入格式、输出格式、可用工具、失败时的处理方式。这份说明书既是给团队看的文档也是写系统提示词的依据。当每个 Agent 的职责都清晰到能写成说明书时你的 Multi-Agent 架构基本就稳了。