这是我的编码 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 负责:
- 消息消费:从 MessageBus 的 inbound 队列消费消息
- 会话路由:按
session_key做任务路由,同一 session 串行,跨 session 并发 - 上下文构建:组装 system prompt、历史记录、memory、skills
- 状态管理:TurnState 状态机
- MCP 连接管理:连接/断开 MCP 服务器
- 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 用事件流(LLMEvent → handleEvent),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 行),是代码库中最复杂的子系统。
它实现了:
- 名称 sanitization:Python 方法名不能有
-,MCP 工具名可能有-,需要做名字映射 - Schema 标准化:MCP 的 JSON Schema 格式和 Nanobot 的 ToolSchema 格式不同,需要转换
- 会话重连:MCP Server 断开后自动重连,带退避策略
- 多 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 通过两种方式自动发现工具:
- pkgutil 扫描:扫描
nanobot.agent.tools包下的所有模块 - 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 更适合。
相关阅读:
- Grok Build 开源!万字拆解 40 万行 Rust 构建的 AI Coding Agent 内部架构
- OpenCode 源码深度拆解:Effect TS 代数效应系统构建的智能编码 Agent
- Pi Agent 深度拆解:TypeScript 多包架构下的 Steering + Follow-up 双队列设计
GitHub: https://github.com/HKUDS/nanobot