前两个月我在做一个智能工单系统的时候最头疼的并不是单个智能体效果不好而是多个智能体怎么配合。明明每个单点能力都过得去合到一起却经常出现互相覆盖、任务反复、上下文丢失最后产出还不如一个老老实实写死的流程。后来我把架构从“一个大模型做所有事”改成“多个异构智能体协作”才逐渐摸到门道。今天围绕“AI多代理系统如何实现异构智能体协作”这个主题把设计思路、实现细节、踩坑过程都摊开讲。内容适合正在做 Agent 系统、或者准备把多个 AI 能力拼到一起的人参考尤其是那种“每个模块都正常但合起来就乱”的团队。我见过的多代理系统失败往往不是因为算法不够高级而是因为连“协作”两个字都没拆清楚。很多人把一堆 AI 接口拉到一个群里让它们自由发言最后得到的只是一场大型噪音。真正有价值的异构智能体协作是需要结构、协议和取舍的。1. 异构智能体到底在协作什么1.1 先拆开“异构”两个字“异构”听起来像学术黑话其实翻译成大白话就是一群出身不同、习惯不同、能力边界不同的个体被放到同一个任务里干活。我在实际项目里见过这么几种成员一两个基于大语言模型的推理智能体负责理解用户意图、生成自然语言回复。一个传统的规则引擎或脚本智能体负责查库、算状态、做精确判断。一个基于历史案例的检索智能体不擅长对话但擅长从历史数据里找出相似工单。一个“人肉智能体”也就是人工审核节点负责花时间少但错误成本高的决策。这些成员的模型架构不同、输出格式不同、响应速度不同连“成功”的定义都不同。比如脚本智能体说“成功”代表数据库字段更新完成大模型智能体说“成功”可能只是它觉得这么回答用户比较合适。所以异构智能体协作的第一件事不是写代码而是先承认它们之间没有共同语言。没有共同语言的团队效率再高也是各干各的。1.2 协作的本质不是调用 API而是对齐预期很多开发者刚开始做多代理系统时习惯把“调用另一个智能体”等同于“协作”。比如在 Prompt 里写一句“请调用售后知识库”然后等着结果回来。这种做法本质上是 API 调用不是协作。协作的关键是对齐预期。具体到工程上要回答几个问题一个智能体发出需求后其他智能体用什么格式响应如果没有人响应或者响应冲突由谁负责任务执行到一半发现信息不足能不能退回重来每个智能体返回的结果可信度怎么表示我常用“跨国项目组”来类比这件事。一个中国开发、一个美国产品、一个印度测试三个人都讲英语但“尽快”这个词在不同文化里含义完全不同。多代理系统也一样如果不提前定义好“响应期限”“结果格式”“置信度范围”一定会出现各种奇怪的偏差。对齐预期最好的方式是先把协议定死再让智能体在协议框架内自由发挥。协议不是用来限制智能体的而是用来保证它们之间的交互可以被理解、被追踪、被管理。1.3 多代理系统不是多线程更不是微服务有一类误解我必须说一下多代理系统听起来像微服务但本质上不是一回事。微服务解决的是“服务的拆分与复用”每个服务做什么、接口长什么样在开发时基本已经确定。调用关系也是静态的。而多代理系统里的智能体具有“自主性”和“协商能力”。同一个任务A 智能体可以拒绝执行B 智能体可以提出替代方案C 智能体可以要求补充信息。这种动态性是微服务架构完全不需要考虑的事情。多线程则更不一样。多线程是同一份代码在多个线程里跑大家共享内存通过锁和队列协调。多代理系统的成员可能是完全不同的代码、不同的模型甚至部署在不同的机器上彼此之间只能通过消息交流。所以如果你习惯用微服务的思路去设计多代理系统很容易掉进一个坑把所有智能体都当成“可以随时调用的工具”。但智能体是有“主观判断”的至少系统设计上要允许它有主观判断否则它跟普通服务没有区别也就没必要叫“多代理系统”了。2. 协作架构怎么搭三种主流模式的取舍2.1 中心化调度简单但容易变成瓶颈最直接的多代理协作模式是有一个调度中心统一负责一切。所有任务先交给调度中心调度中心自己做任务分解再把每个子任务分配给对应的智能体。智能体执行完把结果返回给调度中心由调度中心做汇总。这种模式的优点是好管理、好追踪而且天然适合强流程场景。比如“先查客户订单再判断退款原因最后生成回复”流程固定每个环节谁来执行是明确的。但我做下来发现它的缺点同样明显。第一调度中心的策略会越来越重所有业务规则都堆在它身上一旦调度逻辑出错整个系统瘫痪。第二智能体之间无法直接沟通如果任务处理过程中需要临时补充信息还得绕回调度中心转一圈延迟很高。第三调度中心成了单点不管是性能还是可用性风险都集中。中心化调度适合成员比较少、任务边界比较清楚的系统。一旦智能体数量超过几十个或者协作关系开始变得动态它就会非常吃力。2.2 去中心化协商灵活但要处理复杂一致性另一种极端是去中心化协商。每个智能体都通过某种消息媒介发布自己的需求其他智能体根据自身能力决定是否响应有点像招投标。这种方式非常灵活智能体可以随时加入、退出任务也能动态撮合。比如一个文本摘要智能体需要知道用户偏好它可以发一条“谁有用户标签数据”的消息有数据的智能体自愿响应。但代价也很明显收敛难。如果没有规则约束多个智能体互相响应很容易出现“你问我、我问你”的死循环。而且“一致性”非常难保证不同智能体对同一个状态的理解可能各不相同没有一个全局视角来解决冲突。去中心化模式更适合那种节点动态变化、需要探索式协作的场景。在工程落地时一般需要额外引入“协商超时”“消息最大跳数”“冲突仲裁规则”等机制否则系统会像一列没有轨道的火车跑得越快越危险。2.3 混合式协作我当前项目的选择真正让我觉得工程上靠谱的是第三种模式混合式协作。简单说就是保留一个“协调者”但不让它事无巨细地管所有事。协调者只负责两件事初始任务拆分和分配也就是引导整个系统进入正确的工作流。在协商失败或结果冲突的时候做最终仲裁。而智能体之间具体的沟通、数据交换、补充信息可以通过消息总线点对点完成。比如意图识别智能体发现信息不足可以直接向用户画像智能体发出请求不需要经过协调者。这种做法的好处是兼顾了稳定和灵活。协调者不会变成“所有请求都过一手”的瓶颈智能体之间可以快速互动。同时它保留了全局管控能力一旦出现异常协调者能介入并恢复秩序。我在工单系统里用的就是这个模式。协调者只负责把工单分发到第一层处理节点后面如果知识库检索智能体需要“用户历史工单”数据它会直接向历史分析智能体发消息。只有在多个候选答案冲突时协调者才站出来做最终选择。2.4 消息格式规范没有共同语言就是鸡同鸭讲不管选哪种架构消息格式都必须统一。这不是技术洁癖而是异构智能体协作的基本前提。我在一开始偷懒觉得“反正都是 JSON字段起什么名字无所谓”。后来发现A 智能体传的created_timeB 智能体读的是timestamp线上经常出现字段为空的诡异问题。还有一些智能体把置信度写成 0 到 1 的小数另一些写成 0 到 100 的整数聚合时直接算错排名。后来我设计了一套简化版的消息模型核心字段包括msg_id消息唯一 ID用于链路追踪。conversation_id整个任务的会话 ID所有相关消息共享这一个 ID。sender和receivers发送方与接收方支持广播。performative消息意图比如ask、propose、accept、reject、inform。payload具体业务数据。reply_by约定最后响应期限。timestamp发送时间。这个模型借鉴了经典的 Agent Communication Language 思路但做了大量简化。工程上不需要完整实现学术标准只需要把“问”“答”“建议”“接受”“拒绝”这几个关键动作定清楚就能覆盖绝大多数协作场景。语义层同样重要。不同智能体对“工单已解决”的理解必须一致否则状态永远对不上。我的做法是定义一份共享词汇表把所有关键字段的取值都写清楚。比如“工单状态”只允许open、processing、resolved、closed四种任何智能体都不能自创状态值。3. 从零实现一个异构多代理协作系统3.1 场景定义一个带不同类型智能体的工单处理系统理论讲再多不如直接上个例子。我以“智能工单处理系统”为背景来拆解整个实现过程。这个系统要完成的任务是用户提交一个工单比如“打印机无法连接报错 0x80004005”系统需要判断问题类型、检索可能解决方案、参考历史处理记录最后生成一段给用户的回复并且在必要时转人工。我设计了这么几个自治程度不同的成员意图识别智能体基于大语言模型负责判断用户问题属于硬件、软件、网络还是账务。知识库检索智能体基于传统文本检索负责从文档库中召回答案片段。历史工单分析智能体基于规则和统计模型负责查找相似历史工单及其处理结果。回答生成智能体基于大语言模型负责把前两步结果整合成可读回复。人工审核节点一个 Web 界面处理低置信度场景。这五个成员没有共同的技术栈也没有共享内部状态只能通过消息协作。这就是典型的异构智能体协作。3.2 代理节点的基础封装为了让不同类型的智能体能被统一管理我给每个节点都做了同一套“外壳”至少暴露三个方法方法作用can_handle(task)判断自己是否适合处理该任务返回能力匹配度和置信度execute(task, context)真正执行任务返回结果对象notify(message)接收异步消息用于处理其他智能体的补充请求写出来大概长这样class BaseAgent: def can_handle(self, task: dict) - bool: raise NotImplementedError def execute(self, task: dict, context: dict) - dict: raise NotImplementedError def notify(self, message: dict) - dict: raise NotImplementedError这套外壳的价值不在于代码量而在于统一边界。不管内部是调用一个大语言模型、跑一段 SQL还是等人工输入对外呈现出来的交互模式是一样的。这样调度层就不需要关心每个智能体内部到底怎么实现。需要说明的是can_handle返回的不能只是布尔值最好带上一个能力评分。比如知识库检索智能体对“打印机错误”的处理匹配度可能是 0.92对“订单退款”的匹配度只有 0.2。有了评分之后任务分发才能做出有依据的选择。3.3 消息总线的核心实现有了节点外壳下一步是实现消息传递层。我用的是一套极简消息总线任何智能体都可以往总线发消息也可以订阅自己关心的消息。消息对象我用数据类来表示from dataclasses import dataclass, field import time import uuid dataclass class AgentMessage: msg_id: str field(default_factorylambda: uuid.uuid4().hex) conversation_id: str sender: str receivers: list[str] field(default_factorylist) performative: str inform # ask, propose, accept, reject, inform payload: dict field(default_factorydict) reply_by: float 0.0 timestamp: float field(default_factorytime.time)这里最关键的是performative字段。它让消息不再只是“数据”而带有“动作意图”。ask表示请求信息或请求执行。propose表示给出方案或候选结果。accept表示接受方案。reject表示拒绝。inform表示单方面通知状态。为什么要这么做因为当智能体能明确表达“我在问你”“我在建议”“我不同意”的时候系统才能有针对性地处理冲突而不是把所有消息都当成普通数据往上传。我又维护了一张“会话路由表”记录conversation_id对应的当前参与节点。新消息进来时总线根据receivers字段投递若receivers为空则广播给所有已注册节点。3.4 任务分解与分配策略调度器拿到一个用户工单后并不是直接把整个工单丢给某个智能体而是先做任务分解。分解有两种方式。一种是规则驱动适合流程非常固定的场景。比如先做意图识别再做知识库检索最后生成答案。另一种是模型驱动让大模型动态决定要拆成哪些步骤。模型驱动看着灵活但可重复性差容易跑偏。我的建议是能写死就写死只有规则覆盖不了时才用模型补。分解完成后进入分配阶段。我把每个任务需要的能力用一个标签表示比如text_classification、document_retrieval、summarization。每个智能体启动时会把自己的能力标签和负载情况注册到能力注册中心。分配时的选择逻辑大概是def dispatch_task(task): required_skill task[required_skill] candidates registry.find_by_skill(required_skill) best min(candidates, keylambda agent: agent.load_score agent.latency) return best这个代码很简单但实际中我会把候选人的“能力匹配度”“当前排队长度”“平均响应时间”综合成一个分数避免把所有任务都堆给响应最快但实际负载已经很重的智能体。3.5 结果聚合与冲突消解当多个智能体都返回结果时麻烦就来了。以我的工单系统为例知识库检索智能体和历史工单分析智能体可能给出两种不同的处理建议。一种建议是“重启打印服务”另一种是“更新打印机驱动”。单独看都有道理但用户不能同时执行两条矛盾指令。聚合阶段我做了一个简单的评分函数def aggregate(answers): for ans in answers: ans[score] ans[confidence] * 0.7 ans[source_priority] * 0.3 return sorted(answers, keylambda x: x[score], reverseTrue)[0]confidence是智能体对自己答案的置信度source_priority是数据源在业务上的可信等级。比如“官方知识库”的优先级高于“历史处理记录”。这样就算一个历史记录置信度很高在权威知识库面前也不至于单独冲上去误导用户。冲突消解的关键在于不能让冲突发生在“生成最终回复”那一刻而要在聚合阶段提前判断。否则回答生成智能体拿到两个矛盾素材很可能会自己“脑补”出一个错误答案。3.6 超时、重试和降级异构智能体协作里最现实的一个问题就是每个成员响应速度完全不同。大模型可能要几秒知识库检索可能只要几十毫秒人工审核可能要几分钟。我在系统里给每个子任务都设置了响应期限。reply_by字段就是干这个用的。调度器发送任务时会附带一个期限如果超过期限没有收到结果就触发兜底逻辑。兜底策略分为三种重试只对“可重试”的任务生效比如网络超时。降级如果某个智能体反应太慢直接用规则模板生成一个基础回复。虽然质量一般但至少不会让用户一直等。转人工当结果置信度低于一定阈值或关键节点执行失败直接进入人工审核队列。这三种策略必须提前配置不能等线上出了事故再临时想。我见过太多系统把“重试”设置在用户请求路径上导致用户等了一个请求又等一个请求体验非常差。4. 协作过程中最常见的坑和排查实录4.1 消息格式不一致导致的“垃圾进垃圾出”这个坑我在前面提过但值得展开讲。有一次线上发现历史工单分析智能体输出的“处理结果”字段在回答生成阶段全部为空。排查过程是这样的先看日志发现历史工单分析智能体明明返回了数据再往下一层看发现回答生成智能体读取的是result_text而发送方给的是text。因为强转类型时失败代码吞掉异常直接返回空值。这类问题最好的解决办法是引入“契约测试”。每个智能体在启动时先加载一份共享的 JSON Schema发送方发送前按 Schema 序列化接收方接受后按 Schema 校验。一旦字段缺失或类型不符直接报出清晰的错误而不是让数据带着错误继续往下走。4.2 环路协作导致的消息风暴去中心化协商模式里最典型的问题就是死循环。A 智能体给 B 发消息B 不太确定答案又去问 CC 觉得 A 可能有上下文于是又去问 A。如果没有限制这条消息会把整个系统拖垮。我的处理办法是在消息总线上增加两个硬约束每条消息必须携带conversation_id和msg_id。同一个会话内的消息跳数有上限比如最多 10 跳。总线每次接受到消息都会检查该会话累计跳数。超过上限时直接把后续消息丢弃并向协调者报警。排查时可以通过msg_id串起整条消息链路一眼就能看出循环发生在哪一步。4.3 上下文漂移和记忆错位多代理系统里几个智能体共同处理一个任务时很容易出现“上下文漂移”。比如一开始意图识别智能体判断工单是“硬件故障”但后来知识库检索智能体给出了“网络设置”的资料后续所有智能体都开始按“网络问题”的口径做事情。问题根源是共享状态失控。所有智能体都在往同一个上下文对象里写数据却没人定义谁有权限写哪个字段。我后来引入了“写入权限矩阵”每个字段都指定了唯一可写方。比如case_type只能由意图识别智能体写入其他智能体如果觉得类型判断有误只能通过“建议修改”消息提出最终由协调者决定是否更新。这个改动让上下文的一致性提升了很多。还有一点是“记忆隔离”。不同智能体应该有自己的工作记忆不能所有历史消息都堆到一个上下文里。共享上下文只存放最终裁决后的状态推理过程放在各自的消息记录里。4.4 性能瓶颈定位与限流降级多代理系统一旦跑起来性能问题会是第一个暴露的问题。表现通常是任务积压、队列变长、响应超时。我排查时一般先看三个地方消息总线堆积量如果异常上涨先看是否有消息风暴。大模型类智能体的调用耗时这是最明显的瓶颈。数据库或外部服务的慢查询可能导致某个智能体整体卡住。发现瓶颈后先做“按会话限流”同一个conversation_id最多允许多少并发。然后再做“动态降低采样”当系统负载过高时自动把一些可选智能体从调用链上摘除。比如非核心的“历史相似度分析”可以暂时关闭等高峰期过去再恢复。下面是我整理的一个速查表故障现象可能原因排查思路消息丢了bus 层无持久化检查消息中间件是否开启 ack必要时落盘回复为空字段名不一致看发送方与接收方 Schema 是否一致任务反复执行幂等键缺失检查是否携带唯一任务 ID处理前查重上下文错乱多智能体写同一字段检查写入权限矩阵响应超时大模型调用过慢增加超时时间或降级系统环路协商没有跳数限制检查 hops 上限和会话追踪5. 让异构代理真正“像一个团队”的进阶技巧5.1 能力注册与服务发现智能体多了以后靠硬编码去配置“谁做什么”会累死人。我建议从一开始就做“能力注册中心”。每个智能体在启动时上报自己的能力描述包括三样技能标签、输入输出格式、服务质量承诺。调度器在分发任务时先根据技能标签过滤再根据当前负载、历史成功率、平均延迟做综合排序。这个机制还能解决“智能体新增替代者”的问题。旧智能体下线时只需要把注册信息改为不可用新智能体上线即可自动承接任务不用改调度代码。5.2 角色权限和信任等级成熟的协作系统一定会给成员定角色。我的系统里分成四类角色规划者、执行者、审查者、协调者。规划者负责拆任务执行者负责干活审查者负责挑错协调者负责仲裁。角色不同权限不同。审查者可以驳回执行者的结果但不能直接修改最终答案。执行者可以访问执行所需的数据但无权修改全局状态。我还引入了一个“信任等级”概念。历史成功率高的智能体在冲突仲裁时拥有更高话语权历史表现差的智能体输出会被降低权重。这个不是用来惩罚而是让系统在动态协作中逐渐学习“谁更靠谱”。5.3 协作过程的可观测性与审计多代理系统一旦出问题最怕的就是“查不到现场”。我要求所有消息都必须被完整记录包括消息内容、发送时间、接收方、处理结果。每个会话结束时汇总成一份“协作轨迹”。这个轨迹有两个作用。一是排障直接看哪个环节出了问题。二是复盘对照最终结果看每一步决策是否合理。实际做的时候我用了一张宽表保存消息记录字段包括会话 ID、消息 ID、发送方、接收方、意图类型、耗时、错误码、摘要等。查询时按照会话 ID 聚合配合一个简单的展示页面基本能还原整个过程。5.4 灰度上线与失败回滚多代理系统里换一个智能体不能像换一个普通模块那样直接发布。因为你没法确定新模型在协作环境里的行为会不会带来连锁反应。我的做法是“影子模式”。新智能体先部署上线但它的输出不会被正式使用只会同时跑一份和旧智能体的结果做对照。观察一段时间后确认新智能体的错误率、响应时间、冲突次数都达标再逐步切流。如果出现问题需要支持一键回滚到旧版本。这个回滚不能只是代码回滚更关键的是把“会话路由配置”恢复原样让新智能体不再接收任务。由于整个系统通过能力注册中心做路由这个操作在管理界面上就能完成。6. 一点个人体会别急着做通用平台做了这么多我最想分享的体会是不要一上来就做一个通用的多代理协作平台。先把一个具体场景跑通哪怕场景小一点、规则多一点都没关系。通用平台需要在大量场景中反复打磨才能沉淀出来否则只会变成一堆无法落地的抽象概念。另外要接受一个事实多代理系统永远会有不确定性。异构智能体协作再流畅也总会有某个模型给出意料之外的输出。因此系统里必须保留“人工介入”的口子。我给每个关键节点都加了一个人工审核开关当整体置信度不高时自动转人工。这不是在否定智能体的能力而是在承认现实的复杂性。最后分享一个小技巧如果你的协作频繁出现“两队交锋、互不相让”不妨先从消息协议里找问题而不是去调模型。绝大多数协作混乱都源于预期没有对齐而不是智能体不够聪明。把消息格式、任务期限、冲突仲裁规则先定清楚系统会立刻顺一大截。