夜雨聆风
源码拆解|AgentLoop 是 nanobot 的发动机,20 次迭代、工具调用与进度流
接着上一篇，我们重点研究 nanobot 的核心引擎 AgentLoop。
打开 nanobot/agent/loop.py，我就盯一件事，当用户发来一句话，系统到底经历了什么，才把一段可交付的结果回复回去？
这条链路真正的价值，我更看重可控性（controllability）。
nanobot 的写法很简单，一个循环、一个迭代上限、一个统一工具注册表，再加一个随时可以发回去的进度流。
这篇我就围绕 AgentLoop 看它的 4 个机制。
-
消息处理，session、slash command、memory window
-
上下文构建，ContextBuilder 把一切拼成 prompt
-
迭代循环，最多 20 次工具调用与模型回合
-
进度流，用户不会一直盯着一个空白屏
PART 01｜入口，_process_message 先解决会话纪律
AgentLoop.run() 只是一个不断消费 inbound queue 的外层循环。真正的核心在 _process_message()。
在 nanobot/agent/loop.py 里，它先把 session 找出来。
key = session_key or msg.session_keysession = self.sessions.get_or_create(key)
你可以把这句话理解成，同一个 chat_id 的对话，永远回到同一个状态容器里。
接下来它处理两个 slash 命令，/new 和 /help（同样在 nanobot/agent/loop.py 里）。这段我觉得很产品化，因为它不仅是交互指令，还是状态管理的开关。
if cmd == "/new":# Capture messages before clearing (avoid race condition with background task)messages_to_archive = session.messages.copy()session.clear()self.sessions.save(session)self.sessions.invalidate(session.key)async def _consolidate_and_cleanup():temp_session = Session(key=session.key)temp_session.messages = messages_to_archiveawait self._consolidate_memory(temp_session, archive_all=True)asyncio.create_task(_consolidate_and_cleanup())return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,content="New session started. Memory consolidation in progress.")
我特别注意到两点。
-
它先 clear 再后台 consolidation，避免在用户继续发消息时产生状态竞态（race condition）
-
consolidation 用 asyncio.create_task 异步跑，不堵塞主链路。
这就是轻量系统的实用主义。先让用户体验顺滑，再慢慢补齐记忆。
PART 02｜上下文，ContextBuilder 把 prompt 做成拼装件
当 session 纪律立住后，AgentLoop 做下一步，构建 LLM 输入。
这段调用发生在 nanobot/agent/loop.py，而 build_messages() 的拼装逻辑在 nanobot/agent/context.py。
initial_messages = self.context.build_messages(history=session.get_history(max_messages=self.memory_window),current_message=msg.content,media=msg.media if msg.media else None,channel=msg.channel,chat_id=msg.chat_id,)
这段看似普通，背后其实是 nanobot 的一个核心取舍，history 只拿最近 memory_window 条，剩下的交给 memory 系统去沉淀。
这和很多无限长 prompt 式的 demo 不一样。它在系统层面承认，上下文是有限的，必须做收敛。
PART 03｜心脏，_run_agent_loop 把工具调用写成一个可控循环
真正的发动机在 _run_agent_loop()（同样在 nanobot/agent/loop.py）。
我建议你直接看它的 while loop，这是 nanobot 的第一性原理。
while iteration < self.max_iterations:iteration += 1response = await self.provider.chat(messages=messages,tools=self.tools.get_definitions(),model=self.model,temperature=self.temperature,max_tokens=self.max_tokens,)if response.has_tool_calls:if on_progress:clean = self._strip_think(response.content)await on_progress(clean or self._tool_hint(response.tool_calls))tool_call_dicts = [{"id": tc.id,"type": "function","function": {"name": tc.name,"arguments": json.dumps(tc.arguments)}}for tc in response.tool_calls]messages = self.context.add_assistant_message(messages, response.content, tool_call_dicts,reasoning_content=response.reasoning_content,)for tool_call in response.tool_calls:tools_used.append(tool_call.name)args_str = json.dumps(tool_call.arguments, ensure_ascii=False)logger.info(f"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)else:final_content = self._strip_think(response.content)break
这段代码有三个工程边界非常重要。
-
迭代上限（max_iterations），默认 20。我更愿意把它理解成 tool-using agent 的安全阀。没有上限的 agent，迟早会把账单打爆。
-
工具定义来自 ToolRegistry，工具不写死在 prompt 里，而是 schema 化地传给模型。工具系统越标准化，系统越可维护。
-
每次工具调用都有 tool_result 回写，模型不会凭空知道你执行了什么。它通过 tool message 继续推理。
我做过偏生产的 agent 系统之后，会对这三件事格外敏感。它们往往比模型更强更关键。
PART 04｜工具调用，把 OpenAI 风格 tool_calls 适配到自己的消息结构
nanobot 这里做了一个很典型的适配。把 provider 返回的 tool_calls 转成 OpenAI/ChatCompletions 兼容的结构，再塞进 messages。
tool_call_dicts = [{"id": tc.id,"type": "function","function": {"name": tc.name,"arguments": json.dumps(tc.arguments)}}for tc in response.tool_calls]messages = self.context.add_assistant_message(messages, response.content, tool_call_dicts,reasoning_content=response.reasoning_content,)
这个动作看起来像胶水，但它决定了一个系统能否长出来。
-
provider 可以换（LiteLLM、OpenAI Codex、未来更多）
-
工具调用结构保持一致
-
session 的持久化也能保留 tool metadata（见 Session.get_history()）
这就是所谓 provider-agnostic（供应商无关）。
PART 05｜进度流，让用户看到系统在做事
nanobot 在工具调用阶段，会把模型的中间输出（或者工具提示）发回去，这就是 progress streaming。
if on_progress:clean = self._strip_think(response.content)await on_progress(clean or self._tool_hint(response.tool_calls))
这里还有一个细节我很喜欢。它会把某些模型吐出来的 <think>…</think> 先剥掉。
@staticmethoddef _strip_think(text: str | None) -> str | None:"""Remove <think>…</think> blocks that some models embed in content."""if not text:return Nonereturn re.sub(r"<think>[\s\S]*?</think>", "", text).strip() or None
这更像产品化的处理。用户不需要看到你的思考过程，用户只需要知道你在推进。
PART 06｜默认工具，工具越有边界越好
AgentLoop 的工具注册在 _register_default_tools() 里完成。它默认给了几类基础能力，文件、shell、web、message、spawn、cron。
def _register_default_tools(self) -> None:"""Register the default set of tools."""# File tools (restrict to workspace if configured)allowed_dir = self.workspace if self.restrict_to_workspace else Noneself.tools.register(ReadFileTool(allowed_dir=allowed_dir))self.tools.register(WriteFileTool(allowed_dir=allowed_dir))self.tools.register(EditFileTool(allowed_dir=allowed_dir))self.tools.register(ListDirTool(allowed_dir=allowed_dir))# Shell toolself.tools.register(ExecTool(working_dir=str(self.workspace),timeout=self.exec_config.timeout,restrict_to_workspace=self.restrict_to_workspace,))# Web toolsself.tools.register(WebSearchTool(api_key=self.brave_api_key))self.tools.register(WebFetchTool())# Message toolmessage_tool = MessageTool(send_callback=self.bus.publish_outbound)self.tools.register(message_tool)# Spawn tool (for subagents)spawn_tool = SpawnTool(manager=self.subagents)self.tools.register(spawn_tool)# Cron tool (for scheduling)if self.cron_service:self.tools.register(CronTool(self.cron_service))
我在这里看到 nanobot 的另一个生产味道，restrict_to_workspace。它承认一件事，工具能力必须被限制在一条安全边界里。
你可以把它理解成一个最低配的 sandbox。
PART 07｜如果你要把这段 loop 用在自己的产品里，优先改什么
我给一个很实操的清单（都是从 loop.py 的写法反推出来的）。
-
把 iteration 上限变成可计费预算，不仅限制次数，还限制 token、工具调用成本、外部 API 次数
-
把工具执行改成可取消，用户撤回、超时、系统 shutdown，都应该能 cancel 掉正在跑的工具
-
把每个回合做成可观测（observable），埋点每次 provider.chat、每次 tool.execute、每次 memory consolidation
-
把进度流做成可控协议，用结构化事件（start/step/log/end），别让它想到哪说到哪
nanobot 的实现是一个很好的起点。它已经把边界做出来了，你要做的是把边界“做硬”。
PART 08｜本文小结
很多人做 agent，会把注意力放在 prompt engineering 上。我越来越觉得 prompt 只是表象，真正决定系统能否长期跑的，是 loop 的纪律。
nanobot 把纪律写进了三行代码，迭代上限、工具 schema、session 持久化。
下一篇我介绍 nanobot 如何处理长期记忆，以及集成 Skills。这两块如果处理不好，agent 不是忘事，就是跑偏；如果处理得好，它会越来越像一个能长期协作的同事。
