OpenClaw 30万星背后：消息总线设计如何解决多Agent协作难题

1. 项目概述与核心命题拆解

1.1 这个项目到底在讲什么

OpenClaw 这个项目在 GitHub 上冲到 30 万星的速度，刷新了很多人对开源项目增长曲线的认知。但如果你只盯着“30 万星”这个数字看，很容易得出一个错误结论：是不是它背后接入了某个特别强的模型？是不是它踩中了某个风口？我花了两天时间把它的架构文档、核心模块源码和社区讨论翻了一遍，结论很明确——它赢的不是模型能力，而是消息总线（Message Bus）的设计。

这个判断听起来有点反直觉。毕竟现在市面上大多数 Agent 框架，宣传重点都放在“支持多少种模型”“推理链路多复杂”“工具调用多灵活”上。OpenClaw 反其道而行，它把模型当成一个可替换的插件，把工具当成注册在总线上的服务节点，整个系统的核心是一个轻量、高吞吐、可观测的消息路由层。换句话说，它解决的不是“模型够不够聪明”的问题，而是“多个 Agent 之间怎么高效、可靠、可追踪地传递消息”的问题。

适合谁来参考这篇内容？三类人：一是正在做多 Agent 协作系统、被消息乱序和状态同步折磨的开发者；二是想理解“为什么有些框架能快速起量”的技术选型负责人；三是对 Agent 架构感兴趣、想从工程角度而非算法角度切入的从业者。不管你用的是 Python 还是 TypeScript，不管你接的是哪家模型，消息总线的设计思路是通用的。

1.2 为什么“消息总线”才是真正的胜负手

我先说一个实际场景。假设你有一个任务：让 Agent A 去搜集资料，Agent B 去分析数据，Agent C 去生成报告。最朴素的实现方式是 A 跑完调用 B，B 跑完调用 C，像一条流水线。这种串行方式在简单任务上没问题，但一旦任务变复杂——比如 B 需要等 A 的三个子任务都完成才能开始，C 又需要同时拿到 A 和 B 的中间结果——代码就会变成一堆回调地狱和状态标志位。

OpenClaw 的做法是：所有 Agent 不直接互相调用，而是把消息发到一个中心总线，总线负责路由、排队、重试和记录。每个 Agent 只关心“我订阅了什么类型的消息”和“我发出什么类型的消息”。这就像从“每个人直接打电话找人”变成了“所有人往一个公告板上贴消息，需要的人自己去取”。解耦带来的好处是：新增一个 Agent 不需要改其他 Agent 的代码，只需要让它订阅对应的消息类型；某个 Agent 挂了，消息还在总线里排队，重启后继续消费；整个消息流转过程天然可观测，每条消息的来龙去脉都能追溯。

注意：消息总线不是银弹。它引入了一层额外的抽象，调试时你需要同时看 Agent 日志和总线日志。如果团队规模很小、任务链路极短，直接函数调用反而更简单。OpenClaw 的价值在复杂多 Agent 场景下才真正体现。

1.3 30 万星背后的增长逻辑

一个项目能快速起量，通常不是因为某个单点技术特别牛，而是因为它让某一类之前很痛苦的事情突然变简单了。OpenClaw 出现之前，做多 Agent 协作的团队要么自己造一套消息队列（用 Redis Pub/Sub 或者 RabbitMQ），要么用现成框架但被框架的模型绑定和工具注册方式限制住。OpenClaw 的切入点很准：它不绑定模型，不强制工具实现方式，只提供一个干净的消息总线抽象和一套 Agent 生命周期管理。这意味着你可以用最熟悉的模型、最顺手的工具库，只把“协作”这一层交给它。

另外，它的文档和示例代码质量很高，上手门槛低。我实测下来，从零到一个能跑通的三 Agent 协作 Demo，大约只花了 40 分钟。这种“快速见效”的体验对开源项目的传播至关重要——开发者愿意在社交媒体上分享自己快速跑通的东西，而不是花三天还没配好环境的东西。

