企业官网建设流程全解析

前两个月我在做一个智能工单系统的时候，最头疼的并不是单个智能体效果不好，而是多个智能体怎么配合。明明每个单点能力都过得去，合到一起却经常出现互相覆盖、任务反复、上下文丢失，最后产出还不如一个老老实实写死的流程。后来我把架构从“一个大模型做所有事”改成“多个异构智能体协作”，才逐渐摸到门道。今天围绕“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_time，B 智能体读的是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 代理节点的基础封装

为了让不同类型的智能体能被统一管理，我给每个节点都做了同一套“外壳”，至少暴露三个方法：

写出来大概长这样：

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_factory=lambda: uuid.uuid4().hex) conversation_id: str = "" sender: str = "" receivers: list[str] = field(default_factory=list) performative: str = "inform" # ask, propose, accept, reject, inform payload: dict = field(default_factory=dict) reply_by: float = 0.0 timestamp: float = field(default_factory=time.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, key=lambda 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, key=lambda x: x["score"], reverse=True)[0]

confidence是智能体对自己答案的置信度，source_priority是数据源在业务上的可信等级。比如“官方知识库”的优先级高于“历史处理记录”。这样就算一个历史记录置信度很高，在权威知识库面前，也不至于单独冲上去误导用户。

冲突消解的关键在于，不能让冲突发生在“生成最终回复”那一刻，而要在聚合阶段提前判断。否则回答生成智能体拿到两个矛盾素材，很可能会自己“脑补”出一个错误答案。

3.6 超时、重试和降级

异构智能体协作里最现实的一个问题，就是每个成员响应速度完全不同。大模型可能要几秒，知识库检索可能只要几十毫秒，人工审核可能要几分钟。

我在系统里给每个子任务都设置了响应期限。reply_by字段就是干这个用的。调度器发送任务时会附带一个期限，如果超过期限没有收到结果，就触发兜底逻辑。

兜底策略分为三种：

• 重试：只对“可重试”的任务生效，比如网络超时。
• 降级：如果某个智能体反应太慢，直接用规则模板生成一个基础回复。虽然质量一般，但至少不会让用户一直等。
• 转人工：当结果置信度低于一定阈值，或关键节点执行失败，直接进入人工审核队列。

这三种策略必须提前配置，不能等线上出了事故再临时想。我见过太多系统把“重试”设置在用户请求路径上，导致用户等了一个请求又等一个请求，体验非常差。

4. 协作过程中最常见的坑和排查实录

4.1 消息格式不一致导致的“垃圾进垃圾出”

这个坑我在前面提过，但值得展开讲。有一次线上发现历史工单分析智能体输出的“处理结果”字段在回答生成阶段全部为空。

排查过程是这样的：先看日志，发现历史工单分析智能体明明返回了数据；再往下一层看，发现回答生成智能体读取的是result_text，而发送方给的是text。因为强转类型时失败，代码吞掉异常，直接返回空值。

这类问题最好的解决办法是引入“契约测试”。每个智能体在启动时先加载一份共享的 JSON Schema，发送方发送前按 Schema 序列化，接收方接受后按 Schema 校验。一旦字段缺失或类型不符，直接报出清晰的错误，而不是让数据带着错误继续往下走。

4.2 环路协作导致的消息风暴

去中心化协商模式里最典型的问题，就是死循环。A 智能体给 B 发消息，B 不太确定答案，又去问 C，C 觉得 A 可能有上下文，于是又去问 A。如果没有限制，这条消息会把整个系统拖垮。

我的处理办法是在消息总线上增加两个硬约束：

• 每条消息必须携带conversation_id和msg_id。
• 同一个会话内的消息跳数有上限，比如最多 10 跳。

总线每次接受到消息，都会检查该会话累计跳数。超过上限时，直接把后续消息丢弃，并向协调者报警。排查时可以通过msg_id串起整条消息链路，一眼就能看出循环发生在哪一步。

4.3 上下文漂移和记忆错位

多代理系统里，几个智能体共同处理一个任务时，很容易出现“上下文漂移”。比如一开始意图识别智能体判断工单是“硬件故障”，但后来知识库检索智能体给出了“网络设置”的资料，后续所有智能体都开始按“网络问题”的口径做事情。

问题根源是共享状态失控。所有智能体都在往同一个上下文对象里写数据，却没人定义谁有权限写哪个字段。

我后来引入了“写入权限矩阵”，每个字段都指定了唯一可写方。比如case_type只能由意图识别智能体写入，其他智能体如果觉得类型判断有误，只能通过“建议修改”消息提出，最终由协调者决定是否更新。这个改动让上下文的一致性提升了很多。

