如果想学 AI Agent 框架，建议看 nanobot。类似的项目OpenClaw、Hermes Agent、NanoClaw、PicoClaw ，思路都是：本地运行、多渠道接入（飞书、Telegram、Discord 等）、多 LLM 适配、可独立部署的个人助手。

不过点进去看几个会很快发现一个规律：代码量要么上到天际，要么小到没什么可学。比如 OpenClaw，能力挺全的，渠道、模型、生态都铺开了。但代码量 40 万行 TypeScript，还没开始读框架就被仓库规模劝退。再小一点的同类项目倒是不用劝退，代码可能就几百行。但功能太单薄，拿来学只能学个概念，学不到一个完整 Agent 该怎么设计。

代码量与功能平衡得刚刚好的是 nanobot。港大数据智能实验室（HKUDS）出的，GitHub 上 44k Star。核心代码大概 4000 行 Python，却支持 8 个聊天平台、10 多个 LLM 厂商，还支持 MCP 协议。

不过读 nanobot 仓库的时候会又遇到个问题——6 万多行 Python、173 个文件，加上 webui、desktop、bridge 一堆子项目。打开 main 分支读一会上不到主线条，被各种工程化代码带偏。

为了把架构理清，我写了个配套教程仓库 nanobot-tutorial（https://github.com/yaoweizhang/nanobot_tutorial），把 nanobot 按架构拆成 14 个递进小章节，每章 60 到 500 行可运行代码，从 s01 到 s14：

这个拆法的好处是：每章只讲一个核心抽象。s01 只讲消息总线、s02 只讲 Tool，读完一章就只多了一块拼图，不会被一堆架构词汇砸晕。

s01-s05 是 Build 阶段，从 30 行代码堆到 500 行的“迷你 nanobot”，能跑通 LLM 对话。s06-s14 是 Guided 阶段，对照 nanobot 真实源文件看。

这是第一篇，s01 章节——消息总线。

1. 解耦"消息来源"和"处理逻辑"

想象一个最简单的场景。你想写一个 Agent，让它能：

• 从 Telegram 收消息
• 从 Discord 收消息
• 从 WebUI 收消息
• 被定时任务触发
• 接收 HTTP API 请求

最直觉的写法是这样：

  
    
    
    
  
  python
def handle_message(source, text):
    return call_llm(text)

每加一个渠道，就在 handle_message 里加一个 if source == "telegram"。加到第 5 个，这个函数已经 200 行了。加到第 8 个，没人想动它。

这种 if-else 堆叠的代码，常见于 Agent 项目的第一版。每次加新渠道都要回头改主函数。

s01 章节给的解法是：用两个队列把"消息入口"和"处理逻辑"分开。

  
    
    
    
  
  python
import asyncio
from dataclasses import dataclass

@dataclass
class Msg:
    text: str

class Bus:
    def __init__(self):
        self.inb: asyncio.Queue[Msg] = asyncio.Queue()
        self.out: asyncio.Queue[Msg] = asyncio.Queue()

就这么一个 Bus 类，10 行不到。

但它的作用是把整个系统的结构重写了。渠道（Channel）只管往 inb 塞消息。Agent 主循环只管从 inb 拿消息、处理、往 out 塞结果。渠道再从 out 把结果拿回去发。

这三类角色互不知道彼此。 新增一个 Telegram 渠道？写一个类，往 inb 塞消息就行。LLM 推理逻辑一行不用改。

s01 章节给了一个 60 行的最小例子，能直接跑（初次运行需要将.env.example复制为.env，配置llm api key）：

  
    
    
    
  
  python
import asyncio
from dataclasses import dataclass

@dataclass
class Msg:
    text: str

class Bus:
    def __init__(self):
        self.inb: asyncio.Queue[Msg] = asyncio.Queue()
        self.out: asyncio.Queue[Msg] = asyncio.Queue()

async def producer(bus, text):
    await bus.inb.put(Msg(text))

async def agent(bus):
    msg = await bus.inb.get()
    response = call_llm(msg.text)
    await bus.out.put(Msg(response))

async def consumer(bus):
    msg = await bus.out.get()
    print(f"[assistant] {msg.text}")

async def main(text):
    bus = Bus()
    await asyncio.gather(
        producer(bus, text),
        agent(bus),
        consumer(bus),
    )

asyncio.gather 把三个协程同时拉起来。producer 塞进去，agent 拿到开始推理，consumer 等结果打印。三者之间没有直接调用，全靠 Bus 通信。

跑起来什么样？控制台里输问题回车，回复就会打印出来。看着挺简陋，但这就是 nanobot 处理每条消息的核心结构。

2. 为什么用 asyncio.Queue，不用 list？

我第一次看到这里也有这个疑问。答案有两个，都挺关键。

第一个是背压保护。

asyncio.Queue 可以设上限，比如 asyncio.Queue(maxsize=100)。当第 101 个消息想塞进去，put 会 await 住——生产者自动被限流。

LLM 推理是慢的，但消息可以涌进来很快。如果用 list.append，队列会无限增长，直到把内存吃光——这是 Agent 项目的常见 OOM 原因。Queue 自带的"满了就等"机制，是天然的流量整形。

第二个是异步安全。

多协程并发 get/put 时，Queue 内部有锁保证原子性。不会出现"取到一半被另一个协程截胡"的诡异 bug。list 在 await 之间被改是经典坑，调试起来能让人抓狂。

3. nanobot 真实代码长什么样

把上面 60 行的玩具放大到 nanobot 真实代码，结构其实没变多少：

• inb

  入口：8 种渠道并发往里塞消息——Telegram bot 收到一条、Discord bot 收到一条、Cron 触发一条，全部互不感知
• out

  出口：每条响应带 channel + chat_id 标记，由 ChannelManager 查表发回对应渠道

• agent 部分：就是 AgentLoop，每条消息走一遍 7 个状态（恢复会话、压缩、处理命令、组装 Prompt、跑 LLM、保存、回复）

nanobot 真实代码里的 MessageBus 本质上就是上面这个 Bus 类的"成人版"：

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

事件从简单的 Msg 升级成 InboundMessage(channel, chat_id, user_id, text, ...)，但结构一点没变。

4. 常见的误区：主函数堆 if-else

很多 Agent 项目的第一版长这样：

  
    
    
    
  
  python
def handle_message(source, text):
    if source == "telegram": ...
    elif source == "discord": ...
    elif source == "feishu": ...
    # ... 加一个渠道就要加一个 elif

加到第 5 个还能忍，加到第 8 个没人想动它。

正确做法：第一步就引入消息总线。哪怕就 10 行 Bus 类，能解决三件事：

• 加新渠道零成本——新增渠道只写一个类，往 inb 塞消息
• 测试变简单——往 inb 塞测试消息，从 out 拿响应断言，不需要起 Telegram bot
• 多渠道并发安全——用 gather 拉起多个渠道，LLM 推理只有一份

这个 10 行的小类，比想象中有价值。

5. 系列预告

• 01 消息总线 ← 你在这里
• 02 Tool 与 Provider 的扩展艺术
• 03 Agent 状态机：一条消息的 7 个状态
• 04 Session 与两阶段记忆压缩
• 05 周边能力：Cron / Command / 沙箱

下一篇看 s02 + s03 章节，讲怎么用 pkgutil + entry_points 实现 Tool 的零修改扩展，以及 LLMProvider + ProviderSpec 怎么让代码"不锁定"具体厂商。

6. 参考资料

• 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 s01_bus/code.py 体验一下"两个队列解决解耦"是什么感觉，再对照 nanobot 源码 nanobot/bus/ 目录看真实实现。