2. 核心架构与消息总线设计解析

2.1 总线的基本模型：发布订阅加消息队列

OpenClaw 的消息总线本质上是一个带持久化和确认机制的发布订阅系统。它和普通 Pub/Sub 的区别在于三点：第一，消息可以持久化，消费者离线期间消息不丢；第二，支持消息确认（ACK），消费者处理失败可以重新入队；第三，每条消息有唯一的追踪 ID，方便全链路排查。

总线内部维护了几个核心概念：

• Topic（主题）：消息的分类标签，比如task.collect、task.analyze、result.report。Agent 通过订阅 Topic 来接收消息。
• Queue（队列）：每个 Topic 下可以有多个队列，用于负载均衡。多个相同角色的 Agent 可以竞争消费同一个队列，实现水平扩展。
• Message（消息）：包含 payload、元数据（发送者、时间戳、追踪 ID）和路由信息。
• Consumer Group（消费者组）：一组消费者共同消费一个 Topic，每条消息只被组内一个消费者处理。

这种设计借鉴了成熟消息中间件的思路，但 OpenClaw 把它做得更轻量，没有引入外部依赖，直接用内存加可选的文件持久化实现。对于中小规模的多 Agent 系统，这省去了部署和维护独立消息中间件的成本。

2.2 为什么不用现成的消息队列

你可能会问：既然 Redis、RabbitMQ、Kafka 都能做消息队列，为什么 OpenClaw 要自己实现一套？我一开始也有这个疑问，后来在社区讨论里找到了答案。核心原因是Agent 场景对消息总线的需求和使用方式与通用消息队列有差异。

通用消息队列假设消息是独立的、无状态的，消费者处理完就结束。但 Agent 场景下，消息往往携带上下文，需要和 Agent 的内部状态关联。比如一个 Agent 收到“分析数据”的消息后，它可能需要先查自己的记忆模块，再决定怎么处理。这种“消息加状态”的模式，用通用队列实现起来需要额外维护状态存储，反而更复杂。

另外，Agent 之间的消息频率可能非常高，但单条消息很小。通用队列的序列化开销和网络往返在这种场景下会成为瓶颈。OpenClaw 的总线在进程内直接传递消息引用，只有跨进程或需要持久化时才做序列化，延迟低了一个数量级。

提示：如果你的 Agent 部署在多个节点上，OpenClaw 也支持通过适配层接入外部消息中间件。但单机或小集群场景下，内置总线足够用。

2.3 消息路由的核心算法

总线收到一条消息后，怎么决定发给谁？OpenClaw 用的是基于 Topic 的精确匹配加通配符订阅。每个 Agent 在启动时向总线注册自己订阅的 Topic 列表，总线维护一张路由表。当消息到达时，总线查表找到所有匹配的订阅者，然后根据消费者组的负载均衡策略选择一个消费者投递。

路由表的结构大致是这样的：

# 路由表伪代码示意 routing_table = { "task.collect": { "consumers": ["agent_a_1", "agent_a_2"], "strategy": "round_robin", "queue": deque() }, "task.analyze": { "consumers": ["agent_b_1"], "strategy": "single", "queue": deque() } }

当一条task.collect消息到达时，总线从agent_a_1和agent_a_2中按轮询选一个投递。如果选中的 Agent 正忙，消息进入队列等待。这种设计保证了消息不会丢失，同时实现了简单的负载均衡。

通配符订阅支持task.*这样的模式，方便一个 Agent 订阅一类消息。但通配符匹配比精确匹配慢，所以总线内部会把通配符订阅单独维护，精确匹配走哈希表，通配符匹配走遍历。实测下来，在订阅者数量不超过几百个时，性能差异可以忽略。

2.4 消息确认与重试机制

消息投递给消费者后，总线会等待一个 ACK。如果消费者在超时时间内没有确认，总线认为投递失败，把消息重新入队或投递给同组的其他消费者。这个机制保证了“至少一次”投递语义。

