ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

LangGraph循环机制优化Agent工作流:从隐式循环到显式图

LangGraph循环机制优化Agent工作流:从隐式循环到显式图 1. 先搞明白一件事你的Agent卡在哪了先说结论绝大多数LangChain Agent跑demo没毛病一上生产就各种“不听话”问题几乎都出在运行循环上。LangChain里那个默认的AgentExecutor内部其实就是一个封装好的while循环——Agent思考一步、调用一次工具、拿到结果、再思考一步循环往复直到得出最终回答。听着没问题对吧但一旦你需要在中间插一步人工审核、需要控制最大调用次数、需要把某次会话中断后恢复接着跑这个“隐形循环”就变成了一堵墙。这也是LangGraph真正被需要的原因。它是LangChain官方团队出的图框架核心思路就是把Agent的运行结构从“隐形的while循环”改成“显式的节点边工作流”。节点是一个个有名字的动作边决定了动作之后下一步去哪个节点循环不过是其中一条“从工具节点指回Agent节点”的边而已。循环成为图形结构中的第一等公民你可以看到它、控制它、甚至给它加条件。这篇文章不是抄文档。我会从一个实际案例出发先把一个用LangChain AgentExecutor写的工作流拆开看再把它完整重构成LangGraph版本讲清楚每一步的取舍和为什么这么干。适合的人群也很明确已经跑通过LangChain的Agent但感觉控制不住节奏或者想在项目里加入人机协同、会话恢复、精确控制调用次数等功能的人。还没碰过LangChain的建议先补一点基础再来读这篇。2. 拆解LangChain Agent的运行逻辑与三个要命的痛点2.1 AgentExecutor内部到底在循环什么想理解LangGraph的价值得先知道LangChain Agent的默认执行器做了什么。在LangChain的经典实现中AgentExecutor接收一个Agent对象和一组Tools然后内部执行一个循环每一步大致是把对话历史、用户输入、工具描述组装成Prompt。让LLM输出一个Action决定调用哪个工具、传入什么参数或Final Answer决定直接回复。如果是Action就执行对应工具把Observation工具返回的结果追加到消息列表里。回到第2步再让LLM基于新的Observation继续决策。直到LLM输出Final Answer或者达到最大迭代次数抛出StopIteration。这里的第4步“回到第2步”就是循环。用大白话说Agent就像个跑腿员每跑一趟回来都要向主管汇报主管再决定是继续跑还是收工。AgentExecutor就是那个自动跑循环的主管但问题在于这个主管的内部逻辑是写死的你基本没法在“跑腿员汇报完”和“主管做决策”之间插入自己的流程。2.2 痛点一循环过程不透明调试全靠猜AgentExecutor把整个循环封装在内部中间的过程对开发者来说是个黑盒。我能看到最终输出但看不到“为什么它要先调这个工具再调那个工具”也不知道它在第几步开始跑偏。一旦工具调用链出了问题最常见的排查手段就是打开debug日志看每一步的输入输出日志一长就看花了眼。LangGraph改掉的就是这件事。每个节点有明确的名字每一条边有明确的走向我可以在任意两个节点之间插入打印、埋点、甚至断点。工作流是画出来的不是猜出来的。2.3 痛点二想插一段人工审批AgentExecutor直接傻眼AgentExecutor的循环是自动跑到底的它没有“暂停”这个概念。但真实业务里人工介入是刚需。比如一个Agent准备调用“发送营销邮件”这个工具谁乐意让它直接发出去我需要在工具执行前加一道人工确认确认完再继续跑。在AgentExecutor里实现这个需求很别扭。有人用“工具内部写input()等待输入”这种法子本地玩玩可以部署成服务就完全没法用。LangGraph天生支持在任意节点处加interrupt把图的执行暂停等人类确认后再从暂停处恢复。原因在于它本身就是一个有状态图节点的执行状态是保存在State里的暂停和恢复不过是状态机的自然操作。2.4 痛点三已有会话无法恢复用户一问三不知我见过一个比较典型的场景Agent执行过程中后端服务突然重启或网络抖动整个会话就这么断了。用户回头再问一句“刚才那个结果出来了吗”Agent一脸茫然。本质上是因为默认的AgentExecutor没有持久化机制所有中间状态都在内存里进程一挂就全没了。LangGraph的解决方式是Checkpointer。每隔一步把State完整保存到内存、数据库或文件里下次通过同一个thread_id进入时可以把完整状态读回来接着上次没跑完的图继续跑。这个能力在AgentExecutor里几乎等于要自己重新造一套轮子。3. LangGraph的循环机制为什么是“设计出来的”而不是“写出来的”3.1 节点、边、State三者如何组成一张工作流图LangGraph的核心抽象只有三个节点Node、边Edge、状态State。节点是所有逻辑的载体。一个节点可以是一个普通的Python函数接收当前State处理后返回一个更新后的State字典。节点内部干什么都行调LLM、执行工具、访问数据库、发HTTP请求。关键是节点要有一个名字因为边要靠名字来连接。边定义节点之间的走向。最普通的是普通边表示“执行完A节点之后无条件去B节点”。真正让LangGraph变得强大的是条件边它在A节点执行完后根据当前State的内容决定下一步去哪个节点甚至可以走到多个分支。State是整个图的数据中枢。它通常是一个TypedDict字段根据自己的业务来定义最常用的是messages列表用来累积所有对话和工具调用信息。每次节点返回一个字典LangGraph会把字典里的字段合并进整体State实现状态的迭代更新。简单理解State就是那个所有节点都能读写的小黑板整个工作流的记忆全挂在这块黑板上。3.2 把“循环”画成图谁看了都明白传统AgentExecutor里那个“回到第2步”的隐性循环画成LangGraph之后大概是这样的结构入口节点是Agent负责让LLM思考并决定动作。Agent之后接一条条件边如果LLM输出了工具调用指令走Tools节点如果输出的是最终回答走END节点。Tools节点执行完工具后接一条普通边回到Agent节点。看到了吗从Tools指回Agent的那条普通边就是循环本身。工具执行完状态回到Agent节点Agent基于新的Observation再决定下一步。循环不再是一个被隐藏的while语句而是图上一条可以看见、可以命名、可以单独做断点的边。3.3 从“顺序执行”到“有状态图”Workflow与Agent的分界线理解LangGraph时很多人会混淆Workflow和Agent。我的理解是这样的Workflow是固定的流水线节点和边的连接关系在编译后就不变了最多在条件边上根据输入走不同分支。它适合流程清晰、步骤固定的业务比如文档处理流水线。Agent则强调动态决策。每一步究竟走哪个分支不是预先写死的而是LLM根据当前状态实时决定的。LangGraph完全能承载两者区别只在于条件边的那套判断逻辑是规则写死的还是LLM驱动的。标题里“循环机制优化工作流”这句话正是LangGraph最核心的用法你把循环显式地画在图上然后所有Agent系的特性动态决策、工具调用、知识检索都放在这个可控的循环骨架里跑。既保留了Agent的灵活性又拿到了Workflow的可控性。4. 实战把LangChain Agent完整重构成LangGraph循环工作流4.1 先写一个最朴素的LangChain Agent作为对照组为了说明问题我先用LangChain经典的AgentExecutor写一个简单的助手它能联网查天气、能做加减乘除四则运算还会根据情况决定要不要调用工具。代码本身不复杂适用于LangChain 0.2以上版本from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper from langchain_core.tools import tool from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder tool def add(a: float, b: float) - float: 两数相加 return a b tool def multiply(a: float, b: float) - float: 两数相乘 return a * b tools [add, multiply] llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个数学助手需要计算时就调用工具。), MessagesPlaceholder(chat_history), (human, {input}), MessagesPlaceholder(agent_scratchpad), ]) agent create_tool_calling_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue, max_iterations5) result executor.invoke({input: 请计算 (23 67) * 2 的结果}) print(result[output])这个版本工作正常但注意AgentExecutor接收的参数max_iterations5是唯一的循环护栏。循环内部跑了多少步、每步决策是什么、在哪个节点耗时最长你只能靠verboseTrue打日志才能看到。它适合快速验证想法一旦涉及复杂业务越往后越难控制。4.2 LangGraph版节点、条件边、循环边逐一落地下面我用LangGraph重写同一个Agent。先定义状态再定义节点最后把图画出来。需要安装langgraph库pip install langgraph langchain-openai完整代码如下我加了详细注释import operator from typing import TypedDict, Annotated, Literal from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition from langgraph.checkpoint.memory import MemorySaver # ---------- 1. 定义状态 ---------- class AgentState(TypedDict): messages: Annotated[list, operator.add] # 消息列表自动累加 # ---------- 2. 定义工具 ---------- tool def add(a: float, b: float) - float: 两数相加 return a b tool def multiply(a: float, b: float) - float: 两数相乘 return a * b tools [add, multiply] # ---------- 3. 定义Agent节点 ---------- def agent_node(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) result llm_with_tools.invoke(state[messages]) return {messages: [result]} # ---------- 4. 组装图结构 ---------- builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, agent) # 条件边agent节点根据输出决定去tools还是直接结束 builder.add_conditional_edges( agent, tools_condition, ) # 循环边工具执行完回到agent这就是循环的核心 builder.add_edge(tools, agent) # ---------- 5. 编译图 ---------- graph builder.compile(checkpointerMemorySaver()) # ---------- 6. 执行 ---------- config {configurable: {thread_id: demo-001}} result graph.invoke( {messages: [{role: user, content: 请计算 (23 67) * 2 的结果}]}, configconfig, ) print(result[messages][-1].content)和AgentExecutor版本一对比结构上的差别很明显agent节点负责“思考”tools节点负责“执行动作”从tools指回agent的那条边就是循环。LLM一旦决定调用工具工作流就会沿着这条边转一圈执行完工具后带着Observation回到agent节点继续思考直到LLM输出不再包含工具调用的回答条件边才把它导向END。4.3 调试循环过程看一看出色的执行轨迹LangGraph比较贴心的一个点是编译后的图自带get_state()和流式输出方法。我可以直接查看每一步执行后的State内容而不用去翻杂乱的日志。比如执行完上述代码后我可以在命令行里打印整张图的结构from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png()))这样能直观看到节点的连接方式尤其能看到从tools回到agent的那条回路。没有这条回路Agent就没法执行多轮工具调用工作流直接退化成一个“调一次工具就结束”的半残版本。4.4 给循环加护栏拦截死循环和暴力调用图结构给你极大的灵活性但没有护栏的循环是会跑的。LangGraph有一个全局参数叫recursion_limit限制编译图最多执行多少步。默认值对一般对话够用但一旦工具数量多、LLM抽风它可能在一个循环里反复调用同一个工具几十次白白消耗Token。我习惯在调用时显式设置这个值result graph.invoke( {messages: [{role: user, content: 请计算 (23 67) * 2 的结果}]}, config{configurable: {thread_id: demo-001}, recursion_limit: 20}, )20的含义是图上所有节点累计最多执行20次。如果Agent实在绕不出来了LangGraph会直接抛一个GraphRecursionError。这个报错说明图的设计循环次数不够或者LLM在反复做无效决策需要回到Prompt层面优化。5. 循环之外的高级玩法人机协同与会话恢复5.1 interrupt机制在工具执行前加一道人工确认LangGraph能精准控制循环的启停这为人工介入创造了条件。比如我想加一个“人工确认后Agent才真正发送邮件”的节点可以在工具节点之前插入一个human_approval节点使用interruptfrom langgraph.types import interrupt def human_approval_node(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: # 暂停执行把待执行的动作抛给人类确认 approved interrupt({ question: 是否允许执行以下工具调用, tool_calls: last_message.tool_calls, }) if not approved: # 人类拒绝返回一条非工具消息替代原消息 return {messages: [{ role: assistant, content: 用户已拒绝工具调用请直接回答。, }]} return {messages: []}当工作流跑到human_approval_node时它会暂停并等待外部传入新的值。这就是图上循环中间踩了一脚刹车。业务上可以在前端弹窗等待审批审批通过后再调用graph.invoke(Command(resumeTrue))图会从暂停的地方继续跑下去。这个能力在AgentExecutor里需要靠全局变量、数据库状态位等方式自己造而在LangGraph里是一条原生的边和节点关系整个流程的上下文都保存在State里恢复之后一点不丢。5.2 MemorySaver和数据库Checkpointer会话断了也能续上checkpointer参数值得一提。上面的代码我只用了MemorySaver它把状态存内存里适合开发和单机Demo。生产环境建议用SqliteSaver或者接入PostgreSQL的AsyncPostgresSaver把每一步State持久化到真正的数据库里进程重启后依然可以从某个thread_id恢复之前的会话。一个很实用的操作是让Agent恢复执行。假设工具调用到一半超时了我重新发起一次完全相同的thread_id调用LangGraph会从最新的checkpoint继续而不是从头开始。这个体验和人工确认组合起来足以构建客服工单、审批流、定时任务恢复等真实业务的Agent后端。5.3 会话保持用户上下文不靠“塞进Prompt”很多团队在LangChain阶段会手动把历史会话塞进Prompt的chat_history字段每次请求都全量重发。LangGraph的State天然承载了消息列表配合Checkpointer只要传入同一个thread_id历史消息会自动累积在State里不需要你再做拼装。这个设计在长会话场景下的代码整洁度提升是肉眼可见的。6. 常见问题与排查技巧实录6.1 一运行就报GraphRecursionError怎么定位收到这个错误先别急着调大recursion_limit绝大多数情况下是Agent在循环里反复调用同一个工具。我先打印出最近几轮的状态消息看看工具的Observation是否一直在报错或者重复。state_snapshot graph.get_state(config) for msg in state_snapshot.values[messages][-10:]: print(f{msg.type}: {msg.content[:200]})看到最后几轮确实都是在调用同一个工具且结果相似时大概率是LLM把一个问题拆成了大量重复步骤或者工具返回的结果格式没法支撑下一次决策。我的处理习惯是给工具加更清晰的描述和输出规范同时在System Prompt里明确要求“步骤不超过3步能计算就直接给结果”。6.2 工具执行报错直接把整个工作流搞崩了LangGraph默认的ToolNode如果捕获到工具异常会把异常信息直接拼进消息让Agent看到之后自我纠正。但这个默认行为有时候不够友好。我习惯自定义一个tools_node用try-except包住工具执行把异常转成一条明确的Observation消息让Agent知道“刚才的调用失败了原因是什么可以换一种方法重试”。这样能让循环更稳定不会因为一次工具异常就导致整段对话断裂。6.3 messages列表无限膨胀Token越用越贵循环机制跑起来之后State里的messages会一直累积。到了第8轮工具调用时整个Prompt会变得很长Token消耗直线上升。我在生产中会做一个“消息压缩节点”放在agent节点之前检查消息条数和总字符数超过阈值就用LLM对历史消息做一次摘要替换掉过长的中间消息。这也是LangGraph图结构带来的好处压缩逻辑也能变成一个独立的节点单独测试、单独优化。6.4 从LangChain Agent迁移到LangGraph的通用节奏如果你手头已经有跑通的LangChain Agent我建议分三步走不要一次性推翻重写。第一步先原样保留AgentExecutor版本写一组涵盖核心场景的测试用例记录当前输出。第二步用LangGraph搭建一个结构相同的工作流把原来AgentExecutor里的Prompt、工具、LLM配置原样搬过来保证节点输入输出对齐。第三步在新旧两版上跑同一组测试用例对比输出质量和执行步骤。确认稳定后再把AgentExecutor下线。这样做的好处是LangGraph的调试和重构风险是可控的遇到问题能立刻回退到旧版本。7. 个人经验补充三个让我少走弯路的判断先说工具选择。LangGraph和LangChain不是替代关系LangGraph更像是在LangChain核心抽象之上长出来的编排层。如果你的项目只需要一个快速原型AgentExecutor完全够用一旦你开始频繁调工具、需要严格步骤控制、要跟人打交道尽早切换到LangGraph越晚迁移成本越高。再说循环设计的直觉。不要把LangGraph的循环想成“让程序反复执行”而是想成“让Agent保持思考直到完成任务”。图上那条从tools回到agent的边本质上是“拿到新信息后重新思考”的语义。每次设计循环边之前先问自己一句“回到这个节点后State有什么新变化”没有新信息就不该循环如果有那这一步循环就是有意义的。最后说调试心态。我见过不少人一上来就拖入LangGraph Studio做可视化调试其实上手期最有效率的调试方式就是print State。把关键节点的State打出来看消息流怎么变化比任何可视化工具都直观。等流程跑熟了再上Studio效率会高得多。LangGraph把Agent的循环从一团乱麻理成了一张能看懂、能控制、能恢复的图。这个设计带来的收益在AgentExecutor跑通第一个Demo时体现得还不明显等你开始做生产级工作流就会发现它几乎是绕不开的方向。
返回列表