nanobot 源码拆解:15K 行代码怎么撑住一个全功能 AI Agent

nanobot 项目封面：轻量级开源个人 AI Agent，MIT 协议

TL;DR

港大 HKUDS 的 Xubin Ren 一个人写了 nanobot，15K 行 Python，2026 年 2 月首发，4 个月迭代到 v0.2.2，18 个 release。10 个聊天渠道、12 个 LLM Provider、MCP 原生支持、记忆系统、WebUI，一条命令安装。

这是一个你可以一个下午读完源码、自己改、自己部署的 Agent 。

如果你正在做以下事情，值得停下来读：

用 LangChain 拼 Agent，被抽象层折磨
用 Dify 做平台，改不了核心逻辑
自己从零写 Agent，聊天渠道和记忆全得造 - 想接 MCP 但不知道怎么落地

2026 年个人 Agent 的三个条件

2026 年之前，个人 Agent 框架没有真正跑起来，缺三个条件。现在三个条件同时成熟了。

条件一：方向验证。 2025 年 SmolAgents、PydanticAI、Mastra 等轻量 Agent 框架集中出现，验证了「小核心 + 工具调用」路线可行。市场不需要又一个 LangChain，需要一个能读完的。nanobot 的 15K 行代码是对这个趋势的极端回应。

条件二：基础设施成熟。 MCP 协议在 2025 年完成标准化，主流 LLM Provider 的 API 趋向统一（OpenAI-compatible 成为事实标准），流式输出和 tool calling 成为标配。nanobot 不需要自己造协议，站在 MCP 和 OpenAI-compatible 两个标准上就行。

条件三：成本下降。开源模型质量追上 GPT-4 级别（DeepSeek、Qwen），API 价格一年降了 80%。个人跑一个 Agent 一天几块钱，不再是实验室特权。nanobot 的「一台机器 + API key」部署模式，在 2024 年还不现实，2026 年刚好。

三个条件缺一个，nanobot 都没有存在的理由。方向验证说明有人愿意用，基础设施成熟说明能接出来，成本下降说明跑得起。

先说结论：该不该用

直接给决策框架，不绕弯子。

适合你，如果：

需要一个接多个聊天平台的个人 Agent（Telegram + 微信 + 飞书 + WebUI）
想要小核心代码可以自己改（15K 行，一个下午能读完）
需要接 MCP 工具生态 - 部署预算接近零（一台机器 + API key） - 能接受一个人维护的项目风险

不适合你，如果：

需要 multi-agent 协作编排（nanobot 是单 Agent）
需要向量检索记忆（nanobot 用 Markdown 文件，没有 embedding） - 需要企业级权限管理和审计日志
需要稳定的 SLA（一个人维护，没有保障）
你的 Agent 需要跑复杂的 DAG 工作流

这个决策框架比任何 feature 对比表都实用。下面展开说为什么，从源码层面拆。

痛点诊断：Agent 框架的三个地狱

做过 Agent 落地的开发者都踩过这三个坑。

地狱一：抽象层套娃

LangChain 的 Agent 是这样跑的：你的消息穿过 Prompt Template，穿过 Output Parser，穿过 Tool Router，穿过 Memory Buffer，穿过 Agent Executor，最后到 LLM。中间任何一层出 bug，你都得翻五层抽象才能找到。

调试一个工具调用失败，你要看：是 Prompt 没传对？是 Output Parser 解析错了？是 Tool Router 没找到工具？是 Agent Executor 的循环逻辑有问题？每一层都有自己的 error handling，错误信息被层层包装，到你看的时候已经面目全非。

nanobot 的做法是砍掉抽象层。核心就两个文件：agent/loop.py 管「消息进来、context 构建、回复发出」，agent/runner.py 管「调模型、执行工具、喂结果回去」。没有 Prompt Template 类，没有 Output Parser 类，没有 Agent Executor 类。

这是设计选择，不是简化。LangChain 的抽象层是为了通用性，nanobot 的扁平结构是为了可读性。你用 LangChain 时是框架的用户，你用 nanobot 时是代码的主人。

地狱二：渠道集成黑洞

你要接一个 Telegram bot，需要：BotFather 注册、webhook 设置、消息格式处理、媒体下载、流式输出、会话隔离。一个渠道至少两天。

然后你要接飞书。飞书的 CardKit 消息格式和 Telegram 完全不同。再两天。然后微信。微信没有官方 bot API，你得用非官方协议，随时可能被封。再一周。

10 个渠道就是两个月纯工程量，和 Agent 逻辑无关。多数团队在这一步就放弃了，只接一个渠道。nanobot 把 10 个渠道全内置了，你改 config 就能接，不用写代码。

地狱三：记忆系统玄学

Agent 记忆是个被过度复杂化的问题。市面上有三种主流方案：无限堆 session（context window 撑不住）、summarization（丢信息）、向量检索（引入 embedding 依赖，复杂度高）。

nanobot 走了第四条路：三层分离加 Dream 整理。这个后面结合源码细说。

架构：一条直线走到底

nanobot 架构全景：Channel -> MessageBus -> AgentLoop -> AgentRunner -> Provider/Tools，Session/Memory 持久化

核心数据流是一条直线：

Channel -> MessageBus -> AgentLoop -> AgentRunner -> Provider/Tools -> 回写

MessageBus 的实现只有 45 行。两个 asyncio.Queue，一个进一个出，完事。没有 Kafka，没有 Redis Streams，没有 RabbitMQ。对于单进程 Agent 运行时，这就够了。过度设计消息中间件是 Agent 框架的常见病症。

class MessageBus:    def __init__(self):        self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue()        self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue()    async def publish_inbound(self, msg: InboundMessage) -> None:        await self.inbound.put(msg)    async def consume_inbound(self) -> InboundMessage:        return await self.inbound.get()  # 阻塞直到有消息

阻塞直到有消息

但简单不等于没有契约。Bus 上跑的是两个结构化事件 InboundMessage 和 OutboundMessage，渠道和 Agent 核心只认这两个类型，互相不知道对方存在。这层解耦带来一个意外的好处：子 Agent 完成任务后，可以伪装成一条 channel="system" 的 InboundMessage 把结果投回 bus，主 Agent 像处理普通消息一样处理它。后面讲子 Agent 时会看到这个技巧。