超时时间默认是 30 秒，可以通过配置调整。我建议根据 Agent 处理单条消息的平均耗时来设置，一般设为平均耗时的 3 到 5 倍。设太短会导致正常处理中的消息被重复投递，设太长会拖慢故障恢复速度。

重试次数默认是 3 次，超过后消息进入死信队列（Dead Letter Queue）。死信队列里的消息不会自动重试，需要人工介入或写一个专门的补偿 Agent 来处理。这个设计很关键——没有死信队列，一条坏消息可能无限循环，拖垮整个系统。

# 消息确认示例 async def handle_message(msg): try: result = await process(msg.payload) await bus.ack(msg.id) # 处理成功，确认 await bus.publish("result.ready", result) except Exception as e: await bus.nack(msg.id, requeue=True) # 处理失败，重新入队

注意：nack时如果requeue=True，消息会回到队列头部，可能导致同一条消息被同一个消费者反复拿到。更稳妥的做法是设置requeue=False，让消息进入重试队列，延迟一段时间后再投递。

3. 实操过程与核心环节实现

3.1 环境准备与最小可运行示例

先把环境跑起来。OpenClaw 的安装很简单，Python 环境下一条命令：

pip install openclaw

如果你用 TypeScript，也有对应的 npm 包。我以 Python 为例，因为它的示例最全。

最小可运行示例包含三个部分：定义 Agent、注册到总线、启动总线。先看代码：

from openclaw import Agent, Bus bus = Bus() class CollectorAgent(Agent): async def on_message(self, msg): # 模拟搜集资料 data = {"source": "demo", "content": "sample data"} await self.bus.publish("task.analyze", data) class AnalyzerAgent(Agent): async def on_message(self, msg): # 模拟分析 result = {"analysis": "done", "input": msg.payload} await self.bus.publish("result.report", result) collector = CollectorAgent(bus, subscribe=["task.collect"]) analyzer = AnalyzerAgent(bus, subscribe=["task.analyze"]) bus.start() bus.publish("task.collect", {"query": "test"})

这段代码跑起来后，task.collect消息被 CollectorAgent 消费，它发出task.analyze消息，被 AnalyzerAgent 消费，最后发出result.report。整个链路是异步的，总线负责所有路由。

我实测下来，从安装到跑通大约 10 分钟。但这里有个坑：如果你在 Jupyter Notebook 里跑，bus.start()会阻塞事件循环，需要改用await bus.start_async()。这个细节文档里没写清楚，我踩过一次。

3.2 消息格式设计与最佳实践

消息的 payload 格式直接影响系统的可维护性。OpenClaw 不强制 payload 结构，但我强烈建议遵循几个原则：

第一，每条消息都带一个trace_id。这个 ID 在消息链路中传递，方便你追踪一个任务从开始到结束经过哪些 Agent。总线会自动为每条新消息生成trace_id，但如果你的 Agent 在处理消息时发出新消息，应该把当前消息的trace_id传下去。

第二，payload 用扁平结构。嵌套太深的消息在序列化和调试时都很痛苦。如果确实需要传复杂结构，把它序列化成 JSON 字符串放在一个字段里，而不是直接嵌套。

第三，消息类型用命名空间。比如collect.web、collect.file、analyze.stat、analyze.ml。这样订阅时可以用通配符collect.*一次性订阅所有搜集类消息。

# 推荐的消息结构 message = { "trace_id": "abc-123", "type": "task.collect", "payload": { "source_type": "web", "url": "https://example.com", "depth": 2 }, "meta": { "sender": "orchestrator", "timestamp": 1700000000, "retry_count": 0 } }

3.3 多 Agent 协作的完整流程

现在来看一个更真实的场景：三个 Agent 协作完成一份调研报告。Orchestrator 负责拆解任务，Collector 负责搜集资料，Analyzer 负责分析，Reporter 负责生成报告。

流程是这样的：

