Featured image of post Nanobot 源码深度拆解:Python 异步消息总线构建的轻量级 AI Agent 框架

Nanobot 源码深度拆解:Python 异步消息总线构建的轻量级 AI Agent 框架

这是我的编码 Agent 开源系列第四篇。前三天我分别拆解了 Grok Build(Rust 50+ crate)、Pi Agent(TypeScript 双队列)和 OpenCode(Effect TS 四轴路由)。今天的主角是 Nanobot

Nanobot 由香港大学数据科学实验室(HKUDS)开发,是一个 Python 3.11+ asyncio 构建的轻量级 AI Agent 框架。它和前三者的最大区别在于:Nanobot 不是编码 Agent,而是通用 Agent 框架。

TL;DR:Nanobot 的 Dream 两阶段记忆合并系统是三者中最完善的记忆方案。它的 MessageBus 架构让 15+ 聊天平台可以即插即用。但 Python 的性能限制让它不适合做代码编辑等延迟敏感的任务。


一、架构全景:MessageBus 异步事件驱动

Nanobot 的核心架构是一句话:一个 async MessageBus 解耦了 Chat Channels 和 Agent Core

Channels Layer (Telegram, Discord, Slack, Feishu, 微信...)
    ↓ InboundMessage (publish_inbound)
┌──────────────────────────────────────────────────┐
│  MessageBus (asyncio.Queue)                      │
│     ↓ consume_inbound                            │
├──────────────────────────────────────────────────┤
│  AgentLoop (调度层)                               │
│     ↓ AgentRunner (执行层 - LLM + Tool 循环)       │
│     ↓ ToolRegistry (工具执行)                      │
│     ↓ SessionManager (JSONL 持久化)               │
│     ↓ Consolidator (Dream 记忆合并)               │
│     ↑ OutboundMessage (publish_outbound)          │
├──────────────────────────────────────────────────┤
│  MessageBus (outbound)                            │
│     ↓ consume_outbound                           │
└──────────────────────────────────────────────────┘
Channels → send() to platform

nanobot/bus/queue.py,只有 44 行:

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

极简主义。没有优先级队列,没有持久化,没有死信队列。

对比其他三个项目的架构模式:

项目 通信模式 通道层 解耦方式
Grok Build ACP 协议 (Gateway) 5 种部署模式 协议层解耦
OpenCode Effect HttpApi TUI + Desktop + Web Service 层解耦
Pi Agent EventEmitter CLI TUI 事件订阅解耦
Nanobot async Queue 15+ Chat 平台 消息总线解耦

Nanobot 是唯一一个原生支持多聊天平台的。Telegram、Discord、Slack、飞书、微信、企业微信、钉钉、Email、Mattermost、Matrix、WhatsApp、QQ、MoChat、Teams、WebSocket——15+ 平台,全部通过同一个 MessageBus 接入。


二、两阶段 Agent Loop:调度层和执行层分离

这是 Nanobot 最核心的架构设计。Agent 的处理被分成了两个清晰的层次。

2.1 AgentLoop — 调度层

nanobot/agent/loop.py,约 2000 行。

AgentLoop 负责:

  1. 消息消费:从 MessageBus 的 inbound 队列消费消息
  2. 会话路由:按 session_key 做任务路由,同一 session 串行,跨 session 并发
  3. 上下文构建:组装 system prompt、历史记录、memory、skills
  4. 状态管理:TurnState 状态机
  5. MCP 连接管理:连接/断开 MCP 服务器
  6. Cron 和自动化任务:协调定时任务和本地触发器

关键设计是会话锁和 pending 队列

# 每个 session 一个锁
lock = self._session_locks.setdefault(session_key, asyncio.Lock())
# 消息路由到 pending 队列
if effective_key in self._pending_queues:
    self._pending_queues[effective_key].put_nowait(pending_msg)

这实现了 mid-turn injection 模式——用户可以在 Agent 思考过程中发送新消息,它们会被注入到正在运行的 LLM 循环中。

对比 Pi Agent 的 Steering 队列:两者都实现了 mid-turn injection,但实现方式不同。Pi Agent 在 Agent Loop 内部维护 Steering 队列,Nanobot 在外部通过 MessageBus 和 session 锁实现。Pi Agent 的方式更精细(可以区分 Steering 和 Follow-up),Nanobot 的方式更通用(不依赖 Agent Loop 实现)。

