☰
Loop Engineering 详解:从单次推理到循环智能的工程方法论
2026/10/7 8:28:03 网站建设 项目流程

1. 什么是 Loop Engineering

Loop 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):answer=call_llm(messages)candidates.append(answer)

3.2 条件循环

循环是否继续由某个条件决定,最典型的是「重试直到成功」。

whilenotis_valid(answer):answer=call_llm(messages)

3.3 自我修正循环

把模型的输出连同「批评意见」一起送回模型,让它自我改进。这是当前提升推理质量性价比最高的手段之一。

3.4 Agent 循环

感知环境 → 思考 → 调用工具 → 观察结果 → 再思考……直到找到答案。ReAct 是它的经典实现。

理解这四种形态后,你就可以像搭积木一样组合出复杂系统。下面进入代码实战部分。

4. 代码实战:构建属于你的 LLM Loop

4.1 环境准备

先安装依赖并初始化客户端。这里使用 OpenAI 兼容接口,你可以换成任何提供 OpenAI 协议的服务。

pipinstallopenai
importosimportjsonimporttimefromtypingimportAny,CallablefromopenaiimportOpenAI# 初始化客户端(支持任何 OpenAI 兼容服务)client=OpenAI(api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_BASE_URL","https://api.openai.com/v1"),)# 统一封装一次 LLM 调用defcall_llm(messages:list[dict],model:str="gpt-4o-mini",temperature:float=0.7,**kwargs,)->str:"""把消息列表发给模型,返回文本内容。"""resp=client.chat.completions.create(model=model,messages=messages,temperature=temperature,**kwargs,)returnresp.choices[0].message.contentor""

4.2 最简循环:重试直到成功

这是条件循环的最朴素实现。假设我们要让模型输出一个合法 JSON,但模型偶尔会在前后加解释文字,我们就反复要求,直到解析成功。

importjsondefgenerate_json_with_retry(prompt:str,max_attempts:int=5)->dict:"""反复请求模型,直到返回合法 JSON 或超过最大次数。"""messages=[{"role":"system","content":"你是 JSON 生成器,只输出 JSON,不要任何解释。"},{"role":"user","content":prompt},]forattemptinrange(1,max_attempts+1):raw=call_llm(messages,temperature=0.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")

调用示例:

result=generate_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:int=3)->str:"""通过「批评-改进」循环打磨一段文字。"""current=draftforround_noinrange(1,max_rounds+1):# 第一步:批评critique=call_llm([{"role":"system","content":SYSTEM_CRITIC},{"role":"user","content":current},],temperature=0.3,)print(f"\n--- 第{round_no}轮批评 ---\n{critique}")if"APPROVED"incritique:print("审稿通过,结束循环。")break# 第二步:按意见重写current=call_llm([{"role":"system","content":SYSTEM_WRITER},{"role":"user","content":f"原稿如下:\n\n{current}\n\n审稿意见:\n{critique}\n\n请根据意见重写。"},],temperature=0.7,)print(f"\n--- 第{round_no}轮重写后 ---\n{current}")returncurrent# 使用示例draft_text="循环工程就是用 while 循环反复调模型,这样答案会更好。"final_text=self_refine(draft_text,max_rounds=3)print("\n===== 最终版本 =====\n",final_text)

这里有两个值得注意的设计点:

  1. 终止条件多样化:既可以是批评家明确说「APPROVED」,也可以是轮数上限兜底,避免无限循环。
  2. 温度差异化:批评家用低温度保持严谨,作者用较高温度保持表达灵活性。循环中的每个角色可以有自己独立的采样策略。

4.4 ReAct Agent 循环:带工具的完整实现

下面实现一个经典的 ReAct(Reasoning + Acting)循环:模型不再凭空答题,而是可以调用计算器、查询词典等工具,把工具结果观察到上下文里,再继续推理。