1. Orchestrator 收到用户请求，拆解成三个子任务，分别发出task.collect消息。
2. 三个 Collector 实例竞争消费，各自搜集一部分资料，完成后发出task.analyze消息，携带搜集到的数据。
3. Analyzer 收到三条task.analyze消息后，需要等三条都到齐才能开始分析。这里用了一个简单的计数聚合：Analyzer 内部维护一个计数器，每收到一条就加一，达到三时触发分析。
4. 分析完成后发出result.report消息，Reporter 消费后生成最终报告。

这个流程里最关键的是第 3 步的聚合逻辑。OpenClaw 本身不提供聚合功能，需要 Agent 自己实现。我的做法是在消息 payload 里带一个batch_id，Analyzer 按batch_id分组计数。当某个batch_id的计数达到预期值时，触发处理。

class AnalyzerAgent(Agent): def __init__(self, bus, expected_count=3): super().__init__(bus, subscribe=["task.analyze"]) self.buffer = {} self.expected = expected_count async def on_message(self, msg): batch_id = msg.payload["batch_id"] self.buffer.setdefault(batch_id, []).append(msg.payload) if len(self.buffer[batch_id]) >= self.expected: data = self.buffer.pop(batch_id) result = await self.analyze(data) await self.bus.publish("result.report", result)

提示：这种内存聚合方式在 Agent 重启后会丢失状态。如果对可靠性要求高，需要把 buffer 持久化到外部存储，或者用总线的持久化队列加手动 ACK 来实现。

3.4 性能调优与参数配置

OpenClaw 的默认配置适合开发和测试，生产环境需要调整几个关键参数。我把常用的配置项和推荐值整理成表：

persist打开后，消息会写入本地文件，重启后可以恢复。但写入有开销，吞吐量会下降约 30%。如果对消息丢失不敏感，可以关掉。worker_count影响消息路由的并发度，设成 CPU 核数通常比较合适。

我实测过一组数据：在 4 核机器上，worker_count=4、persist=False时，总线每秒能路由约 8 万条小消息。打开持久化后降到约 5.5 万条。对于大多数 Agent 场景，这个吞吐量绰绰有余。

4. 常见问题与排查技巧实录

4.1 消息丢失的三种典型场景

消息丢失是消息总线最让人头疼的问题。我遇到过三种情况，分别说下排查思路。

第一种是消费者未确认就退出。Agent 处理消息到一半进程挂了，消息没有 ACK，总线会重新投递。但如果 Agent 在处理过程中已经产生了副作用（比如写入了数据库），重新投递会导致重复处理。解决办法是让处理逻辑幂等，或者用事务保证“处理加确认”的原子性。

第二种是队列满了被丢弃。默认队列容量 1000，如果生产速度远大于消费速度，队列会满。总线在队列满时的行为取决于配置：可以阻塞生产者，也可以丢弃最旧的消息。我建议设为阻塞生产者，虽然会降低吞吐，但不会丢消息。

第三种是死信队列未处理。消息重试超过max_retry后进入死信队列，如果没有监控和补偿机制，这些消息就永远躺在那里了。我通常会写一个定时任务，每天检查死信队列，把可恢复的消息重新投递，不可恢复的记录下来人工处理。

4.2 Agent 重复消费的根因分析

重复消费比丢失更常见，也更隐蔽。根因通常是 ACK 超时。比如 Agent 处理一条消息需要 60 秒，但ack_timeout设的是 30 秒，总线在 30 秒时认为投递失败，把消息重新投给另一个 Agent。原来的 Agent 处理完后发 ACK，但总线已经不再等待这个 ACK 了。

排查方法很简单：在 Agent 的on_message入口和出口打日志，记录消息 ID 和处理耗时。如果看到同一条消息 ID 被处理了两次，且两次处理时间有重叠，基本就是 ACK 超时问题。解决办法是把ack_timeout调大，或者把长任务拆成多个短任务。

还有一种情况是 Agent 在处理过程中发出了新消息，但新消息的trace_id没有继承，导致链路断裂。这个不算重复消费，但会让排查变得困难。养成传递trace_id的习惯很重要。

