目录
- 1 Agent Loop 核心逻辑分析:为什么 Agent 能自动调用 Tool 和 LLM?
1 Agent Loop 核心逻辑分析:为什么 Agent 能自动调用 Tool 和 LLM?
分析项目:learn-claude-code(Python)、nanobot(Python)、nullclaw(Zig)、openclaw(TypeScript)
1.1 一句话总结
Agent 的"自动"本质上是一个while循环:不断调用 LLM → 检查 LLM 是否要求调用工具 → 执行工具 → 把结果喂回 LLM → 直到 LLM 说"我说完了"。
1.2 核心原理:LLM 的 Tool Use 协议
所有 4 个项目都依赖同一个关键机制 ——LLM 原生的 Tool Use(Function Calling)协议:
┌─────────────────────────────────────────────────────────────────┐ │ 调用 LLM 时,同时传入: │ │ 1. messages(对话历史) │ │ 2. tools(工具定义列表:名称 + 描述 + 参数 JSON Schema) │ │ │ │ LLM 返回时,有两种可能: │ │ A. stop_reason = "end_turn" → 纯文本回复,循环结束 │ │ B. stop_reason = "tool_use" → 返回要调用的工具名 + 参数 │ │ → Agent 执行工具 → 把结果追加到 messages → 再次调用 LLM │ └─────────────────────────────────────────────────────────────────┘关键洞察:Agent 并不"理解"工具,它只是把工具的 JSON Schema 描述传给 LLM,由 LLM 决定是否调用、调用哪个、传什么参数。Agent 只负责"执行"和"回传结果"。
1.3 四个项目的核心循环对比
1.3.1 learn-claude-code(Python,最简实现,~30 行)
📁 核心文件:
learn-claude-code/agents/s01_agent_loop.py
defagent_loop(messages:list):whileTrue:# ① 调用 LLM(带上工具定义)response=client.messages.create(model=MODEL,system=SYSTEM,messages=messages,tools=TOOLS,max_tokens=8000,)# ② 把 LLM 的回复追加到消息历史messages.append({"role":"assistant","content":response.content})# ③ 判断:LLM 是否要求调用工具?ifresponse.stop_reason!="tool_use":return# 不需要 → 循环结束# ④ 执行工具,收集结果results=[]forblockinresponse.content:ifblock.type=="tool_use":output=run_bash(block.input["command"])results.append({"type":"tool_result","tool_use_id":block.id,"content":output,})# ⑤ 把工具结果作为 "user" 消息追加(LLM 协议要求)messages.append({"role":"user","content":results})# → 回到 ① 继续循环工具注册(s02_tool_use.py):
# 工具定义:告诉 LLM 有哪些工具可用TOOLS=[{"name":"bash","description":"Run a shell command.","input_schema":{"type":"object","properties":{"command":{"type":"string"}},"required":["command"]}},{"name":"read_file",...},{"name":"write_file",...},{"name":"edit_file",...},]# 工具分发:名称 → 处理函数TOOL_HANDLERS={"bash":lambda**kw:run_bash(kw["command"]),"read_file":lambda**kw:run_read(kw["path"],kw.get("limit")),"write_file":lambda**kw:run_write(kw["path"],kw["content"]),"edit_file":lambda**kw:run_edit(kw["path"],kw["old_text"],kw["new_text"]),}1.3.2 nanobot(Python,生产级实现)
📁 核心文件:
nanobot/nanobot/agent/loop.py
asyncdef_process_message(self,msg:InboundMessage)->OutboundMessage|None:# 构建消息上下文messages=self.context.build_messages(history=session.get_history(),current_message=msg.content,)iteration=0final_content=None# ① 主循环(最多 max_iterations 次,默认 20)whileiteration<self.max_iterations:iteration+=1# ② 调用 LLM(传入工具定义 + tool_choice="auto")response=awaitself.provider.chat(messages=messages,tools=self.tools.get_definitions(),# 所有注册工具的 JSON Schemamodel=self.model,)# ③ 判断:LLM 是否要求调用工具?ifresponse.has_tool_calls:# 添加 assistant 消息(包含 tool_calls)messages=self.context.add_assistant_message(messages,response.content,tool_call_dicts,)# ④ 执行每个工具调用fortool_callinresponse.tool_calls:result=awaitself.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=response.contentbreakreturnOutboundMessage(content=final_content)工具注册机制(Registry 模式):
# 工具抽象基类classTool(ABC):@property@abstractmethoddefname(self)->str:...@property@abstractmethoddefparameters(self)->dict[str,Any]:...@abstractmethodasyncdefexecute(self,**kwargs:Any)->str:...defto_schema(self)->dict[str,Any]:return{"type":"function","function":{"name":self.name,"description":self.description,"parameters":self.parameters,}}# 工具注册表classToolRegistry:def__init__(self):self._tools:dict[str,Tool]={}defregister(self,tool:Tool)->None:self._tools[tool.name]=tooldefget_definitions(self)->list[dict]:return[tool.to_schema()fortoolinself._tools.values()]asyncdefexecute(self,name:str,params:dict)->str:tool=self._tools.get(name)returnawaittool.execute(**params)LLM 调用(LiteLLM 统一接口):
# 关键:传入 tools 和 tool_choice="auto"kwargs={"model":model,"messages":messages,"tools":tools,# ← 工具定义列表"tool_choice":"auto",# ← 让 LLM 自动决定是否调用工具}response=awaitacompletion(**kwargs)1.3.3 nullclaw(Zig,高性能实现)
📁 核心文件:
nullclaw/src/agent/root.zig
pub fn turn(self: *Agent, user_message: []const u8) ![]const u8 { var iteration: u32 = 0; // ① 主循环(最多 max_tool_iterations 次,默认 25) while (iteration < self.max_tool_iterations) : (iteration += 1) { // 中断检查 if (self.isInterruptRequested()) return self.interruptedReply(); // 构建消息 const messages = try self.buildProviderMessages(arena); // ② 调用 LLM(传入工具定义) var response = self.provider.chat( self.allocator, .{ .messages = messages, .model = self.model_name, .tools = if (native_tools_enabled) turn_tool_specs else null, // ... }, self.model_name, self.temperature, ) catch |err| { /* 错误处理和重试 */ }; // ③ 解析工具调用(支持两种格式) var parsed_calls: []ParsedToolCall = &.{}; if (response.hasToolCalls()) { // 优先:原生 JSON 格式(OpenAI 兼容) parsed_calls = try dispatcher.parseStructuredToolCalls( self.allocator, response.tool_calls, ); } else { // 回退:XML 标签格式 <invoke>...</invoke> const xml_parsed = try dispatcher.parseToolCalls( self.allocator, response_text, ); parsed_calls = xml_parsed.calls; } // ④ 判断:没有工具调用 → 循环终止 if (parsed_calls.len == 0) { return final_text; // 返回最终结果 } // ⑤ 执行工具调用 for (parsed_calls) |call| { const result = self.executeTool(arena, call); try results_buf.append(self.allocator, result); } // ⑥ 将工具结果格式化并追加到历史记录 const formatted_results = try dispatcher.formatToolResults(arena, results_buf.items); try self.history.append(self.allocator, .{ .role = .user, .content = formatted_results, // 工具结果作为 user 消息 }); // → 回到 ① 继续循环 } }工具接口(Zig 的 VTable 多态):
pub const Tool = struct { ptr: *anyopaque, vtable: *const VTable, pub const VTable = struct { execute: *const fn (ptr: *anyopaque, allocator: Allocator, args: JsonObjectMap) anyerror!ToolResult, name: *const fn (ptr: *anyopaque) []const u8, description: *const fn (ptr: *anyopaque) []const u8, parameters_json: *const fn (ptr: *anyopaque) []const u8, }; };工具执行(遍历查找 + 调用):
fn executeTool(self: *Agent, allocator: Allocator, call: ParsedToolCall) ToolExecutionResult { for (self.tools) |t| { if (std.ascii.eqlIgnoreCase(t.name(), call.name)) { const args = std.json.parseFromSlice(..., call.arguments_json, ...); const result = t.execute(allocator, args); return result; } } return .{ .output = "Unknown tool", .success = false }; }1.3.4 openclaw(TypeScript,SDK 封装实现)
📁 核心文件:
openclaw/src/agents/pi-embedded-runner/run/attempt.ts
OpenClaw 将循环逻辑封装在Pi Agent SDK中,上层代码只需配置:
// ① 创建 Agent Session(注册工具 + 配置 LLM)const{session}=awaitcreateAgentSession({tools:builtInTools,// 内置工具customTools:allCustomTools,// 自定义工具model:params.model,// ...});// ② 设置 LLM 调用函数activeSession.agent.streamFn=streamSimple;// Anthropic API// ③ 发送 Prompt → SDK 内部自动执行 Agent Loop// LLM 调用 → 工具检测 → 工具执行 → 结果回传 → 循环工具注册(pi-tools.ts):
exportfunctioncreateOpenClawCodingTools(options){return[createExecTool({...}),// bash 执行createProcessTool({...}),// 进程管理createApplyPatchTool({...}),// 补丁应用createOpenClawReadTool({...}),// 文件读取createSandboxedWriteTool({...}),// 文件写入(沙箱)createSandboxedEditTool({...}),// 文件编辑(沙箱)// ...];}SDK 内部的循环逻辑(伪代码,与 learn-claude-code 一致):
// Pi Agent SDK 内部实现asyncfunctionagentLoop(messages,tools){while(true){constresponse=awaitstreamFn(messages,tools);messages.push({role:"assistant",content:response.content});if(response.stop_reason!=="tool_use")return;// 终止for(constblockofresponse.content){if(block.type==="tool_use"){constoutput=awaitexecuteTool(block.name,block.input);messages.push({role:"user",content:[{type:"tool_result",tool_use_id:block.id,content:output}],});}}}}1.4 统一流程图
┌──────────────────────┐ │ 用户输入消息 │ └──────────┬───────────┘ │ ┌──────────▼───────────┐ │ 构建 messages 数组 │ │ (system + history │ │ + user message) │ └──────────┬───────────┘ │ ┌────────────────▼────────────────┐ │ │ │ ┌──────────────────────────┐ │ │ │ 调用 LLM API │ │ │ │ 传入: messages + tools │ │ │ └────────────┬─────────────┘ │ │ │ │ │ ┌────────────▼─────────────┐ │ │ │ LLM 返回 response │ │ │ └────────────┬─────────────┘ │ │ │ │ │ ┌────────────▼─────────────┐ │ │ │ stop_reason == tool_use?│ │ │ └──┬──────────────────┬────┘ │ │ │ YES │ NO │ │ │ │ │ │ ┌──▼───────────┐ ┌──▼─────┐ │ │ │ 执行工具 │ │ 返回 │ │ │ │ 收集结果 │ │ 最终 │ │ │ └──┬───────────┘ │ 文本 │ │ │ │ └────────┘ │ │ ┌──▼───────────┐ │ │ │ 追加结果到 │ │ │ │ messages │ │ │ └──┬───────────┘ │ │ │ │ │ └───────────┐ │ │ │ 继续循环 │ └──────────────────┘ │ │ ┌─────────────────────────────┘ │ ▼ ┌──────────────────┐ │ 输出最终回复 │ └──────────────────┘1.5 消息格式详解:为什么 LLM 能"自动"调用工具?
秘密在于messages 数组的结构。每一轮循环,messages 都在增长:
[{"role":"system","content":"你是一个编程助手..."},{"role":"user","content":"帮我创建一个 hello.py 文件"},// === 第 1 轮 LLM 返回:要求调用工具 ==={"role":"assistant","content":[{"type":"text","text":"好的,我来创建文件"},{"type":"tool_use","id":"call_1","name":"write_file","input":{"path":"hello.py","content":"print('hello')"}}]},// === Agent 执行工具后,把结果追加 ==={"role":"user","content":[{"type":"tool_result","tool_use_id":"call_1","content":"File written successfully"}]},// === 第 2 轮 LLM 返回:不再调用工具 ==={"role":"assistant","content":[{"type":"text","text":"已经创建好 hello.py 文件了!"}]}// stop_reason = "end_turn" → 循环结束]LLM 之所以"知道"要调用工具,是因为:
- 工具定义通过
tools参数传给了 LLM,LLM 知道有哪些工具可用 - LLM 经过训练,能根据用户意图判断是否需要调用工具
- LLM 返回结构化的工具调用请求(工具名 + 参数),而不是自己执行
- Agent 负责实际执行,然后把结果以特定格式回传
1.6 四个项目对比总结
| 维度 | learn-claude-code | nanobot | nullclaw | openclaw |
|---|---|---|---|---|
| 语言 | Python | Python | Zig | TypeScript |
| 代码量 | ~30 行核心 | ~100 行核心 | ~500 行核心 | SDK 封装 |
| 循环方式 | while True | while iter < max | while iter < max | SDK 内部 |
| 终止判断 | stop_reason != "tool_use" | has_tool_calls == False | parsed_calls.len == 0 | stop_reason != "tool_use" |
| 最大迭代 | 无限制 | 20 次 | 25 次 | 32-160 次 |
| 工具注册 | 字典映射 | Registry 模式 | VTable 多态 | SDK 工厂函数 |
| LLM 调用 | Anthropic SDK | LiteLLM(多家) | 自实现 HTTP | Pi Agent SDK |
| 工具格式 | Anthropic 格式 | OpenAI 格式 | 原生 + XML 回退 | Anthropic 格式 |
| 异步 | 同步 | asyncio | 同步(多线程) | async/await |
| 特色 | 教学用,极简 | 消息总线解耦 | Arena 内存管理 | 沙箱隔离 |
1.7 核心结论
Agent 的全部秘密就是一个 while 循环 + LLM 的 Tool Use 协议。
while LLM 说"我要调用工具": 执行工具 把结果喂回 LLM 返回 LLM 的最终回复生产级 Agent(nanobot、nullclaw、openclaw)在这个核心循环之上,叠加了:
- 安全层:工具权限控制、沙箱隔离、速率限制
- 可靠性层:最大迭代次数、超时控制、错误重试
- 上下文管理:历史压缩、内存检索、会话隔离
- 可扩展性:工具注册表、Provider 抽象、插件机制
但循环本身始终不变。理解了这个循环,就理解了所有 AI Agent 的核心。