【Loop Engineering】智能体Loop工程

你是否想过为什么你的 AI 智能体总是“差那么一点”它能完成任务但不够稳定它能输出结果但需要人工反复审核它能运行但无法融入你的工作流。答案可能在于你只构建了一层循环而优秀的智能体需要四层循环的协同工作。本文源自 swyx 的精彩文章《Loopcraft: The Art of Stacking Loops》我们将深入拆解这四层循环架构并结合 LangChain 的原生组件为你展示如何构建一个真正“生产级”的智能体系统。无论你是 AI 应用开发者、技术架构师还是对智能体工程化感兴趣的读者这篇文章都将为你提供可落地的架构思路。文章结构速览本文将依次介绍智能体循环 → 验证循环 → 事件驱动循环 → 爬山算法循环最后给出四层关系总结和最佳实践建议。Loop 工程的艺术以下是对这种“循环栈”的理解以及如何利用 LangChain 的原生组件Primitives来实现每个层级的架构。1.环层 1智能体循环 (Loop 1: The Agent)从本质上讲智能体就是一个模型在循环中不断调用工具直到任务完成。这就是 LangChain 的 create_agent 所实现的功能。挑一个模型接入工具你就得到了一个可以运行的智能体循环。正是工具赋予了智能体在现实世界中采取行动的能力。下面是一个最小可运行的智能体循环示例使用 LangChain 的create_react_agent实现fromlangchain_openaiimportChatOpenAIfromlanggraph.prebuiltimportcreate_react_agentfromlangchain_core.toolsimporttooltooldefsearch_docs(query:str)-str:搜索内部文档库returnf找到与{query}相关的文档片段请参考 docs/guide.mdtooldefwrite_file(path:str,content:str)-str:写入文件到仓库withopen(path,w)asf:f.write(content)returnf已写入{path}llmChatOpenAI(modelgpt-4o,temperature0)agentcreate_react_agent(llm,[search_docs,write_file])# 运行智能体循环——模型会自动规划、调用工具、直到任务完成resultagent.invoke({messages:[(human,请搜索智能体架构相关文档并将结果写入 summary.md)]})print(result[messages][-1].content)以我们内部的文档智能体为例我们将在接下来的文章中一直用它作为引导示例。在第一层循环中它接收到改进文档的请求模型开始规划并起草修改内容然后使用工具去克隆仓库、读取文件、编写文档、发起拉取请求Pull Request等。2.环层 2验证循环 (Level 2: Verification loop)智能体循环确实能把工作做完但它并不总能在第一次尝试时就产出正确或一致的结果。当工作质量的一致性至关重要时通常需要在其外部包裹一层验证循环用来检查输出并在结果不达标时将反馈送回模型。验证循环引入了一个评分器Grader它会根据既定的规则评分标准Rubric来检查智能体的输出。如果未通过就会带着反馈意见将结果打回重做。评分器可以是确定性的代码也可以是基于智能体的例如经典的LLM 担任裁判。下面是用 Python 实现验证循环的示例使用after_agent钩子接入评分逻辑fromtypingimportDict,Anyfromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,LiteralclassAgentState(TypedDict):messages:listattempts:intmax_attempts:intdefgrader(state:AgentState)-Literal[rework,accept]:评分器检查输出是否包含关键信息last_msgstate[messages][-1].content# 确定性评分规则checks[代码示例inlast_msg,配置说明inlast_msg,len(last_msg)200,]ifall(checks):returnaccept# 未通过则打回附带反馈state[messages].append((system,输出缺少必要内容请补充代码示例和配置说明))returnrework# 构建带验证循环的图builderStateGraph(AgentState)builder.add_node(agent,lambdastate:state)# 实际接入 create_react_agentbuilder.add_node(grader,grader)builder.set_entry_point(agent)builder.add_conditional_edges(agent,grader,{rework:agent,# 不合格 → 重新执行accept:END# 合格 → 结束})RubricMiddleware 组件专门处理这种模式或者你也可以通过 create_agent 上的 after_agent 钩子Hook来实现。在我们的文档编写智能体示例中评分器会在每次尝试后运行测试检查所有链接是否能正常访问、所有 CI持续集成检查是否通过以及代码差异Diff是否仅局限于被请求修改的范围。这样一来无需人工审核就能拦截这类常见错误。权衡利弊 引入验证循环会增加单次运行的延迟和成本。但当质量比速度更重要时这几乎涵盖了绝大多数生产环境的落地场景这么做是非常值得的。3.环层 3事件驱动循环 (Level 3: Event driven loop)智能体开发中最重要的环节之一是集成层Integrations Layer将你的智能体连接到现有的生态系统中使其能够在后台自主运行。事件驱动循环将你的智能体与生态系统紧密相连。一旦某个事件被触发比如上传了新文档、触发了定时任务、或者接收到了 Webhook网络钩子智能体就会开始运行。此时智能体不再需要你手动去调用而是作为一个持续运行的组件嵌入在一个更大的系统内。下面是用 FastAPI LangGraph 实现事件驱动智能体的示例支持 Webhook 触发和定时任务fromfastapiimportFastAPI,Requestfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.graphimportStateGraphimportasyncio appFastAPI()agent_graphNone# 接入你的 create_react_agentapp.post(/webhook/docs)asyncdefhandle_docs_request(request:Request):Webhook 触发收到 Slack 消息后启动智能体payloadawaitrequest.json()channelpayload.get(channel,#docs-plz)messagepayload.get(text,)config{configurable:{thread_id:fdocs-{channel}}}resultagent_graph.invoke({messages:[(human,f请处理文档请求{message})]},configconfig)return{status:processing,result:result[messages][-1].content}app.on_event(startup)asyncdefstart_cron_agent():定时任务每天凌晨 2 点自动检查文档更新asyncdefcron_job():whileTrue:nowasyncio.get_event_loop().time()# 实际项目中用 APScheduler 或 Celery Beatawaitasyncio.sleep(86400)# 每 24 小时agent_graph.invoke({messages:[(human,检查所有文档是否需要更新)]})asyncio.create_task(cron_job())LangSmith Deployment 支持这种触发器基础设施包括对 Cron 定时任务和 Webhook 的支持。其中一个广受欢迎的定时任务应用案例是 openclaw 中的心跳Heartbeats机制它能让你的智能体变成一个全天候在线、主动提供帮助的助手。我们的文档智能体是由我们免代码智能体构建工具 Fleet 驱动的。Fleet 的频道Channels和调度表Schedules可以完美处理事件驱动和 Cron 风格的触发器。我们设置了一个频道只要有人在我们的 Slack#docs-plz 频道中发送消息就会自动触发文档智能体去干活。4.环层 4爬山算法循环 (Level 4: Hill climbing loop)前三个循环实现了工作的自动化而第四个循环可以说也是最重要的一个实现了优化的自动化智能体的每一次运行都会产生一个追踪记录Trace其中记录了模型做了什么、调用了哪些工具、评分器的反馈等等。这些追踪记录包含了极具价值的信号能反映出哪些做法有效哪些无效。爬山算法循环Hill climbing loop会安排一个分析智能体去审查这些追踪记录并利用分析结果来重写和优化外壳框架的配置。优化的内容可以包括提示词/工具的微调或是评分标准的修改。下面是一个分析追踪记录并自动优化提示词的实现示例fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportPromptTemplate# 1. 收集追踪记录traces[{task:编写 API 文档,tools_used:[search,write],score:0.6,error:缺少参数说明},{task:编写 API 文档,tools_used:[search,write],score:0.7,error:格式不符合规范},{task:更新 README,tools_used:[read,write],score:0.9,error:None},]# 2. 分析智能体审查追踪记录analyzerChatOpenAI(modelgpt-4o,temperature0.3)analysis_promptPromptTemplate.from_template( 分析以下智能体运行追踪记录找出系统性问题并给出优化建议 追踪记录 {traces} 请输出 1. 高频错误模式 2. 建议修改的提示词片段 3. 建议调整的评分标准 )analysisanalyzer.invoke(analysis_prompt.format(tracestraces))print(分析结果,analysis.content)# 3. 自动优化——更新智能体的系统提示词defupdate_system_prompt(analysis:str)-str:根据分析结果自动重写系统提示词if缺少参数说明inanalysis:return(你是文档编写智能体。在编写 API 文档时必须包含以下部分\n- 接口描述\n- 请求参数含类型、必填、说明\n- 响应示例\n- 错误码说明)return你是文档编写智能体请按规范输出。new_promptupdate_system_prompt(analysis.content)print(优化后的系统提示词,new_prompt)在 LangChain 中你可以使用我们专门用于追踪分析的智能体 Engine来落地这第四层循环。回到文档智能体的例子我们让 Engine 审查文档智能体的追踪记录以检测潜在问题。当多个追踪记录都释放出同一个潜在问题的信号时系统就会自动提交一个 Issue请求对出错的提示词或工具进行修改。这里的关键一着在于返回的箭头不仅仅是循环回顶部而是直接深入系统内部更新了智能体循环本身。外层循环的每一次迭代都会让内层循环变得更加高效。展望未来提示词和工具的配置是最容易进行自动改进的但它们绝非唯一的选择。对于运行开源/权重开放Open-weight模型的团队来说爬山算法循环可以接入强化学习微调RL Fine-tuning将追踪记录或评估结果作为训练信号来直接提升模型本身的能力。诸如记忆Memory和检索到的技能Retrieved skills等辅助上下文也可以通过同样的方式进行优化。循环是一种模式至于具体优化什么完全取决于你。上述4层的关系梳理环层1 和 2主动调用模式顺序关系1产生原始输出2检查输出质量反馈修正环层 3触发方式变了不是顺序关系是一种事件驱动的方式环层 4分析前三个循环产生的追踪记录Trace反过来优化环层1的配置提示词、工具、评分标准