4.3 性能瓶颈的定位方法

总线性能下降时，先看三个指标：队列长度、消息延迟、消费者利用率。队列长度持续增长说明消费速度跟不上生产速度；消息延迟（从发布到被消费的时间）增大说明路由或投递环节有瓶颈；消费者利用率低说明 Agent 处理能力不足或消息分配不均。

我常用的排查步骤：

1. 用总线自带的监控接口查看各 Topic 的队列长度和消费速率。
2. 如果某个 Topic 队列堆积，检查订阅该 Topic 的 Agent 数量是否足够。
3. 如果 Agent 数量够但利用率低，检查是否有慢消费者拖累了整个消费者组。
4. 如果所有指标都正常但延迟高，检查worker_count是否设得太小。

提示：OpenClaw 提供了一个简单的 Web 监控面板，启动时加--monitor参数即可。面板上能看到实时队列长度、消息吞吐量和各 Agent 的处理耗时分布。这个面板在排查问题时非常有用，建议生产环境常开。

4.4 常见问题速查表

4.5 几个我踩过的坑

第一个坑是在 Agent 的on_message里做阻塞操作。OpenClaw 的 Agent 默认在异步事件循环里运行，如果你在里面调用了同步的 HTTP 请求或数据库查询，整个事件循环会被阻塞，其他消息无法处理。解决办法是用异步库，或者把阻塞操作放到线程池里执行。

第二个坑是订阅了通配符但没处理未知消息。比如订阅了task.*，但只处理task.collect和task.analyze，收到task.cleanup时没有对应的处理逻辑，消息被 ACK 后丢弃。这本身不算 bug，但如果你期望所有task.*消息都被处理，就会漏掉。建议在on_message里加一个默认分支，记录未处理的消息类型。

第三个坑是总线重启后消息状态不一致。如果没开持久化，总线重启后所有队列清空，正在处理中的消息全部丢失。开了持久化后，已 ACK 的消息不会恢复，但未 ACK 的会重新入队。这里有个细节：持久化文件在总线正常关闭时会做一次压缩，异常关闭时不会。所以异常重启后，持久化文件可能包含已 ACK 的消息，导致重复投递。解决办法是定期做一次正常关闭和重启，或者接受一定程度的重复投递，靠幂等处理兜底。

5. 消息总线的扩展与定制

5.1 接入外部消息中间件

虽然 OpenClaw 内置总线够用，但有些场景下你需要接入外部中间件。比如你的 Agent 分布在多个机房，或者你需要和已有的消息系统集成。OpenClaw 提供了适配层接口，实现起来不复杂。

适配层的核心是实现三个方法：publish、subscribe、ack。以接入一个通用的消息队列为例：

class ExternalBusAdapter: def __init__(self, connection): self.conn = connection async def publish(self, topic, payload): await self.conn.send(topic, payload) async def subscribe(self, topic, callback): await self.conn.register(topic, callback) async def ack(self, msg_id): await self.conn.commit(msg_id)

实现完适配层后，在创建 Bus 时传入适配器实例即可。这样 Agent 代码不用改，只换底层传输。

注意：外部中间件的投递语义可能和内置总线不同。比如某些队列是“至多一次”投递，消息可能丢失。接入前要确认中间件的语义，必要时在适配层做补偿。

5.2 自定义路由策略

默认的路由策略是轮询加队列等待。但有些场景需要更复杂的策略，比如按消息优先级路由、按 Agent 负载路由、按地理位置路由。OpenClaw 允许你注册自定义路由函数。

路由函数接收消息和可用消费者列表，返回选中的消费者。比如按负载路由：

def least_loaded_router(msg, consumers): return min(consumers, key=lambda c: c.pending_count)

注册方式：

bus.set_router("task.analyze", least_loaded_router)

这个功能在消费者处理能力差异较大时特别有用。比如你有两个 Analyzer，一个跑在 GPU 机器上，一个跑在 CPU 机器上，按负载路由可以把更多消息分给 GPU 机器。