InboundMessage 里有个关键字段 session_key_override。正常消息的 session key 由 channel:chat_id 拼出来，但注入消息可以显式指定要进哪个 session 的待处理队列。这是「把结果路由到正确会话」和「作为一个独立任务竞争执行」之间的分界线。

状态机：不是简单的 while 循环

你以为 AgentLoop 就是个 while True？看源码，它是一个显式状态机：

class TurnState(Enum):    RESTORE = auto()    # 加载 session 历史    COMPACT = auto()    # 检查是否需要压缩    COMMAND = auto()    # 路由 slash 命令    BUILD = auto()      # 构建 context    RUN = auto()        # 调 AgentRunner    SAVE = auto()       # 持久化 session    RESPOND = auto()    # 发回复到渠道    DONE = auto()       # 结束_TRANSITIONS = {    (TurnState.RESTORE, "ok"): TurnState.COMPACT,    (TurnState.COMPACT, "ok"): TurnState.COMMAND,    (TurnState.COMMAND, "dispatch"): TurnState.BUILD,    (TurnState.COMMAND, "shortcut"): TurnState.DONE,    (TurnState.BUILD, "ok"): TurnState.RUN,    (TurnState.RUN, "ok"): TurnState.SAVE,    (TurnState.SAVE, "ok"): TurnState.RESPOND,    (TurnState.RESPOND, "ok"): TurnState.DONE,}

每个状态有对应的 handler，handler 返回一个事件字符串，驱动器查转移表决定下一步。这是真状态机，有转移表、有 trace 记录、有错误恢复路径。

你可能会问：一个消息处理需要状态机吗？需要。因为每个状态可能失败，失败后需要知道当前在哪一步，才能做正确的恢复。比如 RUN 阶段 LLM 超时，你需要跳到 SAVE 保存已有结果，再跳到 RESPOND 告诉用户出错了。如果是一个 200 行的大函数，这种恢复逻辑会变成意大利面条。

AgentLoop vs AgentRunner：职责拆分

AgentLoop（1866 行）管渠道面向的事：收消息、确定 session key、构建 context（SOUL.md + USER.md + MEMORY.md + session 历史 + 技能 + 渠道元数据）、调 AgentRunner、存 session、发回复。

AgentRunner（1570 行）管模型面向的事：调 Provider、处理流式响应、执行工具调用、喂结果回模型、判断什么时候停。

做过 Agent 开发的人都知道，渠道逻辑和模型逻辑混在一起是灾难。渠道要处理消息格式、媒体下载、流式输出、鉴权，和模型调用的逻辑完全不同。这个拆分让调试变得简单：渠道问题查 loop.py，模型问题查 runner.py。

你可能会想，1866 行的 loop.py 也叫小核心？这里有个细节：AgentLoop 的 __init__ 方法有 30 多个参数，包括 provider、workspace、session_manager、cron_service、subagent_manager、consolidator、auto_compact。这是组合根，不是上帝类。每个子系统是独立模块，AgentLoop 负责把它们串起来。真正的逻辑分散在各自的模块里。

Hook 系统：用生命周期回调隔离副作用

AgentRunner 的核心循环只管「调模型、执行工具、喂回去」。但实际运行时有一堆横切关注点：流式输出要推到渠道、子 Agent 要更新状态、SDK 调用要捕获结果、WebUI 要收到 reasoning 增量。如果把这些全塞进 runner，它会迅速变成上帝类。

nanobot 的解法是 agent/hook.py 的生命周期 Hook。AgentHook 定义了一组空实现的回调点，runner 在关键节点触发它们：

class AgentHook:    def wants_streaming(self) -> bool: return False    async def before_run(self, ctx): ...    async def before_iteration(self, ctx): ...    async def on_stream(self, ctx, delta): ...        # 流式 token    async def emit_reasoning(self, content): ...       # 思考过程增量    async def before_execute_tools(self, ctx): ...    async def after_iteration(self, ctx): ...    async def after_run(self, ctx): ...    async def on_error(self, ctx): ...    async def on_finally(self, ctx): ...    def finalize_content(self, ctx, content): return content

AgentHookContext 是每次迭代的可变状态快照（iteration、messages、tool_calls、usage、stop_reason），通过引用传给所有 hook，谁需要看什么自己取。

真正精巧的是 CompositeHook，它把多个 hook 扇出成一个：

async def _for_each_hook_safe(self, method_name, *args, **kwargs):    for h in self._hooks:        if getattr(h, "_reraise", False):            await getattr(h, method_name)(*args, **kwargs)            continue        try:            await getattr(h, method_name)(*args, **kwargs)        except Exception:            logger.exception("AgentHook.{} error in {}", method_name, type(h).__name__)

注意这里的错误隔离：一个有 bug 的自定义 hook 抛异常，只会被 log 下来，不会让整个 Agent 循环崩溃。这是「插件不能拖垮宿主」的经典实现。但 finalize_content 例外，它是个 pipeline（一个 hook 的输出是下一个的输入），刻意不做隔离，因为这里的 bug 应当暴露出来而不是被吞掉。

这个设计让渠道流式、子 Agent 状态、SDK 捕获三套完全不同的副作用，用同一组接口挂到 runner 上，runner 本身一行渠道代码都不用知道。

设计哲学：写进仓库的架构铁律

nanobot 能保持 15K 行不膨胀，不是靠运气，是靠写死在仓库里的约束。.agent/design.md 是给 AI 协作者和人类贡献者看的设计宪法，几条铁律值得任何做 Agent 框架的人抄走。

铁律一：核心小，边缘扩展。原文：「新能力应当通过 channels/、tools/、skills 或 MCP server 添加。agent/loop.py 和 agent/runner.py 是关键核心路径，改动必须最小且有充分理由。如果一个功能能放进渠道适配器、工具或外部 MCP server，就不应该内联进 agent loop。」这条规则把「核心」和「能力」物理隔离，新功能默认往边缘走，核心两个文件被严格保护。

