0x00 概要OpenClaw 应该有40万行代码阅读理解起来难度过大因此本系列通过Nanobot来学习 OpenClaw 的特色。Nanobot是由香港大学数据科学实验室(HKUDS)开源的超轻量级个人 AI 助手框架定位为Ultra-Lightweight OpenClaw。非常适合学习Agent架构。Agent 是“业务执行者”解决“消息怎样变成模型调用、工具执行和最终回复”。它们具备独立的上下文与主对话隔离和可使用的特定工具并且具备定义明确的角色和方法论。每次 Agent 收到消息时运行的核心推理周期如下从总线接收消息组装上下文推理该做什么这是 LLM 调用根据决定行动调用工具、执行命令观察结果保存状态判断我完成了吗还是再循环一次完成后回复注本系列借鉴的文章过多可能在参考文献中有遗漏的文章如果有还请大家指出。0x01 原理1.1 Agent负责“执行”一个 Agent 一个完整的 AI “大脑实例”每个 Agent 都拥有独立资源。Agent 是“执行平面”解决“消息怎样变成模型调用、工具执行和最终回复”。具体如下图来自MiniClaw。Feishu Cloud | | HTTP POST /feishu/events | (im.message.receive_v1) v [ESP32 Webhook Server :18790] | | message_bus_push_inbound() v [Message Bus] ── [Agent Loop] ── [Message Bus] (Claude/GPT) | | outbound dispatch v [feishu_send_message()] | | POST /im/v1/messages v Feishu API下图是Agent 的最小循环。每个 AI Agent 都需要这个循环。模型决定何时调用工具、何时停止。代码只是执行模型的要求。An agentic loop is the full “real” run of an agent: intake → context assembly → model inference → tool execution → streaming replies → persistence. THE AGENT PATTERN User -- messages[] -- LLM -- response | stop_reason tool_use? / \ yes no | | execute tools return text append results loop back ----------------- messages[]1.2 Pi-Agent框架OpenClaw所使用的引擎是Pi-Agent框架它是一个仅有四个工具、系统提示词不到1000个token秉持“精简至上”原则的AI编程Agent。与其他编程Agent相比Pi的工程设计和决策机制极为简洁形成了鲜明对比。下图是 OpenClaw 的循环概要。runEmbeddedPiAgent() └── while (true) { // 主重试循环 ├── 检查重试次数限制 (MAX_RUN_LOOP_ITERATIONS) ├── 调用 runEmbeddedAttempt() // 单次推理尝试 ├── 处理 context overflow → 自动压缩 ├── 处理 auth failure → profile轮换 ├── 处理 timeout → 重试或报错 └── 成功则返回 payloads }Pi的设计理念可以总结为不是为LLM打造一个复杂的“控制台”而是给它一把“多功能小刀”——工具虽少但实用提示虽简但明确让模型的原生能力成为主导而不是被框架的复杂性所掩盖。Pi 这种设计理念是基于一个关键事实——经过强化学习训练的前沿LLM模型已经具备了很强的理解和执行能力。它们能明确知道“编码Agent”的主要任务是什么根本不需要长篇大论的系统提示词和复杂的辅助模块来“指导”它们工作。从数据层面分析Pi的系统提示词加上工具定义总长度还不到1000个token仅仅是Claude Code的十分之一内置工具也只有4个远少于同类产品。这说明Pi在主流Agent都在强化的方面几乎都做了简化系统提示词简短明了内置工具数量精简没有复杂的规划模式和多代理通信协议Plan Mode和MCP支持更没有难以监控的子AgentPi的核心策略是去除冗余辅助模块让LLM模型发挥核心作用用最简洁的结构实现最核心的功能。或许有人会问如此简单的设计真的能应对复杂的编码任务吗实际上Pi的简洁并非“简陋”而是“精准”。接下来我们详细解析这4个内置工具的设计思路——read、write、edit、bash工具主要功能read读取文件、审查代码、获取上下文信息write创建文件、写入内容edit修改代码、进行增量更新bash执行命令、操作环境、通过自我调用来拆分任务这四个工具几乎涵盖了编码Agent的所有核心需求。特别是bash工具的引入既实现了复杂任务的拆分和执行保证了功能的完整性又避免了引入子Agent可能带来的不可预测性和监控难题——这就是Pi敢于放弃子Agent架构的原因。同时Pi使用简短的系统提示词并非降低了对LLM的引导标准而是充分信任前沿LLM的能力。正如Mario Zechner所倡导的与其用大量token去“教导”LLM如何成为Agent不如用简洁的提示词明确其核心任务让LLM充分发挥自身的理解和执行能力。这种设计思路带来了三大好处节省上下文空间——降低推理成本提高运行效率行为更加灵活自主——LLM能根据实际情况动态调整策略不受冗长规则限制更好的适应性——简洁的结构意味着更低的认知负担和更强的泛化能力0x02 AgentLoopAgentLoop 是nanobot Agent运行的核心。智能体循环是区分聊天机器人和智能体的关键。2.1 架构AgentLoop 类的架构如下2.2 流程下面是一个 AI Agent智能体的消息处理流程图展示了从消息接收到响应发送的完整链路包括 LLM 交互、工具调用循环等核心机制。入口消息到达InboundMessage ↓ AgentLoop.run() - 监听并接收消息 ↓ AgentLoop._dispatch() - 分派处理 ↓ AgentLoop._process_message() - 主要处理逻辑 ↓ ContextBuilder.build_messages() - 构建上下文 ↓ AgentLoop._run_agent_loop() - 核心代理循环 ↓ Provider.chat() - LLM交互 ↓ ← 判断是否有工具调用 ↓ 否 ← 返回最终内容 ↓ 是 ← 执行工具调用 ↓ ContextBuilder.add_tool_result() - 添加工具结果 ↓ ← 继续循环直到没有更多工具调用 ↓ AgentLoop._save_turn() - 保存交互记录 ↓ 通过MessageBus发布OutboundMessage - 发送响应部分环节详细拆解如下2.3 定义和初始化AgentLoop 的定义和初始化代码如下class AgentLoop: The agent loop is the core processing engine. It: 1. Receives messages from the bus 2. Builds context with history, memory, skills 3. Calls the LLM 4. Executes tool calls 5. Sends responses back def __init__( self, bus: MessageBus, # 消息总线用于接收/发送消息 provider: LLMProvider, # LLM提供者如OpenAI/本地模型 workspace: Path, # Agent工作目录用于隔离文件操作 model: str | None None, # 使用的LLM模型名称 max_iterations: int 40, # Agent最大迭代次数防止无限循环 temperature: float 0.1, # LLM温度参数越低越确定 max_tokens: int 4096, # LLM最大生成Token数 memory_window: int 100, # 记忆窗口大小会话历史最大条数 brave_api_key: str | None None, # Brave搜索API密钥用于网页搜索工具 exec_config: ExecToolConfig | None None, # 命令执行工具配置 cron_service: CronService | None None, # 定时任务服务可选 restrict_to_workspace: bool False, # 是否限制Agent仅操作工作区 session_manager: SessionManager | None None, # 会话管理器可选 mcp_servers: dict | None None, # MCP服务器配置可选 channels_config: ChannelsConfig | None None, # 通道配置可选 ): # 解决循环导入问题仅运行时导入ExecToolConfig from nanobot.config.schema import ExecToolConfig # 基础属性初始化 self.bus bus # 消息总线实例 self.channels_config channels_config # 通道配置 self.provider provider # LLM提供者实例 self.workspace workspace # 工作目录路径 # 模型名称优先传入值否则使用LLM提供者默认模型 self.model model or provider.get_default_model() self.max_iterations max_iterations # 最大迭代次数 self.temperature temperature # LLM温度 self.max_tokens max_tokens # LLM最大Token数 self.memory_window memory_window # 记忆窗口大小 self.brave_api_key brave_api_key # Brave API密钥 # 执行工具配置默认空配置 self.exec_config exec_config or ExecToolConfig() self.cron_service cron_service # 定时任务服务 self.restrict_to_workspace restrict_to_workspace # 工作区限制开关 # 核心组件初始化 self.context ContextBuilder(workspace) # 上下文构建器构建LLM输入上下文 # 会话管理器优先传入实例否则创建新实例 self.sessions session_manager or SessionManager(workspace) self.tools ToolRegistry() # 工具注册表管理所有可用工具 # 子Agent管理器用于生成子Agent处理子任务 self.subagents SubagentManager( providerprovider, workspaceworkspace, busbus, modelself.model, temperatureself.temperature, max_tokensself.max_tokens, brave_api_keybrave_api_key, exec_configself.exec_config, restrict_to_workspacerestrict_to_workspace, ) # 运行状态与资源管理属性 self._running False # Agent循环是否运行 self._mcp_servers mcp_servers or {} # MCP服务器配置 self._mcp_stack: AsyncExitStack | None None # MCP连接上下文栈 self._mcp_connected False # MCP是否已连接 self._mcp_connecting False # MCP是否正在连接 self._consolidating: set[str] set() # 正在进行记忆合并的会话Key集合 self._consolidation_tasks: set[asyncio.Task] set() # 记忆合并任务集合 self._consolidation_locks: dict[str, asyncio.Lock] {} # 会话记忆合并锁 self._active_tasks: dict[str, list[asyncio.Task]] {} # 活跃任务session_key - 任务列表 self._processing_lock asyncio.Lock() # 全局消息处理锁防止并发冲突 self._register_default_tools() # 注册默认工具 def _register_default_tools(self) - None: Register the default set of tools. 注册默认工具集 # 确定文件工具的允许目录如果限制工作区则为工作目录否则为None无限制 allowed_dir self.workspace if self.restrict_to_workspace else None # 注册文件系统工具读/写/编辑/列目录 for cls in (ReadFileTool, WriteFileTool, EditFileTool, ListDirTool): self.tools.register(cls(workspaceself.workspace, allowed_dirallowed_dir)) # 注册命令执行工具 self.tools.register(ExecTool( working_dirstr(self.workspace), # 工作目录 timeoutself.exec_config.timeout, # 执行超时时间 restrict_to_workspaceself.restrict_to_workspace, # 工作区限制 path_appendself.exec_config.path_append, # 环境变量PATH追加 )) # 注册网页相关工具搜索/爬取 self.tools.register(WebSearchTool(api_keyself.brave_api_key)) self.tools.register(WebFetchTool()) # 注册消息发送工具回调函数为消息总线发布出站消息 self.tools.register(MessageTool(send_callbackself.bus.publish_outbound)) # 注册子Agent生成工具 self.tools.register(SpawnTool(managerself.subagents)) # 如果有定时任务服务注册定时任务工具 if self.cron_service: self.tools.register(CronTool(self.cron_service)) async def _connect_mcp(self) - None: Connect to configured MCP servers (one-time, lazy). 连接MCP服务器懒加载仅一次 # 跳过条件已连接/正在连接/无MCP配置 if self._mcp_connected or self._mcp_connecting or not self._mcp_servers: return self._mcp_connecting True # 标记为正在连接 from nanobot.agent.tools.mcp import connect_mcp_servers # 延迟导入MCP连接函数 try: # 创建异步上下文栈用于管理MCP连接资源 self._mcp_stack AsyncExitStack() await self._mcp_stack.__aenter__() # 进入上下文栈 # 连接MCP服务器将工具注册到MCP await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack) self._mcp_connected True # 标记为已连接 except Exception as e: # 连接失败记录日志下次消息处理时重试 logger.error(Failed to connect MCP servers (will retry next message): {}, e) if self._mcp_stack: try: await self._mcp_stack.aclose() # 关闭上下文栈 except Exception: pass self._mcp_stack None finally: self._mcp_connecting False # 清除正在连接标记 def _set_tool_context(self, channel: str, chat_id: str, message_id: str | None None) - None: Update context for all tools that need routing info. 更新需要路由信息的工具上下文 # 消息工具设置通道/聊天ID/消息ID用于消息发送路由 if message_tool : self.tools.get(message): if isinstance(message_tool, MessageTool): message_tool.set_context(channel, chat_id, message_id) # 子Agent生成工具设置通道/聊天ID if spawn_tool : self.tools.get(spawn): if isinstance(spawn_tool, SpawnTool): spawn_tool.set_context(channel, chat_id) # 定时任务工具设置通道/聊天ID if cron_tool : self.tools.get(cron): if isinstance(cron_tool, CronTool):