还有一点是“记忆隔离”。不同智能体应该有自己的工作记忆，不能所有历史消息都堆到一个上下文里。共享上下文只存放最终裁决后的状态，推理过程放在各自的消息记录里。

4.4 性能瓶颈定位与限流降级

多代理系统一旦跑起来，性能问题会是第一个暴露的问题。表现通常是：任务积压、队列变长、响应超时。

我排查时一般先看三个地方：

• 消息总线堆积量：如果异常上涨，先看是否有消息风暴。
• 大模型类智能体的调用耗时：这是最明显的瓶颈。
• 数据库或外部服务的慢查询：可能导致某个智能体整体卡住。

发现瓶颈后，先做“按会话限流”，同一个conversation_id最多允许多少并发。然后再做“动态降低采样”，当系统负载过高时，自动把一些可选智能体从调用链上摘除。比如非核心的“历史相似度分析”可以暂时关闭，等高峰期过去再恢复。

下面是我整理的一个速查表：

5. 让异构代理真正“像一个团队”的进阶技巧

5.1 能力注册与服务发现

智能体多了以后，靠硬编码去配置“谁做什么”会累死人。我建议从一开始就做“能力注册中心”。

每个智能体在启动时，上报自己的能力描述，包括三样：技能标签、输入输出格式、服务质量承诺。调度器在分发任务时，先根据技能标签过滤，再根据当前负载、历史成功率、平均延迟做综合排序。

这个机制还能解决“智能体新增替代者”的问题。旧智能体下线时，只需要把注册信息改为不可用，新智能体上线即可自动承接任务，不用改调度代码。

5.2 角色权限和信任等级

成熟的协作系统一定会给成员定角色。我的系统里分成四类角色：规划者、执行者、审查者、协调者。规划者负责拆任务，执行者负责干活，审查者负责挑错，协调者负责仲裁。

角色不同，权限不同。审查者可以驳回执行者的结果，但不能直接修改最终答案。执行者可以访问执行所需的数据，但无权修改全局状态。

我还引入了一个“信任等级”概念。历史成功率高的智能体，在冲突仲裁时拥有更高话语权；历史表现差的智能体，输出会被降低权重。这个不是用来惩罚，而是让系统在动态协作中逐渐学习“谁更靠谱”。

5.3 协作过程的可观测性与审计

多代理系统一旦出问题，最怕的就是“查不到现场”。我要求所有消息都必须被完整记录，包括消息内容、发送时间、接收方、处理结果。每个会话结束时，汇总成一份“协作轨迹”。

这个轨迹有两个作用。一是排障，直接看哪个环节出了问题。二是复盘，对照最终结果看每一步决策是否合理。

实际做的时候，我用了一张宽表保存消息记录，字段包括会话 ID、消息 ID、发送方、接收方、意图类型、耗时、错误码、摘要等。查询时按照会话 ID 聚合，配合一个简单的展示页面，基本能还原整个过程。

5.4 灰度上线与失败回滚

多代理系统里换一个智能体，不能像换一个普通模块那样直接发布。因为你没法确定新模型在协作环境里的行为会不会带来连锁反应。

我的做法是“影子模式”。新智能体先部署上线，但它的输出不会被正式使用，只会同时跑一份，和旧智能体的结果做对照。观察一段时间后，确认新智能体的错误率、响应时间、冲突次数都达标，再逐步切流。

如果出现问题，需要支持一键回滚到旧版本。这个回滚不能只是代码回滚，更关键的是把“会话路由配置”恢复原样，让新智能体不再接收任务。由于整个系统通过能力注册中心做路由，这个操作在管理界面上就能完成。

6. 一点个人体会：别急着做通用平台

做了这么多，我最想分享的体会是：不要一上来就做一个通用的多代理协作平台。先把一个具体场景跑通，哪怕场景小一点、规则多一点，都没关系。通用平台需要在大量场景中反复打磨才能沉淀出来，否则只会变成一堆无法落地的抽象概念。

另外，要接受一个事实：多代理系统永远会有不确定性。异构智能体协作再流畅，也总会有某个模型给出意料之外的输出。因此，系统里必须保留“人工介入”的口子。我给每个关键节点都加了一个人工审核开关，当整体置信度不高时，自动转人工。这不是在否定智能体的能力，而是在承认现实的复杂性。

最后分享一个小技巧：如果你的协作频繁出现“两队交锋、互不相让”，不妨先从消息协议里找问题，而不是去调模型。绝大多数协作混乱，都源于预期没有对齐，而不是智能体不够聪明。把消息格式、任务期限、冲突仲裁规则先定清楚，系统会立刻顺一大截。