多智能体协作系统实战：从单Agent困境到分工调度与校验设计

上周我把一套多智能体协作系统从原型阶段直接推翻重做原因是单Agent模式撑不住了。用户把一个问题拆成三个子问题丢进来模型东一句西一句地回答情绪一乱连上下文都丢了。那段时间我几乎每天都在和提示词较劲最后意识到问题根本不是单个Agent不够聪明而是我们试图让一个Agent同时扮演太多角色。后来我参考了多智能体协作系统里常见的“分工—调度—校验”思路把整个流程拆成了四个角色重新设计协作协议才真正把项目跑通。这个案例太典型了值得完整复盘一遍。不管你是做AI应用开发、做企业级落地还是面试时想拿项目讲方案多智能体协作系统都是绕不开的话题。接下来我就按自己的实际拆解过程来聊先讲为什么单Agent不够用再讲角色怎么分、协作怎么调度、Agent之间怎么“对账”最后把我踩过的三个经典坑毫无保留地放出来。内容偏工程实践不绕理论。1. 单Agent忙不过来的真正原因为什么要上多智能体协作系统1.1 我的判断标准任务是不是“多工种”组合先说个最简单的判断方法。你要做的事情如果只靠一个人从头做到尾需要不停切换身份那它就是多工种任务。比如客服工单处理先理解用户诉求再查知识库匹配方案接着判断退款还是换货最后还要调接口执行操作。这中间至少有四件事每一件事的能力要求都不一样。我在早期版本里把这件事全交给一个Agent做。提示词里写了“你是客服助手请先理解用户问题再检索资料然后判断如何处理最后执行动作”。听起来很完整实际上效果很差。模型既要维护对话情绪又要做精准检索还要做业务决策提示词越长各段能力互相干扰越严重。经常出现一种情况前面还在好好分析后面突然自己造了一个“退款成功”的结论实际上什么都没执行。我后来换了个思路才真正理解多智能体协作系统的价值。核心不是“多个Agent更聪明”而是“把复杂任务拆成多个专业岗位每个岗位只干一件事并且干得可观察、可验证”。1.2 拆成多Agent之后我拿到的三个实际收益拆分之后最明显的收益不是效果提升而是问题可定位。以前单Agent出错你不知道错在理解、检索还是决策环节。拆成多个Agent后每个Agent有独立的输入输出日志一眼就能看出是哪一环出了问题。这个变化对线上调试来说价值极高。第二个收益是可替换性。如果知识库检索Agent的效果不好我只优化它一个模块不动其他代码。模型迭代也是一样。最早我用的是一个通用模型做意图识别后来换成了更擅长中文理解的大模型只改了Gateway Agent的模型配置其他Agent一点没动。第三个收益是可并行。四条工单进来如果每个工单都走“理解→检索→决策→执行”链路理论上四个工单可以被四个Agent管道并行处理。虽然受限于底层接口速度但调度层已经具备扩展能力。这一点对后面做吞吐量优化非常关键。1.3 不适合拆的三种任务先泼一盆冷水多智能体协作系统不是银弹。我自己也踩过“为拆而拆”的坑。总结下来有三种任务不适合上多Agent第一种是简单问答。用户问“你们营业时间是多少”一个Agent查一下就能答拆成四个角色纯粹是增加延迟和成本。第二种是强实时交互场景。比如语音助手每轮对话要求几百毫秒内返回。多Agent协作涉及多次模型调用时间不可控。硬件再优化也很难和单Agent相比。第三种是预算非常紧张的小型项目。每多一个Agent就多一次大模型调用成本成倍增加。如果场景简单真没必要为了追热点而上多智能体。提示判断要不要上多Agent先看你单Agent版本是不是遇到了“角色冲突”或“问题不可定位”。如果没有这两个痛点拆了反而更糟。2. 先把角色分清楚五类Agent和职责边界设计2.1 案例背景售后工单自动处理系统的五类角色我做的这个系统是处理售后工单的。用户在电商平台提交“申请退款”“商品破损”之类的问题系统自动理解诉求、检索售后政策、给出处理决策并执行退货或退款操作。在这个案例里我把多智能体协作系统划分为五类角色每一类对应一个Agent实例Agent名称核心职责关键输入关键输出Gateway Agent理解用户原始诉求清洗文本识别意图用户留言原文结构化意图、情感标签、关键字段Retrieval Agent从知识库/政策库检索匹配条目结构化意图政策条目ID列表、匹配置信度Decision Agent基于检索结果做出业务决策政策条目、用户订单信息决策结论、需要执行的动作列表Executor Agent调用外部接口执行退款/换货/人工审核决策结论执行结果回执Audit Agent审计整个决策链路发现异常并标记全部中间记录审计结论、风险标记这个案例里的五个Agent分工非常清楚。Gateway Agent不负责回答“能不能退”它只负责把用户乱七八糟的语言变成系统能处理的结构化字段。Retrieval Agent不看用户情绪它只负责“找出匹配的政策条款”。Decision Agent更纯粹拿到政策列表和订单数据输出一个决策结论。Executor Agent是最“无脑”的一个它接到明确的动作指令就执行执行结束原样上报结果。2.2 职责边界设计的两条铁律把角色分对了协作就成功了一半。但我在第一版踩过一个很典型的坑Decision Agent自作主张调用了退款接口。原因是提示词里写了一句话“如果符合退款条件请帮助用户完成退款。”模型真就调了接口结果用户的情况其实符合换货退款接口一调后续流程全乱了。所以我在重构时给自己立了两条铁律以后所有多智能体协作系统都遵守。第一条数据改动权唯一。任何一份数据在一个环节里只能有一个Agent可以修改。Decision Agent可以建议“退款50元”但真正把“退款状态”改成“已退款”的只有Executor Agent。Decision Agent连写权限的影子都不能有。第二条每个Agent只能依赖上游结果不能越级。Gateway Agent不能直接去查知识库Retrieval Agent不能自己决定“这个工单应该转人工”那属于Decision Agent的职责。一旦越级多个Agent同时对同一件事发表看法系统就会混乱。2.3 给Agent写“岗位说明书”而不只是提示词“岗位说明书”这个概念是我自己总结出来的。它的思路是你招聘一个员工不会只给他发一段“你要做一个好人”的口号。你会告诉他这个岗位的职责是什么、有什么权限、有什么工作流程、遇到啥情况可以上报。给Agent写提示词也一样。下面是我给Decision Agent写的岗位说明书模板可以做个参考你是一个售后决策Agent只负责一件事基于政策列表和订单信息输出退款/换货/转人工的决策。 输入数据 - policy_list: 匹配到的政策条目列表每一项包含policy_id、policy_title、policy_rule - order_info: 用户订单信息 - user_request: 结构化用户诉求 输出格式严格JSON { decision: approve_refund | approve_exchange | manual_review, reason: 决策依据说明, amount: 退款金额单位为分, policy_ids: [引用到的政策ID列表] } 禁止事项 1. 禁止执行任何实际接口调用退款/换货由 Executor Agent 执行 2. 禁止补充政策列表中没有的信息 3. 禁止在不确定时强行下结论叫 manual_review 转人工即可。这套写法的好处是边界感极强。Agent即使模型能力弱一些只要严格按输入输出规范来也不会做出太离谱的事。说实话我用这套提示词模板之后模型出错率比之前写长作文式的提示词低很多。3. 协作不用靠自觉任务队列、状态同步与消息回执的落地写法3.1 三种主流协作拓扑我的选择和理由多智能体协作系统的拓扑结构网上讨论很多简化下来无非三类中心化调度、链式传递、自由协商。中心化调度所有Agent不直接通信统一通过一个调度中心分配任务。优点是好管理、好排查、状态集中缺点是调度中心可能成为瓶颈。链式传递Agent A处理完传给BB传给C。适合流水线式任务缺点是任何一环挂了全局就停。自由协商Agent之间直接对话、互相请求。最灵活但状态管理混乱不适合生产级系统。我的选择是中心化调度加任务队列。尤其是在售后工单这个场景流程链路固定没有太多灵活协商的需求。中心化调度让整个系统的状态足够清晰。注意自由协商模式看起来很酷但别在生产环境轻易尝试。Agent之间的对话没有结构化协议约束你很难确认它们的会话到底走到了哪一步。3.2 调度器的核心代码骨架调度器的作用很简单接收任务、找到合适的Agent、派发任务、跟踪结果。下面是我重构后的调度器核心骨架Python代码可以当作参考模板。import uuid import time from dataclasses import dataclass, field from enum import Enum from queue import Queue class TaskStatus(str, Enum): PENDING pending RUNNING running DONE done FAILED failed NEEDS_REVIEW needs_review dataclass class Task: task_id: str field(default_factorylambda: uuid.uuid4().hex) agent: str payload: dict field(default_factorydict) status: TaskStatus TaskStatus.PENDING result: dict field(default_factorydict) created_at: float field(default_factorytime.time) updated_at: float field(default_factorytime.time) class Coordinator: def __init__(self): self.queue Queue() self.agents {} self.state {} def register(self, agent_name, agent_func): self.agents[agent_name] agent_func def submit(self, agent_name, payload) - str: task Task(agentagent_name, payloadpayload) self.state[task.task_id] task self.queue.put(task) return task.task_id def run_once(self) - Task: task self.queue.get() task.status TaskStatus.RUNNING task.updated_at time.time() try: result self.agents[task.agent](task.payload) task.result result task.status TaskStatus.DONE except Exception as exc: task.result {error: str(exc)} task.status TaskStatus.FAILED task.updated_at time.time() self.state[task.task_id] task return task def trigger_pipeline(self, initial_payload): 按工单处理流程编排各 Agent task_id self.submit(gateway_agent, initial_payload) while True: task self.run_once() if task.status TaskStatus.FAILED: return task if task.agent gateway_agent: self.submit(retrieval_agent, task.result) elif task.agent retrieval_agent: self.submit(decision_agent, task.result) elif task.agent decision_agent: if task.result.get(decision) manual_review: return task self.submit(executor_agent, task.result) elif task.agent executor_agent: return task这段代码里有几个关键设计值得展开。一个是task_id贯穿始终。一条工单从Gateway到Executor所有环节都通过同一个任务链路串起来。后面排查问题的时候拿着task_id去翻日志一切一目了然。另一个是状态机设计。我只保留了pending、running、done、failed、needs_review五个状态。没有做复杂的分布式状态管理因为这个场景还不需要。状态由调度器统一更新Agent不直接修改全局状态避免并发写冲突。3.3 我为什么坚持“任务回执”而不是Agent直接互相调用早期版本里我是让Agent直接互相调用的Gateway Agent处理完直接把结果传给Retrieval Agent。看起来省事但很快出了问题。最典型的问题是这个链路没有超时控制。Retrieval Agent连着调用了三次大模型接口超时了Gateway Agent还在傻等。由于它们之间是同步直接调用整个流程被一个子任务拖死没有中间层来干预。改成任务队列之后调度器成了掌控者。任何一个Agent超时或出错调度器捕获异常把任务状态置为failed或者转入needs_review。这个能力比Agent之间直接通信可靠得多。所以在多智能体协作系统里我的建议很明确Agent之间尽量不要直连所有通信都经过调度器这样问题才不会失控。4. Agent之间如何取得共识输入输出协议和双重校验设计4.1 结构化输出是第一版就该定的事Agent之间交流靠什么不靠自然语言靠结构化数据。很多初学者会把Agent的输出直接当作文本塞给下一个Agent这是大忌。自然语言本身有歧义下一个Agent理解上一段的“输出文本”时大概率会加戏。比如Retrieval Agent输出“用户符合七天无理由退换货条件”Decision Agent可能把它理解成“可以退款”但业务规则里“退换货”和“退款”是两件事情。我的做法是定义一个统一Schema所有Agent的输出都严格遵循这个结构。下面是一个标准返回示例{ task_id: a1b2c3d4, agent: retrieval_agent, status: done, result: { matched_ids: [policy_001, policy_014], policy_summary: 用户符合七天无理由退换货条件, confidence: 0.92 } }字段名称、类型、取值范围都要提前定义好。Decision Agent拿到这份JSON不需要解析自然语言直接读matched_ids就能判断该引用哪几个政策条目。这样即使某一次模型输出飘了至少JSON结构不会变下游解析不会崩。4.2 数字对账让Agent回答前先“自证清白”这是我在多智能体协作系统里最引以为傲的一个设计思路也是实测中最能拦住幻觉的手段。我称之为“数字对账”。具体做法是Retrieval Agent输出政策条目时必须给出policy_id列表。Decision Agent在决策时必须在policy_ids字段里引用它用到的政策ID同时给出置信度。调度器在中间做一次强制校验。如果Decision Agent引用的policy_id在Retrieval Agent返回的列表中不存在就直接标记为needs_review并给出原因“决策引用了未检索到的政策条目”。这个逻辑很像审计对账。采购单要有采购审批号报账单要有发票号两边对不上就要退货。Agent之间也必须有这种互验机制不能谁说了算。我甚至把校验逻辑抽出来做成了一个小函数在调度器里复用def validate_decision(retrieval_result, decision_result): retrieved_ids set(retrieval_result.get(matched_ids, [])) referenced_ids set(decision_result.get(policy_ids, [])) missing referenced_ids - retrieved_ids if decision_result.get(decision) manual_review: return True, manual_review if missing: return False, f引用了检索结果中没有的 policy_id: {missing} return True, ok实际跑下来这个机制至少拦住了两三成“看似合理、实则编造”的决策。4.3 失败信息不要被“包装”成成功另一个容易踩坑的地方是错误处理。Agent在调用外部接口失败时经常会把错误信息包装成正常输出返回。比如Executor Agent调用退款接口失败模型可能生成“用户已指导完成退款”这种结果但系统根本没有执行成功。所以我给所有Agent的错误返回都做了统一规范。错误必须带上错误码不能只是文本描述。错误码含义使用场景E_NOT_FOUND没有找到匹配数据知识库无对应政策E_TOOL_ERROR外部接口调用失败退款API超时、订单查询失败E_POLICY_REQUIRED需要人工审核高风险退款、金额异常E_LOW_CONFIDENCE模型置信度过低检索结果匹配度不足E_TIMEOUT处理超时Agent调用大模型超时这些错误码会随任务状态一起记录。调度器看到E_TOOL_ERROR直接进入重试逻辑看到E_POLICY_REQUIRED则转人工。宁可多转人工也不能把错误当成成功往下传。5. 实测翻车现场三个真实踩坑案例的完整排查链路再理论的内容不碰一碰坑都白搭。下面三个案例是我在调试这套多智能体协作系统期间真实遇到过的也是多Agent架构里最典型的翻车现场。5.1 幻觉传染一个Agent胡说另一个Agent跟着走现象是这样的。某个用户的订单金额是199元业务规则是金额低于200元的订单可以自动退款。Retrieval Agent不知道是不是在检索时漏掉了政策错误返回了一条“金额低于500元可自动退款”的虚构政策。Decision Agent信了因为它只看了Retrieval Agent的输出没有和原始政策库对账。它判断“199元小于500元符合条件”输出approve_refund。然后Executor Agent真的把退款指令发出去了。这个问题的核心在于错误在某一个环节产生后被下游当作“有效依据”层层放大全链路都没有拦截就像流言在人际传播中被不断加工。排查时我一开始以为是Decision Agent的问题单独测试它发现它只要拿到正确的政策列表就不会误判。后来把Retrieval输出和原始知识库比对才发现源头在Retrieval Agent的幻觉条目。修复方案有两层。第一层是加了来源ID对账——就是第4节提到的validate_decision逻辑Decision Agent引用政策时必须带上policy_id这些ID必须真实存在于检索结果中。第二层是限定Decision Agent只能基于给定的政策内容做判断不允许补充“常识性”业务规则。这两层叠加之后同类问题再没出现过。5.2 隐性死锁两个Agent互相等对方先表态这个坑比较隐蔽。业务里有一种场景用户买了两件商品申请部分退款。Decision Agent要判断的具体金额依赖Retrieval Agent返回的每件商品的价格而Retrieval Agent返回的价格又依赖Gateway Agent结构化出来的“申请退款清单”。从数据流上看这是一条链本身没毛病。但我在执行时用了两组并行任务一组处理退款资格一组处理金额计算。这两组任务在调度器里互等对方的结果。结果出现了A任务等B任务、B任务等A任务的死锁任务在队列里永远处于pending状态直到超时。排查时我直接去看调度器状态发现一条工单的处理时间从5秒涨到了120秒任务状态全是pending。再一拆看发现两个任务形成了循环依赖。这也是为什么我后来坚持使用简单线性的流水线编排Gateway→Retrieval→Decision→Executor坚决不在一个流程里搞并行子任务。如果真要并行就在调度器里显式定义依赖关系并使用asyncio或线程池来控制而不是让多个任务盲等。5.3 上下文爆炸Agent把整个对话历史传得到处都是第三个坑是成本与性能问题。早期我的实现很粗暴为了让Gateway Agent保留用户信息我让它在输出结构化意图的同时也把原始对话全文塞进回执Retrieval Agent收到后又决定保留这段对话“以备后用”又把它传给Decision Agent。结果每经过一个Agentpayload就膨胀一圈。到最后Executor Agent收到的上下文里包含了全部原始工单、多次重复的历史记录有时甚至超过了模型窗口长度导致调用报错。更危险的是上下文混淆。两个不同的工单如果被同一个Agent实例处理而上下文没有彻底隔离Agent甚至会把上一张工单的信息混进下一张工单的判断里。有一次系统差点把A用户的退款金额用B用户的订单金额去算。修复其实很简单每个Agent只保留自己需要的字段。Gateway输出结构化信息后原始对话就封存了不再往下传。下游需要什么字段就传什么字段其他一律不传。我专门写了一个字段白名单机制每个Agent在注册时声明自己需要的字段调度器按白名单过滤后再派发任务。这个改动同时解决了上下文爆炸和上下文串味两个问题。5.4 排查工具task_id贯穿所有日志三个坑排查下来我有一个共同心得日志里没有贯穿全链路的标志字段排查就是噩梦。现在的做法是从Gateway Agent收到工单那一刻起就生成一个task_id塞进所有日志、所有回执、所有错误信息里。每个Agent的每次调用都记录一条包含task_id、agent_name、上游task_id、耗时、返回摘要的结构化日志。import logging logger logging.getLogger(agent_trace) def log_agent_call(task_id, agent_name, status, duration_ms, summary): logger.info({ task_id: task_id, agent: agent_name, status: status, duration_ms: duration_ms, summary: summary, })找问题的时候就拿task_id去过滤能一口气把那件事从头到尾的走向全部拉出来。这算是所有高级机制落地前最值得先做好的基础设施。6. 这套系统还能扩展成什么样复用思路与个人建议6.1 从工单系统到通用协作模板做完这套系统后我最大的感受是多智能体协作系统本身是一种通用架构思想。把“理解—检索—决策—执行—审计”这条链路抽象出来可以套到非常多场景里。比如内容审核场景。Gateway Agent负责把用户上传的文本拆词和识别敏感类别Retrieval Agent去规则库匹配审核规则Decision Agent判断内容是否违规或转人工Executor Agent执行删除或提示修改。再比如数据处理场景。Gateway Agent理解用户查询意图Retrieval Agent去大数据平台取数Decision Agent生成报表口径Executor Agent把报表推送给用户。复用这套架构时变化最大的就是每个Agent内部用的提示词和业务数据源调度和通信框架基本不用动。这也是我觉得多智能体协作系统最大的价值它不是解决某一个具体场景而是提供了一套可以复用的工程范式。6.2 我的个人建议与技术路线上的一点感想最后说几个基于实操的建议。第一个建议是先手工模拟一遍流程再写代码。我自己第一版写太急直接写调度器结果流程漏洞百出。后来我在纸上把一条工单从入口到出口每一步用什么数据、什么格式、什么条件跳转都画清楚再动手写代码效率反而高很多。第二个建议是Agent数量宁少勿多。三个能解决的问题别用五个。每一个Agent都意味着额外的大模型调用延迟和费用。先跑通三个Agent链路再加复杂角色也不迟。第三个建议是一定要给每个Agent配置“不确定就说不确定”的策略。我在Decision Agent的岗位说明书里明确写了一条如果知识库里没有明确的政策依据输出manual_review转人工处理。这一句话帮系统拦住了大量潜在的误退款和客诉风险。第四个建议是别只盯着大模型的“聪明程度”多花时间在调度器、校验逻辑、状态同步这些工程细节上。多智能体协作系统的瓶颈往往不在单个Agent的理解能力而在系统能不能优雅地处理单个Agent的失败。这是我前后踩了两个多月坑才沉淀下来的一套体会。多智能体协作系统这个方向还会继续演进但底层那些“角色边界清晰、状态集中管理、输出可对账、失败可追踪”的原则我相信会长期适用。