1. nanobot 源码拆解Agent 主循环到底在跑什么nanobot 是港大 HKUDS 发布的一个轻量级个人 AI 助手框架定位是 OpenClaw 的平替实现核心代码大约 4000 行相比 OpenClaw 的 43 万行压缩了 99%。它能做什么一句话概括把 Telegram、Discord、微信这类聊天渠道接进来用一个 Agent 主循环驱动 LLM 做工具调用、记忆管理和定时任务。适合谁适合想读懂轻量 Agent 框架内部机制、又不想被几十万行代码淹没的开发者。我第一次翻 OpenClaw 源码的时候翻了两天还在 channels 层打转后来转去看 nanobot一个下午就把主链路摸清了。这篇文章就按三条主线来拆Agent 主循环、工具调用、LLM 调度最后把 endpoint 和 auth.json 改到 TaoToken 统一通道本地跑通验证。nanobot 的内部分层非常清晰先看整体结构模块职责核心文件AgentAI 代理核心逻辑loop.py, context.pyChannels聊天平台适配层base.py, telegram.py, discord.pyBus消息路由与队列events.py, queue.pyProvidersLLM 提供商抽象base.py, litellm_provider.pyTools工具执行系统base.py, registry.py, mcp.pyMemory持久化记忆memory.pyConfig配置管理schema.py, loader.py每一层都值得单独看但如果你时间有限优先看 Agent 和 Providers 这两层因为它们决定了整个框架的行为边界。1.1 外部处理流程从消息进来到回复出去先看最外层的处理逻辑在nanobot/agent/loop.py的_process_message方法里async def _process_message(self, msg: InboundMessage): 核心处理逻辑 # 1. 获取或创建会话 session self.sessions.get_or_create(msg.session_key) # 2. 加载历史消息 history session.get_history(max_messagesself.memory_window) # 3. 构建上下文系统提示 历史 当前消息 initial_messages self.context.build_messages( historyhistory, current_messagemsg.content, mediamsg.media, channelmsg.channel, chat_idmsg.chat_id, ) # 4. 运行 Agent 循环LLM ↔ 工具调用 final_content, _, all_msgs await self._run_agent_loop( initial_messages, on_progress_bus_progress ) # 5. 保存会话历史 self._save_turn(session, all_msgs, skip_count) self.sessions.save(session) return OutboundMessage(...)get_or_create就是去 memory 里找这个对话是否已有 session有则取出没有就新建一个用session_key标识。Session 类的定义长这样dataclass class Session: A conversation session. Stores messages in JSONL format for easy reading and persistence. Important: Messages are append-only for LLM cache efficiency. The consolidation process writes summaries to MEMORY.md/HISTORY.md but does NOT modify the messages list or get_history() output. key: str # channel:chat_id messages: list[dict[str, Any]] field(default_factorylist) created_at: datetime field(default_factorydatetime.now) updated_at: datetime field(default_factorydatetime.now) metadata: dict[str, Any] field(default_factorydict) last_consolidated: int 0 # Number of messages already consolidated to files注意last_consolidated这个字段它是记忆整合的游标后面讲 Memory 模块时会重点说。get_history是一个历史消息的拼接只要完整的、限制最长的、剔除非客户的、结构化的信息def get_history(self, max_messages: int 500) - list[dict[str, Any]]: Return unconsolidated messages for LLM input, aligned to a user turn. unconsolidated self.messages[self.last_consolidated:] sliced unconsolidated[-max_messages:] # Drop leading non-user messages to avoid orphaned tool_result blocks for i, m in enumerate(sliced): if m.get(role) user: sliced sliced[i:] break out: list[dict[str, Any]] [] for m in sliced: entry: dict[str, Any] {role: m[role], content: m.get(content, )} for k in (tool_calls, tool_call_id, name): if k in m: entry[k] m[k] out.append(entry) return out这里有个细节值得注意for i, m in enumerate(sliced)那段是为了避免出现孤立的 tool_result 块。因为如果切片刚好从一条 tool 消息开始LLM 会报错说找不到对应的 tool_call。这个坑我在自己写 Agent 的时候踩过报错信息是messages with role tool must be a response to a preceding message with tool_calls排查了半天才发现是切片位置的问题。build_messages的任务是构造大模型适配的内容来使用def build_messages( self, history: list[dict[str, Any]], current_message: str, skill_names: list[str] | None None, media: list[str] | None None, channel: str | None None, chat_id: str | None None, ) - list[dict[str, Any]]: Build the complete message list for an LLM call. # 运行时的信息拼接 runtime_ctx self._build_runtime_context(channel, chat_id) # 用户消息拼接 user_content self._build_user_content(current_message, media) # Merge runtime context and user content into a single user message # to avoid consecutive same-role messages that some providers reject. if isinstance(user_content, str): merged f{runtime_ctx}\n\n{user_content} else: merged [{type: text, text: runtime_ctx}] user_content return [ {role: system, content: self.build_system_prompt(skill_names)}, *history, {role: user, content: merged}, ]这里把 runtime context 和用户内容合并成一条 user 消息是为了避免连续两条同角色消息被某些 provider 拒绝。这个兼容性处理很实用因为不同厂商的 API 对消息序列的校验严格程度不一样。构造好信息后就开始进入_run_agent_loop循环开始跑任务了。_save_turn和self.sessions.save(session)就是把现有的结果信息整理记录下来内部包含很多清晰的工作如清除空 assistant 消息、截断超长工具结果、多模态信息处理等。1.2 Agent 内部循环ReAct 是怎么转起来的接下来讲_run_agent_loop这是 Agent 内部处理任务的流程。核心是这么几个步骤初始化循环状态进入迭代循环调用大模型工具调用分支调用工具获取结果更新到消息历史最终回复分支生成最终回复返回结果调用大模型是直接封装了一个 provider 来处理的response await self.provider.chat( messagesmessages, toolsself.tools.get_definitions(), modelself.model, temperatureself.temperature, max_tokensself.max_tokens, reasoning_effortself.reasoning_effort, )这个 provider 就是调用大模型。这里以最简单的 LiteLLMProvider 为例讲讲在nanobot/providers/litellm_provider.pyasync def chat( self, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None None, model: str | None None, max_tokens: int 4096, temperature: float 0.7, reasoning_effort: str | None None, ) - LLMResponse: Send a chat completion request via LiteLLM. ...这里有个细节发现一个还挺好的工具litellm可以简化大模型的调用from litellm import acompletion response await acompletion(**kwargs)它的后处理也有些说法除了解析文本还会解析可能会用到的 tool这里是专门弄了个 Response 的类来装这个def _parse_response(self, response: Any) - LLMResponse: Parse LiteLLM response into our standard format. choice response.choices[0] message choice.message tool_calls [] if hasattr(message, tool_calls) and message.tool_calls: for tc in message.tool_calls: # Parse arguments from JSON string if needed args tc.function.arguments if isinstance(args, str): args json_repair.loads(args) tool_calls.append(ToolCallRequest( id_short_tool_id(), nametc.function.name, argumentsargs, )) usage {} if hasattr(response, usage) and response.usage: usage { prompt_tokens: response.usage.prompt_tokens, completion_tokens: response.usage.completion_tokens, total_tokens: response.usage.total_tokens, } reasoning_content getattr(message, reasoning_content, None) or None thinking_blocks getattr(message, thinking_blocks, None) or None return LLMResponse( contentmessage.content, tool_callstool_calls, finish_reasonchoice.finish_reason or stop, usageusage, reasoning_contentreasoning_content, thinking_blocksthinking_blocks, )注意json_repair.loads(args)这一行因为 LLM 返回的工具参数 JSON 经常不合法比如多一个逗号、少一个引号用 json_repair 能自动修复。这个库在 Agent 场景里几乎是必备的。LLMResponse 的类是这样的dataclass class LLMResponse: Response from an LLM provider. content: str | None tool_calls: list[ToolCallRequest] field(default_factorylist) finish_reason: str stop usage: dict[str, int] field(default_factorydict) reasoning_content: str | None None # Kimi, DeepSeek-R1 etc. thinking_blocks: list[dict] | None None # Anthropic extended thinking property def has_tool_calls(self) - bool: Check if response contains tool calls. return len(self.tool_calls) 0如果大模型判断要调用工具那就要执行if response.has_tool_calls: # 判断是否有还在进行的指令如果有则会给一个正在进行的回复。 if on_progress: clean self._strip_think(response.content) if clean: await on_progress(clean) await on_progress(self._tool_hint(response.tool_calls), tool_hintTrue) # 把工具转为对应的标准格式。 tool_call_dicts [ { id: tc.id, type: function, function: { name: tc.name, arguments: json.dumps(tc.arguments, ensure_asciiFalse) } } for tc in response.tool_calls ] # 保存助手的完整响应含工具调用到消息历史 messages self.context.add_assistant_message( messages, response.content, tool_call_dicts, reasoning_contentresponse.reasoning_content, thinking_blocksresponse.thinking_blocks, ) # 依次开始执行工具 for tool_call in response.tool_calls: tools_used.append(tool_call.name) args_str json.dumps(tool_call.arguments, ensure_asciiFalse) logger.info(Tool call: {}({}), tool_call.name, args_str[:200]) result await self.tools.execute(tool_call.name, tool_call.arguments) messages self.context.add_tool_result( messages, tool_call.id, tool_call.name, result )流程讲解都在注释里比较浅显。需要注意的是result await self.tools.execute(tool_call.name, tool_call.arguments)这里可以注意到执行工具是同步的使用的是 await。如果是非工具调用一般就是准备最终的回复了这个比较简单else: clean self._strip_think(response.content) # Dont persist error responses to session history — they can # poison the context and cause permanent 400 loops (#1303). if response.finish_reason error: logger.error(LLM returned error: {}, (clean or )[:200]) final_content clean or Sorry, I encountered an error calling the AI model. break messages self.context.add_assistant_message( messages, clean, reasoning_contentresponse.reasoning_content, thinking_blocksresponse.thinking_blocks, ) final_content clean break这里有个注释值得注意错误响应不写入 session history因为会污染上下文导致永久 400 循环。这个 issue #1303 的修复思路很值得借鉴我在自己的项目里也遇到过类似问题——一次失败的请求被存进历史后后续每次请求都带着这条错误消息LLM 一直返回 400。2. TaoToken 前置把 LLM 调度统一到一个通道nanobot 默认走 LiteLLM支持 OpenAI、Anthropic、DeepSeek 等多家 provider。但实际用的时候你会发现一个问题每换一个模型就要改一次配置API Key 散落在各个地方调试的时候很难统一管理。我试过把 endpoint 和 auth.json 改到 TaoToken 统一通道好处是所有模型走一个 Base URLKey 也只管一个切换模型只改 Model ID 就行。TaoToken 的官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。nanobot 的 provider 配置在nanobot/config/schema.py里核心字段是base_url、api_key、model。LiteLLM 本身支持自定义api_base所以只要把这三个字段指向 TaoToken 就行。先看 nanobot 的配置文件结构。默认配置在~/.nanobot/config.json你也可以在项目根目录放一个config.json覆盖。关键字段长这样{ providers: { default: { type: litellm, base_url: https://taotoken.net/api, api_key: sk-你的TaoToken密钥, model: claude-sonnet-4-5, max_tokens: 4096, temperature: 0.7 } }, agent: { memory_window: 50, max_iterations: 20 } }这里base_url填https://taotoken.net/api注意不要带 UTM 参数API 调用只需要干净的端点。model字段填你要用的模型 IDTaoToken 支持的模型列表可以在控制台里看。如果你用的是 Codex 风格的auth.json格式是这样的{ OPENAI_API_KEY: sk-你的TaoToken密钥, OPENAI_BASE_URL: https://taotoken.net/api, model: claude-sonnet-4-5 }三件套必须写全Base URL、Key、Model ID。少任何一个都会报 401 或者 model not found。如果你用 Claude Code 或者 Cline 这类工具配置方式类似。Claude Code 的 settings.json 里{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的TaoToken密钥, ANTHROPIC_MODEL: claude-sonnet-4-5 } }Cline 的 MCP 配置里如果是走 OpenAI 兼容协议{ mcpServers: { taotoken: { command: npx, args: [-y, modelcontextprotocol/server-openai], env: { OPENAI_API_KEY: sk-你的TaoToken密钥, OPENAI_BASE_URL: https://taotoken.net/api } } } }这里要提醒一句MCP 直连生产库是禁止的上面这个配置只是演示协议格式实际用的时候不要把 MCP server 指向你的生产数据库。配置好之后nanobot 启动时会读取这个配置LiteLLM 会用base_url作为api_base发起请求。你可以在nanobot/providers/litellm_provider.py里加一行日志确认logger.info(Using base_url: {}, self.base_url)如果日志里打印的是https://taotoken.net/api说明配置生效了。3. 可复制配置从零跑通 nanobot TaoToken这一节给你一份可以直接复制的配置从 clone 代码到跑通第一条消息。3.1 环境准备与依赖安装先 clone 代码git clone https://github.com/HKUDS/nanobot.git cd nanobot创建虚拟环境并安装依赖python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -e .nanobot 的核心依赖包括litellm、asyncio、pydantic、json_repair等。如果你要用 Telegram 或 Discord 渠道还需要额外装python-telegram-bot和discord.py。3.2 配置文件config.json 完整示例在项目根目录创建config.json{ providers: { default: { type: litellm, base_url: https://taotoken.net/api, api_key: sk-你的TaoToken密钥, model: claude-sonnet-4-5, max_tokens: 4096, temperature: 0.7, reasoning_effort: medium } }, agent: { memory_window: 50, max_iterations: 20, workspace: ./workspace, restrict_to_workspace: true }, channels: { cli: { enabled: true, send_tool_hints: true, send_progress: true } }, tools: { exec: { timeout: 30, path_append: } } }几个关键字段说明base_url填https://taotoken.net/api这是 TaoToken 的 API 端点。api_key填你在控制台生成的密钥格式一般是sk-开头。model填模型 ID比如claude-sonnet-4-5、gpt-4o、deepseek-chat等具体支持列表看 TaoToken 控制台。memory_window控制加载多少条历史消息默认 50。max_iterations是 Agent 循环的最大迭代次数防止死循环。restrict_to_workspace设为 true 时文件工具只能操作 workspace 目录内的文件这个安全设置建议开启。3.3 auth.json 配置Codex 风格如果你用的是 Codex 风格的认证文件在~/.nanobot/auth.json里写{ OPENAI_API_KEY: sk-你的TaoToken密钥, OPENAI_BASE_URL: https://taotoken.net/api, model: claude-sonnet-4-5 }nanobot 启动时会优先读auth.json如果不存在再读config.json里的 providers 配置。两个文件都写的话auth.json优先级更高。3.4 启动与验证启动 CLI 模式python -m nanobot.cli.commands或者用入口脚本nanobot启动后你会看到类似这样的输出Agent loop started Using base_url: https://taotoken.net/api Model: claude-sonnet-4-5然后就可以输入消息了。先试一条简单的你好请介绍一下你自己如果配置正确你会看到 Agent 的回复。如果报错看下一节的排查清单。4. 验证请求确认 LLM 调度真的走通了配置好之后怎么确认请求真的发到了 TaoToken 而不是别的地方有几个验证方法。4.1 日志验证在nanobot/providers/litellm_provider.py的chat方法开头加一行logger.info(LLM request: model{}, base_url{}, model, self.base_url)启动后发一条消息看日志输出。如果打印的是base_urlhttps://taotoken.net/api说明配置生效。4.2 抓包验证可选如果你有抓包工具可以看请求的 Host 是不是taotoken.net。不过更简单的方法是看响应里的 usage 字段TaoToken 返回的 usage 格式和 OpenAI 一致usage { prompt_tokens: 123, completion_tokens: 45, total_tokens: 168, }在_parse_response里加一行日志打印 usage确认有值就说明请求成功了。4.3 工具调用验证发一条需要工具调用的消息比如帮我读一下 README.md 文件的内容Agent 应该会触发read_file工具。你会在日志里看到Tool call: read_file({path: README.md})然后 Agent 会把文件内容返回给你。这一步验证了工具调用链路是通的。4.4 记忆验证发几条消息后检查workspace/MEMORY.md和workspace/HISTORY.md是否有内容cat workspace/MEMORY.md cat workspace/HISTORY.md如果HISTORY.md里有对话记录说明记忆模块在工作。MEMORY.md是长期记忆需要触发 consolidate 才会写入默认是消息数超过memory_window时触发。5. 本篇常见错排查401、local proxy failed、reading choices这一节列出实际跑的时候最容易遇到的几个报错以及对应的排查方法。5.1 401 Unauthorized报错信息litellm.exceptions.AuthenticationError: OpenAIException - Error code: 401 - {error: {message: Invalid API key, type: invalid_request_error}}原因API Key 不对或者 Base URL 配错了。排查步骤第一确认api_key字段填的是 TaoToken 的密钥不是 OpenAI 的。TaoToken 的密钥格式一般是sk-开头但和 OpenAI 的密钥不通用。第二确认base_url填的是https://taotoken.net/api不要带 UTM 参数不要带尾部斜杠。有些人会填成https://taotoken.net/api/或者https://taotoken.net/api?utm_sourcexxx都会导致 401。第三确认auth.json和config.json没有冲突。如果两个文件都写了auth.json优先。检查一下auth.json里的OPENAI_API_KEY是不是旧的。5.2 local proxy failed报错信息litellm.exceptions.APIConnectionError: OpenAIException - local proxy failed原因网络连不上或者 Base URL 不可达。排查步骤第一确认网络能访问taotoken.net。在终端里跑curl -I https://taotoken.net/api如果返回 200 或 401说明网络通。如果超时检查你的网络设置。第二确认没有配置额外的代理。LiteLLM 会读环境变量HTTP_PROXY和HTTPS_PROXY如果这两个变量指向了一个不可用的代理就会报 local proxy failed。检查一下echo $HTTP_PROXY echo $HTTPS_PROXY如果有值且不是你想要的unset 掉unset HTTP_PROXY unset HTTPS_PROXY第三确认base_url没有拼写错误。常见错误是写成https://taotoken.net/v1或者https://api.taotoken.net正确的端点是https://taotoken.net/api。5.3 reading choices 报错报错信息KeyError: choices或者IndexError: list index out of range原因LLM 返回的响应格式不对_parse_response里response.choices[0]取不到值。排查步骤第一确认模型 ID 正确。如果model字段填了一个 TaoToken 不支持的模型返回的可能是错误信息而不是标准的 chat completion 响应。检查 TaoToken 控制台的模型列表确认你填的模型 ID 在列表里。第二在_parse_response开头加一行日志logger.info(Raw response: {}, response)看看实际返回的是什么。如果返回的是{error: model not found}那就是模型 ID 的问题。第三确认max_tokens没有超过模型限制。有些模型的最大输出是 4096你填 8192 会报错。5.4 OAuth 相关报错报错信息OAuth token expired或者Failed to refresh OAuth token原因如果你用的是 OAuth 认证而不是 API Keytoken 过期了。排查步骤第一确认你用的是 API Key 而不是 OAuth。TaoToken 支持 API Key 认证在config.json里填api_key字段就行不需要 OAuth。第二如果确实需要 OAuth检查auth.json里的 token 是否过期。重新生成一个 API Key 更简单。第三确认auth.json的格式正确。OAuth 相关的字段和 API Key 字段不能混用。5.5 工具调用报错报错信息Error: Tool xxx not found原因LLM 调用了一个不存在的工具。排查步骤第一检查_register_default_tools里注册了哪些工具。默认注册的工具包括read_file、write_file、edit_file、list_dir、exec、web_search、web_fetch、message、spawn、cron。第二如果 LLM 调用了一个没注册的工具检查 system prompt 里是否包含了这个工具的定义。有时候 LLM 会幻觉出一个工具名。第三在ToolRegistry.execute里加日志logger.info(Executing tool: {} with params: {}, name, params)看看实际调用的是什么工具。6. 语义一致 CTA把 nanobot 接到 TaoToken 统一通道nanobot 的源码拆解到这里就差不多了。三条主线——Agent 主循环、工具调用、LLM 调度——都过了一遍。如果你想把 nanobot 跑起来最省事的方式是把 endpoint 和 auth.json 改到 TaoToken 统一通道所有模型走一个 Base URLKey 也只管一个。具体操作第一去 TaoToken 控制台生成一个 API Key地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。第二把 Key 填到config.json的api_key字段base_url填https://taotoken.net/apimodel填你要用的模型 ID。第三启动 nanobot发一条消息验证。如果你在配置过程中遇到报错先看第 5 节的排查清单。401 一般是 Key 或 Base URL 的问题local proxy failed 一般是网络或代理的问题reading choices 一般是模型 ID 的问题。想验证模型是否可用可以直接在 TaoToken 的模型对话页面测试 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。如果你打算长期用 nanobot 做编码或 Agent 任务可以考虑 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入文档在这里 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。Claude Code 的接入配置参考 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 。最后说一个实际经验nanobot 的 Memory 模块是本地文件化的MEMORY.md和HISTORY.md都存在 workspace 目录下。如果你在多台机器上跑记忆不会同步。生产环境用的话建议把记忆层换成数据库或者至少把 workspace 挂到共享存储上。这个坑我在部署的时候踩过A 机器上聊了半天换到 B 机器发现记忆全没了。