铁律二：宁可重复，不要过早抽象。这条反直觉。原文明确允许各个渠道和 Provider 重复相似逻辑（发送重试、媒体处理、消息分片），并禁止「为了消除重复而引入复杂基类或共享 helper」。每个渠道文件必须独立可读。代价是代码行数变多（飞书 2356 行、微信 1586 行有不少相似逻辑），收益是你改飞书渠道时不会意外搞坏 Telegram，也不用在五层基类里追逻辑。对一个人维护的项目，这个权衡是对的。

铁律三：显式优于魔法。配置必须在 config/schema.py 的 Pydantic 模型里显式声明，错误处理要抛清晰异常而不是静默纠正坏输入。Provider 自动检测可以存在，但「每一条解析路径都必须能从 factory 追溯到具体的 provider 类」。这就是为什么 Provider 系统用注册表而不是一堆 if-else 散落各处。

这三条放在一起，解释了 nanobot 所有架构选择的底层逻辑：可读性和可追溯性优先于 DRY 和通用性。这是「给人读的代码库」和「给框架用的代码库」的根本分野。

Context 治理：隐藏的复杂度

这是 nanobot 源码里最值得读的部分。

你以为 AgentRunner 的核心循环就是「发消息给 LLM，拿回复」？在每次调 LLM 之前，Runner 会跑一个 5 步 context 治理管线：

for iteration in range(spec.max_iterations):    # 5 步 context 治理，每次调 LLM 前都跑    messages_for_model = self._drop_orphan_tool_results(messages)    messages_for_model = self._backfill_missing_tool_results(messages_for_model)    messages_for_model = self._microcompact(messages_for_model)    messages_for_model = self._apply_tool_result_budget(spec, messages_for_model)    messages_for_model = self._snip_history(spec, messages_for_model)    # 再清一次 orphan（snip 可能产生新的）    messages_for_model = self._drop_orphan_tool_results(messages_for_model)    messages_for_model = self._backfill_missing_tool_results(messages_for_model)    response = await self._request_model(spec, messages_for_model, ...)

这 5 步解决的是 Agent 在生产环境跑时的真实问题。做过 Agent 落地的人一看就懂：

第一步 _drop_orphan_tool_results：删除没有对应 tool_calls 的 tool 角色消息。为什么会出孤儿？因为 context 压缩时可能把 assistant 消息删了但留下了它对应的 tool result。OpenAI 和 Anthropic 的 API 会拒绝包含孤儿 tool result 的请求。

第二步 _backfill_missing_tool_results：反过来，如果 assistant 消息有 tool_calls 但对应的 tool result 丢了（比如执行中断），插入一条合成的错误结果。这步是防止 API 报「tool_use without tool_result」错误。

第三步 _microcompact：把旧的 read_file、exec、grep、web_search 等工具的结果替换成一行摘要 [tool_name result omitted from context]，只保留最近 10 条完整结果。做过浏览器标签页管理的人都知道这个感觉：关掉旧的标签页，只留最近看过的 10 个。你跑一个 Agent 让它读了 20 个文件，每个文件 5000 字，context 一下就爆了。microcompact 在不调 LLM 的前提下压缩 context，成本为零。

第四步 _apply_tool_result_budget：把超长的 tool result 截断到 max_tool_result_chars 限制。这是防止一个 exec 命令输出 50KB 日志把 context 撑爆。

第五步 _snip_history：估算整个 prompt 的 token 数，如果超过 context window 减去输出预算，从最旧的消息开始删，保留 system 消息和最近的对话。删的时候找 user 消息作为切割点，保证删完后第一条非 system 消息是 user（有些 provider 比如 GLM 会拒绝 system→assistant 的消息序列）。

这 5 步是 nanobot 在生产环境能跑起来的关键。没有这 5 步，Agent 跑几轮就会因为 context 超限、孤儿消息、API 格式错误而崩溃。像人走路时大脑自动处理平衡一样，这 5 步在后台默默处理 context 的脏污。我在生产环境跑 Agent 最大的教训就是：context 管理比 Agent 逻辑本身重要 10 倍，nanobot 是少数把这个做对的框架。

错误恢复链

除了 context 治理，AgentRunner 还有三层错误恢复：

错误类型	恢复策略	上限
空回复	重试，可能是模型偶发空输出	2 次
输出截断 (finish_reason="length")	追加一条「请继续」消息让模型续写	3 次
达到最大迭代	去掉 tools 再调一次模型做总结	1 次

还有一个 arrears 检测：如果 API key 余额不足，直接返回「请充值」提示，不重试。这个细节很实际，国产 API 经常余额不足，不检测的话会白白重试 3 次。

中途注入：边跑边接消息

Agent 在跑一个多轮工具调用任务时，用户又发来一条新消息怎么办？粗暴的做法是排队等当前任务跑完，或者打断重来。nanobot 走第三条路：在迭代循环的安全点把新消息注入进去。

AgentRunner 每轮迭代会调 _try_drain_injections()，通过 injection_callback 拉取待处理的用户消息，规范化成 user message 追加到对话里，让模型在下一轮自然地看到。关键是几个上限的钳制：

async def _try_drain_injections(self, spec, messages, assistant_message,                                injection_cycles, *, allow_goal_continue=False):    injections = []    if injection_cycles < _MAX_INJECTION_CYCLES:      # 防止无限注入        injections = await self._drain_injections(spec)    # 没有真实注入但目标仍活跃时，合成一条"继续"消息    if not injections and allow_goal_continue and assistant_message is not None:        if spec.goal_active_predicate and spec.goal_active_predicate():            injections = [self._build_goal_continue_message(spec)]    ...

_MAX_INJECTION_CYCLES 限制一次 turn 里最多注入几轮，_MAX_INJECTIONS_PER_TURN 限制单次拉取多少条，超出的会被 log 记录而不是静默丢弃。这两个上限防止「用户狂发消息把 Agent 卡在注入循环里」。

同一套注入机制还服务于「持续目标」（sustained goal）：当一个长任务还没完成、用户也没发新消息时，goal_active_predicate() 返回 True，runner 合成一条 continuation 消息让 Agent 自己接着干。用户插话和系统自驱，复用同一条注入通道，这是很干净的抽象复用。

记忆系统：三层分离加 Dream 整理

nanobot 内置的 10 个聊天渠道，每个都支持媒体和流式输出

nanobot 的记忆分三层：

