1. 项目概述从源码视角理解AgentLoop最近在深入研究OpenClaw这个项目特别是其核心执行引擎Nanobot的源码。很多朋友对AI Agent的开发框架感兴趣但往往停留在调用API的层面对于其内部如何调度、如何循环、如何管理状态知之甚少。这次我们就聚焦在Nanobot源码中一个非常核心的模块——AgentLoop。这个模块简单来说就是驱动整个AI Agent“思考-行动-观察”循环的心脏。它决定了Agent如何接收指令、调用工具、处理大模型响应、更新内部状态并最终完成任务。理解AgentLoop就相当于拿到了打开Agent系统黑盒的钥匙无论是想深度定制自己的Agent还是想借鉴其设计思想构建自己的框架都至关重要。2. AgentLoop的核心架构与设计哲学2.1 什么是AgentLoop在AI Agent的语境下AgentLoop并非一个新鲜概念它本质上是实现“感知-规划-行动”Perception-Planning-Action循环的代码实体。但在Nanobot的实现中它被赋予了更具体、更工程化的内涵。它不是一个简单的while循环而是一个状态机驱动的、可插拔的、具备容错与回溯能力的执行管道。你可以把它想象成一个高度智能的流水线控制器。用户的一个请求比如“帮我查一下北京明天的天气然后推荐是否适合户外跑步”进入这条流水线。AgentLoop负责调度流水线上的各个“工位”先由“理解工位”LLM解析用户意图并生成计划然后“执行工位”Tool调用天气查询API获取结果后再由“评估工位”LLM判断信息是否完整、是否需要进一步行动比如查询空气质量最后“汇报工位”整理答案并返回给用户。AgentLoop确保这个流程有序、可靠地进行并处理中间可能发生的任何异常如API调用失败、模型返回格式错误。2.2 Nanobot中AgentLoop的模块化设计阅读Nanobot的源码你会发现AgentLoop被设计得非常清晰和解耦。它通常不作为一个庞大的单体类存在而是由几个关键组件协同工作状态管理器 (State Manager)这是Loop的核心记忆单元。它维护着当前会话的完整上下文包括用户初始目标、已执行的动作历史、工具调用结果、LLM的思考过程Chain-of-Thought、以及当前的执行状态如PLANNING,EXECUTING,OBSERVING,FINISHED。在Nanobot中这个状态对象往往是不可变的Immutable每次循环迭代都会产生一个新的状态对象这为实现时间旅行调试回溯到之前的某个状态和并发安全提供了便利。动作执行器 (Action Executor)负责具体执行Agent“规划”出的动作。最常见的就是工具调用。执行器需要解析动作参数。找到并实例化对应的工具Tool。安全地执行工具通常会在沙箱或受限环境中。捕获执行结果或异常并将其格式化为标准的观察Observation对象反馈给状态管理器。规划器 (Planner)/决策引擎这是Agent的“大脑”。它基于当前状态决定下一步做什么。在Nanobot中规划器通常就是与大语言模型LLM交互的模块。它将状态信息如目标、历史构造成提示词Prompt发送给LLM并解析LLM的返回将其转化为一个或多个明确的“动作”Action例如ToolCall(‘get_weather’, {‘city’: ‘北京’})或FinalAnswer(‘...’)。观察处理器 (Observation Processor)当动作执行器完成工作后会产生一个原始结果。观察处理器负责对这个结果进行加工比如提取关键信息、判断结果是否成功、是否包含错误信息、是否需要触发重试等。处理后的“观察”才会被正式加入到状态历史中供下一轮规划使用。循环控制器 (Loop Controller)这是驱动整个循环的逻辑。它定义了循环的终止条件如收到最终答案、达到最大迭代次数、超时、用户中断等并按照“规划 - 执行 - 观察 - 更新状态 - 再规划...”的顺序协调上述组件工作。它还需要处理循环控制逻辑比如是否开启“自我反思”ReAct模式中的Think步骤等。注意这种高度模块化的设计带来了极大的灵活性。例如你可以轻松替换默认的基于GPT的规划器换成Claude或本地部署的模型你也可以为特定的工具设计自定义的执行器或观察处理器而不影响其他部分。2.3 设计中的关键考量与取舍在构建一个生产可用的AgentLoop时Nanobot的源码体现出了几个关键的设计权衡同步 vs 异步Agent的动作尤其是调用外部API可能是耗时的。一个健壮的AgentLoop必须支持异步操作以避免阻塞主线程。Nanobot的源码中大量使用了async/await语法确保在等待LLM响应或工具执行时系统资源不会被浪费。状态管理的复杂性状态是Agent的记忆但记忆不是越多越好。无限制地增长上下文会消耗大量Token增加成本并可能降低模型性能。因此AgentLoop中通常集成有“上下文窗口管理”策略比如只保留最近N轮交互或对历史进行智能摘要Summarization。错误处理与鲁棒性这是区分玩具项目和实用系统的关键。AgentLoop必须能妥善处理各种异常LLM返回非结构化内容、工具调用超时或返回错误、网络中断等。常见的策略包括动作重试、规划步骤回退让LLM重新规划、以及优雅降级返回部分结果并说明情况。可观测性 (Observability)一个运行中的Agent在想什么、做了什么对开发者来说必须是透明的。好的AgentLoop会生成结构化的日志和追踪Trace信息方便调试和优化。这在源码中体现为在各个关键节点插入日志记录和指标收集。3. 深入AgentLoop源码核心流程拆解让我们暂时抛开抽象的模块深入到类似Nanobot的AgentLoop核心执行流程中看看代码是如何一步步运转的。以下是一个高度简化但体现了核心逻辑的伪代码流程它可以帮助你建立直观的认识。class AgentLoop: async def run(self, initial_input: str) - str: # 1. 初始化状态 state AgentState(goalinitial_input, history[]) # 2. 循环控制 for step in range(self.max_iterations): # 2.1 检查终止条件 if self._should_stop(state): break # 2.2 规划阶段决定下一步做什么 action await self.planner.plan(state) # 记录规划动作到状态 state state.add_to_history(typeplan, contentaction) # 2.3 执行阶段执行规划出的动作 if action.type tool_call: raw_observation await self.executor.execute(action) # 2.4 处理观察结果 processed_observation await self.observer.process(raw_observation) elif action.type final_answer: return action.content # 循环结束返回最终答案 # 2.5 更新状态将观察结果加入历史形成新状态 state state.add_to_history(typeobservation, contentprocessed_observation) # 可选在这里进行状态摘要或修剪防止上下文过长 state self._maybe_compress_history(state) # 3. 循环结束可能因超限或错误 return self._handle_timeout_or_error(state)这个流程看似简单但每个步骤都隐藏着大量的细节和设计选择。3.1 规划阶段与LLM的深度交互规划是AgentLoop中最具“魔法”的部分。在源码中planner.plan(state)函数内部通常是这样工作的构建提示词将state中的历史对话、工具列表、当前目标等按照预定义的模板组织成一个结构化的提示词。这个模板的质量直接决定了LLM的表现。例如你是一个助手。你的目标{state.goal}。你可以使用的工具{tool_descriptions}。之前的步骤{formatted_history}。请根据以上信息决定下一步是调用工具还是直接回答。如果调用工具请严格按照JSON格式输出。调用LLM并解析将提示词发送给配置好的大模型。这里的关键在于输出格式的约束。为了稳定地解析出结构化的动作ActionNanobot这类框架通常会采用以下一种或多种技术函数调用 (Function Calling)利用OpenAI等原生支持的function calling功能让LLM返回一个标准的函数调用请求。JSON模式 (JSON Mode)在提示词中严格要求LLM以指定JSON格式回复并在调用时开启response_format{ type: json_object }。输出解析器 (Output Parser)使用像Pydantic这样的库定义期望的输出数据结构然后结合LangChain等框架的解析器对LLM的文本输出进行强制的结构化提取和校验。动作生成将LLM的返回解析成一个内部的Action对象。这个对象包含了动作类型工具调用/最终回答和所有必要的参数。实操心得规划阶段的稳定性是Agent可靠性的基石。在实际开发中LLM“胡言乱语”或不按格式输出的情况时有发生。一个健壮的规划器必须有重试和降级机制。例如第一次解析失败后可以将错误信息连同原提示再次发给LLM要求其纠正如果多次失败则降级为让Agent输出“我无法处理这个请求”之类的安全回复。3.2 执行与观察阶段工具的可靠调用当规划器产出一个ToolCall动作后执行器便登场了。工具路由与加载执行器根据工具名称如get_weather从一个注册中心Tool Registry找到对应的工具定义。这个定义包括工具的函数、参数schema、描述等。Nanobot的源码通常会维护一个全局的工具字典。参数验证与安全执行在执行前必须用JSON Schema或Pydantic模型验证传入的参数是否合法。这是防止无效调用和潜在安全风险的重要一步。执行本身可能发生在隔离环境或带有超时、资源限制的包装器中。观察处理工具返回的可能是任何东西——一个字典、一个字符串、甚至一个异常对象。观察处理器的任务是将这些“原材料”转化为对Agent“思考”有用的信息。例如它可能标准化将不同工具的返回格式统一。提取关键信息从一大段HTML或JSON中提取出核心数据。判断成功与否根据HTTP状态码或返回结构标记此次观察是成功、失败还是需要重试。格式化将结果格式化为一段自然语言描述方便LLM在下轮规划中理解。3.3 状态更新与上下文管理每一轮循环结束后新的“动作-观察”对会被添加到状态历史中。但随着对话轮次增加历史会越来越长。直接将整个历史扔给LLM不仅成本高而且可能因超出上下文长度导致模型性能下降或报错。因此_maybe_compress_history(state)这个函数至关重要。常见的策略有滑动窗口只保留最近K轮交互。摘要压缩当历史达到一定长度时调用另一个LLM将早期对话总结成一段简短的摘要然后用“摘要近期详细历史”的方式替代完整历史。这需要在信息丢失和Token节省之间取得平衡。重要性筛选尝试识别并保留与最终目标最相关的历史片段。在Nanobot的源码中这部分逻辑可能被抽象成一个独立的ContextManager或Memory组件由AgentLoop在每轮迭代后调用。4. 高级特性与实战技巧理解了基础循环后我们来看看一个成熟的AgentLoop如Nanobot所实现的通常还具备哪些高级特性以及在实际使用中的技巧。4.1 支持复杂工作流多Agent与子任务简单的单循环Agent能处理的任务有限。复杂的任务需要分解和协作。AgentLoop可以升级为支持层次化或协同式的工作流。子任务分解主Agent的规划器在遇到复杂目标时可以生成一个“创建子任务”的动作。AgentLoop会为此实例化一个新的、拥有独立状态的子Agent循环。子循环执行完毕后将结果作为“观察”返回给主循环。这在源码中可能体现为AgentLoop类本身可以被递归或嵌套调用。多Agent协同多个拥有不同技能工具集的Agent同时运行并通过一个共享的“黑板”Blackboard或消息队列进行通信。一个中央协调器Orchestrator或另一个负责规划的Agent来管理它们之间的交互。此时的AgentLoop可能演变为一个更复杂的、事件驱动的架构。4.2 提升可靠性的关键验证、重试与回滚一个在生产环境运行的Agent绝不能是脆弱的。AgentLoop必须内置强大的可靠性机制。动作验证在执行工具前除了参数格式校验还可以进行语义校验。例如调用“预订餐厅”工具前先检查“时间”参数是否在未来。自动重试对于网络超时、速率限制等暂时性错误执行器应自动进行指数退避重试。重试逻辑需要仔细设计避免无限循环。规划回滚如果LLM连续几步都走进了死胡同比如反复调用一个不存在的工具AgentLoop可以主动回滚到之前某个“检查点”状态并尝试不同的规划路径。这需要状态管理器支持状态快照。4.3 可观测性与调试让Agent透明化调试一个“胡思乱想”的Agent是痛苦的。因此AgentLoop的每个步骤都应该产生丰富的日志和追踪数据。结构化日志记录每轮循环的输入状态、输出的动作、执行结果、新状态。使用JSON格式便于后续分析。追踪链生成一个完整的追踪链记录Agent完整的“思考过程”。这不仅是调试的利器也是后续进行效果评估Evaluation和微调Fine-tuning的数据基础。许多框架会将这些追踪信息输出为标准格式如OpenAI的Compatible格式方便接入LangSmith等可视化平台。关键指标收集循环次数、工具调用耗时、Token消耗、成功率等指标用于监控和优化成本与性能。避坑指南在开发初期就集成可观测性。不要等到出了问题才加日志。一个简单的做法是在AgentLoop的每个主要函数入口和出口处都记录下关键信息。使用像logging模块的DEBUG级别可以在需要时详细输出在正常运行时关闭避免日志泛滥。5. 从源码学习到自主实践构建你的简易AgentLoop读懂了Nanobot的设计思想后最好的巩固方式就是自己动手实现一个简化版的AgentLoop。下面是一个基于OpenAI API和简单工具调用的实践路线。5.1 环境准备与基础定义首先定义最核心的数据结构AgentState和Action。from typing import List, Dict, Any, Optional from pydantic import BaseModel from enum import Enum class ActionType(str, Enum): TOOL_CALL tool_call FINAL_ANSWER final_answer class Action(BaseModel): type: ActionType name: Optional[str] None # 工具名 args: Optional[Dict[str, Any]] None # 工具参数 content: Optional[str] None # 最终答案内容 class AgentState(BaseModel): goal: str history: List[Dict[str, Any]] # 存储每一步的 action 和 observation step: int 05.2 实现核心循环组件接下来实现规划器、执行器和循环控制器。import openai import asyncio import json class SimplePlanner: def __init__(self, llm_client, tools): self.llm llm_client self.tools tools # 工具描述列表 async def plan(self, state: AgentState) - Action: # 1. 构建提示词 prompt self._build_prompt(state) # 2. 调用LLM response await self.llm.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], response_format{ type: json_object } # 强制JSON输出 ) # 3. 解析响应 try: result json.loads(response.choices[0].message.content) if result.get(action) final_answer: return Action(typeActionType.FINAL_ANSWER, contentresult.get(answer)) elif result.get(action) use_tool: return Action(typeActionType.TOOL_CALL, nameresult.get(tool), argsresult.get(args)) else: raise ValueError(Invalid action from LLM) except (json.JSONDecodeError, KeyError, ValueError) as e: # 解析失败返回一个安全动作比如请求澄清 return Action(typeActionType.FINAL_ANSWER, content我遇到了理解上的困难请重新表述您的问题。) def _build_prompt(self, state): # 这里是一个简化的提示词模板 tools_desc \n.join([f- {t[name]}: {t[description]} (参数: {t[parameters]}) for t in self.tools]) history_str \n.join([fStep {i}: {h.get(action)} - {h.get(observation)} for i, h in enumerate(state.history[-5:])]) # 只保留最近5步 return f 目标{state.goal} 可用工具 {tools_desc} 最近步骤 {history_str} 请决定下一步。你必须以严格的JSON格式回复只包含以下两种之一 1. 调用工具{{action: use_tool, tool: 工具名, args: {{参数名: 值}}}} 2. 最终回答{{action: final_answer, answer: 你的回答内容}} 现在请输出JSON class SimpleExecutor: # 一个简单的工具实现示例 tools_dict { get_current_time: { func: lambda **kwargs: {current_time: 2023-10-27 15:30:00}, description: 获取当前时间 }, calculate: { func: lambda expression: {result: eval(expression)}, # 注意生产环境切勿使用eval此处仅为演示 description: 计算数学表达式如 3 5 * 2 } } async def execute(self, action: Action) - Dict[str, Any]: if action.type ! ActionType.TOOL_CALL: return {error: Not a tool call action} tool_name action.name if tool_name not in self.tools_dict: return {error: fTool {tool_name} not found} try: # 执行工具函数 func self.tools_dict[tool_name][func] result func(**(action.args or {})) return {success: True, data: result} except Exception as e: return {success: False, error: str(e)} class SimpleAgentLoop: def __init__(self, planner, executor, max_steps10): self.planner planner self.executor executor self.max_steps max_steps async def run(self, goal: str) - str: state AgentState(goalgoal, history[]) for step in range(self.max_steps): print(f\n Step {step} ) # 规划 action await self.planner.plan(state) print(fPlanned Action: {action}) state.history.append({step: step, action: action.dict()}) # 检查是否为最终答案 if action.type ActionType.FINAL_ANSWER: return action.content # 执行与观察 raw_obs await self.executor.execute(action) print(fRaw Observation: {raw_obs}) # 简化处理直接将结果转为字符串作为观察 observation_str str(raw_obs) state.history.append({step: step, observation: observation_str}) # 简单上下文管理如果历史太长截断最早的部分非生产环境策略 if len(state.history) 20: # 保留最多10轮交互每轮actionobservation state.history state.history[-20:] # 循环超限 return f任务未在{self.max_steps}步内完成。最新状态{state.history[-1] if state.history else 无}5.3 运行你的第一个AgentLoop最后将它们组合起来并运行。async def main(): # 1. 定义可用工具 tools [ {name: get_current_time, description: 获取系统当前时间, parameters: 无}, {name: calculate, description: 计算一个数学表达式, parameters: {expression: 字符串}} ] # 2. 初始化组件 (需要设置你的OPENAI_API_KEY) client openai.AsyncOpenAI(api_keyyour-api-key) planner SimplePlanner(client, tools) executor SimpleExecutor() loop SimpleAgentLoop(planner, executor, max_steps6) # 3. 运行Agent goal 现在几点了如果是下午就计算一下(35)*2等于多少。 result await loop.run(goal) print(f\n最终结果: {result}) # 运行 if __name__ __main__: asyncio.run(main())这个简易实现涵盖了AgentLoop的核心状态、规划、执行、观察、循环。运行它你会看到Agent一步步地“思考”先规划调用get_current_time得到时间后再规划调用calculate最后给出最终答案。6. 常见问题与排查技巧实录在实际开发和运行基于AgentLoop的Agent时你会遇到各种各样的问题。以下是一些典型问题及其排查思路很多都是我在实践中踩过的坑。6.1 LLM不按格式输出或“胡言乱语”现象规划器解析LLM响应时频繁报错JSONDecodeError或得到意料之外的动作类型。原因提示词Prompt不够清晰没有强约束输出格式。上下文历史过长或混乱干扰了LLM。模型能力不足如使用了过于基础或不适配的模型。解决方案强化提示词约束在提示词开头和结尾明确强调输出格式。使用“你必须”、“只能”、“严格按照以下JSON格式”等强指令。提供多个清晰的正反示例Few-shot。启用JSON Mode如果使用支持该功能的API如OpenAI务必开启response_format{ type: json_object }这能极大提高输出稳定性。使用输出解析库采用LangChain的PydanticOutputParser或类似库它们能提供更鲁棒的解析并在解析失败时自动尝试“修复”LLM的输出。实施重试机制在规划器代码中捕获解析异常将错误信息和原提示重新发送给LLM要求其纠正。通常重试1-2次能解决大部分问题。精简和清理上下文确保传给LLM的历史是干净、相关的。及时进行历史摘要或滑动窗口截断。6.2 Agent陷入死循环或无效动作现象Agent反复调用同一个工具或在一系列无意义的动作中打转无法推进任务。原因工具结果未被正确理解观察处理器没有从工具返回的原始数据中提取出对规划有用的信息导致LLM基于错误或模糊的观察做出重复决策。缺乏任务进度感知Agent没有机制判断当前是否更接近目标容易在原地踏步。工具能力不足或描述不清LLM误以为某个工具能做它实际上做不到的事情。解决方案优化观察处理确保观察处理器输出的信息是简洁、准确、面向目标的。例如不要直接把一大段JSON扔回去而是总结成“查询成功北京明天晴气温5-15度”。在状态中引入进度标记可以在状态中显式地维护一个“已完成子目标”的列表或在提示词中让LLM每次规划时都评估一下当前进度。设置最大迭代次数这是最后的安全网。在AgentLoop中必须有一个硬性的max_steps限制并在达到时优雅失败给出已尝试的步骤记录方便分析。改进工具描述在工具的描述中明确指出其限制和边界条件。6.3 工具执行超时或失败现象工具调用长时间无响应或返回网络错误、5xx错误等。原因外部API不稳定、网络问题、工具本身有缺陷。解决方案为执行器添加超时和重试使用asyncio.wait_for或httpx.Timeout为每个工具调用设置合理的超时时间。对于网络错误、5xx错误实现指数退避重试逻辑。区分错误类型不是所有错误都值得重试。4xx错误如参数错误、权限不足应立即失败并反馈给LLM让LLM调整策略。5xx和超时才进行重试。实现熔断器如果某个工具连续失败多次可以暂时将其“熔断”标记为不可用在后续几轮规划中避免调用它过一段时间后再恢复。6.4 上下文长度超限与成本失控现象任务执行到后期变慢成本激增甚至收到API的上下文超长错误。原因历史对话未经管理无限增长。解决方案强制滑动窗口这是最简单有效的方法。只保留最近N轮如10轮的详细交互。动态摘要实现一个“摘要器”组件。当历史达到一定长度如Token数超过阈值调用一个成本较低的模型如gpt-3.5-turbo将早期的对话总结成一段简短的摘要。后续循环使用“摘要 近期详细历史”作为上下文。这需要在每次循环中判断是否触发摘要。选择性记忆尝试设计更智能的算法只保留与核心目标高度相关的历史片段。但这实现起来比较复杂。监控与告警在AgentLoop中记录每轮消耗的Token数并设置成本预算和告警。通过对Nanobot等开源框架AgentLoop源码的深入学习并将其核心思想付诸实践你不仅能构建出功能强大的AI Agent更能深刻理解智能体系统稳定、高效运行背后的工程逻辑。从清晰的状态管理到鲁棒的循环控制从灵活的规划到安全的执行每一个细节都关乎最终体验的成败。