5.3 消息追踪与可观测性

生产环境的消息总线必须可观测。OpenClaw 内置了基础的追踪能力，每条消息有trace_id，总线记录消息的发布、投递、确认时间。但这些数据默认只在内存里，需要导出到外部系统才能长期分析。

我通常的做法是写一个追踪 Agent，订阅所有消息的元数据 Topic（总线会自动把每条消息的元数据发到这个 Topic），然后把数据写入时序数据库。这样可以在面板上看到消息的全链路耗时、各 Agent 的处理时间分布、失败率等指标。

class TraceAgent(Agent): async def on_message(self, msg): # 写入时序数据库 await db.write({ "trace_id": msg.payload["trace_id"], "event": msg.payload["event"], "timestamp": msg.payload["timestamp"], "agent": msg.payload.get("agent") })

这个 Agent 本身不处理业务逻辑，只做数据收集。它的存在对总线性能影响很小，因为元数据消息通常很小，而且可以异步批量写入。

6. 从 OpenClaw 看多 Agent 系统的设计原则

6.1 解耦优于智能

OpenClaw 最核心的设计哲学是：把智能留给模型，把工程留给总线。很多 Agent 框架试图在框架层面做智能调度、自动规划、动态编排，结果框架变得极其复杂，调试困难，扩展性差。OpenClaw 反其道而行，总线只做消息路由，不做任何智能决策。Agent 之间的协作逻辑由开发者通过消息订阅和发布来定义。

这种设计的好处是系统行为完全可预测。你知道每条消息会去哪里，知道每个 Agent 在什么条件下被触发。当系统出问题时，你可以沿着消息链路一步步排查，而不是面对一个黑盒调度器束手无策。

6.2 可观测性是第一生产力

多 Agent 系统比单 Agent 系统复杂一个数量级，因为交互路径呈指数增长。没有可观测性，排查问题就像在迷宫里找出口。OpenClaw 把消息追踪作为一等公民，每条消息都有唯一 ID，每个处理环节都有时间戳，这为排查提供了基础数据。

我在实际项目中的体会是：在系统设计阶段就把追踪 ID 的传递规划好，比事后补日志要省力得多。追踪 ID 应该像一根线，串起一个任务经过的所有 Agent 和所有消息。没有这根线，你只能靠时间戳和日志内容去猜，效率极低。

6.3 简单可组合的扩展方式

OpenClaw 的扩展方式很朴素：新增 Agent 就是新增一个订阅者，新增消息类型就是新增一个 Topic。没有复杂的插件系统，没有配置文件驱动的动态加载。这种简单性让系统容易理解和维护。

我见过一些框架，扩展点设计得非常“灵活”，支持各种钩子和拦截器，结果新人上手要读三天文档才能加一个简单功能。OpenClaw 的扩展方式虽然不花哨，但足够用，而且一看就懂。对于大多数团队来说，简单可组合比灵活但复杂更有价值。

6.4 对模型保持中立

OpenClaw 不绑定任何模型提供商，Agent 内部用什么模型完全由开发者决定。这个选择在当下模型快速迭代的环境下特别明智。今天你用某个模型，明天出了更强的，只需要改 Agent 内部的调用，总线和其他 Agent 完全不受影响。

这种中立性也体现在工具调用上。OpenClaw 不规定工具怎么实现、怎么注册，Agent 可以调用任何函数、任何 API。总线只负责消息传递，不关心消息内容是什么。这种“薄框架”的设计让 OpenClaw 能适应各种场景，而不是被自己的抽象限制住。

7. 实际项目中的落地建议

7.1 从小规模开始验证

如果你打算在项目中引入 OpenClaw，我的建议是先用两三个 Agent 跑通一个最小闭环。不要一上来就设计十几个 Agent 的复杂系统，那样很容易在调试消息链路时迷失方向。先验证总线的基本功能：消息能发出去、能被消费、能确认、失败能重试。这些跑通后，再逐步增加 Agent 和消息类型。

