多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

Loop Engineering实战:从零构建带反思循环的AI Agent调研系统

Loop Engineering实战:从零构建带反思循环的AI Agent调研系统 做 AI Agent 应用这一年多我先把话放在前面如果只让我挑一个最影响项目成败的工程概念我会选 Loop Engineering中文圈里现在也常叫它“循环工程”。起因是我接的一个竞品调研项目翻车了。第一版我天真地以为把需求写进一个 Prompt 调用大模型就能自动产出一份靠谱报告结果产品上线没几天就被用户追着吐槽信息全是旧的、数据有编造成分、报告读起来像高中生临时拼凑的作文。后来我把整个链路拆开来看才发现问题根本不在模型不行而在我完全没搞懂“让模型在一个任务上反复循环直到干完”这件事。Loop Engineering 做的事情简单说就是把“推理—行动—观察—再推理”这个循环本身当作核心工程对象去设计、调优和维护。它解决的是 LLM 单次调用能力很强、但连续完成多步骤任务时容易跑偏、忘记上下文、不会自我纠错的问题。这篇文章不打算讲太多虚头巴脑的概念我会用一套完整的保姆级实战带你看清楚一个带反思循环的调研 Agent 是怎么从零写出来的也把我在真实项目里踩过的坑、总结出来的经验一并分享。无论你是刚接触大模型应用开发还是已经写过不少自动化脚本这篇文章应该都能给你一些可以直接拿去用的思路。1. 一个竞品调研 Agent 的翻车现场一次 Prompt 调用不够用1.1 首版实现把所有需求塞进一次 Prompt我一开始的做法非常“新手友好”就是用一条超长 Prompt 承载所有需求。简单来说把调研目标、要分析的维度、输出格式全部写进去然后期望大模型“一次出活”。伪代码如下from llm_sdk import complete def first_version(company_names: list[str]) - str: prompt f 你是资深行业分析师。请调研以下竞品{company_names} 分析维度包括产品功能、定价策略、客户评价、融资动态。 输出一份 2000 字中文调研报告每个维度都要有数据支持。 response complete(prompt, max_tokens4000) return response初看完全没问题Prompt 里约束给得很全模型回答得也像模像样。但一旦落到真实业务里问题立刻暴露模型为了“完成任务”会自动脑补数据和事实因为它没有真正查询过任何外部信息源即使我后来把搜索工具接了进来也只是在中间插入了一轮“检索”检索结果和最终报告之间没有任何校验关系模型可能直接忽略搜到的内容整个流程是单向管道没有反馈回路模型不知道自己哪里没查清楚更不会回头补查。这个版本最后在评估集上的表现非常糟糕10 个竞品里有 6 个的功能描述明显过时3 个定价数据是错的还有 1 个公司名完全是模型幻觉出来的。这不是模型能力的问题而是流程设计的问题。1.2 翻车原因拆解LLM 没有“回头看”的能力为什么一次调用搞不定因为大模型的本质是“给定上文续写下文”它不是数据库不具备实时校验能力也没有主动“发现缺失—补齐缺口”的机制。你可以把它想象成一个表达能力很强但记性不太稳定的助手你问它“这家公司最近融资了吗”它只能依据训练时学到的知识回答而这个知识很可能是半年以前的。真正靠谱的调研流程应该是先确定调研问题围绕问题去查资料根据查到的资料判断线索是否足够不够就换关键词继续查最后基于所有证据写报告。这个流程天然就是一个循环。Loop Engineering 的核心就是把这种人类分析师的工作方式用代码和模型推理显式地组织起来。1.3 Loop Engineering 到底在解决什么问题Loop Engineering 这个名字近年在开发圈子里越来越常见你可以把它理解为一套“给大模型装配反馈回路”的工程方法。一个最小可用的循环包括四步模型根据当前状态生成下一步动作这个动作要非常明确比如选择搜索、读文件或执行代码系统执行这个动作拿到真实结果把动作结果返回给模型让它看到“世界真实发生了什么”模型把结果融合进上下文决定是继续循环还是输出最终答案。只要最后一个条件不满足循环就不结束。但它和传统软件工程里的循环很不一样LLM 循环每一步都是文本没有天然语义所以“停止条件”和“纠错机制”必须由工程师显式设计。这也是为什么我会说Loop Engineering 不是“写个 for 循环包住 API”而是一整套关于状态、终止、预算和稳定性的工程方法。2. 五种循环形态先搞清楚你要的是哪一种 Loop不要一上来就堆代码先看形态。Loop Engineering 里的“循环”其实有很多种不同循环解决的是不同层面的问题设计思路也不同。2.1 简单重试循环最基础的一种调用失败就重试。它通常用于处理瞬时错误比如网络超时、返回格式不合法、服务限流。简单重试循环只解决“偶发失败”的问题不解决“能力不足”的问题。写起来也很简单for attempt in range(3): try: result call_model(...) break except Exception: time.sleep(2 ** attempt)注意这里的退避策略我用的是指数退避第一次等 1 秒第二次等 2 秒第三次等 4 秒。这样不容易把本来就不稳定的服务打挂。2.2 工具调用循环这是 Agent 应用里最常见的一种模型决定调用哪个工具、传什么参数系统执行完把结果返回模型再根据结果决定下一步。典型场景包括让 Agent 去搜索网页、查询数据库、调用内部 API。工具调用的结果直接注入对话上下文成为下一轮推理的依据。这种循环最需要关注的是“工具返回质量”。如果搜索结果只是一堆杂乱摘要模型可能要多搜好几轮才能拿全信息如果返回的是结构化字段模型往往一两轮就能收敛。工具调用循环是整个体系的地基后面说的反思和记忆都建立在它之上。2.3 自我反思循环工具调用循环解决“拿到外部信息”的问题但 Agent 可能拿到一堆信息却不知道怎么用或者生成了一份质量很差的初稿。自我反思循环的做法是先生成一份结果然后换一个“批评者”视角重新审视结果找出缺失和错误再让模型带着批评意见重新生成。如今很多长文写作工具、代码生成工具内核都有类似机制。举个例子初稿的结论是“竞品 A 定价偏高”反思层可能会批评说“你没有给出具体价格区间也没有提到和竞品 B 的对比”。模型收到这个意见后会带着这个缺口去补全而不是自己盲目发挥。2.4 记忆聚合循环当任务被拆成很多步之后每一步的中间结果不能全部塞进上下文否则 token 会迅速爆炸。记忆聚合循环负责把分散的观察结果提炼成结构化记忆比如抽取出“竞品 A 价格区间、融资轮次、客户口碑关键词”存成摘要或 JSON 字段在下一轮需要时重新注入。这个循环往往不是单独存在的而是嵌入在工具调用循环和反思循环里。它的关键问题是“聚合的粒度”聚得太粗会丢信息聚得太细又失去省 token 的意义。我一般会按任务维度聚合比如调研项目就按“功能、定价、口碑、动态”四个槽位去维护。2.5 多智能体协商循环更复杂的架构里会有多个 Agent 循环互相“递话”一个 Agent 负责检索信息一个 Agent 负责质疑检索结果一个 Agent 负责汇总一个负责最终把关。它们之间形成一条多角色反馈环路。这个形态适合高难度任务但工程复杂度最高通信、终止、成本都可能失控一般项目不建议一开始就上。循环类型解决的核心问题典型场景终止条件设计重点简单重试循环瞬时失败网络抖动、格式解析失败最大重试次数、退避时长工具调用循环获取外部实时信息搜索、SQL 查询、订单操作工具是否完成、结果是否满足约束自我反思循环初稿质量不足长文写作、代码生成、报告生成反思轮次上限、评分阈值记忆聚合循环上下文过长、信息碎片化多步调研、长期任务记忆条目数量、覆盖度评分多智能体协商循环多角色协作复杂任务拆解与审查协作轮次、共识条件这里要特别强调一句五种循环不是彼此替代的关系而是经常层层叠加。一个生产级 Agent 往往就是“工具调用循环 自我反思循环 记忆聚合循环”的组合。下一章的实战项目就是这三者的组合。3. 保姆级实战从零实现一个带反思循环的调研 Agent3.1 环境准备与技术选型这个项目的目标很明确输入一个竞品公司名Agent 输出一份带来源引用和置信度的调研报告。技术选型上我用的是 Python 3.11LLM 调用统一走 OpenAI 兼容接口。为什么强调“兼容接口”因为 Loop Engineering 的核心工程对象是循环控制逻辑不是具体模型能力。把模型调用封装成一个薄接口后面想换模型、从付费切到本地开源模型都只用改一行配置不用动业务逻辑。真实项目里这个抽象非常重要因为不同模型在“反思”任务上的表现差异很大后续调优时几乎一定会换模型。依赖很少常规环境只需要两个库pip install openai python-dotenv我给 LLM 调用写了一个极简封装实际项目里你可以在这个基础上加超时、重试和 token 统计# llm.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client OpenAI( base_urlos.getenv(LLM_BASE_URL), api_keyos.getenv(LLM_API_KEY), ) class LLM: def complete(self, prompt: str, max_tokens: int 1000) - str: resp client.chat.completions.create( modelos.getenv(LLM_MODEL), messages[{role: user, content: prompt}], max_tokensmax_tokens, temperature0.2, ) return resp.choices[0].message.content这里我把 temperature 设置成 0.2。原因是生成动作指令和反思意见时我们希望结果稳定可复现过高的随机性会让循环在同样的状态上做出不同决策非常难调试。3.2 数据与工具层Agent 需要两个工具search_web(query)负责返回搜索结果列表每个结果包含 title、url、snippetsearch_news(query)聚焦新闻类信息。为了让你能直接跑通我写一个带缓存的伪实现。真实项目中你只需要把query_web_search换成自己对接的搜索 API 即可# tools.py import time from hashlib import md5 _CACHE: dict[str, list] {} def _cache_key(query: str) - str: return md5(query.encode(utf-8)).hexdigest() def query_web_search(query: str) - list[dict]: key _cache_key(query) if key in _CACHE: return _CACHE[key] # 这里对接你的真实搜索服务下面只是占位返回。 time.sleep(0.2) results [ { title: f关于 {query} 的示例结果, url: fhttps://example.com/search?q{query}, snippet: f这里是通过搜索接口返回的摘要内容真实项目中会来自外部索引。, } ] _CACHE[key] results return results def search_web(query: str) - list[dict]: return query_web_search(query)不要小看这个缓存。循环一旦跑起来模型很有可能会用几乎相同的关键词重复搜索。缓存不仅节省费用还能明显降低单次任务的延迟。我做得更狠一点会在缓存里同时记录搜索结果的时间戳超过一定时效就强制重查避免信息过旧。3.3 带反思循环的核心逻辑核心循环我拆成三段规划、执行、反思。为了让循环保持清晰我先把状态集中到一个 dataclass 里# agent.py from dataclasses import dataclass, field from tools import search_web dataclass class AgentState: company: str observations: list[str] field(default_factorylist) reflection_history: list[str] field(default_factorylist) iters: int 0 done: bool False final_report: str 状态集中在同一个对象里比散落在全局变量里好维护得多也方便统一加日志和预算统计。接下来是循环核心。每一轮我都让模型做一个显式决策SEARCH 关键词、REFLECT还是FINISH。注意这里我没有让模型自由写一段话而是强制它输出结构化的动作前缀这样后续解析和错误处理都更简单class ResearchAgent: MAX_ITERS 6 def __init__(self, llm): self.llm llm def step(self, state: AgentState) - AgentState: action self.llm.complete( f你是调研 Agent。目标是调研公司{state.company}\n f当前已有观察{state.observations}\n f请决定下一步动作只能输出以下三类\n fSEARCH 关键词 / REFLECT / FINISH ).strip() if action.startswith(SEARCH): keyword action.removeprefix(SEARCH ).strip() results search_web(keyword) state.observations.append( f[{keyword}] | .join(r[snippet] for r in results) ) elif action.startswith(REFLECT): critique self.llm.complete( f请审查当前的调研证据指出仍然缺失的关键信息\n{state.observations} ) state.reflection_history.append(critique) missing_query self.llm.complete( f根据上述批评给出一个最需要补充搜索的关键词\n{critique} ).strip() results search_web(missing_query) state.observations.append( f[反思补充] | .join(r[snippet] for r in results) ) elif action.startswith(FINISH): state.final_report self.llm.complete( f基于以下证据撰写调研报告每个结论都要标注对应证据编号\n{state.observations} ) state.done True state.iters 1 if state.iters self.MAX_ITERS: state.done True return state def run(self, company: str) - str: state AgentState(companycompany) while not state.done: state self.step(state) return state.final_report这段代码虽然简化了不少但已经具备一个循环工程的最小要素状态对象、动作决策、动作执行、观察结果回填、终止条件、最大轮数兜底。每一步之后模型都会“看到”真实的搜索结果而不是自己脑补。3.4 完整运行效果与输出示例跑起来之后日志大概是这样的省略了真实 API 输出只保留动作轨迹Iteration 1: SEARCH 某公司 最新融资 Iteration 2: SEARCH 某公司 产品定价 Iteration 3: REFLECT - 缺少客户评价建议搜索“某公司 用户口碑” Iteration 4: SEARCH 某公司 用户口碑 Iteration 5: FINISH最终报告会带着具体证据编号和来源链接用户能直接追溯到信息出处。有了这个基础版本我们才有资格去谈优化。我见过太多人上来就搭很复杂的 Agent 框架结果连最基础的循环都不稳定。4. 循环最容易失控的地方终止条件、成本与超时4.1 一次真实事故忘记 max_iters 的代价我在早期版本里犯过一个现在想起来都肉疼的错误为了让 Agent “尽量多搜索”我把最大轮数设得很大。结果某个客户案例里模型在“搜索→反思→再搜索”之间反复横跳所有轮次全部跑满中间还因为搜索接口偶发报错触发了大量重试。最后那个任务消耗的 token 是正常任务的三倍多而且报告质量并没有因此变好——最后几轮完全是在用相似的关键词换着花样搜索同一批结果。教训很直接循环必须终止而且终止条件不能只靠一个。你必须同时设好几层保险。4.2 终止条件设计三类条件一起上一个稳定的循环至少要有三种终止条件成功条件Agent 自己认为已经获得足够信息输出 FINISH。但依赖模型自身判断是不够的我会额外加校验规则比如最终报告必须引用至少 3 个不重复的信息源否则强制回到反思轮次最大轮数MAX_ITERS是硬性保险。我现在的默认值是 8宁可少跑也不许失控Token/时间预算循环开始前估算本轮可用 token 上限超预算就强制切换到“基于已有证据输出”模式。可以把这想象成开车成功条件是“到目的地”最大轮数是“油箱红线”token 预算是“导航里的预计到达时间”。三者缺一个你就有可能在路上瞎转悠。4.3 工具调用失败的退避策略与“换路”机制工具调用失败和模型调用失败要分开处理。模型调用失败通常用指数退避重试第一次等 1 秒、第二次等 2 秒、第三次等 4 秒到上限后放弃。工具调用失败则有两条路短时重试适用于临时故障比如搜索接口超时换路适用于工具本身不适用比如模型反复用同一个错误的公司名搜索结果永远为空。这时候应该让反思层介入提示模型换更宽泛的关键词或者改为搜索竞品所在的赛道。很多循环工程跑崩不是模型不行而是工具失败时的兜底逻辑没做。一个搜索接口超时可能导致整个循环卡死 30 秒然后重试三次再失败最终用户等来一片空白。这种体验一次就足以让人对整个 Agent 失去信任。4.4 反思循环的度什么情况该停止自我怀疑反思是一把双刃剑。好的反思能补全缺失信息过度的反思却会让 Agent 陷入“自我怀疑黑洞”永远觉得证据不够永远在生成补充搜索关键词最后把预算和时间全烧光。我的经验是给反思层加一个“预期收益判断”。具体做法是维护一个reflection_history列表当连续两轮反思生成的补充查询关键词高度相似时说明反思已经收敛不动了直接跳到 FINISH。代码层实现得粗糙一点可以直接比较字符串的相似度或重合度要精致一点就把关键词向量化后算余弦相似度。核心意图是不要让 Agent 为了反思而反思。5. 让循环稳定交付的工程化手段日志、状态机与逃生通道5.1 循环可观测性把思考过程变成可回放的日志Loop 的调试难度远高于普通代码。普通代码跑挂了有栈和报错信息循环跑歪了可能从头到尾都“正常”只是结果不对。所以第一件事就是把每一轮的核心信息落盘包括模型每次输出的决策原文动作类型和参数工具返回的关键字段当前轮次、累计 token、耗时。我习惯用 JSON Lines 按行写日志每条日志包含 iteration、phase、action、observation、cost、latency 等字段。后期可以用一个小脚本分析哪一步耗时最长、哪类动作失败率最高、哪个关键词被搜索次数最多。这些数据对优化循环的价值极高甚至可以反推出 Prompt 应该在哪一句话上调整。5.2 状态机化别让循环退化成多个 if 嵌套循环工程做得越多越会发现它本质上是一个“状态机 策略”问题。状态有PLAN、ACTION、OBSERVE、REFLECT、FINISH转移条件一部分来自代码一部分来自模型输出。如果把所有逻辑写成一大串 if/else三个月之后你自己都改不动。推荐做法是把状态转移收敛成一个step()方法每次循环只做一件事读当前状态、调模型、执行动作、更新状态、判断是否终止。这样单测会好写很多也方便在每一个状态入口加日志。实测下来状态机化的代码比散装 if 版本调试效率高出一大截。5.3 记忆如何闭环从循环中提炼长期记忆工具调用循环每轮都会产生新的观察结果如果全部原样堆进上下文字段很快 token 就爆了。这时候需要记忆聚合用一个独立的 reduce 步骤把前面几轮的观察压缩成结构化摘要。我的做法很务实每完成一轮搜索就让模型把“新增信息”合并进之前的记忆对象。比如调研项目的记忆可以设计成 JSON{ 功能: [], 定价: [], 口碑: [], 融资动态: [] }每轮搜索后把新的 snippet 交给模型让它只更新对应字段而不是把原文全部保留。这既减少了后续轮次的 token也让 Agent 的决策更聚焦。真实项目里这个记忆对象最好持久化到 Redis 或 SQLite。因为任务往往不会只跑一次下次调研同一家公司时可以直接复用记忆既省成本又更稳定。5.4 逃生通道与人工接管再强的循环也会遇到边界情况。比如搜索接口完全不可用、模型连续三次输出非法动作、所有轮次跑完但证据严重不足。在设计系统时我会提前预留三类逃生通道降级输出即使证据不足也强制生成一份明确标注“证据不足”的报告而不是返回空结果人工接管把 Agent 的中间日志和失败原因封装成一条消息发给运营人员由人来补充操作切换模型或工具如果主模型连续出错自动切换备用模型重新跑循环。逃生通道不是不信任 Agent而是承认单次循环有边界。生产系统必须接受这个现实否则一个异常就能把整条业务链路拖垮。5.5 成本与延迟预算循环的性价比公式最后给一个我一直在用的预估公式可以帮你在设计阶段就判断循环参数合不合理单轮成本 ≈ prompt_tokens / 1000 * 输入单价 completion_tokens / 1000 * 输出单价 循环总成本 ≈ 单轮成本 * 平均轮数举个具体例子一个调研任务平均跑 5 轮每轮大约消耗 3000 输入 token 和 500 输出 token。如果输入单价是 0.01 美元/1K token、输出单价是 0.03 美元/1K token那么单任务成本就是(3000 / 1000 * 0.01 500 / 1000 * 0.03) * 5 (0.03 0.015) * 5 0.045 * 5 0.225 美元这个数字不是用来精确记账的而是用来帮你看趋势当一个循环平均要跑 20 轮时你首先要优化的不是 Prompt 措辞而是工具返回质量。把搜索结果从稀疏片段换成结构化字段Agent 往往一两轮就能拿够信息这比任何 Prompt 技巧都管用。6. 从单个 Loop 到多 Loop 编排什么时候需要升级架构6.1 任务分解循环 执行循环的分层设计单 Agent 循环能解决单个任务但真实业务往往是“任务集合”。比如调研 10 个竞品、每个竞品再细分多个维度这时如果只靠一个循环顺序硬跑可能要跑几十轮延迟和成本都不可控。我实际用的做法是两层循环上层是任务分解循环负责把大目标拆成子任务清单并为每个子任务定义验收标准下层是执行循环每个子任务独立跑一套“工具调用 反思”循环把结果回传。上层循环跑完后如果发现某些子任务质量不达标会有选择地对那几个子任务重新调度。这个模式其实借鉴了操作系统里的分治思想先拆分再并行最后合并。工程上要注意上下层的通信协议一定要稳定最好都走统一的 JSON 格式不要给每个循环自定义一种返回结构。6.2 把循环配置化用一份 YAML 描述 Agent 行为循环逻辑稳定之后就不该频繁改代码了应该把策略参数外置。我现在是用 YAML 描述 Agent 行为启动时读取配置把它组装进AgentStateagent: name: research_agent max_iters: 8 terminal_conditions: min_sources: 3 max_tokens: 12000 tools: - search_web - search_news reflection: enabled: true max_reflections: 3这样做的好处是运营同学也能在不看代码的前提下调整轮数上限和反思开关项目可维护性高很多。我在实际项目里发现把这类参数从 hardcode 里解放出来之后调优效率提升非常明显因为你会敢于做实验把 max_iters 从 6 改成 8中间结果有什么变化这类对比测试只有在配置化之后才能低成本完成。6.3 什么时候完全不需要 Loop Engineering最后说一个反直觉的经验不是所有任务都需要 Loop Engineering。如果任务本身是一次性的比如“把下面这段文本翻译成英文”“把这段 JSON 转成 Markdown 表格”你加循环反而增加延迟和不确定性甚至让模型在简单任务上“过度设计”。我的判断标准很简单这个任务是否依赖外部信息或者多轮校验是否要求结果必须完整、可引用、可修正如果两个答案都是否直接用一次 Prompt 调用就好。Loop Engineering 是用来解决复杂任务的而不是用来给所有调用都披上一层花架子的。我个人现在接手任何 Agent 项目第一件事永远是问自己这个任务需要几轮循环每轮必须拿到什么证据才算完成终止条件写在需求文档里再动手写代码。这个习惯帮我省了太多莫名其妙的调试时间也希望你踩过几次坑之后能在一开始就把循环设计这件事当回事。
返回列表