本文回答什么问题:MessageBus 是怎么实现的?为什么用双向 asyncio.Queue?怎样扩展 MessageBus 支持新通道?
目标读者:想深入了解基础设施层的 LLM Agent 开发者 / 系统架构师 预计阅读时间:10 分钟 源码版本:GitHub HKUDS/nanobot main 分支主线代码(仓库相对路径)
MessageBus 是 nanobot 整套架构的"信息动脉"。本章聚焦一个核心文件 nanobot/bus/queue.py(约 100 行),用 3 步拆解双向 asyncio.Queue 的设计与实现。
1. 整体定位:为什么需要 MessageBus
如果没有 MessageBus,通道(如 Telegram 通道)就要直接调 AgentLoop——通道与 AgentLoop 强耦合,新增通道要改 AgentLoop 代码,测试时还要 mock 整个 AgentLoop。MessageBus 把通道和 AgentLoop 解耦——通道只关心"我发布出去 / 我消费传入"。
核心要点速查(建议收藏)
- 核心文件:nanobot/bus/queue.py(约 100 行,L8-L42 核心实现,L42 之后是测试桩)
- 数据结构:2 个 asyncio.Queue(inbound + outbound),都是 FIFO 无界队列
- 4 个公开方法:publish_inbound / consume_inbound / publish_outbound / consume_outbound
- 3 个内部方法:_close / closed / __aiter__(异步迭代支持)
- 单实例 vs 多实例:默认 1 个全局 MessageBus;多通道共享,通过 nanobot.bus.queue.get_bus() 单例工厂获得
2. 3 步拆解 MessageBus 实现
步骤 1:定义双队列数据结构(nanobot/bus/queue.py L8-L20)
class MessageBus:
def __init__(self):
self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue()
self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue()
为什么用 2 个独立的 Queue 而非单条 + 标签?
- 通道只关心"我发出的有没有被消费 / 给我的要不要消费"——天然双向
- 单 Queue + 标签会增加路由复杂度("按 channel 过滤"要写额外的逻辑)
- 测试时可独立 mock 单边
步骤 2:4 个 publish/consume 方法(L20-L42)
async def publish_inbound(self, msg: InboundMessage) –> None:
await self.inbound.put(msg)
async def consume_inbound(self) –> InboundMessage:
return await self.inbound.get()
async def publish_outbound(self, msg: OutboundMessage) –> None:
await self.outbound.put(msg)
async def consume_outbound(self) –> OutboundMessage:
return await self.outbound.get()
每个方法就是 put 或 get,没有其他逻辑——保持最小 API,易测试。
步骤 3:关闭语义 + 单例工厂(L42-L100)
async def _close(self) –> None:
"""Producer 发完后,等所有 consumer 消费完。"""
await self.inbound.join()
await self.outbound.join()
self._closed = True
# 关键:`queue.join()` 要求每条 `put()` 都有对应的 `task_done()`
# 否则永久 hang(详见 §6 Q&A)
@property
def closed(self) –> bool:
return self._closed
_BUS: MessageBus | None = None
def get_bus() –> MessageBus:
global _BUS
if _BUS is None:
_BUS = MessageBus()
return _BUS
单例模式的好处:多通道(Telegram + Discord + Slack 共 17 个)共享同一 MessageBus——保证 outbound 投递顺序一致。
3. 逻辑总览图
U
AgentRunner
AgentLoop
MessageBus
Channel_Telegram_CLI
U
AgentRunner
AgentLoop
MessageBus
Channel_Telegram_CLI
#mermaid-svg-DYs8nrm47E3ILsAn{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-DYs8nrm47E3ILsAn .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-DYs8nrm47E3ILsAn .error-icon{fill:#552222;}#mermaid-svg-DYs8nrm47E3ILsAn .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-DYs8nrm47E3ILsAn .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-DYs8nrm47E3ILsAn .marker{fill:#333333;stroke:#333333;}#mermaid-svg-DYs8nrm47E3ILsAn .marker.cross{stroke:#333333;}#mermaid-svg-DYs8nrm47E3ILsAn svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-DYs8nrm47E3ILsAn p{margin:0;}#mermaid-svg-DYs8nrm47E3ILsAn .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-DYs8nrm47E3ILsAn text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-DYs8nrm47E3ILsAn .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-DYs8nrm47E3ILsAn .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-DYs8nrm47E3ILsAn .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-DYs8nrm47E3ILsAn .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-DYs8nrm47E3ILsAn #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-DYs8nrm47E3ILsAn .sequenceNumber{fill:white;}#mermaid-svg-DYs8nrm47E3ILsAn #sequencenumber{fill:#333;}#mermaid-svg-DYs8nrm47E3ILsAn #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-DYs8nrm47E3ILsAn .messageText{fill:#333;stroke:none;}#mermaid-svg-DYs8nrm47E3ILsAn .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-DYs8nrm47E3ILsAn .labelText,#mermaid-svg-DYs8nrm47E3ILsAn .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-DYs8nrm47E3ILsAn .loopText,#mermaid-svg-DYs8nrm47E3ILsAn .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-DYs8nrm47E3ILsAn .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-DYs8nrm47E3ILsAn .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-DYs8nrm47E3ILsAn .noteText,#mermaid-svg-DYs8nrm47E3ILsAn .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-DYs8nrm47E3ILsAn .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-DYs8nrm47E3ILsAn .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-DYs8nrm47E3ILsAn .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-DYs8nrm47E3ILsAn .actorPopupMenu{position:absolute;}#mermaid-svg-DYs8nrm47E3ILsAn .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-DYs8nrm47E3ILsAn .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-DYs8nrm47E3ILsAn .actor-man circle,#mermaid-svg-DYs8nrm47E3ILsAn line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-DYs8nrm47E3ILsAn :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}
publish_inbound InboundMessage
consume_inbound
run_turn spec
final OutboundMessage
publish_outbound OutboundMessage
consume_outbound
send via platform API
4. 关键流程逐步拆解
4.1 阶段 ①:Inbound 接入
以 Telegram 通道为例(nanobot/channels/telegram/runtime.py),收到 update.message.text 后:
async def on_message(self, update):
await self.bus.publish_inbound(InboundMessage(
channel=self.name, # "telegram"
sender_id=str(update.effective_user.id),
chat_id=str(update.effective_chat.id),
content=update.message.text,
media=[],
))
4.2 阶段 ②-③:AgentLoop 拉取
AgentLoop 在 _dispatch 主循环里:
while not self._closing:
msg = await self.bus.consume_inbound() # 阻塞等待
try:
await self._handle_one_turn(msg)
finally:
self.bus.inbound.task_done()
4.3 阶段 ④:Outbound 投递
AgentLoop 把最终回复推入 outbound:
await self.bus.publish_outbound(OutboundMessage(
channel="telegram",
chat_id=chat_id,
content=final_content,
event=TurnEndEvent(latency_ms=elapsed_ms),
))
4.4 阶段 ⑤:通道 send
ChannelManager._dispatch_outbound 找到目标通道:
async def _dispatch_outbound(self, msg: OutboundMessage):
ch = self._channels[msg.channel] # 按 channel 名字查
await ch.send(msg)
5. 为什么这样设计:3 个核心决策
决策 1 · 双向 asyncio.Queue 而非单条 + 标签
决策依据:通道与 AgentLoop 是天然的"双向契约",用 2 条队列让契约显式化。
决策 2 · 单例工厂 get_bus()
决策依据:17 通道 + 1 个 AgentLoop + 1 个 AgentRunner 共享同一 MessageBus,保证:
- 投递顺序一致(同一 chat_id 的回复顺序与发送顺序一致)
- 单元测试可注入 mock bus(覆盖 get_bus() 工厂即可)
决策 3 · 用 asyncio.Queue 而非 multiprocessing.Queue
决策依据:nanobot 是单进程 asyncio 应用,所有通道共享事件循环,不需要跨进程。asyncio.Queue 比 multiprocessing.Queue 轻量 100 倍。
6. 常见问题 / 避坑
Q:inbound / outbound 会无限增长导致 OOM 吗?
A:理论上 asyncio.Queue() 无界,实际由"消费速度 >= 生产速度"保证。AgentLoop 单线程消费,Channel 单线程生产,正常情况下不会堆积。如果某通道 L L 异常慢,可用 asyncio.Queue(maxsize=N) 限流。
Q:怎么注入 mock bus 做单元测试?
A:测试用 monkeypatch.setattr("nanobot.bus.queue._BUS", MockBus()) 替换全局工厂返回的实例。
Q:MessageBus 和 nanobot/cli/commands.py 启动流程怎么连?
A:nanobot gateway 命令会调 get_bus() 创建单例,再传给 AgentLoop(bus=bus) 和各 Channel(bus=bus) 构造。
7. 小结
- 核心文件:nanobot/bus/queue.py(100 行内)
- 数据结构:2 个 asyncio.Queue
- 4 个方法:publish/consume × inbound/outbound
- 单例工厂:get_bus() 保证 17 通道共享
- 设计决策:双向解耦 / 单例共享 / asyncio 单进程
本文要点速查
按角色推荐
- 系统架构师 / 安全审计:必读(§5 的 3 决策 why 是基础设施层核心)
- LLM Agent 开发者:必读(后续 10-16 章所有 Agent 行为都走 MessageBus,这里打基础)
- LLM Provider 适配者:选读(知道 inbound/outbound 流向即可)
- 聊天通道开发者:必读(通道实现必读 §4.1 + §4.4 的 publish/consume 流程)
- Tool / MCP 工具开发者:选读(知道 Tool 上下文如何通过 MessageBus 传播即可)
下一步
- 第 05 章《事件协议:InboundMessage + OutboundEvent》 —— 总线里的"消息体"长什么样(主题群"架构与基础设施",第 1 周)
- 第 10 章《AgentLoop 编排核心》 —— 6 阶段流水线完整源码(主题群"Agent 核心",第 3 周)
- 第 17 章《LLMProvider 抽象》 —— 跳出基础设施,看 LLM 层(主题群"LLM Provider 适配",第 4 周)
tags:#nanobot #AI Agent #LLM #Python #源码解析 #消息总线 #asyncio
