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 启动时自动发现，新渠道就跟内置的一样工作。

为什么不直接内置？三个原因：

核心安装不臃肿——17 个 SDK（telegram-bot、discord.py、slack-sdk、飞书 SDK……）全打包，依赖几百 MB。第三方按需装，核心保持轻量。
企业内部渠道——大公司有自研 IM（华为 welink、字节飞书、阿里钉钉），不可能每个都进 nanobot 核心。entry_points 让公司自己包。
快速实验——社区可以快速试新协议（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 个方法。