这两年做AI应用开发,躲不开一个词:Loop Engineering。每次看到群里有人讨论,总有人问:这不就是while循环吗?还真不是。如果你要做的是那种“能自己决定下一步干啥、干完这步看结果再决定下一步”的智能体应用,核心就是Loop Engineering——把感知、推理、行动、观察串成一个闭环,让程序在循环里不断自我修正。它解决的问题很具体:让模型不再只回答一次,而是能连续工作直到把任务做完。这篇教程适合两类人:一类是刚接触智能体开发、被各种概念绕晕的入门者;另一类是已经会调大模型接口、但一做多步任务就卡壳的开发者。我会从零开始,一步步带你搭出一个真正能跑的项目。
1. Loop Engineering的本质:把“一次性问答”变成“带反馈的工作流”
1.1 一句话定义:让程序“边干边看、边看边调整”
Loop Engineering直译过来是“循环工程”,但你别被这个名字吓住。它本质上就是一件事:设计一个由模型驱动的循环回路,让AI在每个循环里经历四个阶段——感知当前状态、推理决定下一步、执行一个动作、观察执行结果,然后带着结果进入下一轮。
你可以把它理解成一个新员工入职第一天的工作方式:老板给一个模糊目标,员工先看手头有什么资源(感知),想想第一步该干啥(推理),动手干(行动),干完看看效果对不对(观察),如果不对就调整思路再干一轮。一个成熟员工之所以靠谱,不是因为一次就能做对,而是因为能快速试错、持续逼近目标。Loop Engineering就是把这套“试错逼近”的机制工程化。
很多人问:为什么不一次性让模型输出完整结果?因为现实任务大多不是“给定输入、吐出输出”的问答,而是多个子任务串起来的流程。比如让AI整理资料:它得先看资料内容是否杂乱,再决定清洗哪些字段,然后归档、写摘要,最后汇总成报告。如果只调用一次模型,它没有机会看到中间结果,也就无法根据实际情况随机应变。循环的价值就在于:每一步的结果都会成为下一步的决策依据,形成一个真正的反馈闭环。
1.2 为什么“循环回写”是灵魂所在
Loop Engineering和普通代码循环最大的区别,是“观察结果回写上下文”。普通while循环写死了执行逻辑,但Agent循环里每一步做什么,是由模型根据最新上下文临时决定的。
我举个生活化的例子:炒菜。普通程序就像你照着菜谱定时翻炒——第1分钟放盐,第3分钟出锅,不管锅里情况如何。而Loop Engineering就像一位厨师——下锅后观察颜色,闻闻香气,尝一口,觉得淡了就加盐,觉得生就再炒一会儿。这里的“尝一口”就是观察,“加盐”就是行动,观察结果直接影响下一轮决策。
所以实现Loop Engineering时,最重要的不是循环写法,而是你怎么把“每次执行完工具得到的结果”重新拼装成模型能理解的文本,塞回对话历史里。这一步做不好,循环就断了,模型就会瞎猜。
1.3 什么时候需要用Loop Engineering
不是所有场景都需要循环。判断标准很简单:任务是否存在“执行—反馈—再决策”的依赖链?
- 需要:多步骤数据处理、资料整理、自动化测试、带工具调用的问答、动态规划类任务。
- 不需要:单次翻译、单次摘要、单轮知识问答。
如果任务本身只需要一次模型调用,你硬套循环,只会增加延迟和成本。我见过不少初学者给所有功能都上Agent循环,结果杀鸡用牛刀,还容易死循环。先想清楚任务类型,再决定要不要加循环。
2. 保姆级环境准备:零依赖搭起最小可运行骨架
2.1 环境依赖清单
老规矩,先把环境跑通,再讲原理。这个项目我不打算让你装一堆重依赖,Python 3.8以上就够,核心代码只用标准库。
- Python 3.8+
- 无第三方包依赖(项目基础版)
- 文本编辑器或IDE,推荐VS Code
为什么不用第三方框架?因为Loop Engineering的核心逻辑其实很朴素:一个while循环、一个上下文列表、一套工具注册表。先把裸金属逻辑搞清楚,以后再用框架会容易得多。很多初学者上来就套现成框架,一旦出了问题,连日志都不知道怎么看。
2.2 Agent循环的最小骨架代码
下面这段代码是一个完整的Agent循环骨架,不依赖任何外部模型服务。为了让你在没有API密钥的情况下也能把整个流程跑通,我用一个模拟函数代替模型调用,重点展示循环调度逻辑。
import json # 模拟模型服务返回结构化动作 def llm_complete(messages): # 真实环境里这里换成你的模型服务调用 # 这里的逻辑:根据已对话轮次返回固定动作序列 call_index = len(messages) - 2 # 去掉system和初始user消息 if call_index == 0: return json.dumps({"action": "clean_text", "params": {"text_id": 1}}, ensure_ascii=False) elif call_index == 1: return json.dumps({"action": "save_note", "params": {"filename": "note_1.md"}}, ensure_ascii=False) elif call_index == 2: return json.dumps({"action": "summarize", "params": {"text_id": 1}}, ensure_ascii=False) else: return json.dumps({"action": "finish", "summary": "已完成一轮整理"}, ensure_ascii=False) # 工具注册表 TOOLS = {} def register_tool(name): def decorator(fn): TOOLS[name] = fn return fn return decorator # 工具实现示例 @register_tool("clean_text") def clean_text(text_id): return f"笔记{text_id}已清洗完成" @register_tool("save_note") def save_note(filename): return f"已保存到{filename}" @register_tool("summarize") def summarize(text_id): return f"笔记{text_id}的摘要内容" # 主循环 def run_agent_loop(user_request, max_steps=5): messages = [ {"role": "system", "content": "你是资料整理助手。请严格按JSON格式输出动作。"}, {"role": "user", "content": user_request}, ] for step in range(1, max_steps + 1): raw = llm_complete(messages) result = json.loads(raw) action = result["action"] print(f"第{step}轮 -> 行动: {action}") if action == "finish": print("最终总结:", result.get("summary")) return result.get("summary") tool = TOOLS.get(action) if not tool: messages.append({"role": "user", "content": f"未知动作: {action},请重新选择"}) continue try: observation = tool(**result.get("params", {})) except Exception as e: observation = f"工具执行失败: {e}" messages.append({"role": "user", "content": f"观察结果: {observation}"}) print(f"观察结果: {observation}") raise RuntimeError("超过最大轮数,循环强制终止") if __name__ == "__main__": run_agent_loop("请把笔记1清理、保存并生成摘要")你直接运行这段代码,会看到每一轮的“思考—行动—观察”过程。这就是Loop Engineering最核心的骨架:模型每次返回一个结构化动作,程序执行动作,把结果作为观察信息追加到上下文,然后进入下一轮。
2.3 关键设计:为什么用JSON约束模型输出
注意看模拟模型返回的内容,全是JSON格式:{"action": "clean_text", "params": {...}}。这不是随手写的格式,而是整套循环能跑通的关键。
模型输出如果不结构化,程序就无法解析“下一步要做什么”。让模型返回JSON,相当于在模型和程序之间建立了一个协议:action指明调哪个工具,params是传给工具的参数。在真实项目里,你还需要在System Prompt里明确写清楚JSON的格式规则,告诉模型哪些字段必须有、值域是什么、怎么表示终止。
我建议所有初学者都先在本地把模拟版本跑通,再去接真实模型服务。为什么?因为调试循环逻辑时,你希望错误源单一——如果接上真实模型后出了问题,你能确定是调度逻辑的问题还是模型输出格式的问题。用模拟器先验证调度逻辑,是效率最高的排错方式。
3. 核心细节解析:感知、推理、行动、观察四要素怎么做
3.1 感知:把用户目标变成模型可判读的上下文
“感知”在代码层面就是messages这个列表。你注入什么样的信息,模型就有什么样的感知边界。这里的门道很多。
第一,System Prompt要写清楚:你是谁、任务目标是什么、可用工具有哪些、输出格式要求是什么。第二,用户初始消息要包含足够的数据信息——如果你让AI整理某个文件,你得把文件内容、文件路径、目标目录都放进去,或者告诉它工具可以从哪里读取。第三,每一轮观察结果也要追加进messages,这样模型才能看到“我刚做的操作产生了什么效果”。
我见过最典型的问题就是:模型在循环里反复调同一个工具,因为它看不到前一轮已经做过这件事了。原因往往是观察结果没回写,或者回写了但信息太冗长,模型没注意到。感知环节要控制上下文的信息密度:不要把几百行日志全部塞进去,让模型读不到重点。
3.2 推理:让模型输出可解析的决策
推理环节就是模型调用本身。你要做的不是简单地“把请求发出去”,而是设计好输出协议。
这里有一个特别容易踩的坑:不要用自然语言让模型“随便输出”,然后尝试用正则解析。一旦模型回复的格式稍微变一下,你的解析器就崩了。正确做法是:
- 在System Prompt中给出严格的JSON Schema示例。
- 要求模型不要输出多余文字,只输出JSON。
- 代码里做好容错:解析失败时,把错误信息作为观察结果回传给模型,让它自己修正格式。
def safe_parse(raw): try: return json.loads(raw) except json.JSONDecodeError: return {"action": "format_error", "message": f"模型输出无效JSON: {raw}"}在真实项目中,我还会在请求参数里设置temperature较低(比如0.2以下),让模型输出更稳定。推理环节的目标不是让模型创新,而是让它做出可靠的下一步决策。
3.3 行动:工具注册与参数校验
工具注册表是整个循环里最务实的一环。上面的代码用了装饰器模式注册工具,每个工具就是一个普通函数。随着工具数量增多,你需要补上两件事:参数校验和错误捕获。
参数校验很重要。模型返回的params是不可信的,它可能把字符串传成数字,可能漏掉必填字段。在调用工具之前做一遍校验,比在工具内部散落一堆防御代码要清晰。校验失败时,不要直接抛异常终止循环,而是把校验失败信息作为观察结果回传给模型,让它重新决策。
错误捕获同理。工具执行失败是常态,比如文件不存在、网络超时、参数越界。要把异常信息转换成模型能理解的文本,追加到messages里。模型看到障碍后,通常会改变策略——这正是循环的价值。
3.4 观察:把工具结果变成下一轮的决策依据
观察环节做得好不好,直接决定循环会不会“原地打转”。工具返回值往往很“脏”:可能是完整文件内容、大段日志、数据库查询结果。你不能原封不动塞给模型,而要做一个提炼。
我常用的策略是:
- 截断长文本:超过500字符的部分用“内容过长已省略”代替。
- 提取关键指标:比如处理成功、生成了几个文件、匹配到几条记录。
- 保留错误信息:失败时一定要原样或概括性保留错误原因。
def format_observation(tool_result, max_len=500): text = str(tool_result) if len(text) > max_len: return text[:max_len] + "...(内容过长已截断)" return text观察环节还有一个容易被忽略的点:观察结果应该以独立消息追加,不要覆盖之前的消息。这样模型能看到完整的决策链——它知道自己做过什么、结果如何、现在是第几步。
3.5 循环收敛:怎么判断任务真正结束
循环必须有明确的终止条件,否则就是灾难。终止条件通常有几种:
- 模型主动输出finish动作。
- 达到最大轮数上限,强制终止。
- 外部信号终止,比如用户手动中断、成本超限。
我建议多种条件并行。即使模型认为自己已经完成,你也要做结果校验——比如检查目标文件是否真的存在、数据是否达标。因为模型判断“完成”有时候只是幻觉,它并没有真正验证过。
if action == "finish": final_status = verify_work(result.get("summary")) if final_status.get("valid"): return final_status else: messages.append({"role": "user", "content": f"结果校验未通过: {final_status['reason']},请继续处理"}) continue这个“结果校验强迫症”是我踩过坑后养成的习惯。早期我做Agent,模型说“已完成”,我直接收工,结果用户拿到报告后发现文件根本没生成。从那以后,任何finish都要经过验证才能放行。
4. 项目实战:从零实现一个资料整理循环Agent
4.1 项目目标与整体设计
现在我把上面所有知识点整合成一个能跑出真实结果的项目。场景:你有一批散乱的笔记,每段笔记可能包含多余空行、乱码符号、无价值碎碎念。Agent需要在循环中完成:逐条清洗笔记、把清洗结果保存到独立文件、生成摘要、最后输出一份汇总报告。
这个任务的巧妙之处在于:它天然需要多步骤协调,而且每一步的结果都影响下一步。清洗前,Agent得知道有哪些笔记;清洗后,才能做摘要;摘要后,才能写汇总。整个过程虽然是模拟的,但流程设计完全贴近真实业务。
整体设计分四层:
- 数据层:NOTES字典存放原始笔记。
- 工具层:清洗、保存、摘要、汇总四个工具。
- 调度层:run_agent_loop主循环。
- 接口层:模拟模型服务(可替换为真实API)。
4.2 数据与工具实现
import json import os import re NOTES = { 1: " 今天天气不错 \n \n 但我的笔记很乱。。 \n\n\n 决定整理一下", 2: "第二个笔记 测试用 \n 内容。有。很多。标点", } def clean_text(text): text = re.sub(r"\s+", " ", text).strip() text = re.sub(r"[。]+", "。", text) return text def save_to_file(content, filename): with open(filename, "w", encoding="utf-8") as f: f.write(content) return f"已保存到{filename}" def quick_summary(text): sentences = re.split(r"[。!?]", text) valid = [s for s in sentences if len(s) > 2] return "。".join(valid[:2]) + "。" def build_report(items): lines = ["# 资料整理汇总报告", f"共处理笔记 {len(items)} 条", ""] for item in items: lines.append(f"## {item['filename']}") lines.append(f"摘要: {item['summary']}") lines.append("") return "\n".join(lines)注意quick_summary这个工具,它是纯规则实现,不需要模型参与。实际项目中可以用模型做摘要,但用规则工具的好处是:演示循环时,你不需要额外调用模型,逻辑更清晰。真实场景中,把工具实现替换成模型调用即可。
4.3 带工具注册表的循环调度器
这部分是真正的“工程核心”。我用字典记录工具返回的结果,让模型能在后续轮次中引用之前的结果。
TOOLS = {} MEMORY = {} def register_tool(name): def decorator(fn): TOOLS[name] = fn return fn return decorator @register_tool("clean_note") def clean_note(note_id): note_id = int(note_id) if note_id not in NOTES: raise ValueError(f"笔记{note_id}不存在") cleaned = clean_text(NOTES[note_id]) MEMORY[f"cleaned_{note_id}"] = cleaned NOTES[note_id] = cleaned return f"笔记{note_id}清洗完成: {cleaned[:50]}" @register_tool("save_result") def save_result(note_id, filename): note_id = int(note_id) content = MEMORY.get(f"cleaned_{note_id}") if not content: raise ValueError("该笔记尚未清洗,请先执行clean_note") path = save_to_file(content, filename) MEMORY[f"file_{note_id}"] = path return path @register_tool("make_summary") def make_summary(note_id): note_id = int(note_id) content = MEMORY.get(f"cleaned_{note_id}", NOTES.get(note_id)) summary = quick_summary(content) MEMORY[f"summary_{note_id}"] = summary return f"摘要: {summary}" @register_tool("generate_report") def generate_report(): items = [] for note_id in sorted(NOTES.keys()): items.append({ "filename": MEMORY.get(f"file_{note_id}", f"note_{note_id}.md"), "summary": MEMORY.get(f"summary_{note_id}", ""), }) report = build_report(items) with open("final_report.md", "w", encoding="utf-8") as f: f.write(report) return f"汇总报告已生成: final_report.md"这里有个细节:save_result工具依赖clean_note先执行,如果模型顺序搞错了,工具会报错“该笔记尚未清洗”。这个错误会作为观察结果回传,模型在下一轮就会调整动作顺序。这就是循环工程比线性脚本强的地方——它允许出错的步骤发生后,通过反馈机制自我纠正。
4.4 模拟模型服务的动作编排
为了让项目演示能自动跑完整套流程,我写一个模拟模型,它会根据当前轮次返回合适的动作序列。这个序列正好模拟了一个聪明模型应该做的决策:先清洗,再保存,再摘要,最后生成报告。
def llm_complete(messages): history_len = len(messages) if history_len == 2: return json.dumps({"action": "clean_note", "params": {"note_id": 1}}, ensure_ascii=False) elif history_len == 3: return json.dumps({"action": "save_result", "params": {"note_id": 1, "filename": "note_1.md"}}, ensure_ascii=False) elif history_len == 4: return json.dumps({"action": "make_summary", "params": {"note_id": 1}}, ensure_ascii=False) elif history_len == 5: return json.dumps({"action": "clean_note", "params": {"note_id": 2}}, ensure_ascii=False) elif history_len == 6: return json.dumps({"action": "save_result", "params": {"note_id": 2, "filename": "note_2.md"}}, ensure_ascii=False) elif history_len == 7: return json.dumps({"action": "make_summary", "params": {"note_id": 2}}, ensure_ascii=False) else: return json.dumps({"action": "generate_report", "params": {}}, ensure_ascii=False)真实项目中,这个函数换成大模型API调用,动作序列就是模型自己“思考”出来的。用模拟器先跑通,是为了让你看清楚每一轮的消息数量变化如何影响模型决策。
4.5 主循环升级版:加入结果校验与容错
主循环在骨架版基础上增加两个能力:解析失败自动纠正、finish前的结果校验。
def run_agent_loop(user_request, max_steps=12): messages = [ {"role": "system", "content": "你是资料整理助手。可用工具: clean_note、save_result、make_summary、generate_report。输出JSON格式动作。"}, {"role": "user", "content": user_request}, ] for step in range(1, max_steps + 1): print(f"\n--- 第{step}轮 ---") raw = llm_complete(messages) try: result = json.loads(raw) except json.JSONDecodeError: messages.append({"role": "user", "content": "输出不是有效JSON,请重新输出"}) continue action = result["action"] params = result.get("params", {}) print(f"行动: {action} | 参数: {params}") if action == "generate_report": report_path = TOOLS["generate_report"]() with open(report_path, encoding="utf-8") as f: report = f.read() if "共处理笔记 2 条" in report: print("结果校验通过,任务完成") return report else: messages.append({"role": "user", "content": "报告内容不完整,请重新生成"}) continue tool = TOOLS.get(action) if not tool: messages.append({"role": "user", "content": f"未知工具:{action},请从可用工具中选择"}) continue try: observation = tool(**params) except Exception as e: observation = f"工具执行失败: {e}" messages.append({"role": "user", "content": f"观察结果: {observation}"}) print(f"观察结果: {observation}") raise RuntimeError("超过最大轮数,任务未完成")运行后会输出的效果大致如下:
--- 第1轮 --- 行动: clean_note | 参数: {'note_id': 1} 观察结果: 笔记1清洗完成: 今天天气不错 但我的笔记很乱。决定整理一下 ... --- 第5轮 --- 行动: generate_report | 参数: {} 结果校验通过,任务完成这样的输出,你一眼就能看清每一轮发生了什么。我在调试真实Agent时,也会保持这种“每轮行动和观察都打印”的习惯,排查问题非常高效。
5. 常见问题与排查技巧实录
5.1 死循环怎么破:三层防护一个都不能少
死循环是Loop Engineering翻车率最高的问题。模型在一个错误动作里反复打转,比如一直调用同一个失败工具,或者不停输出finish但结果一校验就失败。
我的建议是三层防护:第一层,代码级最大轮数限制,超了就强制终止;第二层,动作去重检测,如果连续N轮输出相同动作且参数相同,直接切断并提示模型换策略;第三层,成本监控,真实部署时统计token消耗,超过阈值自动熔断。
我曾经跑过一个自动化任务,模型因为工具返回的日期格式不符合预期,连续五轮都在重试同一个解析动作,直到我用最大轮数强制终止才停下。事后分析,是工具报错信息太笼统,模型根本不知道要怎么改。
5.2 上下文膨胀:模型越来越“瞎”
循环每跑一轮,观察结果就往messages里追加一条。几十轮之后,messages可能膨胀到几万字,模型反而找不到关键信息。
对付这个问题有两个思路:一是做上下文压缩,把历史消息中不重要的内容缩写成一行;二是只保留最近N轮消息,更早的决策过程用一段摘要代替。我在实际项目中常用第二种,简单直接。
另一种膨胀来自工具输出过长。一定要在观察环节做截断,把冗长的日志、文件内容、中间数据压缩后再回传。否则模型会忽略真正重要的信息,在无关内容上做文章。
5.3 模型输出格式不稳定:让循环自己纠错
模型偶尔会不按JSON格式输出,或者输出JSON但action是乱写的。不要让你的程序直接崩溃,而是把解析错误作为反馈信息回传:告诉模型“你输出的内容无法解析,请严格按JSON格式重新输出”。
这个“反馈纠错”机制本身就是Loop Engineering的体现——把模型本身的错误也当成一种需要观察并修正的状态。实际上,通过一两次纠错,大多数模型都能回到正轨。如果连续三轮格式错误,说明System Prompt写得不清楚,而不是模型有问题。
提示:System Prompt里的JSON示例要写全字段,最好直接放一个完整样例。我习惯在Prompt里放一段“好的输出示例”和一段“坏的输出示例”,模型基本不会再犯格式错误。
5.4 结果校验与任务终止的边界
判断“任务是否真的完成”是个哲学问题。模型认为完成,不代表实际完成。所以我的原则是:能自动校验就自动校验,校验不过就继续循环。
比如保存文件后,校验文件是否存在且有内容;生成摘要后,校验摘要是否覆盖了正文要点;汇报完成后,校验报告里是否包含所有任务项。校验失败时,把失败原因作为反馈回传,模型通常会补做遗漏的步骤。
这个习惯的价值在真实项目中甚至超过代码本身。很多“翻车现场”不是因为Agent没能力,而是因为缺少“做完检查一遍”的强制性机制。
6. 从骨架到真实项目的扩展经验
跑通上面这个项目,你已经掌握了Loop Engineering的核心。再往深走一步,我分享几条从骨架项目到真实项目必须跨过的坎。
第一,把模拟模型替换成真实API时,注意超时与重试。模型服务调用不像本地函数,可能几秒才返回,也可能超时。你的循环调度器要能容忍网络抖动,超时后把错误信息作为观察结果回传。
第二,工具设计要尽量“小而专”。一个工具只做一件事,参数明确,返回简短可读的结果。大而全的工具容易让模型混乱,也不容易复用。
第三,给每轮循环加上日志跟踪。我建议直接把每轮的思考摘要、动作、参数、观察结果都落到文件里,方便事后复盘。调试Agent不比调试普通代码,它带着一定随机性,单靠断点不好使。
最后分享一个我自己的体会:Loop Engineering上手不难,难得是控制闭环里的不确定性。你要时刻记住,循环里的每一步都是模型自己决定的,你的职责是设计好边界、协议、终止条件和反馈机制。程序员的角色不是替模型做决策,而是保证它无论怎么决策,都不会跑偏、不会失控、不会无限循环。这种“设计边界”的思维方式,才是Loop Engineering真正值得学的东西。