openclaw平替之nanobot源码解析（二）：agent命令、消息总线与循环引擎

在上一篇中，我们完成了 nanobot 的“启动仪式”。现在，你的 Agent 已经可以跑起来了。但你可能会好奇：当你输入 nanobot agent 并发送一条消息时，代码底层到底发生了什么？为什么它既能支持终端对话，又能轻松扩展到 Telegram 或飞书？今天，我们就通过 nanobot agent 命令来拆解 nanobot 的心脏——MessageBus（消息总线），以及驱动对话的核心引擎 AgentLoop。
1. agent 命令：通往智能体的入口

agent 命令的执行逻辑位于 nanobot/cli/commands.py 中的 agent 方法。它是用户与 Agent 交互的最直接方式。
参数详解：通过 Typer 框架，nanobot 为我们提供了丰富的参数：
参数缩写类型默认值说明--message-mstrNone发送给 Agent 的消息。提供此参数将进入 单次消息模式。--session-sstrcli:direct会话 ID。用于区分不同的对话上下文。--workspace-wstrNone工作区目录。指定 Agent 运行时的根目录。--config-cstrNone配置文件路径。默认为 ~/.nanobot/config.json。--markdownboolTrue是否以 Markdown 格式渲染 Agent 的输出。--logsboolFalse是否在聊天过程中显示运行时日志（调试神器）。运行模式：单次任务 vs 持续交互
agent 命令根据是否提供 m / -message 参数，会自动切换两种运行模式：

• 单次消息模式 (Single-Message Mode)：当你运行 nanobot agent -m "你好" 时，nanobot 会直接处理这条消息，打印回复后立即退出。这种模式非常适合在脚本中使用，或者进行简单的单次查询。
  
• 交互模式 (Interactive Mode)：如果你只运行 nanobot agent 而不带 -m 参数，nanobot 会进入一个持续的交互式会话。你可以像在聊天软件中一样，连续发送多条消息，Agent 会保留之前的对话上下文。在交互模式下，你可以输入 exit、quit 或按 Ctrl+C 来退出。
  

2. 核心架构：极简消息总线（MessageBus）

nanobot 之所以能支持多渠道，核心在于它实现了一个极简的 生产者-消费者模型。这个模型将“消息来源”（Telegram、终端等）与“消息处理”（Agent 核心）完全解耦。
源码分析：nanobot/bus/queue.py
你会惊讶于它的实现竟然如此简单（不到 50 行代码）：

1. class MessageBus:
2.         def __init__(self):
3.                 self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue()
4.                 self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue()
6.         async def publish_inbound(self, msg: InboundMessage):
7.                 await self.inbound.put(msg)
9.         async def consume_inbound(self) -> InboundMessage:
10.                 return await self.inbound.get()
12.         # ... outbound 同理

复制代码

inbound 队列：所有渠道（CLI、Telegram 等）将用户输入封装成 InboundMessage 丢进这里。
outbound 队列：Agent 处理完后，将回复封装成 OutboundMessage 丢进这里，由对应的渠道负责取走并发送给用户。
这种设计的精妙之处在于：
Agent 根本不需要知道消息是从哪来的。它只管从 inbound 取消息，处理完丢进 outbound。这为后续接入新渠道提供了极大的灵活性。
3. 内置工具集：Agent 的“瑞士军刀”

相关代码位于 nanobot/agent/tools 目录下。nanobot 默认提供了一套精简而强大的内置工具，让 Agent 能够与现实世界进行交互。这些工具被划分为三大类：

• 文件系统 (Filesystem)：赋予 Agent 读写代码和文档的能力。
  
  
  
  • read_file：读取指定文件的内容。
    
  • write_file：创建新文件或覆盖现有文件。
    
  • edit_file：通过精确的行匹配和替换逻辑，安全地修改文件内容。
    
  • list_dir：列出目录下的文件和子目录，帮助 Agent 探索项目结构。
    
  
  
• 网页能力 (Web)：让 Agent 能够获取实时信息。
  
  
  
  • web_search：基于 Brave Search 引擎进行联网搜索。
    
  • web_fetch：抓取指定 URL 的网页内容并转换为 Markdown 格式，方便 LLM 阅读。
    
  
  
• 系统与编排 (System & Orchestration)：这是 nanobot 实现复杂任务的核心。
  
  
  
  • exec：在 shell 中执行任意命令（如运行测试、安装依赖）。
    
  • spawn：创建一个子 Agent（Subagent）来异步处理耗时任务，实现“多智能体协作”。
    
  • message：发送主动消息，用于向用户汇报进度或在多渠道间传递信息。
    
  • cron：管理定时任务，让 Agent 具备“时间观念”，能够按计划执行操作。
    
  
  

4. 智能体循环：AgentLoop 的“五步走”

如果说 MessageBus 是血管，那么 AgentLoop 就是心脏。它负责驱动整个对话的生命周期。在 nanobot/agent/loop.py 中，你可以看到它的核心逻辑被拆解为五个关键步骤：
1）第一步：接收 (Receive)
Agent 持续监听 MessageBus.inbound 队列。一旦有新消息进入（无论是来自终端、Telegram 还是系统任务），Agent 就会将其取出并启动处理流程。
2）第二步：构建上下文 (Context)
这是 AI 思考的基础。ContextBuilder 会将以下信息“缝合”在一起：

• 对话历史：最近的聊天记录。
  
• 长期记忆：从 MEMORY.md 中提取的事实。
  
• 当前任务：用户刚发送的消息。
  
• 技能索引：可用工具的简要说明。
  