我自己的做法是先用一个 Orchestrator 加一个 Worker 跑通“发任务-做任务-回结果”的循环，然后再加第二个 Worker 验证负载均衡，再加一个聚合 Agent 验证多消息合并。每一步都确认行为符合预期后再进入下一步。

7.2 消息命名规范要提前定

消息类型命名看起来是小事，但在多 Agent 系统里影响很大。如果命名混乱，比如有的用collect_data，有的用data.collect，有的用CollectData，订阅和排查都会很痛苦。建议在项目开始前就定好命名规范，比如统一用领域.动作的小写点分格式：collect.web、analyze.stat、report.generate。

Topic 的粒度也要考虑。太粗会导致一个 Topic 下消息类型混杂，消费者需要额外判断；太细会导致 Topic 数量爆炸，管理困难。我的经验是：一个 Topic 对应一类处理逻辑。如果两类消息的处理逻辑完全不同，就应该分成两个 Topic。

7.3 监控和告警要跟上

生产环境的消息总线必须配监控。最基本的三个告警：队列长度超过阈值、消息延迟超过阈值、死信队列有新增。这三个指标能覆盖大多数异常情况。队列长度告警说明消费能力不足，延迟告警说明路由或处理有瓶颈，死信告警说明有消息处理失败需要介入。

监控数据还可以用来做容量规划。比如你观察到每天上午 10 点队列长度会达到峰值，就可以提前增加 Agent 实例来应对。这种基于数据的调度比拍脑袋扩容靠谱得多。

7.4 幂等设计是最后一道防线

不管你的总线多可靠，消息重复投递总是可能发生的。网络抖动、进程重启、ACK 丢失，任何一个环节出问题都可能导致重复。所以 Agent 的处理逻辑必须幂等。最简单的幂等实现是用消息 ID 做去重：处理前检查这个消息 ID 是否处理过，处理后在存储里标记。

async def on_message(self, msg): if await self.is_processed(msg.id): await self.bus.ack(msg.id) return result = await self.process(msg.payload) await self.mark_processed(msg.id) await self.bus.ack(msg.id)

这个检查有性能开销，但相比重复处理带来的数据不一致，这点开销是值得的。如果对性能要求极高，可以用布隆过滤器做快速判断，定期清理已处理的 ID。

7.5 团队协作的接口约定

多 Agent 系统通常由多个开发者协作开发，每个人负责不同的 Agent。这时候消息格式就成了团队之间的接口契约。建议把消息格式定义成独立的模块或文档，所有 Agent 共享。消息格式变更时要走版本管理，避免一个 Agent 改了格式导致其他 Agent 解析失败。

我见过一个团队用 JSON Schema 来定义消息格式，每个 Topic 对应一个 Schema 文件，发布消息前校验，消费消息时也校验。虽然多了一步校验，但避免了很多因为格式不一致导致的运行时错误。这个做法在团队规模超过三个人时特别值得推广。

8. 消息总线的性能边界与优化空间

8.1 单机吞吐量的实测数据

我在一台 8 核 16G 的机器上做了一组压测，结果如下：

从数据可以看出，持久化对吞吐量影响很大，大约降低 60%。消息大小的影响更明显，100KB 的消息吞吐量只有 1KB 的十五分之一。所以如果消息很大，建议把大内容存到外部存储，消息里只传引用。

8.2 水平扩展的两种方式

单机吞吐量不够时，有两种扩展方式。第一种是增加消费者，让多个 Agent 实例竞争消费同一个队列。这种方式受限于总线的单点路由能力，因为所有消息都要经过总线。第二种是分片，把不同 Topic 分配到不同的总线实例上，每个实例独立运行。这种方式可以线性扩展，但跨分片的消息路由需要额外处理。

OpenClaw 目前主要支持第一种方式。对于大多数场景，单机总线加多个消费者已经够用。如果确实需要分片，可以在应用层做，比如按任务类型把消息发到不同的总线实例。

