我一直有个执念:让 AI 不再只是“被调用”,而是学会自己找人商量。OpenClaw 项目第一次让我觉得这个想法变得可落地。它不是什么宏大平台,而是一个偏向单机或中小规模集群的智能体运行框架;核心亮点是内置了一条支持智能体之间互相通信的消息总线。简单说,你配置两个智能体,它们各自的提示词和工作流不需要强行塞进同一个 prompt 里,而是通过订阅、发送、回应消息来完成协作。这篇文章就是我在调通这套机制后的完整复盘,既讲设计思路,也贴可以直接抄的配置,最后还整理了我在实际调试中踩过的坑。
如果你正在做智能客服、自动化运维、研究助手,或者只是好奇多个 AI 角色怎么一起干活,这篇文章应该对你有用。我尽量用“两个人怎么配合工作”的语气来写,把通信协议、消息路由、失败重试这些听起来很硬的东西拆开看。放心,不需要你之前写过任何多智能体代码,只要会看 YAML 和 JSON,就能跟着搭出一个最小可跑的例子。
1. 为什么需要智能体相互对话
1.1 单智能体的天花板
单智能体并不是不行,而是做得越大越痛苦。拿一个“库存管理助手”举例:你希望它接收入库单、判断是否缺货、生成补货建议、预排配送时间。如果全部塞进一个系统里,通常有三种结果。
第一种是提示词越来越长。你需要把所有业务规则、历史数据、工具描述、输出格式全写进系统提示词,很快上下文窗口就被占满,模型反而会忽略中间的关键规则。第二种是责任边界模糊。同一个智能体既要理解仓库语言,又要理解运输调度语言,遇到冲突时它会“和稀泥”,而不是像两个专业角色那样各自坚持判断。第三种是状态难以隔离。一个任务出错会把整个会话状态污染掉,比如一次错误的时间判断会让之后的补货建议全部偏移。
OpenClaw 处理这个问题的方式很直接:把大智能体拆成小智能体,让它们像同事一样通过消息协作。你不追求一个脑子什么都懂,而是让一个脑子只管自己的领域,然后把领域之间的衔接交给通信层。
1.2 通信让能力变成组合
我常用一个比喻:单智能体像一个全能实习生,什么都做,但每件都做不深;多智能体沟通则像一个项目组,有人写文案,有人算数据,有人催进度,各干各的,再通过会议同步信息。
通信的价值不在于“能互相发消息”这个动作本身,而在于组合能力。比如一个情感分析智能体输出的结果,可以直接作为另一个风险提示智能体的输入;一个调度智能体发现车辆不足,可以抛出一个“运力缺口事件”,让采购智能体决定是否临时租车。每个智能体只维护自己的知识图谱、工具列表和决策规则,但通过消息组合,整体能力会超出任何单一模型。
这种模式还有一个隐藏优势:可以单独升级某一个智能体。如果新版本模型在调度上更强,你只需要替换调度智能体,不用动仓库智能体。通信协议不变,模型内部怎么换都行。
1.3 谁真正需要这套机制
不是所有场景都该上多智能体。如果单个 prompt 加几个工具就能搞定,硬拆反而会导致消息过多、问题难排查。我建议你考虑通信机制,一般是因为出现了这三个信号中的至少一个:
- 业务涉及多个领域,每个领域都有独立的术语和判断标准,比如财务、物流、客服。
- 有多个不同来源的任务,需要按优先级或条件交给不同角色处理,比如工单分派。
- 你希望让每个角色的行为可独立测试、独立复盘,而不是在一个大黑盒里猜来猜去。
我见过一个团队用 OpenClaw 做工单系统,客服智能体负责理解用户诉求,技术智能体负责定位代码问题,财务智能体负责判断退款路径。三个智能体之间通过一张消息网协作,比原来一个巨型智能体稳定很多。这正是通信层的价值所在。
2. OpenClaw 的通信模型与设计思路
2.1 一个简单的“对话层”:消息总线
OpenClaw 的核心通信组件叫 AgentBus,可以理解为一个在智能体之间传递消息的内部邮箱。所有智能体不直接握着对方的连接句柄,而是把消息投递到总线上,总线根据主题把消息分发给订阅者。
为什么不直接让智能体互相调用接口?因为直接调用接口会把系统耦合成一张蜘蛛网。A 要调用 B,B 要调用 C,C 又要回掉 A,任何一个人接口变更都会连锁爆炸。总线模式更像编辑部:每个人把稿子放到公共投稿区,谁负责哪一类就去取,没有人需要知道别人坐在哪张桌子。
在 OpenClaw 里,AgentBus 既支持内存模式,适合本地调试;也支持持久化模式,适合记录所有历史消息。我一般调试时用内存模式,部署到长期运行的服务里再开启持久化。
2.2 消息协议:信封 + 内容
智能体之间的消息长什么样?我用了一个信封加内容的模型。信封负责路由,内容负责表达。一个典型消息如下:
{ "message_id": "8f1f2a3b-77c4-4d2e-9a12-88f9e2d3d001", "sender": "inventory_agent", "topic": "warehouse.inbound", "recipient": "dispatch_agent", "reply_to": "inventory_agent", "priority": "high", "timestamp": "2025-04-12T09:30:00+08:00", "correlation_id": "order-20250412-0001", "payload": { "sku": "A100", "quantity": 20, "location": "A-3-02", "need_transport": true } }信封里的topic是最重要的,它决定了消息该送给谁。sender和recipient分别是发起方与目标方,reply_to则留下回信地址。correlation_id用于把一组相关消息串起来,方便事后追溯整条协作链路。
内容部分我建议尽量只放结构化数据,不要放一大段自然语言。两个智能体之间说“你好,请问可以帮我送一下吗”很容易产生歧义,反而没有{"need_transport": true}来得可靠。自然语言可以放在备注字段里给人类留痕,但路由和判断尽量用结构化字段。
2.3 为什么不用直接调函数
你可能会有疑问:既然两个智能体部署在同一套系统里,为什么不直接定义一个 Python 函数,让 A 调用 B 的assign_truck()方法?
直接调函数的问题在于调用时机的耦合。如果一个函数同步等待结果,那么 A 就被 B 的计算时间卡住。当 B 需要调用外部接口、等待模型返回时,A 就白等在那里。而消息机制天然是异步的,A 发出消息后可以继续处理自己的事情,等 B 的回信到达时再继续。
另一个问题是故障传播。函数调用失败会直接抛出异常,让 A 的处理流程中断;消息失败则只会让这条消息进入重试队列,不会影响 A 当前正在处理的其他任务。对于长时间运行的多智能体系统来说,这种解耦非常值钱。
2.4 任务编排 vs 自由闲聊
OpenClaw 同时支持两种通信风格:一种是“任务编排”,由中央调度指定谁做什么;另一种是“自由活动”,智能体根据订阅主题自己去接活。
我更推荐你优先用订阅发布做粗调度,再用定向消息做细控制。先让每个智能体声明自己关心什么主题,比如库存智能体订阅warehouse.inbound、调度智能体订阅transport.assign,然后系统自然会把消息送到该去的地方。等逻辑复杂后,再引入一个中央协调智能体来拆解大型目标,避免彼此争论谁先谁后。
自由闲聊看起来更酷,但也更容易失控。两个智能体在一个主题上反复回应对方,可能形成死循环。所以我在所有通信配置里都加上max_hops,限制一条消息最多被转手几次。
3. 核心落地:让两个智能体在 OpenClaw 里对话
3.1 最小环境准备
OpenClaw 的安装不算复杂。它基于 Python 3.10 以上版本,安装时只需要拉取代码仓库,创建虚拟环境,再执行依赖安装。你不需要 GPU 也能跑通通信机制,因为消息路由本身不消耗模型推理;只有在智能体实际调用模型时,才需要本地或远程的模型服务。
我建议至少准备两个不同的配置目录,分别代表两个智能体。这个隔离很重要:如果公用一个配置目录,两个智能体可能会共享状态,调试时很难判断谁做了什么。
3.2 定义两个智能体
先看第一个智能体,我把它定义为库存管理员:
name: inventory_agent description: "负责处理入库消息和库存查询" llm: provider: local model: local-chat listen: topics: - warehouse.inbound - stock.query respond: default: "处理订阅主题中的消息,并输出结构化结果" context: system_prompt: "你是一名仓库管理员,只负责入库登记和库存核对。"再看第二个智能体,负责物流调度:
name: dispatch_agent description: "负责根据入库消息安排车辆" llm: provider: local model: local-chat listen: topics: - transport.assign - warehouse.inbound respond: default: "收到入库消息后,判断是否需要安排车辆" actions: - name: notify_driver type: http url: "http://localhost:9101/driver/notify" method: POST headers: content-type: application/json注意,两个智能体都订阅了warehouse.inbound,但处理逻辑不同。库存管理员只登记库存,调度员则判断是否需要派车。这种“同一事件,两个角色各自反应”的模型非常灵活。
3.3 开启 AgentBus 并注册智能体
先启动总线服务。在 OpenClaw 安装目录下执行:
openclaw bus start --port 7801然后分别注册两个智能体:
openclaw agent register --file agents/inventory_agent.yaml openclaw agent register --file agents/dispatch_agent.yaml注册完成后,openclaw status可以看到两个智能体都在线,并且各自订阅的主题已经挂载到总线上。此时总线已经知道“谁对什么主题感兴趣”。
3.4 设置订阅规则
订阅规则可以写在多个地方。一个最直接的方式是在 YAML 里用listen字段。但如果你希望动态调整,也可以在运行时用命令增加订阅:
openclaw subscribe --agent dispatch_agent --topic warehouse.inbound openclaw unsubscribe --agent dispatch_agent --topic stock.query我特别想强调:订阅不是越多越好。智能体每多订阅一个主题,它就多一分上下文混杂的风险。比如调度员订阅了所有仓库事件,结果看到一条“保洁签到”的消息也尝试处理,就会浪费大量 token。尽量让订阅范围窄而明确。
3.5 发送消息与查看响应
现在手动投递一条入库消息到总线,模拟外部系统传入事件:
openclaw topic send warehouse.inbound --payload \ '{"sku":"A100","quantity":20,"location":"A-3-02","need_transport":true}' \ --id order-20250412-0001执行后,库存管理员会先响应,登记当前库存;调度员随后判断需要派车,并调用notify_driver动作。你可以在终端看到类似下面的日志:
[agentbus] warehouse.inbound <- inventory_agent [inventory_agent] 入库登记完成:sku=A100, quantity=20 [agentbus] warehouse.inbound <- dispatch_agent [dispatch_agent] 需要安排车辆,车型建议:小型厢式车 [dispatch_agent] action notify_driver success这就是一次完整的智能体间通信。消息从外部进入总线,被多个智能体各自订阅并处理,每个智能体只回应自己职责范围内的部分。
4. 实战实录:一个“快递中转”场景
4.1 场景设定
我们做一个更完整的场景:某仓库收到一批货物,需要智能体之间协作完成入库、派车、通知收货人三个动作。整个链路我设计了三个角色,但在 OpenClaw 里角色越多,通信协议越需要提前设计好。
- 仓库智能体:负责入库登记,更新库存。
- 调度智能体:负责判断车型和派车时间。
- 通知智能体:负责向收货人发送消息。
三个角色之间不是串行调用,而是通过事件驱动。仓库智能体发出一条warehouse.inbound,调度智能体收到后计算运力,再发出transport.assigned。通知智能体订阅了transport.assigned,最后发送通知。
4.2 消息流拆解
第一步,仓库智能体接管入库事件。它不需要知道后面是谁在处理运输,只需要把事实告诉总线:
{ "topic": "warehouse.inbound", "correlation_id": "order-20250412-0001", "payload": { "sku": "B200", "quantity": 50, "pickup": "A区-03", "destination": "滨河路仓库", "urgent": false } }第二步,调度智能体订阅了warehouse.inbound,计算得到“需要一辆 4.2 米厢式车”,随后发送新消息:
{ "topic": "transport.assigned", "correlation_id": "order-20250412-0001", "sender": "dispatch_agent", "payload": { "vehicle": "truck_4m2", "driver": "driver_1024", "departure": "2025-04-13T08:00:00+08:00" } }第三步,通知智能体收到transport.assigned后,给收货人发送短信或者应用内通知。注意,通知智能体没有直接收到最初的入库消息,它只关心“是否已经分配了车”。通过correlation_id,它可以把这条消息和之前的事件串起来。
4.3 用一张配置实现场景
你只需要定义一个总入口配置,把三个智能体串起来:
agents: - file: agents/warehouse_agent.yaml - file: agents/dispatch_agent.yaml - file: agents/notify_agent.yaml bus: type: persistent port: 7801 ack_timeout: 30 max_retries: 3 max_hops: 3这里的max_hops: 3非常关键。如果不限制跳数,万一通知智能体对调度智能体的消息又回复了一条transport.assigned.ack,而调度智能体又对这个确认消息再次响应,就会形成循环。限制跳数是兜底方案,即使业务逻辑写错了,通信链路也不会无限转圈。
4.4 失败与补偿
现实中消息不是总能一次成功。比如通知智能体调用外部短信服务超时,此时 OpenClaw 会让消息进入重试队列,按照max_retries重试。如果三次后仍然失败,系统会生成一条delivery.failed事件,让调度智能体知道需要换一个通知渠道,而不是让任务默默失败。
这种补偿机制让整个链路更接近真实团队协作:同事们不会因为一次没联系上就散伙,而是会另找办法。智能体间通信也应该具备同样的容错能力。
5. 常见问题与排查技巧实录
5.1 智能体收了消息却不回应
这是最常见的问题。你看到消息已经被总线接收,但某个智能体就是没有后续动作。我排查时一般按这个顺序:
- 确认智能体是否订阅了正确主题。订阅是精确匹配还是带通配符,需要看
listen里的配置。 - 确认智能体是否处于忙碌状态。如果它正在处理一条长耗时消息,新消息会排队。
- 确认模型调用是否成功。本地模型并发能力不够时,会因为排队而显得“不回应”。
- 确认消息的
priority是否被策略丢弃。低优先级消息在高负载时确实可能被延时处理。
在 OpenClaw 里,openclaw bus log --topic warehouse.inbound可以查看所有经过总线的消息记录。只要能看到消息被投递,但智能体没有响应日志,问题多半出在智能体自身或模型调用上。
5.2 消息风暴与循环对话
我在调试阶段经历过一个经典事故:两个智能体都订阅了system.heartbeat,库存智能体收到后回复了一条stock.status,调度智能体收到stock.status后又回了一条transport.check,库存智能体又因为订阅了transport.check再次回复。几分钟内,系统产生了上千条消息。
解决办法有三个。第一,给每个主题都加上单向上传标签,明确该主题事件是由外部系统产生还是由内部智能体产生。第二,允许多跳,但绝不能允许同一correlation_id下出现同主题重复处理。第三,设置max_hops: 3以及每分钟最大处理条数。
现在我把所有智能体配置里的无限重试全部改成了带次数上限的重试。这不是怕系统崩,而是怕调试时被消息洪水淹没,什么线索都找不到。
5.3 上下文丢失问题
智能体在连续处理多条消息时,容易把前面的对话忘记。尤其当它订阅了多个主题,每处理一条新消息都要读取系统提示词和最近几条历史,上下文窗口很快占满。
我的做法是给每个智能体配置“摘要记忆”。处理完一段业务后,让智能体把关键结果压缩成两三段摘要,再跟消息 ID 一起存起来。下次处理同一correlation_id时,只需读取摘要,而不是把所有历史消息都塞进提示词。这能显著减少 token 消耗,也避免模型被无关历史干扰。
OpenClaw 的配置里可以这样开启:
memory: strategy: summary key_field: correlation_id max_summary_length: 500这样每条业务链路只保留最重要的信息,而不是把所有原始消息堆在上下文里。
5.4 消息安全:别让任何人冒充任何人
消息总线如果不做鉴权,智能体之间很容易出现“冒充发送”的风险。比如某个恶意脚本伪造一条warehouse.inbound,调度智能体就会产生错误派车。我在生产配置里做了两件事。
第一,为每个智能体生成独立的访问密钥,总线只接受携带正确密钥的消息。第二,在消息信封里加入expected_recipient字段,接收方校验这个字段是否与自己匹配,不匹配就丢弃并告警。
还有一点容易忽略:不要把敏感数据直接放在消息内容里。外部通知服务可能拿到整条消息,所以密码、手机号、内部地址都应该用引用 ID 代替,真实验证时再由对应智能体查询。
5.5 性能调优的几个参数
如果你发现通信链路跑得慢,先别急着升级机器,先看消息总线瓶颈。我常用的调优参数如下:
| 参数 | 默认值 | 调优方向 |
|---|---|---|
queue_size | 1024 | 单主题消息积压较多时调大 |
worker_count | 4 | 智能体 CPU 充足时调大 |
ack_timeout | 30 | 模型响应不稳定时调大 |
batch_max_size | 10 | 允许批量处理时调大 |
max_hops | 3 | 不要随意调大,越大越危险 |
实测下来,最先卡住的往往不是模型,而是消息总线在处理持久化和消息确认时的 IO。把bus.type从persistent切成memory,测试速度会快很多。所以线上需要持久化,但本地调试尽量用内存模式省时间。
6. 经验心得与后续扩展
6.1 把通信协议当成一等公民
很多人做多智能体,注意力全放在写提示词上,却忘了定义“消息方言”。我踩过几次坑后,现在会先画一张消息表:每条消息从哪个角色发出、发送到哪个主题、载荷有哪些字段、由谁消费。这张表比任何架构图都管用,因为通信的语义一开始就清晰了,后面写智能体才不会各说各话。
如果两个智能体对quantity字段理解不同,一个认为是箱数,一个认为是件数,那通信越顺畅反而越容易造成业务事故。推荐在启动任何智能体之前,先统一单位、时间格式和字段命名。
6.2 小步快跑,别一上来搞全互联
全互联听起来很高端,但调试是灾难。我建议先跑通两个智能体,再逐步加第三个。每次加一个,都要单独验证新增的订阅关系和消息格式,避免一次引入多个变量。用最少角色跑通一条端到端链路,哪怕它只是在本地打印日志,也比一个看起来复杂但一运行就报错的系统有价值得多。
我用 OpenClaw 通常都是先起两个智能体,手动发送一条消息,观察日志;确认无误后,再让一个智能体自动回复另一个智能体的消息;最后才把外部系统接进来。每个阶段都留好回滚点。
6.3 记录所有消息,回放是调试利器
持久化总线最大的好处不是“数据更安全”,而是可以回放。遇到问题时,我会把某条correlation_id对应的所有历史消息导出来,按时间排序看每一步发生了什么。很多时候,问题不是“消息没送到”,而是“消息被处理了两次”或“回复主题写错了”。
建议你在写智能体逻辑时,顺手把输入消息的关键字段打进日志。不需要记录完整 prompt,只需要记录主题、发送方、接收方、载荷摘要和处理结果。这样排查问题时,你能很快从一堆消息中找到断点。
6.4 后续还能怎么扩展
通信基础一旦跑通,玩法就多了。比如可以加一个群聊频道,让多个智能体对一个共同目标发表各自意见,再用仲裁智能体做最终决策;也可以加记忆共享,让智能体之间交换“长期记忆”,避免重复劳动;还可以把外部工具动作包装成消息动作,让任何智能体都能触发同一条工具链。
我个人最看好“仲裁模式”:多个智能体各自生成方案,枚举可能的风险,最后由一个裁判智能体根据规则选择最优解。这种模式在真实业务里最实用,因为它不要求智能体之间完全互相理解,只需要它们把自己的专业判断摆到桌面上来,让裁判做决策。
我在实际使用 OpenClaw 的过程中,最深的体会是:智能体间通信不是给机器写的,而是给人和人之间的协作方式建模。消息协议定得好,系统会越跑越顺;协议混乱,再多模型能力都会被淹没在无效对话里。如果你也在做多智能体,不妨先从小项目开始,把“谁能给谁发什么消息”这件事真正想透,再让模型上场发挥。