拓冰网站定制 
1. 什么是 Loop EngineeringLoop Engineering循环工程是一套围绕「循环」这一核心模式进行 AI 系统设计的方法论它不再把大模型当作一次性的「输入 → 输出」函数而是把模型置于一个受控的循环中通过多轮调用、自我修正、工具交互与外部反馈逐步逼近更高质量的答案。如果说 Prompt Engineering 研究的是「如何写好一句话」那么 Loop Engineering 研究的就是「如何设计好一轮又一轮的协作过程」。前者优化单次推理后者优化整个推理系统的行为。一个最直观的对比维度单次推理Prompt Engineering循环推理Loop Engineering调用次数1 次N 次N 由条件决定能否自我修正不能可以能否使用工具不能可以能否吸收反馈不能可以输出质量上限受单次生成限制随循环逐步提升用一句话概括Loop Engineering 是把「一次性的答案生成」升级为「有目标、有状态、有反馈的持续求解过程」。2. 为什么循环比单次推理更强单次推理的局限非常明显无法纠错模型生成了错误答案没有第二次机会。无法探索面对多步问题数学推导、代码调试一次生成往往顾此失彼。无法利用外部信息不能查资料、跑代码、调 API只能凭训练时的知识作答。无法适应反馈用户说「不对我要的是 A 不是 B」单次推理只能整体重来。循环恰好逐一补上这些短板。以代码生成为例模型第一次写出代码 → 运行报错 → 把报错信息喂回模型 → 模型修复 → 再次运行。这个「写 → 跑 → 改」的循环让模型从「会写代码的样子」变成「真的能写出可运行的代码」。这种能力的提升并非来自于更大的模型而是来自于循环结构本身让模型和环境发生了交互。这也是为什么 Agent、RAG、自我反思等一系列前沿范式本质上都建立在循环之上。3. 循环的基本形态循环并不神秘工程上无非是四种基本形态的组合。3.1 固定次数循环循环次数在代码里写死最常见的是「让模型生成 N 个候选再从中选优」。foriinrange(3):answercall_llm(messages)candidates.append(answer)3.2 条件循环循环是否继续由某个条件决定最典型的是「重试直到成功」。whilenotis_valid(answer):answercall_llm(messages)3.3 自我修正循环把模型的输出连同「批评意见」一起送回模型让它自我改进。这是当前提升推理质量性价比最高的手段之一。3.4 Agent 循环感知环境 → 思考 → 调用工具 → 观察结果 → 再思考……直到找到答案。ReAct 是它的经典实现。理解这四种形态后你就可以像搭积木一样组合出复杂系统。下面进入代码实战部分。4. 代码实战构建属于你的 LLM Loop4.1 环境准备先安装依赖并初始化客户端。这里使用 OpenAI 兼容接口你可以换成任何提供 OpenAI 协议的服务。pipinstallopenaiimportosimportjsonimporttimefromtypingimportAny,CallablefromopenaiimportOpenAI# 初始化客户端支持任何 OpenAI 兼容服务clientOpenAI(api_keyos.getenv(OPENAI_API_KEY),base_urlos.getenv(OPENAI_BASE_URL,https://api.openai.com/v1),)# 统一封装一次 LLM 调用defcall_llm(messages:list[dict],model:strgpt-4o-mini,temperature:float0.7,**kwargs,)-str:把消息列表发给模型返回文本内容。respclient.chat.completions.create(modelmodel,messagesmessages,temperaturetemperature,**kwargs,)returnresp.choices[0].message.contentor4.2 最简循环重试直到成功这是条件循环的最朴素实现。假设我们要让模型输出一个合法 JSON但模型偶尔会在前后加解释文字我们就反复要求直到解析成功。importjsondefgenerate_json_with_retry(prompt:str,max_attempts:int5)-dict:反复请求模型直到返回合法 JSON 或超过最大次数。messages[{role:system,content:你是 JSON 生成器只输出 JSON不要任何解释。},{role:user,content:prompt},]forattemptinrange(1,max_attempts1):rawcall_llm(messages,temperature0.3)try:returnjson.loads(raw)exceptjson.JSONDecodeErrorase:print(f[尝试{attempt}/{max_attempts}] 解析失败:{e})# 关键把错误信息喂回模型让它知道错在哪messages.append({role:assistant,content:raw})messages.append({role:user,content:f上面的输出无法解析为 JSON错误是{e}。请只输出合法的 JSON。,})raiseRuntimeError(超过最大重试次数仍未能得到合法 JSON)调用示例resultgenerate_json_with_retry(生成一个包含名字、年龄、城市的用户信息 JSON)print(result)# {name: 张三, age: 28, city: 杭州}这个循环揭示了一个 Loop Engineering 的核心心法把失败本身也作为输入喂回模型。循环不是为了重复而重复而是让每一步都携带新的信息。4.3 自我修正循环撰写 → 批评 → 重写这是最经典的自我修正模式Critic-Refine。模型先扮演「批评家」挑毛病再扮演「作者」按意见改进循环往复直到批评家满意。SYSTEM_WRITER你是一位严谨的技术写作专家撰写清晰、准确、有深度的内容。SYSTEM_CRITIC(你是一位严格的审稿人。请针对用户给出的文本指出至少一个问题涉及准确性、逻辑漏洞、表达不清或遗漏关键点。如果文本已经很好只回复 APPROVED。)defself_refine(draft:str,max_rounds:int3)-str:通过「批评-改进」循环打磨一段文字。currentdraftforround_noinrange(1,max_rounds1):# 第一步批评critiquecall_llm([{role:system,content:SYSTEM_CRITIC},{role:user,content:current},],temperature0.3,)print(f\n--- 第{round_no}轮批评 ---\n{critique})ifAPPROVEDincritique:print(审稿通过结束循环。)break# 第二步按意见重写currentcall_llm([{role:system,content:SYSTEM_WRITER},{role:user,content:f原稿如下\n\n{current}\n\n审稿意见\n{critique}\n\n请根据意见重写。},],temperature0.7,)print(f\n--- 第{round_no}轮重写后 ---\n{current})returncurrent# 使用示例draft_text循环工程就是用 while 循环反复调模型这样答案会更好。final_textself_refine(draft_text,max_rounds3)print(\n 最终版本 \n,final_text)这里有两个值得注意的设计点终止条件多样化既可以是批评家明确说「APPROVED」也可以是轮数上限兜底避免无限循环。温度差异化批评家用低温度保持严谨作者用较高温度保持表达灵活性。循环中的每个角色可以有自己独立的采样策略。4.4 ReAct Agent 循环带工具的完整实现下面实现一个经典的 ReActReasoning Acting循环模型不再凭空答题而是可以调用计算器、查询词典等工具把工具结果观察到上下文里再继续推理。# ---------- 工具定义 ----------defcalculator(expr:str)-str:计算数学表达式如 3*72。注意这里仅作教学演示生产环境勿用 eval。try:resulteval(expr,{__builtins__:{}},{})returnf结果{result}exceptExceptionase:returnf计算错误{e}defword_length(word:str)-str:返回单词或短句的字符数含空格。returnf{word} 共{len(word)}个字符AVAILABLE_TOOLS:dict[str,dict[str,Any]]{calculator:{description:计算数学表达式输入如 3*72,parameters:{expr:数学表达式字符串},func:calculator,},word_length:{description:统计一段文本的字符数,parameters:{word:要统计的文本},func:word_length,},}TOOL_PROMPT你可以使用以下工具来帮助回答问题。每次只调用一个工具。 可用工具 {tool_desc} 你必须严格按照以下 JSON 格式输出你的下一步不要输出其他内容 {{thought: 你的思考过程, action: 工具名或 finish, action_input: {{参数}} 或 最终答案}} - 当 action 是工具名时action_input 是对应参数字典字符串形式。 - 当你已经有足够信息回答时action 设为 finishaction_input 填最终答案。 defbuild_tool_desc()-str:lines[]forname,metainAVAILABLE_TOOLS.items():params, .join(f{k}:{v}fork,vinmeta[parameters].items())lines.append(f-{name}({params}):{meta[description]})return\n.join(lines)# ---------- Agent 循环 ----------defrun_agent(question:str,max_steps:int8)-str:ReAct 主循环Thought → Action → Observation → ... → Finish。system_promptTOOL_PROMPT.format(tool_descbuild_tool_desc())messages[{role:system,content:system_prompt},{role:user,content:question},]# 观测历史模拟 ReAct 的 Observation 步骤observations:list[str][]forstepinrange(1,max_steps1):print(f\n Step{step})responsecall_llm(messages,temperature0.0)# 解析模型输出的 JSONtry:parsedjson.loads(response)exceptjson.JSONDecodeError:messages.append({role:assistant,content:response})messages.append({role:user,content:输出格式错误请严格按照要求的 JSON 格式输出。})continuethoughtparsed.get(thought,)actionparsed.get(action,)action_inputparsed.get(action_input,)print(f 思考{thought})print(f 动作{action}| 输入{action_input})# 把 assistant 的本次输出记入上下文messages.append({role:assistant,content:response})# 终止条件模型认为已经可以作答ifactionfinish:print(✅ 循环结束得到最终答案。)returnstr(action_input)# 执行工具ifactionnotinAVAILABLE_TOOLS:observationf错误工具 {action} 不存在。可用工具{list(AVAILABLE_TOOLS.keys())}else:tool_funcAVAILABLE_TOOLS[action][func]try:ifisinstance(action_input,dict):observationtool_func(**action_input)else:observationtool_func(action_input)exceptExceptionase:observationf工具调用异常{e}print(f 观测{observation})observations.append(observation)# 关键把 Observation 作为 user 消息送回模型messages.append({role:user,content:fObservation:{observation}})# 可选控制上下文长度只保留最近若干轮# messages messages[:1] messages[-12:]return达到最大步数仍未得到答案。# 使用示例answerrun_agent(请计算 (17*23) (5*9) 的结果并告诉我结果有几个字符)print(\n 最终答案 \n,answer)这段代码完整复刻了 ReAct 论文的核心结构Thought模型先思考下一步该做什么。Action选择调用工具或直接作答。Observation工具返回结果作为新信息进入上下文。循环往复直到模型输出finish。4.5 带记忆与多层终止条件的完整 Agent把上面的能力整合起来加入「多轮对话记忆」「步数上限」「连续失败熔断」三层终止保护形成接近生产可用的骨架。fromdataclassesimportdataclass,fielddataclassclassAgentState:Agent 的完整状态包括历史消息、步数与失败计数。messages:list[dict]field(default_factorylist)steps:int0consecutive_failures:int0defadd_user(self,content:str):self.messages.append({role:user,content:content})defadd_assistant(self,content:str):self.messages.append({role:assistant,content:content})defadd_observation(self,content:str):self.messages.append({role:user,content:fObservation:{content}})deftrim(self,keep_last:int16):只保留系统提示与最近 N 条消息防止上下文爆炸。self.messagesself.messages[:1]self.messages[-keep_last:]classRobustAgent:带记忆、步数上限与连续失败熔断的循环 Agent。def__init__(self,max_steps:int10,max_consecutive_failures:int3):self.max_stepsmax_steps self.max_consecutive_failuresmax_consecutive_failures self.stateAgentState(messages[{role:system,content:TOOL_PROMPT.format(tool_descbuild_tool_desc())}])defrun(self,question:str)-str:self.state.add_user(question)whileself.state.stepsself.max_steps:self.state.steps1print(f\n Step{self.state.steps})responsecall_llm(self.state.messages,temperature0.0)try:parsedjson.loads(response)exceptjson.JSONDecodeError:# 格式错误也算一次失败self.state.consecutive_failures1ifself._should_abort():return连续多次输出异常已熔断终止。self.state.add_assistant(response)self.state.add_user(输出必须是合法 JSON请重试。)continue# 解析成功则重置失败计数self.state.consecutive_failures0actionparsed.get(action)self.state.add_assistant(response)ifactionfinish:returnstr(parsed.get(action_input))# 执行工具tool_funcAVAILABLE_TOOLS.get(action)iftool_funcisNone:observationf错误工具 {action} 不存在。self.state.consecutive_failures1else:try:action_inputparsed.get(action_input)observationtool_func(**action_input)ifisinstance(action_input,dict)elsetool_func(action_input)exceptExceptionase:observationf工具调用异常{e}self.state.consecutive_failures1print(f{observation})self.state.add_observation(observation)# 上下文超长时裁剪iflen(self.state.messages)20:self.state.trim(keep_last14)ifself._should_abort():return连续多次失败已熔断终止。return达到步数上限。def_should_abort(self)-bool:returnself.state.consecutive_failuresself.max_consecutive_failures# 使用示例agentRobustAgent()print(\n 第一问 )print(答案:,agent.run(3 乘以 17 加上 24 等于多少))# 状态保留支持追问这是循环工程带来的「记忆」能力print(\n 追问 )print(答案:,agent.run(刚才那个结果的字符数是多少请用工具算一下。))这里的几个工程细节是把「能跑」变成「能用」的关键状态对象化消息历史、步数、失败次数都放进AgentState便于持久化与 Loop Engineering 详解：从单次推理到循环智能的工程方法论
1. 什么是 Loop EngineeringLoop Engineering循环工程是一套围绕「循环」这一核心模式进行 AI 系统设计的方法论它不再把大模型当作一次性的「输入 → 输出」函数而是把模型置于一个受控的循环中通过多轮调用、自我修正、工具交互与外部反馈逐步逼近更高质量的答案。如果说 Prompt Engineering 研究的是「如何写好一句话」那么 Loop Engineering 研究的就是「如何设计好一轮又一轮的协作过程」。前者优化单次推理后者优化整个推理系统的行为。一个最直观的对比维度单次推理Prompt Engineering循环推理Loop Engineering调用次数1 次N 次N 由条件决定能否自我修正不能可以能否使用工具不能可以能否吸收反馈不能可以输出质量上限受单次生成限制随循环逐步提升用一句话概括Loop Engineering 是把「一次性的答案生成」升级为「有目标、有状态、有反馈的持续求解过程」。2. 为什么循环比单次推理更强单次推理的局限非常明显无法纠错模型生成了错误答案没有第二次机会。无法探索面对多步问题数学推导、代码调试一次生成往往顾此失彼。无法利用外部信息不能查资料、跑代码、调 API只能凭训练时的知识作答。无法适应反馈用户说「不对我要的是 A 不是 B」单次推理只能整体重来。循环恰好逐一补上这些短板。以代码生成为例模型第一次写出代码 → 运行报错 → 把报错信息喂回模型 → 模型修复 → 再次运行。这个「写 → 跑 → 改」的循环让模型从「会写代码的样子」变成「真的能写出可运行的代码」。这种能力的提升并非来自于更大的模型而是来自于循环结构本身让模型和环境发生了交互。这也是为什么 Agent、RAG、自我反思等一系列前沿范式本质上都建立在循环之上。3. 循环的基本形态循环并不神秘工程上无非是四种基本形态的组合。3.1 固定次数循环循环次数在代码里写死最常见的是「让模型生成 N 个候选再从中选优」。foriinrange(3):answercall_llm(messages)candidates.append(answer)3.2 条件循环循环是否继续由某个条件决定最典型的是「重试直到成功」。whilenotis_valid(answer):answercall_llm(messages)3.3 自我修正循环把模型的输出连同「批评意见」一起送回模型让它自我改进。这是当前提升推理质量性价比最高的手段之一。3.4 Agent 循环感知环境 → 思考 → 调用工具 → 观察结果 → 再思考……直到找到答案。ReAct 是它的经典实现。理解这四种形态后你就可以像搭积木一样组合出复杂系统。下面进入代码实战部分。4. 代码实战构建属于你的 LLM Loop4.1 环境准备先安装依赖并初始化客户端。这里使用 OpenAI 兼容接口你可以换成任何提供 OpenAI 协议的服务。pipinstallopenaiimportosimportjsonimporttimefromtypingimportAny,CallablefromopenaiimportOpenAI# 初始化客户端支持任何 OpenAI 兼容服务clientOpenAI(api_keyos.getenv(OPENAI_API_KEY),base_urlos.getenv(OPENAI_BASE_URL,https://api.openai.com/v1),)# 统一封装一次 LLM 调用defcall_llm(messages:list[dict],model:strgpt-4o-mini,temperature:float0.7,**kwargs,)-str:把消息列表发给模型返回文本内容。respclient.chat.completions.create(modelmodel,messagesmessages,temperaturetemperature,**kwargs,)returnresp.choices[0].message.contentor4.2 最简循环重试直到成功这是条件循环的最朴素实现。假设我们要让模型输出一个合法 JSON但模型偶尔会在前后加解释文字我们就反复要求直到解析成功。importjsondefgenerate_json_with_retry(prompt:str,max_attempts:int5)-dict:反复请求模型直到返回合法 JSON 或超过最大次数。messages[{role:system,content:你是 JSON 生成器只输出 JSON不要任何解释。},{role:user,content:prompt},]forattemptinrange(1,max_attempts1):rawcall_llm(messages,temperature0.3)try:returnjson.loads(raw)exceptjson.JSONDecodeErrorase:print(f[尝试{attempt}/{max_attempts}] 解析失败:{e})# 关键把错误信息喂回模型让它知道错在哪messages.append({role:assistant,content:raw})messages.append({role:user,content:f上面的输出无法解析为 JSON错误是{e}。请只输出合法的 JSON。,})raiseRuntimeError(超过最大重试次数仍未能得到合法 JSON)调用示例resultgenerate_json_with_retry(生成一个包含名字、年龄、城市的用户信息 JSON)print(result)# {name: 张三, age: 28, city: 杭州}这个循环揭示了一个 Loop Engineering 的核心心法把失败本身也作为输入喂回模型。循环不是为了重复而重复而是让每一步都携带新的信息。4.3 自我修正循环撰写 → 批评 → 重写这是最经典的自我修正模式Critic-Refine。模型先扮演「批评家」挑毛病再扮演「作者」按意见改进循环往复直到批评家满意。SYSTEM_WRITER你是一位严谨的技术写作专家撰写清晰、准确、有深度的内容。SYSTEM_CRITIC(你是一位严格的审稿人。请针对用户给出的文本指出至少一个问题涉及准确性、逻辑漏洞、表达不清或遗漏关键点。如果文本已经很好只回复 APPROVED。)defself_refine(draft:str,max_rounds:int3)-str:通过「批评-改进」循环打磨一段文字。currentdraftforround_noinrange(1,max_rounds1):# 第一步批评critiquecall_llm([{role:system,content:SYSTEM_CRITIC},{role:user,content:current},],temperature0.3,)print(f\n--- 第{round_no}轮批评 ---\n{critique})ifAPPROVEDincritique:print(审稿通过结束循环。)break# 第二步按意见重写currentcall_llm([{role:system,content:SYSTEM_WRITER},{role:user,content:f原稿如下\n\n{current}\n\n审稿意见\n{critique}\n\n请根据意见重写。},],temperature0.7,)print(f\n--- 第{round_no}轮重写后 ---\n{current})returncurrent# 使用示例draft_text循环工程就是用 while 循环反复调模型这样答案会更好。final_textself_refine(draft_text,max_rounds3)print(\n 最终版本 \n,final_text)这里有两个值得注意的设计点终止条件多样化既可以是批评家明确说「APPROVED」也可以是轮数上限兜底避免无限循环。温度差异化批评家用低温度保持严谨作者用较高温度保持表达灵活性。循环中的每个角色可以有自己独立的采样策略。4.4 ReAct Agent 循环带工具的完整实现下面实现一个经典的 ReActReasoning Acting循环模型不再凭空答题而是可以调用计算器、查询词典等工具把工具结果观察到上下文里再继续推理。# ---------- 工具定义 ----------defcalculator(expr:str)-str:计算数学表达式如 3*72。注意这里仅作教学演示生产环境勿用 eval。try:resulteval(expr,{__builtins__:{}},{})returnf结果{result}exceptExceptionase:returnf计算错误{e}defword_length(word:str)-str:返回单词或短句的字符数含空格。returnf{word} 共{len(word)}个字符AVAILABLE_TOOLS:dict[str,dict[str,Any]]{calculator:{description:计算数学表达式输入如 3*72,parameters:{expr:数学表达式字符串},func:calculator,},word_length:{description:统计一段文本的字符数,parameters:{word:要统计的文本},func:word_length,},}TOOL_PROMPT你可以使用以下工具来帮助回答问题。每次只调用一个工具。 可用工具 {tool_desc} 你必须严格按照以下 JSON 格式输出你的下一步不要输出其他内容 {{thought: 你的思考过程, action: 工具名或 finish, action_input: {{参数}} 或 最终答案}} - 当 action 是工具名时action_input 是对应参数字典字符串形式。 - 当你已经有足够信息回答时action 设为 finishaction_input 填最终答案。 defbuild_tool_desc()-str:lines[]forname,metainAVAILABLE_TOOLS.items():params, .join(f{k}:{v}fork,vinmeta[parameters].items())lines.append(f-{name}({params}):{meta[description]})return\n.join(lines)# ---------- Agent 循环 ----------defrun_agent(question:str,max_steps:int8)-str:ReAct 主循环Thought → Action → Observation → ... → Finish。system_promptTOOL_PROMPT.format(tool_descbuild_tool_desc())messages[{role:system,content:system_prompt},{role:user,content:question},]# 观测历史模拟 ReAct 的 Observation 步骤observations:list[str][]forstepinrange(1,max_steps1):print(f\n Step{step})responsecall_llm(messages,temperature0.0)# 解析模型输出的 JSONtry:parsedjson.loads(response)exceptjson.JSONDecodeError:messages.append({role:assistant,content:response})messages.append({role:user,content:输出格式错误请严格按照要求的 JSON 格式输出。})continuethoughtparsed.get(thought,)actionparsed.get(action,)action_inputparsed.get(action_input,)print(f 思考{thought})print(f 动作{action}| 输入{action_input})# 把 assistant 的本次输出记入上下文messages.append({role:assistant,content:response})# 终止条件模型认为已经可以作答ifactionfinish:print(✅ 循环结束得到最终答案。)returnstr(action_input)# 执行工具ifactionnotinAVAILABLE_TOOLS:observationf错误工具 {action} 不存在。可用工具{list(AVAILABLE_TOOLS.keys())}else:tool_funcAVAILABLE_TOOLS[action][func]try:ifisinstance(action_input,dict):observationtool_func(**action_input)else:observationtool_func(action_input)exceptExceptionase:observationf工具调用异常{e}print(f 观测{observation})observations.append(observation)# 关键把 Observation 作为 user 消息送回模型messages.append({role:user,content:fObservation:{observation}})# 可选控制上下文长度只保留最近若干轮# messages messages[:1] messages[-12:]return达到最大步数仍未得到答案。# 使用示例answerrun_agent(请计算 (17*23) (5*9) 的结果并告诉我结果有几个字符)print(\n 最终答案 \n,answer)这段代码完整复刻了 ReAct 论文的核心结构Thought模型先思考下一步该做什么。Action选择调用工具或直接作答。Observation工具返回结果作为新信息进入上下文。循环往复直到模型输出finish。4.5 带记忆与多层终止条件的完整 Agent把上面的能力整合起来加入「多轮对话记忆」「步数上限」「连续失败熔断」三层终止保护形成接近生产可用的骨架。fromdataclassesimportdataclass,fielddataclassclassAgentState:Agent 的完整状态包括历史消息、步数与失败计数。messages:list[dict]field(default_factorylist)steps:int0consecutive_failures:int0defadd_user(self,content:str):self.messages.append({role:user,content:content})defadd_assistant(self,content:str):self.messages.append({role:assistant,content:content})defadd_observation(self,content:str):self.messages.append({role:user,content:fObservation:{content}})deftrim(self,keep_last:int16):只保留系统提示与最近 N 条消息防止上下文爆炸。self.messagesself.messages[:1]self.messages[-keep_last:]classRobustAgent:带记忆、步数上限与连续失败熔断的循环 Agent。def__init__(self,max_steps:int10,max_consecutive_failures:int3):self.max_stepsmax_steps self.max_consecutive_failuresmax_consecutive_failures self.stateAgentState(messages[{role:system,content:TOOL_PROMPT.format(tool_descbuild_tool_desc())}])defrun(self,question:str)-str:self.state.add_user(question)whileself.state.stepsself.max_steps:self.state.steps1print(f\n Step{self.state.steps})responsecall_llm(self.state.messages,temperature0.0)try:parsedjson.loads(response)exceptjson.JSONDecodeError:# 格式错误也算一次失败self.state.consecutive_failures1ifself._should_abort():return连续多次输出异常已熔断终止。self.state.add_assistant(response)self.state.add_user(输出必须是合法 JSON请重试。)continue# 解析成功则重置失败计数self.state.consecutive_failures0actionparsed.get(action)self.state.add_assistant(response)ifactionfinish:returnstr(parsed.get(action_input))# 执行工具tool_funcAVAILABLE_TOOLS.get(action)iftool_funcisNone:observationf错误工具 {action} 不存在。self.state.consecutive_failures1else:try:action_inputparsed.get(action_input)observationtool_func(**action_input)ifisinstance(action_input,dict)elsetool_func(action_input)exceptExceptionase:observationf工具调用异常{e}self.state.consecutive_failures1print(f{observation})self.state.add_observation(observation)# 上下文超长时裁剪iflen(self.state.messages)20:self.state.trim(keep_last14)ifself._should_abort():return连续多次失败已熔断终止。return达到步数上限。def_should_abort(self)-bool:returnself.state.consecutive_failuresself.max_consecutive_failures# 使用示例agentRobustAgent()print(\n 第一问 )print(答案:,agent.run(3 乘以 17 加上 24 等于多少))# 状态保留支持追问这是循环工程带来的「记忆」能力print(\n 追问 )print(答案:,agent.run(刚才那个结果的字符数是多少请用工具算一下。))这里的几个工程细节是把「能跑」变成「能用」的关键状态对象化消息历史、步数、失败次数都放进AgentState便于持久化与