1. 从一次线上事故说起单体 Agent 到底卡在哪去年冬天我接手了一个内部工单系统的智能化改造需求说起来不复杂让 Agent 自动读取用户提交的问题描述判断问题类型然后调用对应的知识库接口、日志查询接口、工单创建接口最后生成一段回复。一开始我用的是最朴素的单体 Agent 方案——一个 ReAct 循环一个系统提示词挂上五六个工具跑得挺欢。前两周一切正常。第三周开始问题来了。用户提交的问题越来越长有的贴了整段报错日志有的把三四个不相关的诉求混在一条工单里。Agent 开始出现几种典型症状工具调用顺序错乱明明该先查日志却先去创建工单上下文里塞了太多历史对话导致它把上一单的信息串到这一单最要命的是当任务需要先分类、再检索、再校验、最后生成这种多阶段流程时单体 Agent 的准确率从 85% 一路掉到 60% 出头。我当时的第一反应是加提示词、加 few-shot 示例、把工具描述写得更细。有效果但很快又撞到天花板。后来我把这个循环拆成了三个独立的 Agent一个专门做意图分类和任务拆解一个专门做信息检索和工具调用一个专门做结果校验和回复生成。每个 Agent 只关心自己那一小段上下文工具集也各自收敛。改完之后同样的测试集准确率回到了 90% 以上而且单次任务的 token 消耗反而降了将近四成。这次经历让我彻底想明白一件事单体 Agent 的天花板本质上不是模型能力的天花板而是上下文管理和职责边界的天花板。复杂任务必然走向 Multi-Agent不是因为 Multi-Agent 更时髦而是因为单体的上下文窗口、注意力分配和错误传播机制在任务复杂度超过某个阈值后就会系统性失效。这篇文章我想把这件事讲透。适合谁看如果你正在用 Python 写 Agent、正在纠结要不要上 Multi-Agent 框架、或者已经被context length 超限agent execution terminated due to error这类报错折磨过那这篇就是写给你的。我会从设计思路、核心原理、实操落地到踩坑排查一层层拆开讲尽量让你看完能直接抄作业。2. 单体 Agent 的四个隐形天花板2.1 上下文窗口不是越大越好而是越乱越糟很多人对上下文的理解还停留在窗口越大越好。现在主流模型动辄 128K、200K 甚至标称百万级 token 的上下文看起来什么都能塞进去。但实际用下来你会发现上下文长度和有效推理能力根本不是线性关系。我做过一个很粗糙但很说明问题的测试让单体 Agent 处理一个需要调用 4 个工具、涉及 3 轮信息回溯的任务。当我把历史对话控制在 2000 token 以内时工具调用正确率约 88%把历史堆到 8000 token 时掉到 72%堆到 20000 token 时直接掉到 55% 以下而且开始出现幻觉调用——调用一个根本不存在的工具名。原因不神秘。Transformer 的注意力机制在长上下文里会被大量无关信息稀释模型对当前该关注什么的判断力下降。你在上下文里塞了十轮历史对话其中九轮跟当前子任务无关模型却要为这九轮分配注意力预算。这就像让一个人一边打电话一边听旁边三个人聊天信息都在耳朵里但真正能处理的有效带宽是有限的。提示判断上下文是否过载不要只看 token 数要看有效信息密度。如果一段上下文里超过一半的内容与当前决策无关那就是过载信号。2.2 职责耦合导致错误传播不可控单体 Agent 最隐蔽的问题是它把所有职责压在一个决策循环里。分类、检索、推理、生成、校验全由同一个 Agent 在同一个上下文里完成。这带来一个致命后果任何一个环节的偏差都会污染后续所有环节。举个具体例子。用户问我上周提交的退款申请为什么还没到账。单体 Agent 在第一步可能把意图误判成查询订单状态于是去调订单接口订单接口返回的数据里恰好有个退款中字段Agent 又顺着这个字段去调退款接口退款接口返回的是一堆状态码Agent 看不懂于是开始编造解释。整个链条里第一步的误判没有被任何机制拦截反而被后续步骤不断放大。Multi-Agent 的价值在这里就体现出来了分类 Agent 的输出是一个明确的结构化标签检索 Agent 只对这个标签负责校验 Agent 会独立检查检索结果是否支撑结论。每个环节都有独立的守门人错误在传播前就被拦截。2.3 工具集膨胀带来的选择困难单体 Agent 挂的工具一多选择困难就来了。我见过一个项目单体 Agent 挂了 20 多个工具从数据库查询到邮件发送到 PDF 生成全都有。结果模型经常在该用 A 还是 B上犹豫甚至把两个工具的参数混着用。这背后是工具描述之间的语义干扰。当工具数量超过 7 到 10 个工具描述之间的相似度开始显著影响选择准确率。模型要在几十个描述里做细粒度区分本质上是在做一次高难度的多分类而上下文里还塞着一堆对话历史难度进一步叠加。Multi-Agent 的解法很直接按职责切分工具集。检索 Agent 只挂检索类工具生成 Agent 只挂生成类工具每个 Agent 面对的工具数量控制在 5 个以内。选择空间小了准确率自然上去。2.4 单点失败没有兜底单体 Agent 还有一个工程上的硬伤它是一个单点。这个循环一旦因为某个工具超时、某个 API 返回 400、某段上下文超限而中断整个任务就挂了。你没法在中间插入重试、降级、人工接管这些机制因为所有逻辑都缠在一个循环里。我踩过最典型的一个坑某个工具接口偶发超时单体 Agent 在超时后没有重试逻辑直接把超时信息当成工具返回结果继续往下推理最后生成了一段基于错误前提的回复。这种问题在单体架构里极难定位因为错误和正常流程混在同一条执行链上。3. Multi-Agent 的核心设计思路拆什么、怎么拆3.1 拆分的本质是上下文隔离而非功能堆叠很多人第一次接触 Multi-Agent会本能地按功能拆一个查数据库的、一个发邮件的、一个写文案的。这么拆不是不行但没抓到重点。Multi-Agent 真正的价值是上下文隔离——让每个 Agent 只携带完成自己那部分任务所需的最小上下文。我现在的拆分原则是如果一个子任务需要的历史信息和另一个子任务需要的历史信息重叠度低于 30%就应该拆成两个 Agent。因为重叠度低意味着它们各自需要的上下文可以独立裁剪拆开之后每个 Agent 的上下文都能大幅瘦身。反过来如果两个子任务高度依赖同一段上下文硬拆反而会增加信息传递成本这时候留在同一个 Agent 里更划算。拆分不是越多越好是要看上下文能不能真正隔离。3.2 三种主流拓扑流水线、主管制、黑板制实际落地时Multi-Agent 的协作拓扑主要有三种各有适用场景。拓扑类型结构适用场景优点缺点流水线A→B→C 顺序执行步骤明确、依赖线性的任务逻辑清晰、易调试灵活性差、无法并行主管制一个主管 Agent 调度多个执行 Agent任务类型多样、需要动态决策灵活、可动态分配主管本身可能成为瓶颈黑板制多个 Agent 共享一块状态区各自读写需要多轮协商、信息互补的任务协作充分状态管理复杂、易冲突我自己的项目里主管制用得最多。原因很实际大部分业务任务的类型是可枚举的主管 Agent 只需要做一次意图识别和任务分发后面的执行交给专门的 Agent。主管的上下文可以压得很小因为它不需要知道执行细节只需要知道这个任务该派给谁。流水线适合那种步骤完全固定的场景比如解析→校验→入库这种 ETL 流程。黑板制我一般只在需要多 Agent 反复协商的场景用比如多个 Agent 对同一份数据给出不同判断、需要投票或辩论才能定论的情况。这种场景实现复杂度高非必要不上。3.3 通信协议结构化消息是生命线Multi-Agent 之间怎么传消息是决定系统稳不稳的关键。我见过太多项目在这里偷懒Agent 之间直接传自然语言字符串结果下游 Agent 要花大量精力去解析上游的模糊表达错误率居高不下。我的做法是强制结构化消息。每个 Agent 的输入输出都定义成明确的 schema用 JSON 传递。比如分类 Agent 的输出必须是{ intent: refund_status_query, confidence: 0.92, entities: { order_id: ORD-20240115-8823, time_range: last_week } }下游 Agent 拿到这个结构不需要猜用户到底想问什么直接按字段取用。结构化消息把理解成本从运行时前移到了设计时这是 Multi-Agent 稳定性的基石。注意schema 一旦定下来就要严格校验。我一般会在每个 Agent 的入口加一层校验字段缺失或类型不对直接拒绝绝不让脏数据流进下游。4. 用 Python 落地一个最小可用的 Multi-Agent 系统4.1 环境准备与依赖选型先说环境。Python 版本我建议 3.10 以上因为要用到一些较新的类型标注语法。依赖方面核心就几个pip install openai pydantic tenacityopenai调用模型接口如果你用的是其他厂商的兼容接口改 base_url 即可。pydantic定义和校验 Agent 之间的结构化消息这是保证通信可靠的关键。tenacity做重试和退避处理工具调用和 API 的偶发失败。框架层面我刻意不引入重型 Agent 框架。原因是我发现很多框架把编排逻辑封装得太深出问题时排查成本极高。先用原生 Python 把核心逻辑跑通理解每个环节在干什么再决定要不要上框架这个顺序不能反。4.2 定义 Agent 基类与消息协议先定义一个所有 Agent 的基类把公共逻辑抽出来from abc import ABC, abstractmethod from pydantic import BaseModel from typing import Any class AgentMessage(BaseModel): sender: str receiver: str task_type: str payload: dict[str, Any] trace_id: str class BaseAgent(ABC): def __init__(self, name: str, model_client): self.name name self.client model_client abstractmethod def handle(self, message: AgentMessage) - AgentMessage: pass def _call_llm(self, system_prompt: str, user_content: str) - str: response self.client.chat.completions.create( modelyour-model-name, messages[ {role: system, content: system_prompt}, {role: user, content: user_content}, ], temperature0.1, ) return response.choices[0].message.content这里有几个设计决策值得说。temperature 设成 0.1是因为 Agent 之间的通信需要稳定可复现不需要创意。trace_id 贯穿所有消息是为了出问题时能把整条链路串起来排查这个字段在调试阶段能救命。4.3 主管 Agent只做分发不做执行主管 Agent 的职责被刻意压到最小识别任务类型决定派给谁。它不碰任何业务工具上下文里只有任务类型清单和分发规则。class SupervisorAgent(BaseAgent): ROUTING_TABLE { refund_status_query: retrieval_agent, order_modification: retrieval_agent, complaint: generation_agent, } def handle(self, message: AgentMessage) - AgentMessage: system_prompt ( 你是一个任务分类器。根据用户输入判断任务类型 只输出类型标签不要输出任何其他内容。 可选类型refund_status_query, order_modification, complaint。 ) task_type self._call_llm(system_prompt, message.payload[user_input]).strip() target self.ROUTING_TABLE.get(task_type, generation_agent) return AgentMessage( senderself.name, receivertarget, task_typetask_type, payloadmessage.payload, trace_idmessage.trace_id, )主管 Agent 的提示词里我特意强调只输出类型标签。这是为了防止模型自作聪明地输出一段解释导致下游解析失败。对主管这类做决策的 Agent输出格式的约束要比执行类 Agent 更严格。4.4 检索 Agent工具调用的收敛与重试检索 Agent 负责调用外部工具拿数据。它的工具集被严格限制在检索类且每个工具的描述都写得非常具体。from tenacity import retry, stop_after_attempt, wait_exponential class RetrievalAgent(BaseAgent): retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max8)) def _call_tool(self, tool_name: str, params: dict) - dict: # 实际项目里这里对接真实工具 if tool_name query_order: return {order_id: params[order_id], status: refunding} raise ValueError(funknown tool: {tool_name}) def handle(self, message: AgentMessage) - AgentMessage: order_id message.payload.get(order_id) result self._call_tool(query_order, {order_id: order_id}) return AgentMessage( senderself.name, receivergeneration_agent, task_typemessage.task_type, payload{raw_result: result, user_input: message.payload[user_input]}, trace_idmessage.trace_id, )tenacity的重试装饰器是这里的重点。工具调用失败是常态指数退避重试能把大量偶发失败消化在内部不让它们冒泡到上层。我一般设 3 次重试间隔从 1 秒指数增长到 8 秒封顶这个参数在大多数接口上够用。4.5 生成 Agent只对结构化输入负责生成 Agent 拿到的是上游整理好的结构化数据它不需要再调工具只负责把数据翻译成人话。class GenerationAgent(BaseAgent): def handle(self, message: AgentMessage) - AgentMessage: raw message.payload[raw_result] system_prompt ( 你是一个客服回复生成器。根据给定的订单数据生成一段简洁的回复 不要编造数据中没有的信息。 ) user_content f订单数据{raw}\n用户问题{message.payload[user_input]} reply self._call_llm(system_prompt, user_content) return AgentMessage( senderself.name, receiveroutput, task_typefinal_reply, payload{reply: reply}, trace_idmessage.trace_id, )生成 Agent 的提示词里那句不要编造数据中没有的信息是我反复调试后加上的。生成类 Agent 最大的风险是幻觉而幻觉往往来自它试图补全上游没给的信息。明确禁止编造能显著降低这类问题。4.6 编排入口把链路串起来最后写一个简单的编排函数把三个 Agent 串起来def run_pipeline(user_input: str, trace_id: str): msg AgentMessage( senderuser, receiversupervisor, task_typeraw_input, payload{user_input: user_input}, trace_idtrace_id, ) supervisor SupervisorAgent(supervisor, client) retrieval RetrievalAgent(retrieval_agent, client) generation GenerationAgent(generation_agent, client) msg supervisor.handle(msg) msg retrieval.handle(msg) msg generation.handle(msg) return msg.payload[reply]这个编排很朴素但胜在每一步都看得见、改得动。等链路稳定了再考虑引入更复杂的调度、并行、条件分支。5. 踩坑实录Multi-Agent 落地时最容易翻车的六个点5.1 上下文超限不是模型的问题是你塞太多api error: 400 this models maximum context length is 1048576 tokens 这类报错我见过太多次。很多人第一反应是换更大窗口的模型但换完发现还是超。真正的原因往往是某个 Agent 把整条链路的历史都带上了。我的排查方法很土但有效在每个 Agent 的入口打印当前上下文的 token 数跑一遍完整流程看哪个 Agent 的上下文异常膨胀。十有八九是某个 Agent 在传递消息时把上游的完整 payload 原样带下去了而不是只带自己需要的那部分。提示给每个 Agent 定义输入白名单只允许它读取明确列出的字段其余一律丢弃。这个约束能挡掉大部分上下文膨胀。5.2 消息格式漂移下游解析失败的元凶Multi-Agent 跑一段时间后经常出现下游 Agent 解析上游消息失败的情况。根因通常是上游 Agent 的输出格式发生了漂移——比如某个字段偶尔返回字符串、偶尔返回列表。解法是在每个 Agent 的出口做 schema 校验不符合就重试或报错绝不放行。用 pydantic 的model_validate就能做成本很低收益很大。5.3 死循环Agent 之间互相踢皮球主管制拓扑里有个经典问题主管把任务派给 AA 处理不了又退回主管主管再派给 A无限循环。我遇到过一次跑了 40 多轮才因为超时中断token 烧了一大片。防御手段有两个一是给每条消息加跳数计数超过阈值强制终止二是在主管的分发逻辑里记录已派发过的 Agent同一个任务不重复派给同一个 Agent。5.4 工具调用参数错位检索 Agent 调工具时参数名和工具定义对不上是高频错误。比如工具要order_idAgent 传了orderId。这类问题在单体 Agent 里也常见但在 Multi-Agent 里更隐蔽因为参数是在 Agent 之间传递的出错点离调用点很远。我的做法是把工具的参数 schema 直接写进检索 Agent 的提示词并且用 pydantic 在调用前校验一次。多一道校验少一堆深夜排查。5.5 错误被当成正常结果继续传播前面提过工具超时后如果没做区分超时信息会被当成正常返回继续往下走。Multi-Agent 里这个问题更严重因为错误会跨 Agent 传播。解法是在消息协议里显式区分成功结果和错误结果。我一般加一个status字段值为ok或error下游 Agent 遇到error就走降级分支而不是硬着头皮往下推。5.6 调试困难链路太长看不清Multi-Agent 最让人头疼的是调试。一条链路经过四五个 Agent出问题时不知道是哪一环。我的经验是trace_id 必须贯穿全链路且每个 Agent 的输入输出都要落日志。日志格式统一成 JSON方便后续用脚本过滤。下面是我常用的排查速查表症状可能原因排查方向上下文超限某 Agent 携带了全链路历史打印各 Agent 入口 token 数下游解析失败上游输出格式漂移检查出口 schema 校验任务无限循环主管重复派发检查跳数计数与派发记录工具调用报错参数名或类型不匹配校验工具参数 schema结果明显错误错误被当正常结果传播检查 status 字段处理定位不到问题日志缺失或格式混乱统一 JSON 日志 trace_id6. 什么时候不该上 Multi-Agent讲了这么多 Multi-Agent 的好我得泼盆冷水不是所有任务都值得拆。我见过一些项目任务本身很简单硬拆成三个 Agent结果通信开销比任务本身还大延迟翻倍稳定性反而下降。我的判断标准很直接当单体 Agent 在测试集上的准确率稳定在 90% 以上且上下文长期不超过窗口的 30%就别拆。拆分的收益只有在单体确实撞到天花板时才显现。过早拆分是另一种形式的过度设计。还有一个容易被忽略的点Multi-Agent 的维护成本是单体的数倍。每多一个 Agent就多一套提示词、一套 schema、一套日志、一套测试。团队人手有限时宁可把单体 Agent 打磨到极致也不要盲目追求架构上的先进。我个人的经验是从单体到 Multi-Agent 的迁移最好由一次真实的失败驱动——比如某次线上事故确实因为单体架构的缺陷导致这时候拆分才有明确的靶子也才容易说服团队。为了拆而拆最后往往是给自己挖坑。最后分享一个我一直在用的小技巧在拆分之前先把单体 Agent 的提示词按职责段落重新组织一遍。很多时候你会发现把提示词里的分类逻辑、检索逻辑、生成逻辑分段写清楚单体 Agent 的表现就能提升一大截。这一步做完还是不够再考虑拆 Agent。这个顺序能帮你省下大量不必要的架构改造。