乐于分享
好东西不私藏

nanobot源码学习(五):多渠道架构

nanobot源码学习(五):多渠道架构

前 4 篇讲了如何把"一个 Agent 怎么跑起来"——消息总线、Tool、Provider、状态机、入口配置。

但有个绕不开的问题——用户从哪来发消息?

如果是命令行(CLI)跑,那就在终端里输。但现实中消息可能有很多来源:用户在飞书群发消息、在 Telegram 私聊机器人、在 Discord 服务器问问题。一个 Agent 想真正"可用",必须接入这些聊天平台。

nanobot 现在支持 17 个聊天平台——Telegram / Discord / Slack / 飞书 / 微信 / 钉钉 / 企业微信 / WhatsApp / Matrix / Signal / 邮件 / QQ / MS Teams 等等。一个 Agent 同时服务所有这些平台,对用户来说体验一致,对开发者来说代码只写一份。

这一篇讲它怎么做到的。s08 Channel 抽象 定规矩,s09 渠道实战 落地。


1. 17 个平台怎么写在一份代码里

先看几种"看起来合理"的写法,以及为什么行不通:

写法 1:在主循环里堆 if-else


  
    
    
    
  
  python
def handle_message(source, text):
    if source == "telegram":
        ...
    elif source == "discord":
        ...
    elif source == "feishu":
        ...
    # ... 17 个 elif

加到第 5 个就 200 行了,加到第 17 个不敢想有多长。

写法 2:每平台一个独立进程


  
    
    
    
  
  bash
python telegram_bot.py &
python discord_bot.py &
python feishu_bot.py &

简单粗暴,但每个进程是独立的 Agent 状态、独立的 Session、独立的工具调用。同一个用户在飞书和 Telegram 找机器人,机器人会"失忆"——因为它们根本不是同一个 Agent。

nanobot 的解法是第三个方案——一个 Agent 进程 + 多个 Channel 适配器:


  
    
    
    
  
  bash
                  ┌──────────┐
Telegram  ──────▶ │          │
Discord   ──────▶ │  Agent   │ ────▶  LLM
飞书     ──────▶  │ Runtime  │
微信     ──────▶  │          │
...              └──────────┘
                  ▲      │
                  │      ▼
                  └───── Response 路由回原平台

所有平台共享同一个 Agent 状态和 Session。用户在飞书说的"我叫 Alice",在 Telegram 找机器人时它也记得。


2. nanobot核心设计:BaseChannel 抽象基类

BaseChannel 规定每个渠道只需实现 4 个方法:


  
    
    
    
  
  python
class BaseChannel:
    async def start(self) -> None:
        """启动平台 SDK,注册 webhook/long-polling。"""
        raise NotImplementedError

    async def stop(self) -> None:
        """清理资源,关闭连接。"""
        raise NotImplementedError

    async def send(self, msg: OutboundMessage) -> None:
        """把响应发回平台。"""
        raise NotImplementedError

    def on_message(self, callback) -> None:
        """收到平台消息时调 callback。"""
        ...

3 个 abstract,1 个基类给默认实现(on_message 把消息推 inbound bus)。

加一个新平台只写一个类继承 BaseChannel,实现这 4 个方法。Agent 主循环一行不动。


3. 真实 InboundMessage 长什么样

s01 教学版 Msg 只有 text 一个字段。真实 nanobot 的 InboundMessage 有 17 个字段:


  
    
    
    
  
  python
@dataclass
class InboundMessage:
    channel: str              # "telegram" / "discord" / "feishu" ...
    chat_id: str              # 群 ID / 私聊 ID
    user_id: str              # 发送者 ID
    text: str                 # 文本内容
    media: list[Media]        # 图片 / 文件 / 音频
    reply_to: str | None      # 引用消息 ID
    thread_id: str | None     # 群里的"主题"或"话题"
    metadata: dict            # 平台特有字段
    timestamp: datetime
    # ... 还有 session_key 等

这些字段是为了统一 17 个平台的所有差异——飞书的 chat_id 跟 Discord 的 channel_id 不是一个东西,但都被映射到 chat_id 字段。下游业务代码只认 InboundMessage 字段,不知道消息来自 Telegram 还是 Discord


4. session_key:决定"哪些算一段对话"

这是最关键的设计。

假设你在飞书群"项目 A 讨论组"里 @ 机器人说"帮我读 README.md"。机器人怎么判断:

  • 这是新对话?
  • 还是接续之前的对话?

答案是 session_key——一个由 channel + chat_id + thread_id + user_id 哈希出来的字符串:


  
    
    
    
  
  python
def session_key(msg: InboundMessage) -> str:
    raw = f"{msg.channel}|{msg.chat_id}|{msg.thread_id or ''}|{msg.user_id}"
    return hashlib.md5(raw.encode()).hexdigest()