# ---------- 工具定义 ----------defcalculator(expr:str)->str:"""计算数学表达式,如 '3*7+2'。注意:这里仅作教学演示,生产环境勿用 eval。"""try:result=eval(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*7+2'","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 设为 "finish",action_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:int=8)->str:"""ReAct 主循环:Thought → Action → Observation → ... → Finish。"""system_prompt=TOOL_PROMPT.format(tool_desc=build_tool_desc())messages=[{"role":"system","content":system_prompt},{"role":"user","content":question},]# 观测历史,模拟 ReAct 的 Observation 步骤observations:list[str]=[]forstepinrange(1,max_steps+1):print(f"\n========== Step{step}==========")response=call_llm(messages,temperature=0.0)# 解析模型输出的 JSONtry:parsed=json.loads(response)exceptjson.JSONDecodeError:messages.append({"role":"assistant","content":response})messages.append({"role":"user","content":"输出格式错误,请严格按照要求的 JSON 格式输出。"})continuethought=parsed.get("thought","")action=parsed.get("action","")action_input=parsed.get("action_input","")print(f"💭 思考:{thought}")print(f"🎬 动作:{action}| 输入:{action_input}")# 把 assistant 的本次输出记入上下文messages.append({"role":"assistant","content":response})# 终止条件:模型认为已经可以作答ifaction=="finish":print("✅ 循环结束,得到最终答案。")returnstr(action_input)# 执行工具ifactionnotinAVAILABLE_TOOLS:observation=f"错误:工具 '{action}' 不存在。可用工具:{list(AVAILABLE_TOOLS.keys())}"else:tool_func=AVAILABLE_TOOLS[action]["func"]try:ifisinstance(action_input,dict):observation=tool_func(**action_input)else:observation=tool_func(action_input)exceptExceptionase:observation=f"工具调用异常:{e}"print(f"👁 观测:{observation}")observations.append(observation)# 关键:把 Observation 作为 user 消息送回模型messages.append({"role":"user","content":f"Observation:{observation}"})# 可选:控制上下文长度,只保留最近若干轮# messages = messages[:1] + messages[-12:]return"达到最大步数仍未得到答案。"# 使用示例answer=run_agent("请计算 (17*23) + (5*9) 的结果,并告诉我结果有几个字符")print("\n===== 最终答案 =====\n",answer)

这段代码完整复刻了 ReAct 论文的核心结构:

  1. Thought:模型先思考下一步该做什么。
  2. Action:选择调用工具或直接作答。
  3. Observation:工具返回结果,作为新信息进入上下文。
  4. 循环往复,直到模型输出finish。

4.5 带记忆与多层终止条件的完整 Agent

把上面的能力整合起来,加入「多轮对话记忆」「步数上限」「连续失败熔断」三层终止保护,形成接近生产可用的骨架。

fromdataclassesimportdataclass,field@dataclassclassAgentState:"""Agent 的完整状态:包括历史消息、步数与失败计数。"""messages:list[dict]=field(default_factory=list)steps:int=0consecutive_failures:int=0defadd_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":f"Observation:{content}"})deftrim(self,keep_last:int=16):"""只保留系统提示与最近 N 条消息,防止上下文爆炸。"""self.messages=self.messages[:1]+self.messages[-keep_last:]classRobustAgent:"""带记忆、步数上限与连续失败熔断的循环 Agent。"""def__init__(self,max_steps:int=10,max_consecutive_failures:int=3):self.max_steps=max_steps self.max_consecutive_failures=max_consecutive_failures self.state=AgentState(messages=[{"role":"system","content":TOOL_PROMPT.format(tool_desc=build_tool_desc())}])defrun(self,question:str)->str:self.state.add_user(question)whileself.state.steps<self.max_steps:self.state.steps+=1print(f"\n===== Step{self.state.steps}=====")response=call_llm(self.state.messages,temperature=0.0)try:parsed=json.loads(response)exceptjson.JSONDecodeError:# 格式错误也算一次失败self.state.consecutive_failures+=1ifself._should_abort():return"连续多次输出异常,已熔断终止。"self.state.add_assistant(response)self.state.add_user("输出必须是合法 JSON,请重试。")continue# 解析成功则重置失败计数self.state.consecutive_failures=0action=parsed.get("action")self.state.add_assistant(response)ifaction=="finish":returnstr(parsed.get("action_input"))# 执行工具tool_func=AVAILABLE_TOOLS.get(action)iftool_funcisNone:observation=f"错误:工具 '{action}' 不存在。"self.state.consecutive_failures+=1else:try:action_input=parsed.get("action_input")observation=tool_func(**action_input)ifisinstance(action_input,dict)elsetool_func(action_input)exceptExceptionase:observation=f"工具调用异常:{e}"self.state.consecutive_failures+=1print(f"👁{observation}")self.state.add_observation(observation)# 上下文超长时裁剪iflen(self.state.messages)>20:self.state.trim(keep_last=14)ifself._should_abort():return"连续多次失败,已熔断终止。"return"达到步数上限。"def_should_abort(self)->bool:returnself.state.consecutive_failures>=self.max_consecutive_failures# 使用示例agent=RobustAgent()print("\n===== 第一问 =====")print("答案:",agent.run("3 乘以 17 加上 24 等于多少?"))# 状态保留,支持追问(这是循环工程带来的「记忆」能力)print("\n===== 追问 =====")print("答案:",agent.run("刚才那个结果的字符数是多少?请用工具算一下。"))

这里的几个工程细节,是把「能跑」变成「能用」的关键:

  • 状态对象化:消息历史、步数、失败次数都放进AgentState,便于持久化与

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询