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 的协作拓扑主要有三种，各有适用场景。

我自己的项目里，主管制用得最多。原因很实际：大部分业务任务的类型是可枚举的，主管 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 tenacity

• openai：调用模型接口，如果你用的是其他厂商的兼容接口，改 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( model="your-model-name", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}, ], temperature=0.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( sender=self.name, receiver=target, task_type=task_type, payload=message.payload, trace_id=message.trace_id, )

主管 Agent 的提示词里我特意强调"只输出类型标签"。这是为了防止模型自作聪明地输出一段解释，导致下游解析失败。对主管这类做决策的 Agent，输出格式的约束要比执行类 Agent 更严格。

4.4 检索 Agent：工具调用的收敛与重试

检索 Agent 负责调用外部工具拿数据。它的工具集被严格限制在检索类，且每个工具的描述都写得非常具体。

from tenacity import retry, stop_after_attempt, wait_exponential class RetrievalAgent(BaseAgent): @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8)) 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(f"unknown 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( sender=self.name, receiver="generation_agent", task_type=message.task_type, payload={"raw_result": result, "user_input": message.payload["user_input"]}, trace_id=message.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( sender=self.name, receiver="output", task_type="final_reply", payload={"reply": reply}, trace_id=message.trace_id, )

生成 Agent 的提示词里那句"不要编造数据中没有的信息"是我反复调试后加上的。生成类 Agent 最大的风险是幻觉，而幻觉往往来自它试图"补全"上游没给的信息。明确禁止编造，能显著降低这类问题。

4.6 编排入口：把链路串起来

最后写一个简单的编排函数，把三个 Agent 串起来：

def run_pipeline(user_input: str, trace_id: str): msg = AgentMessage( sender="user", receiver="supervisor", task_type="raw_input", payload={"user_input": user_input}, trace_id=trace_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 model's 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 之间互相踢皮球

主管制拓扑里有个经典问题：主管把任务派给 A，A 处理不了又退回主管，主管再派给 A，无限循环。我遇到过一次，跑了 40 多轮才因为超时中断，token 烧了一大片。

防御手段有两个：一是给每条消息加跳数计数，超过阈值强制终止；二是在主管的分发逻辑里记录已派发过的 Agent，同一个任务不重复派给同一个 Agent。

5.4 工具调用参数错位

检索 Agent 调工具时，参数名和工具定义对不上，是高频错误。比如工具要order_id，Agent 传了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，方便后续用脚本过滤。

下面是我常用的排查速查表：

6. 什么时候不该上 Multi-Agent

讲了这么多 Multi-Agent 的好，我得泼盆冷水：不是所有任务都值得拆。我见过一些项目，任务本身很简单，硬拆成三个 Agent，结果通信开销比任务本身还大，延迟翻倍，稳定性反而下降。

我的判断标准很直接：当单体 Agent 在测试集上的准确率稳定在 90% 以上，且上下文长期不超过窗口的 30%，就别拆。拆分的收益只有在单体确实撞到天花板时才显现。过早拆分是另一种形式的过度设计。

还有一个容易被忽略的点：Multi-Agent 的维护成本是单体的数倍。每多一个 Agent，就多一套提示词、一套 schema、一套日志、一套测试。团队人手有限时，宁可把单体 Agent 打磨到极致，也不要盲目追求架构上的"先进"。

我个人的经验是，从单体到 Multi-Agent 的迁移，最好由一次真实的失败驱动——比如某次线上事故确实因为单体架构的缺陷导致，这时候拆分才有明确的靶子，也才容易说服团队。为了拆而拆，最后往往是给自己挖坑。

最后分享一个我一直在用的小技巧：在拆分之前，先把单体 Agent 的提示词按"职责段落"重新组织一遍。很多时候你会发现，把提示词里的分类逻辑、检索逻辑、生成逻辑分段写清楚，单体 Agent 的表现就能提升一大截。这一步做完还是不够，再考虑拆 Agent。这个顺序能帮你省下大量不必要的架构改造。