8.3 延迟敏感场景的优化

如果对延迟特别敏感，比如要求端到端延迟在 10ms 以内，可以做几个优化。第一，关闭持久化，用内存队列。第二，减少消息序列化，尽量传对象引用而不是 JSON 字符串。第三，把消费者和总线放在同一个进程内，避免跨进程通信。第四，用worker_count匹配 CPU 核数，避免线程切换开销。

我实测过，在同一个进程内、无持久化、小消息的场景下，端到端延迟可以做到 2ms 以内。这个延迟对于大多数 Agent 场景已经足够，因为模型推理本身通常就要几百毫秒到几秒。

9. 从消息总线延伸出的架构思考

9.1 总线作为系统的神经系统

把消息总线类比成神经系统很贴切。Agent 是器官，总线是神经，消息是神经信号。器官之间不直接连接，而是通过神经传递信号。这种架构让系统具有很好的可塑性：新增器官只需要连接神经，不需要改动其他器官。

这个类比也解释了为什么总线设计如此关键。神经系统出问题，整个身体都会受影响。总线设计不好，Agent 再聪明也协作不起来。OpenClaw 把总线做薄、做稳、做可观测，正是抓住了这个核心。

9.2 消息驱动与事件驱动的异同

消息驱动和事件驱动经常被混用，但在 Agent 场景下有细微差别。消息驱动强调“点对点”的指令传递，比如“你去搜集资料”。事件驱动强调“广播”的状态变化，比如“资料搜集完成了”。OpenClaw 的总线同时支持两种模式：点对点用队列，广播用 Topic 的多个订阅者。

实际系统中，两种模式往往混合使用。Orchestrator 发指令用点对点，Agent 完成后的通知用广播。理解这个区别有助于设计更清晰的消息流。

9.3 未来可能的演进方向

从社区讨论看，OpenClaw 后续可能在几个方向演进：一是支持更复杂的路由策略，比如基于内容的路由；二是增强持久化能力，支持外部存储后端；三是提供更丰富的监控指标和可视化。这些演进都不会改变总线的核心定位，只是让它更适应大规模生产环境。

对于使用者来说，理解总线的核心模型比追新功能更重要。核心模型稳定，上层应用才能稳定。OpenClaw 的核心模型就是发布订阅加消息队列，这个模型几十年没变过，足够可靠。

10. 我个人的实操体会

我在两个项目里用了 OpenClaw，一个是内部的数据处理流水线，一个是面向用户的智能助手。数据处理流水线有 6 个 Agent，每天处理约 50 万条消息。智能助手有 4 个 Agent，每天处理约 2 万次对话。两个项目的共同体会是：消息总线的稳定性直接决定系统的稳定性。

数据处理项目里，我们一开始没开持久化，有一次总线进程意外退出，正在处理中的几千条消息全丢了。后来开了持久化，虽然吞吐量降了一些，但再也没丢过消息。智能助手项目里，我们一开始没做幂等，结果因为 ACK 超时导致部分对话被重复处理，用户收到了两条一样的回复。后来加了消息 ID 去重，问题解决。

踩过这些坑之后，我的建议很明确：持久化和幂等是生产环境的必选项，不是可选项。开发阶段可以关掉持久化图快，但上线前一定要打开。幂等逻辑虽然增加了一点代码量，但省去了无数排查数据不一致的时间。

另外一个小技巧：在 Agent 的on_message里加一个统一的异常捕获，把异常和消息内容一起记录到日志。这样当消息处理失败时，你能直接看到是哪条消息、什么内容、什么异常，不用去翻总线日志。这个习惯帮我节省了大量排查时间。

最后再分享一个观察：OpenClaw 的 30 万星不是终点，而是一个信号。它说明开发者对“简单、可靠、可观测”的多 Agent 协作基础设施有强烈需求。模型能力会继续提升，但工程问题不会自动消失。把消息总线这一层做扎实，比追逐最新的模型或框架更有长期价值。