统一消息总线与聊天频道

统一消息总线与聊天频道

学习目标

理解 Nanobot 如何通过 asyncio.Queue 消息总线解耦 18 个聊天频道与 Agent 核心。

前置知识

• IM 网关适配器架构 — 多平台消息网关通用模式

下文假设你已理解上述概念，直接聚焦 Nanobot 的具体实现。

项目实践

消息总线

class MessageBus:
    def __init__(self):
        self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue()
        self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue()

极简设计：两个 asyncio.Queue 分别处理入站和出站消息。

入站消息：InboundMessage(channel, chat_id, sender_id, content, media, metadata) 出站消息：OutboundMessage(channel, chat_id, content, metadata)

AgentLoop 消费管线

async def run(self):
    while self._running:
        msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
        # 处理消息

使用 wait_for(..., timeout=1.0) 而非纯阻塞，允许在空闲期间执行后台任务（如会话过期检查、自动压缩）。

频道自动发现

18 个聊天频道通过 pkgutil 扫描 + entry-point 插件注册：

频道	实现文件	特色功能
Telegram	telegram.py	内联按钮、长消息分割、媒体 URL 发送
Discord	discord.py	线程会话、长消息分割
Slack	slack.py	线程隔离、完成 emoji 反应
飞书	feishu.py	流式消息、CardKit、回复
微信	weixin.py	语音、打字指示、媒体
WhatsApp	whatsapp.py	媒体去重、bridge 刷新
Email	email.py	自循环防护、附件
Matrix	matrix.py	媒体处理
钉钉	dingtalk.py	出站媒体链接安全
企业微信	wecom.py	完整媒体支持
QQ	qq.py	群聊支持
MS Teams	msteams.py	陈旧引用清理
WebSocket	websocket.py	多路复用、WebUI
MoChat	mochat.py	企业微信变体
Signal	signal.py	隐私消息

统一会话 vs 独立会话

通过 unified_session 配置切换：

{"agents": {"defaults": {"unifiedSession": true}}}

• 统一会话：所有频道的消息共享同一个 session key（unified:default）
• 独立会话：每个 channel:chat_id 组合有独立 session

运行时 session key 计算：

def _effective_session_key(self, msg):
    if self._unified_session and not msg.session_key_override:
        return UNIFIED_SESSION_KEY
    return msg.session_key

设计取舍

为什么用 asyncio.Queue 而非 Redis/Kafka

原因：Nanobot 定位为”个人 Agent”，单机单进程部署场景下 asyncio.Queue 足够。如果未来需要多进程/分布式部署，可以替换为 Redis Pub/Sub。

参考来源

• Nanobot MessageBus 源码
• Nanobot Channels 目录
• IM 网关适配器架构