3）第三步：调用模型 (LLM Call)
将构建好的上下文发送给大模型（如 Gemini 或 GPT-4）。此时，Agent 会告诉模型：“这是目前的状况，请决定下一步该做什么：是直接回答用户，还是调用某个工具？”
4）第四步：执行工具 (Tool Execution)
如果模型决定调用工具（如 read_file 或 web_search），AgentLoop 会代为执行。
循环往复：工具执行的结果会再次作为“新信息”添加回上下文中，然后回到第三步让模型继续思考。这个过程会一直持续，直到模型认为信息足够，给出最终答复。
5）第五步：发送回复 (Respond)
当模型给出最终答案后，AgentLoop 将其封装成 OutboundMessage 发回 MessageBus.outbound。随后，对应的渠道（如 Telegram 插件）会将其推送给用户。
实战演练：Agent 是如何思考的？
为了让你更直观地理解这个循环，我们来看一个真实的 Debug 案例。当用户问：“为什么这个项目可以直接使用 nanobot 命令？”时，Agent 经历了三次循环：
1）第一次循环：探索环境
输入：

• messages：两条 messages，其中系统提示词会读取工作空间下的 USER.md、SOUL.md 等文件内容，拼接成一个大的提示词。篇幅有限，这里就不把所有的内容都贴出来了。
  
  
  
  
• tools：符合 OpenAI Function Call 规范的工具定义（JSON Schema）
  

1. [
2.         {
3.                 "function": {
4.                         "description": "Read the contents of a file at the given path.",
5.                         "name": "read_file",
6.                         "parameters": {
7.                                 "properties": {
8.                                         "path": {
9.                                                 "description": "The file path to read",
10.                                                 "type": "string"
11.                                         }
12.                                 },
13.                                 "required": ["path"],
14.                                 "type": "object"
15.                         }
16.                 },
17.                 "type": "function"
18.         }
19. ]

复制代码

调用大模型，提供 messages、tools、temperature、max_tokens 等参数：

1. response = await self.provider.chat(
2.         messages=messages,
3.         tools=self.tools.get_definitions(),
4.         model=self.model,
5.         temperature=self.temperature,
6.         max_tokens=self.max_tokens,
7.         reasoning_effort=self.reasoning_effort,
8. )

复制代码

模型思考：模型意识到需要先了解项目结构，于是决定调用工具。
模型回复：发出 list_dir 指令。

1. LLMResponse(
2.         content=None,
3.         tool_calls=[
4.                 ToolCallRequest(
5.                         id="PyPXnKIhI",
6.                         name="list_dir",
7.                         arguments={"path": "/Users/chaoxu.ren/PycharmProjects/nanobot"},
8.                 )
9.         ],
10.         finish_reason="tool_calls",
11.         usage={"prompt_tokens": 3388, "completion_tokens": 357, "total_tokens": 3745},
12.         reasoning_content=None,
13.         thinking_blocks=None,
14. )

复制代码

执行：Agent 执行 list_dir，并将结果（文件列表）追加至上下文。现在上下文中有 4 条消息。

2）第二次循环：定位关键文件
输入：之前的 4 条 messages、tools JSON schema、temperature 等参数。
模型思考：模型在列表中看到了 pyproject.toml，判断这是关键配置文件。
模型回复：发出 read_file 指令。

1. LLMResponse(
2.         content=None,
3.         tool_calls=[
4.                 ToolCallRequest(
5.                         id="undktOZkf",
6.                         name="read_file",
7.                         arguments={"path": "/Users/chaoxu.ren/PycharmProjects/nanobot/pyproject.toml"},
8.                 )
9.         ],
10.         finish_reason="tool_calls",
11.         usage={"prompt_tokens": 3565, "completion_tokens": 103, "total_tokens": 3668},
12.         reasoning_content=None,
13.         thinking_blocks=None,
14. )

复制代码

执行：Agent 读取文件内容，并将 pyproject.toml 内容追加至 messages 中。现在上下文中有 6 条消息。

3）第三次循环：给出最终答案
输入：之前的 6 条 messages、tools JSON schema、temperature 等参数。
模型思考：模型在文件中找到了 [project.scripts] 配置，逻辑闭环，可以回答了。
模型回复：直接给出最终的文字解释，不再调用工具。

1. LLMResponse(
2.         content="在这个项目中可以直接使用 `nanobot` 命令，是因为在项目的 `pyproject.toml` 配置文件中定义了命令行脚本入口（Entry Points）。\n\n具体来说，在 `pyproject.toml` 文件中有一段这样的配置：\n\n```toml\n[project.scripts]\nnanobot = "nanobot.cli.commands:app"\n```\n\n它的工作原理是：\n1. **`[project.scripts]`** 是 Python 打包标准（PEP 621）中用来定义全局命令行工具的部分。\n2. **`nanobot = ...`** 表示要在系统或虚拟环境中注册一个名为 `nanobot` 的可执行命令。\n3. **`nanobot.cli.commands:app`** 指定了当你在终端输入 `nanobot` 时，Python 应该去执行 `nanobot/cli/commands.py` 文件中的 `app` 对象（通常是一个 Typer 或 Click 的应用实例）。\n\n当你使用 `pip install -e .` 或者 `uv pip install -e .` 在当前目录安装这个项目时，包管理器会自动在你的 Python 环境的 bin 目录（或 Windows 的 Scripts 目录）下生成一个名为 `nanobot` 的可执行文件，从而让你能够在终端里直接调用它。",
3.         tool_calls=[],
4.         finish_reason="stop",
5.         usage={"prompt_tokens": 4788, "completion_tokens": 397, "total_tokens": 5185},
6.         reasoning_content=None,
7.         thinking_blocks=None,
8. )

复制代码

结束：Agent 将最终回复发送给用户，循环结束。同时将大模型回复添加到上下文中，以便后续轮次的问答。现在上下文中有 7 条消息。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！