从单体到Multi-Agent：工单系统Agent架构重构实战复盘

1. 从一次线上事故说起单体 Agent 到底卡在哪去年冬天我接手了一个内部工单系统的智能化改造项目需求说起来不复杂让 Agent 自动读取用户提交的问题描述判断问题类型然后调用对应的知识库接口和工单接口最后生成一份处理建议。当时我的想法很朴素——一个 ReAct 循环挂上五六个工具配上足够长的系统提示词应该就能跑起来。第一版确实跑通了。在测试环境里我拿二十条样例数据喂进去Agent 能正确识别问题类型能按顺序调用工具生成的建议也像模像样。我当时觉得这事成了准备上线。然后线上流量一进来问题就炸了。最先暴露的是Context 膨胀。工单系统的知识库返回内容很长一次检索动辄三四千 token加上工具调用的中间结果、历史对话、系统提示词单次请求的上下文很快就逼近模型的上限。我用的模型上下文窗口是 128K听起来很大但实际跑起来一个稍微复杂的工单处理流程五轮工具调用之后上下文就吃掉了一大半。再往后模型开始出现遗忘——前面明明已经确认过的问题类型后面又搞错了已经调用过的接口它又调了一遍。接着是错误累积。ReAct 的机制是思考—行动—观察循环每一步都依赖上一步的输出。单体 Agent 里所有步骤共享同一个上下文一旦某一步的观察结果有噪声或者模型某一次推理跑偏这个错误会像滚雪球一样往下传。我遇到过最离谱的一次Agent 在第二步把退款问题误判成了物流问题结果后面所有工具调用全部走错分支最后生成了一份完全牛头不对马嘴的建议而它自己还信心满满地认为任务完成了。再往后是工具选择的混乱。当工具数量超过十个单体 Agent 的选择准确率会明显下降。这不是模型不行而是把所有工具的说明、参数、返回值格式全塞进一个提示词里模型需要在一次推理中同时完成理解任务、回忆工具、选择工具、构造参数四件事认知负荷太高。我实测过工具数量从 5 个增加到 15 个单体 Agent 的工具选择准确率从 92% 掉到了 67% 左右。那次事故之后我花了大概两周时间重构把单体 Agent 拆成了 Multi-Agent 架构。拆完之后同样的工单处理流程工具选择准确率回到了 90% 以上上下文峰值下降了将近 60%而且最关键的是——错误不再跨阶段传播了。每个 Agent 只负责自己那一小段出了问题能定位、能回滚、能单独优化。这篇文章就是那次重构的完整复盘。我会从为什么单体 Agent 有天花板讲起拆解 Multi-Agent 的核心设计思路、通信机制、上下文管理策略然后给出可直接参考的 Python 实现骨架和踩坑记录。如果你正在做 Agent 开发或者正在纠结要不要拆 Multi-Agent这篇应该能帮你少走一些弯路。2. 单体 Agent 的三个硬天花板在动手拆之前我先花了一周时间做归因分析把单体 Agent 在复杂任务上的失败案例全部拉出来逐条标注失败原因。最后归成了三类这三类问题不是调参能解决的而是架构层面的硬约束。2.1 Context 窗口不是越大越好很多人有个误区模型上下文窗口大单体 Agent 就能处理复杂任务。我一开始也这么想直到我做了个对比实验。我拿同一个复杂任务需要 8 轮工具调用的工单处理分别在 32K、128K、200K 上下文窗口的模型上跑每组跑 50 次统计任务完成率和平均 token 消耗。结果很有意思上下文窗口任务完成率平均 token 消耗平均耗时32K54%28K12s128K71%89K31s200K73%142K52s上下文窗口从 128K 涨到 200K完成率只涨了 2 个百分点但 token 消耗涨了 60%耗时涨了 68%。这说明什么说明瓶颈不在窗口大小而在信息密度。单体 Agent 的上下文里塞了大量当前步骤用不到的信息前面几轮的工具返回结果、已经处理完的子任务细节、无关的历史对话。这些信息对当前推理来说是噪声模型需要在海量 token 里捞出真正相关的那几条注意力被稀释了。业界有个说法叫 lost in the middle——模型对上下文中间部分的信息召回率明显低于开头和结尾。单体 Agent 的上下文越长中间被遗忘的关键信息就越多。Multi-Agent 的核心优势之一就是每个 Agent 只持有自己子任务相关的上下文。工单分类 Agent 不需要知道知识库返回了什么知识库检索 Agent 不需要知道工单接口的参数格式。上下文被切分到各个 Agent 里每个 Agent 的上下文都短而精信息密度高模型注意力集中。2.2 错误传播一步错步步错ReAct 循环的本质是串行依赖。第 N 步的输入是第 N-1 步的输出第 N-1 步的输出又依赖第 N-2 步……这种链式结构在单体 Agent 里没有任何隔离带。我统计过那次事故的失败案例发现一个规律如果前 3 步内出现一次错误最终任务失败的概率超过 80%。而且错误类型越靠前失败率越高。第一步就错的基本没救。更麻烦的是单体 Agent 没有自我纠错的机制。它不会在发现矛盾时停下来重新审视而是会强行自圆其说。我见过 Agent 把用户要求退款理解成用户要求查询退款进度然后调用查询接口返回无退款记录它居然把这个结果解释成用户尚未申请退款最后建议用户先申请退款。整个逻辑链条自洽但起点就错了。Multi-Agent 的解法是引入检查点和角色分离。比如让一个专门的验证 Agent在关键节点检查上游输出或者让执行 Agent和审核 Agent分离执行 Agent 负责干活审核 Agent 负责挑刺。错误在传播到下一阶段之前有机会被拦截。2.3 工具爆炸后的选择困难工具数量对单体 Agent 的影响我做了个梯度实验。固定任务不变逐步增加工具数量观察工具选择准确率工具数量选择准确率平均推理轮次594%3.2888%4.11276%5.81567%7.32051%9.6工具从 5 个涨到 20 个准确率直接腰斩推理轮次翻了 3 倍。原因很简单所有工具的 schema 都要塞进系统提示词模型每次推理都要在几十个工具描述里做选择。这就像让一个人同时记住 20 个不同部门的办事流程然后随机抽一个任务让他立刻判断该去哪个窗口——认知负荷太高出错是必然的。Multi-Agent 的思路是按职能分组工具。分类 Agent 只挂分类相关的工具检索 Agent 只挂检索工具执行 Agent 只挂执行工具。每个 Agent 面对的工具数量控制在 3-5 个选择准确率自然就上去了。3. Multi-Agent 的架构选型不是所有任务都值得拆拆 Multi-Agent 不是免费的。每多一个 Agent就多一次模型调用、多一层通信开销、多一个可能出错的环节。我见过一些项目明明是个简单任务硬拆成五六个 Agent结果延迟翻了三倍稳定性还不如单体。所以第一步是判断这个任务到底值不值得拆。3.1 什么任务适合 Multi-Agent我总结了一个简单的判断标准满足以下任意两条就值得考虑拆任务步骤超过 5 步且步骤之间有明确的阶段划分工具数量超过 8 个且工具可以按职能分组单次上下文峰值超过模型窗口的 60%任务失败后需要定位到具体阶段而不是整体重跑不同阶段需要不同的模型比如分类用便宜的小模型生成用贵的大模型反过来如果任务步骤少于 3 步、工具少于 5 个、上下文很宽裕那单体 Agent 完全够用拆了反而添乱。3.2 三种常见的 Multi-Agent 拓扑业界常见的 Multi-Agent 拓扑大概有三种我分别说说适用场景和坑。第一种流水线式Pipeline。Agent 按顺序排列A 的输出是 B 的输入B 的输出是 C 的输入。这种最简单也最可控。适合步骤明确、阶段划分清晰的任务比如分类→检索→生成→审核。缺点是串行执行延迟是各阶段之和而且中间某个 Agent 挂了整条链就断了。第二种主管-工人式Supervisor-Worker。一个主管 Agent 负责拆解任务、分配子任务、汇总结果多个工人 Agent 各自执行。这种适合子任务可以并行、或者子任务数量不固定的场景。缺点是主管 Agent 本身可能成为瓶颈而且主管的拆解质量直接决定整体效果。第三种辩论式Debate。多个 Agent 对同一问题给出不同答案然后通过多轮讨论收敛。这种适合需要高置信度的场景比如事实核查、风险评估。缺点是 token 消耗大延迟高而且可能收敛不了。我那个工单项目用的是流水线 局部主管的混合拓扑主流程是分类→检索→生成→审核的流水线但在检索阶段内部用一个主管 Agent 协调多个检索工人 Agent 并行查不同知识库。这样既保证了主流程的可控性又在检索阶段拿到了并行的效率。3.3 通信机制消息传递还是共享状态Multi-Agent 的通信机制主流有两种消息传递和共享状态。消息传递就是 Agent 之间通过显式的消息对象通信A 把结果打包成消息发给 B。这种方式的优点是解耦彻底每个 Agent 只需要知道自己上游的消息格式不需要知道上游是谁。缺点是消息格式需要严格定义而且消息在传递过程中可能丢失上下文。共享状态是维护一个全局的 State 对象所有 Agent 读写同一个 State。优点是上下文天然共享不需要显式传递。缺点是耦合度高一个 Agent 改了 State 的某个字段可能影响其他 Agent调试起来很痛苦。我实测下来混合方案最稳主流程用消息传递保证阶段之间的解耦阶段内部用共享状态方便工人 Agent 之间交换中间结果。具体来说每个阶段有一个自己的 State阶段之间通过消息对象传递消息对象里只放下一阶段需要的最小信息集。这里有个坑消息对象千万别放全量上下文。我一开始图省事把上游 Agent 的完整上下文塞进消息里传给下游结果下游 Agent 的上下文又爆了。后来改成只传结论 必要参数上下文峰值直接降了一半。4. 手写一个 Multi-Agent 工单处理系统理论说完了上代码。我用 Python 写了一个简化版的工单处理 Multi-Agent 系统去掉了业务细节保留了核心架构。你可以直接拿去改。4.1 基础 Agent 类的设计先定义 Agent 的基类。核心是三个东西name标识、tools可用工具、run执行入口。from abc import ABC, abstractmethod from typing import Any import json class BaseAgent(ABC): def __init__(self, name: str, llm_client, tools: list None): self.name name self.llm llm_client self.tools tools or [] self.max_iterations 5 abstractmethod def build_prompt(self, input_data: dict) - str: 构造该 Agent 的系统提示词 pass abstractmethod def parse_output(self, raw_output: str) - dict: 解析模型输出为结构化结果 pass def run(self, input_data: dict) - dict: prompt self.build_prompt(input_data) messages [{role: system, content: prompt}] messages.append({role: user, content: json.dumps(input_data, ensure_asciiFalse)}) for i in range(self.max_iterations): response self.llm.chat(messages) parsed self.parse_output(response) if parsed.get(type) final: return parsed[result] if parsed.get(type) tool_call: tool_name parsed[tool] tool_args parsed[args] tool_result self._execute_tool(tool_name, tool_args) messages.append({role: assistant, content: response}) messages.append({role: user, content: f工具 {tool_name} 返回{tool_result}}) return {error: max iterations reached, agent: self.name} def _execute_tool(self, tool_name: str, args: dict) - Any: for tool in self.tools: if tool[name] tool_name: return tool[func](**args) return f工具 {tool_name} 不存在这个基类做了几件事把 ReAct 循环封装在run里子类只需要实现build_prompt和parse_output。max_iterations是硬性熔断防止 Agent 陷入死循环——这个参数很关键我见过没有熔断的 Agent 在工具报错时反复重试同一个调用烧了几百万 token。4.2 分类 Agent只做一件事分类 Agent 的职责非常单一读工单描述输出问题类型。它不需要任何工具只需要一个清晰的分类体系。class ClassifierAgent(BaseAgent): CATEGORIES [退款, 物流, 商品质量, 账号, 其他] def build_prompt(self, input_data: dict) - str: return f你是一个工单分类助手。请将用户的问题分类到以下类别之一 {, .join(self.CATEGORIES)} 输出格式严格 JSON {{type: final, result: {{category: 类别名, confidence: 0.0-1.0, reason: 简短理由}}}} 只输出 JSON不要有其他内容。 def parse_output(self, raw_output: str) - dict: try: return json.loads(raw_output) except json.JSONDecodeError: return {type: final, result: {category: 其他, confidence: 0.0, reason: 解析失败}}分类 Agent 的提示词里我特意加了confidence字段。这个字段后面有用——如果置信度低于 0.6主流程会走人工兜底分支而不是硬着头皮往下走。这是单体 Agent 很难做到的单体 Agent 没有我不确定这个选项它只会继续往下编。4.3 检索 Agent主管-工人模式检索 Agent 是唯一用了主管-工人模式的地方。主管负责决定查哪些知识库工人负责实际检索。class RetrievalSupervisor(BaseAgent): def build_prompt(self, input_data: dict) - str: category input_data[category] return f你是检索主管。当前工单类别是「{category}」。 你有以下知识库可用 - refund_kb: 退款政策、退款流程 - logistics_kb: 物流时效、配送范围 - product_kb: 商品参数、质量说明 - account_kb: 账号安全、密码找回 请决定需要查询哪些知识库可多选输出格式 {{type: final, result: {{kbs: [kb1, kb2], query: 检索关键词}}}} def parse_output(self, raw_output: str) - dict: return json.loads(raw_output) class RetrievalWorker: def __init__(self, kb_name: str, kb_client): self.kb_name kb_name self.kb kb_client def retrieve(self, query: str, top_k: int 3) - list: return self.kb.search(self.kb_name, query, top_ktop_k)主管 Agent 只输出查哪些库 查什么词工人 Agent 并行执行检索。这样主管的上下文很短只有类别和知识库列表工人的上下文也很短只有查询词和检索结果两边都不会爆。4.4 生成 Agent 与审核 Agent生成 Agent 负责把检索结果和工单信息合成一份处理建议。审核 Agent 负责检查生成结果是否引用了不存在的政策、是否遗漏了关键信息。class GeneratorAgent(BaseAgent): def build_prompt(self, input_data: dict) - str: return f你是工单处理建议生成助手。 工单描述{input_data[description]} 问题类别{input_data[category]} 检索到的知识{input_data[retrieved_docs]} 请生成一份处理建议要求 1. 引用知识库中的具体政策条款 2. 给出明确的处理步骤 3. 如果知识库信息不足以支撑建议明确说明信息不足 输出格式{{type: final, result: {{suggestion: ..., cited_docs: [...], confidence: 0.0-1.0}}}} def parse_output(self, raw_output: str) - dict: return json.loads(raw_output) class ReviewerAgent(BaseAgent): def build_prompt(self, input_data: dict) - str: return f你是审核助手。请检查以下处理建议是否存在问题 建议内容{input_data[suggestion]} 引用的知识{input_data[retrieved_docs]} 检查项 1. 建议中引用的政策是否在知识库中存在 2. 建议步骤是否完整 3. 是否存在与工单描述矛盾的地方 输出格式{{type: final, result: {{passed: true/false, issues: [...], revised_suggestion: ...}}}} def parse_output(self, raw_output: str) - dict: return json.loads(raw_output)审核 Agent 是错误拦截的关键。我实测下来加了审核 Agent 之后最终输出的事实性错误引用不存在的政策、步骤缺失从 18% 降到了 4% 左右。代价是整体延迟增加了约 30%但考虑到工单处理不是实时对话场景这个代价可以接受。4.5 主流程编排最后把所有 Agent 串起来class WorkflowOrchestrator: def __init__(self, classifier, retrieval_sup, workers, generator, reviewer): self.classifier classifier self.retrieval_sup retrieval_sup self.workers workers self.generator generator self.reviewer reviewer def process(self, ticket: dict) - dict: # 阶段1分类 cls_result self.classifier.run({description: ticket[description]}) if cls_result.get(confidence, 0) 0.6: return {status: need_human, reason: 分类置信度过低} # 阶段2检索 retrieval_plan self.retrieval_sup.run({category: cls_result[category]}) docs [] for kb_name in retrieval_plan[kbs]: worker self.workers.get(kb_name) if worker: docs.extend(worker.retrieve(retrieval_plan[query])) # 阶段3生成 gen_result self.generator.run({ description: ticket[description], category: cls_result[category], retrieved_docs: docs }) # 阶段4审核 review_result self.reviewer.run({ suggestion: gen_result[suggestion], retrieved_docs: docs }) if not review_result[passed]: return { status: revised, suggestion: review_result[revised_suggestion], issues: review_result[issues] } return {status: ok, suggestion: gen_result[suggestion]}整个流程里每个 Agent 的输入都是上一阶段的结论 必要参数没有任何一个 Agent 拿到全量上下文。这是上下文控制的核心。5. 上下文管理Multi-Agent 的命门拆完 Multi-Agent 之后我发现新的瓶颈不在 Agent 本身而在上下文在 Agent 之间的传递。传多了下游爆传少了下游信息不足。这块我踩了不少坑单独拎出来说。5.1 消息对象的最小信息集原则我一开始的消息对象是这样的message { full_context: ..., # 上游 Agent 的完整上下文 result: {...} }结果下游 Agent 拿到消息后上下文直接翻倍。后来改成message { stage: classification, result: {category: 退款, confidence: 0.85}, metadata: {ticket_id: 12345} }只传结论和必要元数据。下游 Agent 如果需要更多信息自己去查而不是从上游消息里继承。这个原则说起来简单做起来需要克制。每次我想多传一点保险都会问自己下游 Agent 真的需要这个字段吗如果不需要就不传。5.2 上下文压缩的三种手段即使遵循最小信息集原则有些场景下还是需要传递较多信息比如检索结果。这时候需要压缩。第一种摘要压缩。让上游 Agent 在输出前把长文本压缩成摘要。比如检索 Agent 拿到 5 篇文档每篇 2000 字不要直接传给生成 Agent而是先让检索 Agent 生成一份 500 字的摘要。第二种结构化压缩。把非结构化文本转成结构化字段。比如把退款政策用户可在签收后 7 天内申请退款需提供订单号和退款原因压缩成{refund_window: 7天, required_fields: [订单号, 退款原因]}。第三种按需检索。不预先传递所有信息而是给下游 Agent 一个检索工具让它需要时自己去查。这种方式上下文最省但增加了下游 Agent 的推理轮次。我实测下来结构化压缩 按需检索的组合最稳。检索 Agent 输出结构化摘要生成 Agent 如果觉得信息不够可以调用检索工具补充。这样既控制了上下文又保留了灵活性。5.3 上下文隔离的边界怎么划上下文隔离不是越彻底越好。划得太细Agent 之间信息不通任务做不下去划得太粗又回到了单体 Agent 的老路。我的经验是按决策独立性划边界。如果一个决策不需要另一个决策的信息就能做那它们就应该在不同的 Agent 里。比如分类和检索是两个独立决策——分类不需要知道检索结果检索不需要知道分类的推理过程只需要知道分类结论。所以它们应该隔离。反过来生成和审核虽然也是两个阶段但审核需要看到生成的完整输出所以它们之间的上下文传递要更充分。6. 踩坑记录与排查速查表这部分是我实际踩过的坑按出现频率排序。6.1 常见问题速查表问题现象可能原因排查方法解决方案下游 Agent 输出与上游矛盾消息传递丢失关键字段打印消息对象对比上下游输入补全消息字段或让下游 Agent 增加校验整体延迟过高Agent 串行过多或某个 Agent 推理轮次过多给每个 Agent 加耗时打点并行化可并行的阶段给 Agent 设 max_iterationstoken 消耗异常高上下文未压缩或消息对象过大统计每个 Agent 的输入 token 数应用摘要/结构化压缩精简消息对象某个 Agent 反复调用同一工具工具返回结果未被正确解析查看该 Agent 的 messages 历史检查 parse_output 逻辑增加工具调用去重审核 Agent 总是通过审核提示词太宽松人工抽查审核结果增加具体检查项要求审核 Agent 给出理由分类 Agent 置信度普遍偏低分类体系不清晰或提示词示例不足统计各类别的置信度分布补充 few-shot 示例细化分类边界6.2 三个我踩得最深的坑第一个坑Agent 之间的礼貌性冗余。我一开始让每个 Agent 在输出时都加一段我已经完成了 XXX接下来请 XXX 处理。结果这些客套话占了不少 token而且下游 Agent 经常被这些客套话干扰。后来全部改成纯结构化输出token 降了 15%准确率还涨了。第二个坑审核 Agent 的老好人倾向。审核 Agent 如果提示词写得太温和它会倾向于通过。我一开始的提示词是请检查建议是否存在问题结果审核通过率 95%。改成请找出建议中至少一个潜在问题如果确实没有说明理由之后通过率降到了 78%但拦截的真实问题多了很多。第三个坑工具报错的处理。单体 Agent 里工具报错就是报错Agent 会重试或者放弃。Multi-Agent 里工具报错可能被某个 Agent 吞掉然后它编一个假结果传给下游。我后来强制要求所有工具调用必须有明确的成功/失败标记失败时 Agent 必须输出{type: error, reason: ...}不允许编造结果。6.3 性能优化的几个实测数据重构前后同一个工单处理任务我做了对比测试各 100 次指标单体 AgentMulti-Agent变化任务完成率71%89%18pp平均 token 消耗89K52K-42%平均延迟31s38s23%事实性错误率18%4%-14pp错误可定位率32%91%59pp延迟涨了 23%这是 Multi-Agent 的固有代价——多了一次模型调用和通信开销。但完成率、token 消耗、错误率、可定位率全面改善。对于工单处理这种准确性优先于延迟的场景这个 trade-off 是划算的。7. 什么情况下不该拆 Multi-Agent最后说点反向的。不是所有任务都值得拆我见过不少项目为了架构先进而拆结果得不偿失。如果你的任务满足以下条件单体 Agent 完全够用别折腾任务步骤少于 3 步且步骤之间没有明确的阶段划分工具数量少于 5 个且不需要按职能分组单次上下文峰值低于模型窗口的 40%任务失败后可以整体重跑不需要定位到具体阶段延迟敏感比如实时对话场景我有个朋友做客服机器人任务就是理解问题→查 FAQ→回复三步两个工具上下文峰值 8K。他一开始也想拆 Multi-Agent我劝住了。单体 Agent 跑得好好的拆了反而增加延迟和故障点。Multi-Agent 是解决复杂任务的工具不是目的。判断标准很简单如果单体 Agent 的失败原因主要是上下文太长工具太多错误传播那拆如果失败原因是模型能力不够提示词没写好那拆了也没用。我在实际项目里的体会是Multi-Agent 的价值不在于更先进而在于把不可控的大问题拆成可控的小问题。每个小问题单独优化、单独测试、单独监控整体系统的可维护性和可观测性都会上一个台阶。但前提是你得先确认那个大问题真的存在而不是为了拆而拆。