2.2 AgentRunner — 执行层

nanobot/agent/runner.py,约 1400 行。

AgentRunner 是一个纯函数式的执行引擎,不知道任何产品层概念(channel、session、用户)。它接收一个 AgentRunSpec 返回 AgentRunResult

核心循环:

for iteration in range(spec.max_iterations):
    # 1. Context Governance — 修复消息格式、压缩历史
    messages_for_model = self.context_governor.prepare_for_model(...)
    # 2. 调用 LLM
    response = await self._request_model(spec, messages_for_model, hook, context)
    # 3. 执行工具
    for tool_call in response.tool_calls:
        result = await tool_registry.execute(tool_call, context)
    # 4. 检查继续条件

对比 Grok Build 的 MvpAgent:Grok Build 的 Agent 是有状态的(通过 SessionHandle 管理),而 Nanobot 的 AgentRunner 是无状态的——状态全部在 AgentRunSpec 中传递。这使得 AgentRunner 可以被单元测试,不需要 mock 任何全局状态。


三、TurnState 状态机:8 个状态的精细控制

Nanobot 的 TurnState 状态机是代码中最值得学习的模式之一:

RESTORE → COMPACT → COMMAND → BUILD → RUN → SAVE → RESPOND → DONE
状态 做了什么 失败时
RESTORE 从磁盘加载会话历史 跳过
COMPACT 检查是否需要压缩长历史 跳过
COMMAND 检查是否是 / 命令 进入 BUILD
BUILD 构建 LLM 请求 错误返回
RUN 调用 LLM + 执行工具 重试或失败
SAVE 持久化结果 继续
RESPOND 通过 MessageBus 发送回复 继续
DONE 清理和完成

对比 Grok Build 的会话 5 状态(Working → IdleResident → Dormant → Completed → DeadFailed):Grok Build 的状态是会话生命周期,Nanobot 的状态是单个 turn 的执行过程。 两者关注不同层面。

对比 OpenCode 的 Processor 事件驱动:OpenCode 用事件流(LLMEventhandleEvent),Nanobot 用显式状态机。状态机更容易理解和调试,事件流更灵活。


四、Dream 两阶段记忆合并

Nanobot 的记忆系统是三者中最完善的。nanobot/agent/memory.py(1167 行)实现了 Dream 两阶段记忆合并:

第一阶段:Consolidation

按 token 数量自动触发的 Incremental 合并:

if session_total_tokens > memory_config.consolidate_threshold:
    # 取出最早的消息块
    # 用 LLM 生成摘要
    # 用摘要替换原始消息块
    # 记录 token 节省量

第二阶段:Dream

定时触发的 LLM 驱动的跨会话总结

# 在后台每隔一段时间:
# 1. 读取所有 session 的 consolidated history
# 2. 用 LLM 生成跨 session 的洞察
# 3. 写入全局 MEMORY.md

为什么这是三者中最完善的?

对比 Grok Build 的 xai-grok-memory crate:Grok Build 的记忆系统是用户手动管理的(memory 工具 + skill 工具),没有自动合并。它靠的是用户主动决定什么时候保存、保存什么。

对比 Pi Agent:Pi Agent 没有长期记忆系统,session 之间的数据不共享。

对比 OpenCode:OpenCode 有 Session Summary 生成(summary.ts),但仅限于单 session 的摘要,没有跨 session 的合并。

Nanobot 的 Dream 是唯一一个「自动的、跨 session 的、LLM 驱动的」记忆系统。 而且它的设计很克制——Consolidation 只做 token 层面的压缩,Dream 才做语义层面的总结,两阶段分开避免 LLM 调用过多。


五、MCP 集成:最复杂的子系统

Nanobot 的 MCP 集成在 nanobot/agent/tools/mcp.py(1435 行),是代码库中最复杂的子系统。

它实现了:

  1. 名称 sanitization:Python 方法名不能有 -,MCP 工具名可能有 -,需要做名字映射
  2. Schema 标准化:MCP 的 JSON Schema 格式和 Nanobot 的 ToolSchema 格式不同,需要转换
  3. 会话重连:MCP Server 断开后自动重连,带退避策略
  4. 多 MCP Server 管理:每个 Server 独立连接池