场景 session_key
同一群里同一人的连续消息 相同——同 session
换群 不同——新 session
换 thread(飞书话题) 不同——新 session
换人 不同——新 session

关键经验:群消息的 session 边界是 (群, 线程, 人) 三元组,不是 (人)

否则就会出现"Alice 在 A 群聊项目 A,在 B 群聊项目 B,机器人把两段对话混在一起"的事故。


5. 渠道实战:怎么写一个真实 Channel

s09 给了一个最简 CLIChannel 完整版(不读 nanobot 真实源码,只跑 code.py):


  
    
    
    
  
  python
class CLIChannel(BaseChannel):
    async def start(self) -> None:
        # 读 stdin,把每行包成 InboundMessage 推 inbound bus
        async for line in self._stdin:
            msg = InboundMessage(
                channel="cli",
                chat_id="default",
                user_id=os.getlogin(),
                text=line.strip(),
                timestamp=datetime.now(),
            )
            await self._inbound.put(msg)

    async def send(self, msg: OutboundMessage) -> None:
        # 打印到 stdout
        print(f"[{msg.channel} → {msg.chat_id}{msg.text}")

核心就两个方法:start 怎么接消息进 bus,send 怎么把响应发回平台。

nanobot 真实代码里的 telegram.py 复杂得多——要处理 bot token、webhook、media、inline keyboard 等。但结构完全一样:start 启 SDK + 注册 handler,send 调 Telegram Bot API 的 sendMessage


6.  entry_points 让第三方接渠道

17 个内置渠道已经够多,但nanobot还设计了 entry_points 让第三方接,例如:

# 第三方渠道包 my_feishu_channel 的 pyproject.toml
[project.entry-points."nanobot.channels"]
feishu_corp = "my_feishu_channel:FeishuCorpChannel"

pip install my_feishu_channel 就能用。nanobot 启动时自动发现,新渠道就跟内置的一样工作。

为什么不直接内置?三个原因:

  1. 核心安装不臃肿——17 个 SDK(telegram-bot、discord.py、slack-sdk、飞书 SDK……)全打包,依赖几百 MB。第三方按需装,核心保持轻量。
  2. 企业内部渠道——大公司有自研 IM(华为 welink、字节飞书、阿里钉钉),不可能每个都进 nanobot 核心。entry_points 让公司自己包。
  3. 快速实验——社区可以快速试新协议(Matrix、IRC、Mattermost),不用合到 nanobot 仓库。

关键经验:plugin 模式是处理"长尾扩展"的标准解法——主流内置,长尾走 entry_points


7. 跟之前几篇的对比

到这里可以看 nanobot 的整体抽象套路了:

机制 抽象类 扩展方式 主流覆盖
消息总线(s01) Bus - 内置
Tool(s02) Tool entry_points 10+ 内置
Provider(s03/s07) LLMProvider ProviderSpec 13+ 内置
Channel(s08/s09) BaseChannel entry_points 17 内置

统一模式:用抽象基类定契约,主流实现内置,长尾走 plugin。

理解了这一点,再看任何 nanobot 子模块都不会迷路——“它一定有个抽象基类”。


8. 常见的误区:每平台一个进程

很多 Agent 项目的一开始这样写:


  
    
    
    
  
  bash
# docker-compose.yml
services:
  telegram-bot:
    command: python telegram_bot.py
  discord-bot:
    command: python discord_bot.py
  feishu-bot:
    command: python feishu_bot.py

代码量小、上线快。但有几个隐藏问题:

  • Session 不共享——用户在 Telegram 说了"我叫 Alice",在 Discord 找机器人它不认识
  • 资源浪费——每个进程都跑一遍 LLM 客户端初始化、Tool 注册
  • 难统一运营——三个进程要分别部署、监控、日志收集

正确做法:跟 nanobot 一样——一个 Agent 进程 + 多个 Channel 适配器,通过 session_key 隔离不同对话的上下文。

如果你要从这种"每平台一进程"的架构升级到 nanobot 风格,关键改造点只有一个:把 Session 存储从"进程内 dict"挪到"进程外持久化"(文件 / SQLite / Redis),让多个 Channel 共享。


9. 下一篇

下一篇进 s10 章节,讲Tool 沙箱——ReadFile / WriteFile / Bash 这些"有破坏力"的工具,nanobot 怎么用白名单 + PTH 保护防止 Agent 误删文件、误跑命令。


10. 参考资料

  • nanobot 源码:https://github.com/HKUDS/nanobot
  • nanobot-tutorial(14 章配套教程):https://github.com/yaoweizhang/nanobot_tutorial
  • nanobot 官方 Roadmap:https://github.com/HKUDS/nanobot/discussions/431

跟读建议:跑 python s09_channel_impl/code.py 体验最简 CLIChannel。翻 nanobot/channels/telegram.py 看真实 Telegram 渠道怎么实现 4 个方法。