层	存储位置	作用	淘汰机制
Session	sessions/*.jsonl	近期对话回放	Consolidator 摘要后清除
Memory	memory/MEMORY.md	长期事实	Dream 周期整理
Dream	memory/history.jsonl	整理源数据	整理后可清理

Consolidator：不是简单压缩

当 session 的 prompt token 超过 context window 的 50%（可配），Consolidator 启动。像读书做笔记：你不会把整本书背下来，而是读完一章把要点记到笔记本上，然后继续读下一章。Consolidator 做的就是这件事：

1.找到 session 中 last_consolidated 偏移量之后的未压缩消息

2.按 user-turn boundary 找切割点，在用户消息边界切，不在 assistant 消息中间切

3.把要删的消息发给 LLM 做摘要

4.摘要写入 history.jsonl

5.更新 last_consolidated 偏移量

如果 LLM 摘要失败，降级为 raw_archive()：直接把原始消息格式化后写入 history.jsonl，加 [RAW] 前缀。不丢数据，只是没压缩。

这个设计的好处是：session 里的消息可以被安全删除，因为关键信息已经被 LLM 提取到 history.jsonl 了。下次 Agent 启动时，ContextBuilder 会从 history.jsonl 读最近的条目注入 system prompt。

Dream：受限的 Agent 运行

Dream 不是简单的 cron 任务。它是一个完整的 Agent run，但工具集被严格限制：

def build_dream_tools(self):    tools = ToolRegistry()    tools.register(ReadFileTool(...))      # 只能读 workspace    tools.register(EditFileTool(...))      # 只能写 skills/ 和 MEMORY.md/SOUL.md/USER.md    tools.register(ApplyPatchTool(...))    # 同上    tools.register(WriteFileTool(...))     # 只能写 skills/    return tools

Dream 没有 shell、没有 web search、没有 MCP。它只能读文件和编辑记忆文件。像睡觉时大脑整理记忆：清醒时不做的事，在后台以低权限默默完成。这是一个安全设计：Dream 在后台跑，如果给它 shell 权限，Agent 可能在你睡觉时执行危险命令。

Dream 的输入是 history.jsonl 中上次 Dream cursor 之后的条目，每条截断到 500 字，最多取 20 条。输出是更新后的 MEMORY.md，可能还有新技能文件。

Dream session 文件会自动清理，只保留最近 10 个（prune_dream_sessions()）。防止 session 目录无限膨胀。

还有一个细节：记忆文件（SOUL.md、USER.md、MEMORY.md、.dream_cursor）通过 GitStore 做 git 版本追踪。每次 Dream 运行后自动 commit。这意味着你可以 git log 看到记忆的演变历史，也可以 git diff 看 Dream 改了什么。不需要数据库就有版本控制，这是文件系统方案的一个意外优势。值得直接抄这个设计。

原子写入：用文件系统做持久化的代价

文件方案最大的风险是写到一半进程挂了，留下半截损坏的文件。nanobot 的 history.jsonl 用经典的「临时文件 + fsync + 重命名」保证崩溃安全：

def _write_entries(self, entries):    tmp_path = self.history_file.with_suffix(self.history_file.suffix + ".tmp")    with open(tmp_path, "w", encoding="utf-8") as f:        for entry in entries:            f.write(json.dumps(entry, ensure_ascii=False) + "\n")        f.flush()        os.fsync(f.fileno())     # 数据落盘    os.replace(tmp_path, self.history_file)   # 原子重命名

os.replace 在同一文件系统内是原子操作：要么看到旧文件，要么看到完整新文件，永远不会看到半截。gotchas.md 里专门写了一条「不要把这个换成普通 open(..., "w")」，因为那会丢掉崩溃保证。

append 历史还有个并发细节。cursor（条目序号）的分配和写入必须在同一把锁里完成：

with self._append_lock:    cursor = self._next_cursor()    # ... 写入 ...

否则两个并发 writer 可能读到同一个 cursor，产生重复序号。这种 bug 在低并发下几个月才出一次，但出了极难查。

读取侧则是「容忍损坏」。_iter_valid_entries() 逐条校验，遇到格式错误的条目跳过并只 warn 一次，不让一条脏数据搞崩整个历史加载。写入严格、读取宽容，这是处理持久化数据的成熟姿态。

AutoCompact：空闲会话主动压缩

agent/autocompact.py 解决的是另一个问题：长期挂着的 gateway 里，很多 session 聊几句就没动静了，但它们的历史还占着 context，下次被唤醒时又要全量加载。AutoCompact 在 session 空闲超过 TTL 后，主动把它压缩成一段摘要。

设计上有两个考究点。一是不碰正在跑的 session：check_expired() 接收一个 active_session_keys 集合，跳过有 in-flight 任务的 session，避免压缩和正在写入的对话打架。二是冷热双路径取摘要：

def prepare_session(self, session, key):    entry = self._summaries.pop(key, None)    if entry:                                  # 热路径：进程没重启，内存里有        return session, self._format_summary(entry[0], entry[1])    meta = session.metadata.get("_last_summary")  # 冷路径：进程重启过，从持久化读    if isinstance(meta, dict):        return session, self._format_summary(meta["text"], ...)    return session, None

进程没重启时摘要在内存 dict 里（热路径，零 IO），重启后从 session metadata 里恢复（冷路径）。Dream 等内部 session 用前缀 dream: 标识，直接跳过压缩。这套机制让一个挂了几个月、积累上千 session 的 gateway 不会被历史拖垮。

聊天渠道：工程量在哪

10 个渠道不是 10 个 HTTP 转发。看 channels/base.py 的接口定义，每个渠道要实现：

•start()：连接平台、监听消息

•stop()：清理资源

•send(msg)：发送消息

•send_delta(chat_id, delta, metadata)：流式输出一个 chunk

•send_reasoning_delta(chat_id, delta, metadata)：流式输出思考过程

•send_file_edit_events(chat_id, edits, metadata)：推送文件编辑事件

后面三个是可选的，但实现了的渠道体验完全不同。Telegram 用 edit_message 做流式，飞书用 CardKit streaming，WebUI 用 WebSocket。每个平台一套方案。

Pairing：DM 渠道的访问控制

一个容易被忽略的设计：is_allowed 方法实现了四级权限检查：

def is_allowed(self, sender_id: str) -> bool:    if "*" in allow_list:        return True              # 1. 通配符，允许所有人    if str(sender_id) in allow_list:        return True              # 2. 白名单    if is_approved(self.name, str(sender_id)):        return True              # 3. Pairing store    return False                 # 4. 拒绝

如果 sender 不在白名单里，且是 DM（私信），渠道会生成一个 pairing code 发给 sender。管理员看到 code 后批准，sender 就被加入 pairing store。这个设计解决了「我想让朋友用我的 bot 但不想开放给所有人」的场景。

渠道隔离的工程问题

不同平台的消息格式差异很大。Telegram 支持 Markdown，飞书用 CardKit，Slack 用 mrkdwn，微信只支持纯文本。nanobot 在渠道层做格式转换，Agent 核心只产出统一格式。这层转换看着简单，做起来很烦，10 个平台就是 10 套格式规则。不是客套话，渠道适配是 Agent 落地中最容易被低估的工期黑洞。

从代码量看，飞书渠道 2356 行，Telegram 1671 行，微信 1586 行，Signal 1402 行。这些是完整的平台适配层，每个平台一套格式规则。

unifiedSession 配置可以让多渠道共享 session。你在 Telegram 问了一半，切到 WebUI 继续问，上下文不丢。关掉它就是独立会话，适合多人共用一个 bot 的场景。

Provider 路由：注册表驱动

nanobot 内置的 12 个 LLM Provider，覆盖云端、本地、OAuth 三种接入方式

Provider 系统的核心是 providers/registry.py，一个 ProviderSpec 元组注册表。加一个新 Provider 只需要两步：在 PROVIDERS 元组里加一个 ProviderSpec，在 config/schema.py 加配置字段。

注册表的顺序就是匹配优先级。Gateway 类 Provider（OpenRouter、OpenCode Zen/Go）排在前面，因为它们能路由任何模型。本地 Provider（Ollama、vLLM）排在后面。

每个 ProviderSpec 有 20 多个字段，控制行为细节：

•detect_by_key_prefix：通过 API key 前缀猜 Provider（sk-or- = OpenRouter）

•thinking_style：thinking 控制方式（DeepSeek 用 thinking_type，DashScope 用 enable_thinking，MiniMax 用 reasoning_split）

•supports_prompt_caching：是否支持 prompt caching（Anthropic 和 OpenRouter 支持）

•strip_model_prefixes：网关型 Provider 需要剥掉模型前缀（opencode/gpt-5 → gpt-5）

•reasoning_effort_remap：不同 Provider 接受的 reasoning_effort 值不同（Mistral 只接受 high/none）

看一个具体的 ProviderSpec 定义，感受适配粒度：

ProviderSpec(    name="deepseek",    keywords=("deepseek",),    env_key="DEEPSEEK_API_KEY",    base_url="https://api.deepseek.com/v1",    thinking_style="thinking_type",     # DeepSeek 用 thinking_type 控制推理    supports_prompt_caching=False,    is_local=False,    is_gateway=False,)

每个字段对应一个真实适配逻辑。thinking_style="thinking_type" 意味着 nanobot 在发请求时会把 thinking 参数包装成 {"thinking": {"type": "enabled"}} 格式，而不是 OpenAI 的 {"reasoning_effort": "high"}。这种细粒度适配是 12 个 Provider 能真正跑起来的原因。

fallbackModels 配置让主模型挂了自动切备用模型。生产环境用国产 API 时这个功能很重要，稳定性不如 OpenAI，有 fallback 能救命。

Provider 类别	代表	接入方式	关键适配
云端直连	OpenAI、Anthropic、DeepSeek、DashScope	API key	thinking_style 各异
网关路由	OpenRouter、OpenCode Zen/Go	API key + 模型名	strip_model_prefixes
本地推理	Ollama、vLLM	localhost URL	无 API key
OAuth 认证	GitHub Copilot、OpenAI Codex	OAuth flow	token 刷新

这四类覆盖了目前所有主流 LLM 接入方式。加一个新 Provider，在注册表里加一行 ProviderSpec 就行。

FallbackProvider：透明包装的故障转移

fallbackModels 不是在调用点写一堆 try/except，而是用装饰器模式包一层。factory.py 的 make_provider() 在解析出主 Provider 后，如果配了 fallback，就用 FallbackProvider 把主备链包起来：

provider = _make_provider_core(config, ...)         # 主 Providerfallback_presets = _resolve_fallback_presets(config, resolved)if fallback_presets:    provider = FallbackProvider(primary=provider, fallbacks=..., ...)return provider

对 runner 来说，它拿到的就是一个普通 LLMProvider，完全不知道背后有没有 fallback。主模型报错时，FallbackProvider 在内部切到下一个备用模型，调用方无感。这又是「显式优于魔法」铁律的体现：故障转移是一个明确的包装层，能从 factory 一路追溯到它包了谁。

provider_signature：配置热重载的指纹

gateway 是长驻进程，用户可能在运行中改 config.json 换模型。每次都重建 Provider 太浪费，但不重建又会用旧配置。nanobot 的解法是给 Provider 链算一个「指纹」：

def provider_signature(config, ...) -> tuple[object, ...]:    return (        resolved.model, resolved.provider, provider_name,        config.get_api_key(...), config.get_api_base(...),        ..., resolved.max_tokens, resolved.temperature,        resolved.reasoning_effort, resolved.context_window_tokens,        tuple(_fallback_signature(fb) for fb in fallback_presets),  # 连 fallback 一起算    )

连 fallback 一起算

provider_signature() 把所有影响 Provider 行为的配置字段（模型、key、base url、生成参数、整条 fallback 链）打包成一个元组。运行时拿新配置算一次指纹，和当前的比，只在指纹变化时才重建 Provider。这是一个轻量的脏检查，避免了「每次请求都重建」和「配置改了不生效」两个极端。

工具系统：内置 + MCP + 技能三层

工具系统分三层，每层的扩展成本不同。

内置工具（agent/tools/）：文件读写、Shell 执行、Web 搜索、Cron 定时、图片生成、子 Agent。Shell 执行带可配置沙箱，Web 搜索带 SSRF 检查。

MCP 工具：在 config 里加 tools.mcpServers，nanobot 启动时自动连接、发现工具、注入到工具列表。不用写代码，不用重启 Agent，改配置就行。

MCP 实现里有个细节值得注意：_MalformedProgressNotificationFilter。有些 MCP 服务器发送的 progress notification 不符合协议规范（缺少 progressToken），nanobot 不会崩溃，而是静默丢弃这些畸形通知。这是生产环境兼容性的体现。

还有一个 _probe_http_url 函数：在连接 HTTP MCP 服务器之前先做 TCP 探测。如果端口没开，直接跳过，不进入 streamable_http_client。因为 anyio task group 的 cleanup 可能抛出逃逸 try/except 的异常，导致事件循环崩溃。这种边界 case 只有踩过坑的人才会处理。

技能系统：workspace 下的 skills/ 目录放技能文件，可以是 prompt 模板、工具组合、自定义逻辑。通过 /skill 命令发现和安装，还支持从 ClawHub 公共技能市场搜索安装。

层级	扩展方式	需要写代码	典型场景
内置工具	Python 类 + 注册	是	文件、Shell、Web
MCP 工具	config.json 加 mcpServers	否	外部服务接入
技能	skills/ 目录放 .md	否	prompt 模板、工作流

三层工具体系的扩展成本递减：内置工具要写 Python，MCP 工具改配置就行，技能连配置都不用改，放个文件就生效。

工具并发执行

_partition_tool_batches 方法实现了工具并发执行：标记为 concurrency_safe 的工具可以批量并行执行，不安全的串行执行。这是一个实用的性能优化，特别是当 Agent 同时调多个 read_file 或 web_search 时。

ToolRegistry 的稳定排序

def get_definitions(self) -> list[dict[str, Any]]:    builtins.sort(key=self._schema_name)    mcp_tools.sort(key=self._schema_name)    self._cached_definitions = builtins + mcp_tools

工具定义的排序是稳定的：内置工具按名字排序在前，MCP 工具按名字排序在后。这是为了 prompt caching。如果工具顺序每次都不一样，prompt 的 hash 就变了，Anthropic 和 OpenRouter 的 cache 就 miss 了。稳定排序 = 省钱。

子 Agent：后台任务与结果回注

agent/subagent.py 的 SubagentManager 是 nanobot 处理「耗时任务不阻塞主对话」的方案。它不是真正的多 Agent 编排，但把单 Agent 的后台委派做得很干净。

隔离的工具集。子 Agent 不继承主 Agent 的全部工具，而是通过 ToolLoader().load(ctx, registry, scope="subagent") 加载一个 scope 受限的注册表。配置里只保留 exec、web、file 三类，并强制套用 restrict_to_workspace。这意味着后台跑的子 Agent 不能再生子 Agent，也不能碰 workspace 外的文件。

asyncio.Task 后台执行。spawn() 不阻塞，它创建一个 asyncio.create_task，立刻返回一句「已启动，完成后通知你」给主 Agent。任务句柄存进 _running_tasks，并按 session 分组进 _session_tasks，这样可以按 session 批量取消：

async def cancel_by_session(self, session_key: str) -> int:    tasks = [self._running_tasks[tid]             for tid in self._session_tasks.get(session_key, [])             if tid in self._running_tasks and not self._running_tasks[tid].done()]    for t in tasks:        t.cancel()    if tasks:        await asyncio.gather(*tasks, return_exceptions=True)    return len(tasks)

add_done_callback 注册的 _cleanup 负责任务结束后从三个字典里清干净，不留泄漏。

结果回注：伪装成系统消息。子 Agent 跑完后，结果怎么回到主对话？这里用了前面讲的 MessageBus 技巧。子 Agent 把结果包成一条 channel="system"、sender_id="subagent" 的 InboundMessage，通过 session_key_override 对齐到主 Agent 的有效 session key，投回 bus：

msg = InboundMessage(    channel="system",    sender_id="subagent",    chat_id=f"{origin['channel']}:{origin['chat_id']}",    content=announce_content,    session_key_override=override,   # 对齐主 Agent session    metadata={"injected_event": "subagent_result", "subagent_task_id": task_id},)await self.bus.publish_inbound(msg)

源码注释点明了为什么必须用 session_key_override：它让结果进入主 Agent 的待处理队列做「中途注入」，而不是作为一个独立任务去和主 Agent 竞争执行。这正好接上了前面 Context 治理讲的注入机制，子 Agent 的结果走的就是 _try_drain_injections 那条路。

失败也要有结构。子 Agent 出错时不是甩一个堆栈，_format_partial_progress() 会提取最后完成的几步和失败点，组织成「Completed steps / Failure」的简报回注，让主 Agent 知道进展到哪、卡在哪，可以决定要不要重试或换路。

安全：不是可选项

Agent 安全是个被低估的问题。很多人跑 Agent 时直接给 shell 权限，不设白名单。Agent 被注入攻击后可以执行任意命令，这是真实风险。

SSRF 防护

security/network.py 的实现比多数项目想得仔细：

_BLOCKED_NETWORKS = [    ipaddress.ip_network("0.0.0.0/8"),    ipaddress.ip_network("10.0.0.0/8"),    ipaddress.ip_network("100.64.0.0/10"),    # 运营商级 NAT    ipaddress.ip_network("127.0.0.0/8"),    ipaddress.ip_network("169.254.0.0/16"),    # 链路本地 / 云元数据    ipaddress.ip_network("172.16.0.0/12"),    ipaddress.ip_network("192.168.0.0/16"),    ipaddress.ip_network("::1/128"),    ipaddress.ip_network("fc00::/7"),           # IPv6 唯一本地    ipaddress.ip_network("fe80::/10"),          # IPv6 链路本地]

注意 169.254.0.0/16：这是云元数据端点。Agent 通过 web_fetch 访问 http://169.254.169.254/latest/meta-data/ 就能拿到云服务器的临时凭证。nanobot 默认阻断这个网段。

还有一个 IPv6-mapped IPv4 绕过防护：::ffff:127.0.0.1 在 Python 的 ipaddress 模块里是 IPv6Address，不匹配 127.0.0.0/8。nanobot 的 _normalize_addr() 函数把 IPv6-mapped IPv4 转回 IPv4 再检查。这种绕过手法在真实攻击中出现过。

validate_resolved_url() 在 HTTP 重定向后再次检查目标 IP。防止攻击者用一个公网 URL 302 跳转到内网地址。

Shell 沙箱

sandbox.py 实现了 bubblewrap（bwrap）后端：

args = ["bwrap", "--new-session", "--die-with-parent", "--setenv", "HOME", str(ws)]args += ["--ro-bind", p, p for p in required]     # /usr 只读args += ["--tmpfs", str(ws.parent),                # 遮蔽 config 目录         "--dir", str(ws),                         # 重建 workspace 挂载点         "--bind", str(ws), str(ws)]               # workspace 可读写

workspace 只读挂载到沙箱里，workspace 的父目录（存放 config.json）被 tmpfs 遮蔽。Agent 在沙箱里看不到 config.json，即使 shell 有权限也读不到 API key。media 目录只读挂载，Agent 能读上传的图片但不能写。

目前只有 bwrap 一个后端，Linux 专用。macOS 和 Windows 没有 sandbox 后端，只能靠 allow_patterns/deny_patterns 做命令过滤。门已经打开了，但只开了一扇。

Workspace 范围

workspace_access.py 用 ContextVar 实现每轮对话的 workspace 隔离。restrict_to_workspace() 开启后，文件工具只能读写 workspace 内的文件。WebUI 可以动态切换 workspace scope，不同 session 可以指向不同的项目目录。

安全这块 nanobot 做得比多数开源项目认真，但还有提升空间。bwrap 只支持 Linux 是硬伤，macOS 用户只能靠命令过滤兜底。

部署：从零到跑起来要几步

pip install nanobot-ainanobot onboard --wizard# 选 Provider，填 API key，选渠道nanobot gateway

WebUI 浏览器访问 http://127.0.0.1:8765。没有 Docker，没有数据库，没有 Redis。一台能跑 Python 3.11+ 的机器就行。

用云端 API 的话，树莓派都能跑。用本地 LLM 的话，一张消费级 GPU 跑 Ollama 或 vLLM 就够。3B 模型在消费级 GPU 上跑，电费几乎可以忽略。当然，本地模型的推理质量不如 claude，但做日常助手够用。

三种运行模式

模式	命令	场景
CLI 一次性	nanobot agent -m "..."	脚本集成、首次验证
Gateway 长驻	nanobot gateway	多渠道、WebUI、cron、Dream
OpenAI API	nanobot serve	嵌入现有系统、给其他 Agent 做后端

Gateway 模式是主力模式。启动后所有渠道同时在线，cron 任务和 Dream 记忆整理也在后台跑。后台运行用 --background，配合 status/logs/restart/stop 管理。

并发控制通过 NANOBOT_MAX_CONCURRENT_REQUESTS 环境变量设置（默认 3）。超过限制的请求会排队等待。每个 session 有独立的 asyncio.Lock，保证同一 session 的消息串行处理。

WebUI：内置在 wheel 里

nanobot WebUI：浏览器工作台，支持多会话、workspace 切换、实时文件编辑可视化、模型切换

WebUI 不需要额外构建步骤，直接打包在 wheel 里。前端 114 个 TypeScript/TSX 文件，用 Vite 构建，产物写入 nanobot/web/dist/ 并打包进 wheel。pip install 后直接有 WebUI，不用装 Node.js，不用跑 npm build。

配置速查

项目	配置
安装	pip install nanobot-ai
Python	>= 3.11（支持 3.14）
配置文件	~/.nanobot/config.json
工作目录	~/.nanobot/workspace/
会话存储	/sessions/*.jsonl
长期记忆	/memory/MEMORY.md
技能目录	/skills/
Cron 存储	/cron/jobs.json
WebUI 端口	8765（WebSocket）
健康检查	18790
协议	MIT
版本	v0.2.2（2026-06-22）
首发日期	2026-02-02
迭代节奏	日更，4 个月 18 个 release
代码量	~15K 行 Python + 114 个前端文件
最大并发	3（环境变量可调）
Session 上限	2000 条消息/文件
Context 压缩比	0.5（超过 50% context window 时触发）
Dream 批量	20 条历史，每条截断 500 字
Dream session 保留	最近 10 个

这张表是部署前的速查参考。接下来和主流框架做横向对比，看 nanobot 在工程权衡上选了哪条路。

和主流框架的工程权衡

nanobot 与 Dify、AutoGPT、Open Interpreter、LangChain 的能力对比

框架	核心抽象	代码量	渠道	MCP	记忆	部署门槛
nanobot	AgentLoop + Runner	~15K	10+	原生	三层+Dream	pip install
LangChain	Chain + Agent Executor	~200K+	无	社区	需自建	拼装多个包
Dify	Workflow + App	~100K+	插件	插件	DB	Docker + Server
AutoGPT	Autonomous Loop	~50K+	无	无	无	Docker
Open Interpreter	Code Execution	~30K+	CLI	无	无	pip install

这是工程权衡表，不是 feature 对比表。每个框架的设计选择都意味着取舍：

LangChain 选了通用性：抽象层多，什么都能做，但调试痛苦。适合需要快速原型验证的场景，不适合需要稳定运行的生产场景。

Dify 选了平台化：功能全，有 UI，但你不拥有代码。适合不想写代码的团队，不适合需要深度定制的场景。

AutoGPT 选了自主性：Agent 自己决定做什么，但经常死循环。适合实验，不适合生产。

nanobot 选了可拥有性：核心小，你能读完、能改、能部署。功能不如 Dify 全，抽象不如 LangChain 多，但代码是你的。

行业格局 2×2：谁在哪

	零代码	全代码
通用平台	Dify：功能全，不拥有代码	LangChain：抽象全，调试痛
个人可拥有	Open Interpreter：CLI 交互，门槛低	nanobot：15K 行可读，渠道全

左上角是 Dify：零代码、通用平台，适合不想写代码的团队。代价是你不拥有代码，核心逻辑改不了。

右上角是 LangChain：全代码、通用平台，什么都能做。代价是抽象层多，调试痛苦。

左下角是 Open Interpreter：零代码、个人可拥有，CLI 交互门槛低。但没有渠道、没有记忆、没有 MCP。

右下角是 nanobot：全代码、个人可拥有，15K 行可读、10 个渠道全内置。代价是功能不如 Dify 全，生态不如 LangChain 大。

对工程师来说，右下角是最舒服的位置：你有代码控制权，又不用从零造渠道和记忆。nanobot 目前是这个格子里的唯一选择。

一个人维护的项目能活多久

项目由港大 HKUDS 的 Xubin Ren（GitHub: re-bin）以个人开源项目发起。HKUDS 此前出过 LightRAG、GraphRAG 等知名项目，nanobot 是他们在 Agent 基础设施方向的尝试。

迭代节奏惊人：4 个月 18 个 release，几乎每天都有 commit。前两个月搭核心（Provider、Channel、MCP、记忆），后两个月打磨稳定性（session 持久化、流式输出、安全边界、WebUI）。

但一个人维护的项目有结构性风险：

Bus factor = 1。只有一个人理解全部代码。作者停更，项目就死了。社区贡献能补功能，但核心架构决策只能靠作者。

渠道维护是持续成本。 10 个渠道意味着 10 套 API 变更要跟。微信和 QQ 的非官方协议随时可能断。一个人跟 10 个平台的变更，精力分散。

功能扩张 vs 核心简洁的矛盾。 nanobot 的卖点是 15K 行小核心。但用户会要求加向量检索、multi-agent、workflow 编排。加还是不加？加了核心膨胀，不加用户流失。

可持续性判断：短期（6 个月）没问题，迭代活跃，社区在增长。中期（1-2 年）取决于作者能否找到可持续的维护模式。长期不确定，这是所有个人开源项目的通病。

局限

没有 multi-agent 协作。 nanobot 是单 Agent 架构，没有 agent-to-agent 通信协议。子 Agent 工具可以做简单委派，但不是真正的多 Agent 编排。如果你需要多个 Agent 协作（一个写代码一个 review），nanobot 不直接支持。

记忆系统没有向量检索。 MEMORY.md 是纯文本，Dream 做的是事实提取不是语义检索。对话历史长了之后，找相关上下文靠 Consolidator 摘要加 session replay，不靠 embedding 搜索。对于长周期记忆场景（「三个月前我说过什么」），这个方案不够。

WebUI 是单体前端。 114 个 TSX 文件打包在 wheel 里虽然方便，但前端和后端耦合。想用 nanobot 后端但自己写前端，需要走 WebSocket 协议自己对接。

渠道维护成本高。 10 个渠道意味着 10 套 API 变更要跟。微信和 QQ 的非官方协议尤其脆弱。从代码量看，飞书渠道 2356 行，微信 1586 行，这些是持续的维护负担。

没有评测体系。 nanobot 是工程项目不是研究项目，没有 benchmark。Agent 质量完全取决于你接的 LLM 和你写的 prompt。你没法量化比较 nanobot 和其他框架的 Agent 效果。

15K 行的代价。每个模块的代码量都很薄。薄意味着好读，也意味着边界 case 覆盖不够。生产环境跑起来会发现各种小问题，需要你自己补。这是「小核心」设计的必然代价。你选择了可读性，就放弃了健壮性。

一个尖锐问题： context 治理管线的 5 步处理看起来完善，但有一个隐患：_snip_history 的 token 估算是近似的。如果估算偏低，实际 prompt 可能超过 context window，导致 API 报错。估算偏高则浪费 context 空间。nanobot 用 estimate_message_tokens 做粗估，没有精确 tokenizer，对于中文等非拉丁字符的估算偏差更大。这是一个已知的 trade-off：精确 tokenizer 需要为每个 Provider 集成对应的库，成本高。

总结

nanobot 的价值不在功能列表，在「你可以真正拥有它」。15K 行代码，一个下午读完，改你想改的，部署到你想部署的地方。

三个信号值得深究：

信号一：Agent 核心不需要很复杂，但 context 治理必须做。 nanobot 的核心循环就是消息进来，LLM 决定调不调工具，调完喂回去，出结果。没有 DAG，没有 planner，没有 multi-agent 通信。但每次调 LLM 前的 5 步 context 治理（orphan 清理、backfill、microcompact、budget 截断、history snip）是 Agent 在生产环境能跑起来的关键。如果你在做 Agent，这 5 步比任何编排框架都重要。

信号二：渠道集成是 Agent 落地的真问题。多数 Agent 框架不处理聊天渠道，但实际落地时渠道集成占了一半工作量。nanobot 把 10 个渠道内置，是工程量，不是技术难度。如果你要做产品化 Agent，渠道集成是你需要提前规划的，别等到最后才发现这是个黑洞。

信号三：记忆系统需要分层不是压缩。 Session + Memory + Dream 的三层设计比单纯 summarization 更合理。Consolidator 在 token 预算触发时做 LLM 摘要，Dream 在后台做事实提取，GitStore 给记忆文件做版本追踪。不需要向量数据库，不需要 embedding，一个 Markdown 文件加 git 就够。如果你在设计 Agent 记忆，考虑分层而不是一味压缩。

对工程师来说，可以做的事：

1.如果你在做 Agent 落地，clone nanobot，读 agent/runner.py 的 _run_core 方法，理解 5 步 context 治理管线

2.如果你需要多渠道 Agent，直接用 nanobot 的渠道系统，不用自己写 10 个平台的 API 对接

3.如果你在设计 Agent 记忆，参考 Dream 的设计：后台 cron + 受限工具集 + GitStore 版本追踪

4.如果你想理解 MCP 落地，nanobot 的 MCP 实现在 agent/tools/mcp.py，注意畸形通知过滤和 TCP 预探测两个细节

5.如果你在选 Agent 框架，用上面的决策框架判断，别看 feature 列表，看你的场景和 nanobot 的取舍是否匹配

15K 行代码撑住一个全功能 Agent 框架，这件事本身就是一个信号：Agent 核心的复杂度下界比我们以为的低得多。但 context 治理、错误恢复、安全边界的复杂度不会消失，只是被移到了你看不到的地方。nanobot 把这些复杂性暴露在源码里，让你能读懂、能修改、能拥有。