对比 Grok Build 的 xai-grok-mcp:Grok Build 的 MCP 集成更底层——它直接操作 rmcp crate 的 transport 层,支持 Streamable HTTP 和子进程。Nanobot 的 MCP 是更高层的封装——通过 Python 的 mcp 库连 MCP Server,然后适配到自己的工具系统。

两者的差异是语言生态的体现:Rust 生态没有成熟的 MCP 客户端库,所以 Grok Build 必须自己实现 transport 层;Python 生态有 mcp 客户端库,Nanobot 可以站在上面做封装。


六、工具系统:自动发现 + 参数校验

Nanobot 的工具系统在 nanobot/agent/tools/ 下,每个工具一个文件。

6.1 自动发现

tools/loader.py 通过两种方式自动发现工具:

  1. pkgutil 扫描:扫描 nanobot.agent.tools 包下的所有模块
  2. entry_points 插件:通过 Python 的 entry_points 机制发现第三方工具

6.2 参数校验

tools/schema.py 实现了完整的 JSON Schema 类型系统:

class ToolSchema:
    type: str           # "string" | "integer" | "object" | ...
    description: str    # 工具描述
    properties: dict    # 参数定义
    required: list      # 必填参数

每个工具通过 ToolSchema 定义参数,registry.py 在调用时做 validate_params → cast_params 两阶段验证。

对比 Grok Build 的 ToolDefinition:Grok Build 的 MCP 工具通过 ToolBridge.register_mcp_tools() 注册,MCP 工具和本地工具共享同一个 FinalizedToolset。Nanobot 的 MCP 工具需要单独适配,不能和本地工具统一注册。


七、Provider 抽象层:Fallback 链式封装

Nanobot 的 Provider 系统在 nanobot/providers/ 中,支持 5 种后端:

  • Anthropic:Claude 系列
  • OpenAI Compat:兼容 OpenAI API 的模型
  • OpenAI Responses:OpenAI 的 Responses API
  • Azure OpenAI:Azure 部署
  • Bedrock:AWS Bedrock
  • GitHub Copilot:Copilot 代理

核心设计是 FallbackProvider

class FallbackProvider:
    def __init__(self, providers: list[Provider]):
        self.providers = providers

    async def complete(self, messages):
        for provider in self.providers:
            try:
                return await provider.complete(messages)
            except Exception:
                continue
        raise AllProvidersFailed()

对比 Pi Agent 的 30+ Provider:Pi Agent 的 Provider 覆盖更广,但都是单 Provider 直连。Nanobot 的 Provider 数量少,但支持 Fallback 链式切换。这是两种不同的哲学:广度 vs 韧性。


八、和三个编码 Agent 的横向对比

维度 Grok Build OpenCode Pi Agent Nanobot
语言 Rust TypeScript/Effect TypeScript Python
定位 编码 Agent 编码 Agent 编码 Agent 通用 Agent 框架
源文件 2804 2499 859 500+
沟通平台 CLI 5 模式 TUI + Desktop CLI TUI 15+ Chat 平台
会话存储 ACP 5 状态 Durable SQLite JSONL 树形 JSONL 文件
记忆系统 手动管理 Session Summary Dream 自动合并
MCP ✅ 完整传输层 ✅ 插件化 ✅ 适配层
子 Agent 进程内 1 层 插件 RPC 进程 消息总线
工具数 60+ 较多 中等 20+
验证系统 Skeptic + Goal
权限系统 ✅ 完整 ✅ Ruleset ❌ 容器化

写在最后

Nanobot 作为一个 Python Agent 框架,它的定位和前三者完全不同。它不是编码 Agent,而是通用 Agent 运行时。它的 15+ 聊天平台集成、Dream 两阶段记忆合并、TurnState 状态机,都是为「长时间、多平台、多会话」的通用 Agent 场景设计的。

如果你需要的是一个可以接入 Telegram、飞书、微信,能跑自动任务,有记忆系统的 Agent 框架,Nanobot 是这四个项目中唯一的选择。

但如果你需要的是高效的代码编辑能力,Python 的异步性能瓶颈和 GIL 会让它力不从心。这时候 Rust 的 Grok Build 或 TypeScript 的 OpenCode/Pi Agent 更适合。

相关阅读:

GitHub: https://github.com/HKUDS/nanobot

By AI博士 万戈