LangGraph 实战教程:手把手教你搭建有状态的循环工作流
LangGraph 实战教程:手把手教你搭建有状态的循环工作流
关键词:LangGraph、有状态Agent、循环工作流、LangChain、图神经网络(提示词中的关系图、不是纯技术GNN)、状态机、Tool使用Agent
摘要:在大语言模型(LLM)Agent开发中,传统线性或条件分支的LangChain链难以处理需要上下文记忆迭代更新、多工具循环验证、复杂逻辑自主决策修正的任务——比如论文摘要翻译+术语一致性校验循环、Bug复现报告生成+复现脚本编写+验证循环、金融交易前风控规则逐条匹配+数据补充+重新匹配循环。2024年LangChain官方推出的LangGraph正是为解决这些痛点设计的框架:它基于有限状态机(FSM) 和有向无环图转有向循环图(DAG→DCG) 的设计理念,允许开发者显式定义Agent的状态结构、节点行为、边的跳转规则,甚至支持图中断点保存与恢复的强状态持久化能力。
本教程将从「为什么必须用LangGraph替代部分LangChain链」的问题背景出发,完全从零开始——不假设读者有任何LangGraph、FSM或Agent深度开发经验——通过「生活化类比拆解核心概念」「数学模型定义状态与转移规则」「Python代码从简单单循环到复杂嵌套循环」「金融风控、论文翻译、Bug复现三大完整生产级实战项目」「最佳实践与常见陷阱」「行业发展与未来趋势」这七个主线章节,每个章节严格保证超10000字的细节深度,手把手带你掌握LangGraph的精髓,最终能够独立搭建生产级的有状态循环工作流Agent。
第一章 问题背景与挑战:为什么LangChain不够,LangGraph非学不可?
核心概念
本节先铺垫后续所有内容的前提:先定义清楚「线性/分支工作流」「循环工作流」「有状态Agent」「无状态Agent」「有限状态机(FSM)」「LangChain LCEL链」「LangGraph图链」这些高频出现但容易混淆的概念,并用最简单的「快递员送疑难件」的生活场景,把所有概念串起来讲透。
核心概念1:无状态Agent vs 有状态Agent
无状态Agent:可以理解为「每次对话都是全新快递员」——你昨天让他送过A小区3单元101的快递,今天再让他送A小区的件,他完全记不住昨天101门口有个需要用蓝色门禁卡的物业岗亭;他也不会记得昨天给你送错了,你说他下次必须打电话确认收件人姓名;甚至他连自己刚才送了哪几家都不知道——每次任务都是孤立的、一次性的,只依赖当前的输入,没有任何历史积累。
从技术上讲,无状态Agent的典型代表是:
- 直接调用OpenAI的
gpt-4o-miniAPI,每次都只传当前的user_prompt,不传任何messages历史; - 用LangChain LCEL写的最简单的链:
prompt_template | llm | output_parser——这个链处理第一条输入后,内存里不会保存任何输出、中间变量、错误信息,处理第二条输入时,和第一次启动链的状态一模一样。
有状态Agent:则是「跟着你干了三年的固定专属快递员」——他记得所有A/B/C小区的门禁规则,记得你喜欢收顺丰生鲜要优先放楼下冷藏柜,记得你昨天漏收了今天下午要再补送,甚至记得上周送B小区的时候差点被狗咬,绕远路走西门更安全;他每次送件都会更新自己的「待送清单」「已送清单」「问题件备注」,这些清单就是他的「状态」——状态会随着每一步操作不断更新,直到任务结束。
从技术上讲,有状态Agent的典型代表是:
- 带
ConversationBufferMemory的LangChain链——这个链会把所有历史对话保存在内存里,每次调用都会拼接历史+当前输入传给LLM; - 但更高级的是——带结构化状态、显式状态更新规则、循环执行能力的Agent,也就是LangGraph要做的事情。
核心概念2:线性/分支工作流 vs 循环工作流
线性工作流:可以理解为「快递员的标准派件路线」——起点是快递站取件,然后依次是A1→A2→A3→B1→B2→B3→终点是下班打卡,路线是固定的、单向的、不会回头的,每一步的输入只依赖上一步的输出,中间没有任何停顿、没有任何修正、没有任何循环。
从技术上讲,线性工作流的典型代表是:
- 直接用LCEL写的论文翻译链:
prompt_template(输入中文论文摘要) | GPT-4o(翻译成英文) | StrOutputParser(输出字符串); - 这个链只能做「一步到位」的翻译,不会检查翻译后的术语是否和用户提供的术语表一致,不会修正翻译错误,不会处理LLM偶尔输出的无关内容(比如开头的「好的,我来帮你翻译」)。
分支工作流:可以理解为「快递员遇到小问题的应急分支」——标准路线是A1→A2→A3,但如果A2的收件人不在家,就走分支路线A2→物业代收→继续A3;如果A3的门禁卡坏了,就走分支路线A3→打电话给收件人→继续B1;分支路线的数量是有限的、预定义的、不会无限循环的,每一步的分支只依赖上一步的输出的某个判断结果(比如用LangChain的RunnableBranch来判断「收件人是否在家」)。
从技术上讲,分支工作流的典型代表是:
- 带术语一致性检查分支的LCEL论文翻译链:先执行翻译,然后用一个判断链判断翻译后的术语是否都在术语表中——如果是,直接输出;如果不是,走分支链让LLM修正术语后再输出;
- 这个链虽然比线性工作流强,但也有致命缺陷:如果修正后的术语还是有问题(比如LLM第一次修正错了),它不会再检查第二次——因为分支是预定义的、单向的,修正后只能直接输出;而且它的状态只有「历史对话」,没有专门的「术语错误列表」「已修正次数」「修正前后的原文/译文对比」这些结构化的状态——如果要加这些结构化状态,用LCEL的
RunnableWithMessageHistory虽然可以勉强实现,但代码会非常混乱、难以维护、难以扩展,更重要的是——无法显式控制循环次数、无法处理嵌套循环、无法保存断点恢复。
循环工作流:可以理解为「快递员处理疑难件的完整闭环」——疑难件是「收件人地址是A小区3单元101,但物业岗亭说这个地址是空的」,快递员的闭环路线是:
- 起点:疑难件标记为「待处理」;
- 节点1:检查自己的记忆(状态),有没有之前送过A小区3单元101的记录——如果有,跳转到节点2;如果没有,跳转到节点3;
- 节点2:给之前的收件人打电话确认地址是否变更——如果变更了,更新自己的待送清单(状态),跳转到节点4(重新送新地址);如果没有变更,跳转到节点3;
- 节点3:给寄件人打电话确认地址是否正确——如果正确,更新自己的问题件备注(状态),跳转到节点5(找物业经理查空置房记录);如果不正确,更新待送清单(状态),跳转到节点4;
- 节点5:找物业经理查空置房记录——如果确实是空置的,更新疑难件标记为「退回寄件人」,跳转到终点;如果不是空置的(比如物业岗亭的系统坏了),更新待送清单(状态),跳转到节点4;
- 循环条件:只有当疑难件标记为「已完成」或者「退回寄件人」时,才会跳转到终点;否则会一直循环执行上述节点——而且每次循环,快递员的记忆(状态)都会不断更新(比如已打过的电话、已查过的记录、已尝试过的送件次数),不会重复做无用功。
从技术上讲,循环工作流的典型代表就是我们后面要学的LangGraph图链——它可以显式定义:
- 结构化的状态字典:比如「疑难件状态(待处理/待核实地址/待查空置房/已完成/退回)」「寄件人信息」「收件人信息」「之前送过的记录列表」「已打过的电话列表」「已查过的记录列表」「已尝试过的送件次数」;
- 每个节点的具体行为:比如「检查记忆节点」「给寄件人打电话节点」「找物业经理查空置房节点」;
- 每个节点的输出更新状态的规则:比如「给寄件人打电话节点」如果返回「地址变更为A小区5单元202」,就更新状态字典里的「收件人信息」和「待送清单」;
- 边的跳转规则:比如「从检查记忆节点,如果有记录→跳转到给之前收件人打电话节点;如果没有→跳转到给寄件人打电话节点」;
- 循环终止条件:比如「当状态字典里的疑难件状态为已完成或退回时→跳转到终点」;
- 更高级的功能:比如「图中断点保存(比如快递员突然下班了,保存当前的状态字典,明天上班时继续执行)」「人类介入(比如物业经理也查不到空置房记录,这时候需要管理员介入,管理员处理完后更新状态字典,图继续执行)」「并行节点(比如同时给寄件人和物业经理发消息,而不是依次发,提高效率)」。
核心概念3:有限状态机(FSM)
有限状态机(Finite State Machine, FSM)是LangGraph的核心理论基础——在深入学习LangGraph之前,我们必须先把FSM的基本概念讲透,因为LangGraph的状态结构、节点行为、边的跳转规则,本质上就是一个「带外部工具调用、带人类介入、带结构化状态扩展」的FSM。
FSM的数学定义(先讲人话,再放公式)
先讲人话:有限状态机就是一个「有限个状态、有限个输入、每个状态在某个输入下会转移到另一个状态(或者保持原状态)、有一个初始状态、有一个或多个终止状态」的数学模型——我们刚才讲的「快递员处理疑难件的完整闭环」就是一个非常典型的FSM。
再放严谨的数学公式(LaTeX格式,看不懂没关系,后面会用生活场景和Mermaid流程图解释每一个符号):
FSM=(Q,Σ,δ,q0,F)FSM = (Q, \Sigma, \delta, q_0, F)FSM=(Q,Σ,δ,q0,F)
其中:
- QQQ:有限状态集合(Finite Set of States)——比如快递员处理疑难件的状态集合是{待处理q1,待核实之前收件人q2,待核实寄件人q3,待查空置房q4,已完成q5,退回寄件人q6}\{待处理q_1, 待核实之前收件人q_2, 待核实寄件人q_3, 待查空置房q_4, 已完成q_5, 退回寄件人q_6\}{待处理q1,待核实之前收件人q2,待核实寄件人q3,待查空置房q4,已完成q5,退回寄件人q6};
- Σ\SigmaΣ:有限输入字母表(Finite Input Alphabet)——比如快递员的输入字母表是{有之前送过的记录σ1,没有之前送过的记录σ2,收件人地址变更σ3,收件人地址未变更σ4,寄件人地址正确σ5,寄件人地址错误σ6,确实是空置房σ7,不是空置房σ8,管理员介入完成σ9}\{有之前送过的记录\sigma_1, 没有之前送过的记录\sigma_2, 收件人地址变更\sigma_3, 收件人地址未变更\sigma_4, 寄件人地址正确\sigma_5, 寄件人地址错误\sigma_6, 确实是空置房\sigma_7, 不是空置房\sigma_8, 管理员介入完成\sigma_9\}{有之前送过的记录σ1,没有之前送过的记录σ2,收件人地址变更σ3,收件人地址未变更σ4,寄件人地址正确σ5,寄件人地址错误σ6,确实是空置房σ7,不是空置房σ8,管理员介入完成σ9};
- δ\deltaδ:状态转移函数(State Transition Function)——这是FSM的核心,它的定义是δ:Q×Σ→Q\delta: Q \times \Sigma \rightarrow Qδ:Q×Σ→Q,意思是「给定一个当前状态q∈Qq \in Qq∈Q和一个当前输入σ∈Σ\sigma \in \Sigmaσ∈Σ,状态转移函数会返回一个下一个状态q′∈Qq' \in Qq′∈Q」——比如δ(q1,σ1)=q2\delta(q_1, \sigma_1) = q_2δ(q1,σ1)=q2,意思是「当前状态是待处理q1q_1q1,输入是有之前送过的记录σ1\sigma_1σ1,下一个状态是待核实之前收件人q2q_2q2」;
- q0q_0q0:初始状态(Initial State)——q0∈Qq_0 \in Qq0∈Q,意思是FSM启动时的第一个状态——比如快递员处理疑难件的初始状态是待处理q1q_1q1;
- FFF:终止状态集合(Final/Accepting States)——F⊆QF \subseteq QF⊆Q,意思是FSM执行到这些状态时就会停止——比如快递员处理疑难件的终止状态集合是{已完成q5,退回寄件人q6}\{已完成q_5, 退回寄件人q_6\}{已完成q5,退回寄件人q6}。
FSM的Mermaid流程图可视化(快递员处理疑难件的例子)
为了让大家更直观地理解FSM的数学定义,我们用Mermaid的状态图(State Diagram)来可视化刚才的「快递员处理疑难件的完整闭环」FSM:
这个Mermaid状态图完全对应了我们刚才的数学定义:
- QQQ:就是图中所有的状态节点(包括临时的重新送件q_temp和人类介入q_manual);
- Σ\SigmaΣ:就是图中所有的边的标签(比如有之前送过的记录σ1、送件成功σ_temp1等);
- δ\deltaδ:就是图中所有的边(比如从待处理q1指向待核实之前收件人q2的边,对应δ(q1,σ1)=q2\delta(q_1, \sigma_1) = q_2δ(q1,σ1)=q2);
- q0q_0q0:就是图中[*]指向的第一个节点(待处理q1);
- FFF:就是图中指向[*]的节点(已完成q5和退回寄件人q6)。
LangGraph对FSM的扩展(为什么说LangGraph是「带LLM的超级FSM」?)
刚才讲的FSM是「纯理论、无外部输入(除了预定义的输入字母表Σ)、无结构化状态扩展、无人类介入、无并行执行」的——而真实世界的LLM Agent开发场景,比如论文摘要翻译+术语一致性校验循环、Bug复现报告生成+复现脚本编写+验证循环、金融交易前风控规则逐条匹配+数据补充+重新匹配循环,都需要解决这些纯FSM的局限性。
LangChain官方推出的LangGraph,正是对传统FSM的六大核心扩展——这也是为什么LangGraph能够替代部分LangChain链,成为生产级有状态循环工作流Agent开发的首选框架:
- 扩展1:结构化状态字典替代有限状态集合Q:传统FSM的Q是「有限个离散的、预定义的状态标签」——比如待处理q1、已完成q5;而LangGraph的状态是「一个可扩展的Python字典(或者Pydantic模型,后面会讲推荐用Pydantic)」——字典里可以包含任意类型的数据:比如字符串、整数、列表、字典、Pydantic模型对象、甚至LangChain的Tool对象、LLM对象;而且状态字典的内容会随着每个节点的执行不断显式更新——你可以完全控制哪些字段要更新、哪些字段要保留、哪些字段要删除;
- 扩展2:LLM生成的输入/外部工具返回的输入替代有限输入字母表Σ:传统FSM的Σ是「有限个离散的、预定义的输入标签」——比如有之前送过的记录σ1、送件成功σ_temp1;而LangGraph的输入是「任意的:可以是LLM生成的文本、可以是外部工具返回的JSON、可以是用户的输入、可以是状态字典里的某个字段」——而且LangGraph允许你在节点里用LLM来动态生成输入的判断条件——比如你不需要预定义「术语是否一致」的判断规则,只需要写一个提示词让LLM来判断,LLM返回的结果(是/否/需要进一步检查)就可以作为边的跳转条件;
- 扩展3:带逻辑的节点函数替代状态转移函数δ:传统FSM的δ是「一个预定义的映射表」——比如给定q1和σ1,直接返回q2;而LangGraph的状态转移是通过「每个节点的Python函数」来实现的——节点函数可以做任意的事情:比如调用LLM、调用外部工具、更新状态字典、甚至根据逻辑动态生成下一个节点的名称;而且节点函数可以返回两个东西:一个是新的状态字典的部分内容(LangGraph会自动合并到旧的状态字典里,你不需要手动合并),另一个是下一个节点的名称(或者下一个节点的名称列表,用于并行执行);
- 扩展4:支持有向循环图(DCG)替代有向无环图(DAG):传统的LangChain LCEL链只能是「有向无环图(DAG)」——也就是说,链里不能有任何循环(否则会无限循环,或者需要用非常混乱的方式控制循环次数);而LangGraph从设计之初就支持「有向循环图(DCG)」——也就是说,你可以显式定义从某个节点跳转到之前的任意节点(比如从术语一致性检查节点跳转到翻译修正节点,再跳转到术语一致性检查节点,直到术语完全一致为止);而且LangGraph提供了「循环次数限制」的功能(比如最多循环10次,超过10次就强制终止),防止无限循环;
- 扩展5:支持人类介入(Human-in-the-Loop, HITL):真实世界的很多任务(比如论文摘要的最终审核、金融交易的最终决策、疑难Bug的复现方向确认)都需要人类介入——传统的LangChain LCEL链很难实现优雅的人类介入(比如需要手动保存状态、手动处理人类输入、手动重启链);而LangGraph从设计之初就支持「人类介入」的功能——你可以显式定义一个「等待人类输入的节点」,当图执行到这个节点时,会自动暂停,保存当前的状态字典;然后你可以通过LangGraph的API来获取当前的状态、提供人类的输入、更新状态字典;最后你可以通过LangGraph的API来继续执行图;
- 扩展6:支持图中断点保存与恢复(Checkpointing):真实世界的很多任务(比如长达数小时的论文全文翻译+术语一致性校验循环、长达数天的金融风险监控循环)都需要「断点保存与恢复」的功能——比如你的程序突然崩溃了、或者你的服务器突然断电了、或者你需要暂停任务去处理其他事情——传统的LangChain LCEL链几乎不可能实现优雅的断点保存与恢复(除非你自己手动编写大量的状态持久化代码);而LangGraph从设计之初就支持「图中断点保存与恢复」的功能——你可以显式定义一个「检查点存储器(Checkpoint Sink)」(比如内存中的字典、本地文件系统、Redis、PostgreSQL等),LangGraph会自动在每个节点执行完成后保存当前的状态字典到检查点存储器;如果任务中断了,你可以通过LangGraph的API从检查点存储器中恢复最新的状态,继续执行图;甚至你可以恢复到任意一个历史检查点,重新执行图(比如你对刚才的翻译结果不满意,可以恢复到翻译前的检查点,重新写一个提示词,再执行翻译)。
问题背景
大语言模型(LLM)Agent的发展历程(从简单到复杂)
为了让大家更深刻地理解「为什么LangGraph非学不可」,我们先来回顾一下LLM Agent的发展历程——从2022年11月OpenAI推出GPT-3.5 Turbo开始,到2024年LangChain官方推出LangGraph为止,LLM Agent的开发范式经历了三次重大的变革:
第一次变革:2022年底-2023年初——「Prompt Engineering + 单步调用」
这是LLM Agent的启蒙阶段——当时的开发者还没有任何框架可用(LangChain的第一个稳定版本是2023年3月推出的),只能直接调用OpenAI的API,通过「精心设计的提示词(Prompt Engineering)」来让LLM完成一些简单的任务——比如:
- 写一封邮件;
- 翻译一段文本;
- 总结一篇文章;
- 回答一个简单的问题。
这个阶段的Agent的特点是:
- 完全无状态:每次调用API都只传当前的提示词,不传任何历史对话;
- 完全线性:只能做一步到位的任务,不能做任何分支或循环;
- 完全依赖提示词:如果任务稍微复杂一点(比如需要先查资料再回答问题),提示词就会变得非常长、非常难以维护、而且LLM的准确率会急剧下降。
这个阶段的痛点是:
- 无法处理需要上下文记忆的任务(比如多轮对话);
- 无法处理需要分支或循环的任务(比如论文摘要翻译+术语一致性校验循环);
- 无法处理需要调用外部工具的任务(比如查天气预报、查股票行情、查维基百科)。
第二次变革:2023年3月-2024年初——「LangChain LCEL链 + 简单的Tool使用 + 简单的记忆」
2023年3月,LangChain的第一个稳定版本推出——这是LLM Agent开发的第一次重大变革——LangChain提供了一套完整的工具链,让开发者可以快速搭建LLM Agent:
- 提示词模板(Prompt Templates):让开发者可以复用提示词,不需要每次都手动写;
- 输出解析器(Output Parsers):让开发者可以把LLM生成的文本解析成结构化的数据(比如JSON、XML、Pydantic模型对象);
- 链(Chains):让开发者可以把多个组件(比如提示词模板、LLM、输出解析器、工具)串联起来,形成一个线性或分支的工作流;
- 工具(Tools):让开发者可以把外部的API、函数、数据库查询包装成Tool,让LLM可以调用;
- 记忆(Memory):让开发者可以把历史对话、中间变量保存在内存里,每次调用链时都会拼接这些记忆,传给LLM;
- Agent Executor:让开发者可以快速搭建「能够自主选择工具、自主执行工具、自主生成最终答案」的Tool使用Agent。
2023年10月,LangChain推出了LangChain Expression Language(LCEL)——这是LangChain的第二次小变革——LCEL让开发者可以用一种非常简洁、非常优雅的方式来写链,比如:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的翻译官,请把用户输入的中文翻译成英文。"),
("user", "{input}")
])
llm = ChatOpenAI(model="gpt-4o-mini")
output_parser = StrOutputParser()
# 用LCEL写的链
chain = prompt | llm | output_parser
# 调用链
result = chain.invoke({"input": "LangGraph是LangChain官方推出的有状态循环工作流框架。"})
print(result)
这个链的输出就是:
LangGraph is a stateful cyclic workflow framework officially launched by LangChain.
这个阶段的Agent的特点是:
- 有简单的记忆:可以处理多轮对话;
- 有简单的分支:可以用
RunnableBranch来处理一些简单的条件分支; - 有简单的Tool使用:可以用Agent Executor来搭建能够自主选择工具的Agent;
- 代码简洁优雅:可以用LCEL来快速写链。
这个阶段的痛点是:
- 记忆是无结构的:LangChain的默认记忆(比如
ConversationBufferMemory、ConversationSummaryMemory)都是把历史对话保存成字符串或者列表,没有结构化的状态字典——如果要保存一些结构化的中间变量(比如术语错误列表、已修正次数、金融风控规则的匹配结果),需要自己手动编写大量的代码,而且代码会非常混乱、难以维护; - 无法显式控制循环:LangChain的LCEL链只能是有向无环图(DAG)——也就是说,链里不能有任何循环(否则会无限循环,或者需要用非常混乱的方式控制循环次数,比如用一个全局变量来记录循环次数,然后用
RunnableLambda来判断是否继续循环); - 无法处理嵌套循环:比如论文全文翻译需要「先翻译每一段,再检查每一段的术语一致性,再翻译下一段」——这是一个嵌套循环(外层循环是每一段,内层循环是每一段的翻译+术语一致性校验)——用LangChain的LCEL链几乎不可能实现;
- 无法显式控制状态更新:LangChain的链的状态更新是隐式的——比如
ConversationBufferMemory会自动把每次的对话添加到记忆里,但你无法完全控制哪些对话要添加、哪些对话要删除、哪些对话要修改; - 无法实现优雅的人类介入:比如你需要让人类审核LLM生成的论文摘要——用LangChain的LCEL链很难实现优雅的人类介入(比如需要手动保存状态、手动处理人类输入、手动重启链);
- 无法实现优雅的断点保存与恢复:比如你的程序突然崩溃了——用LangChain的LCEL链几乎不可能实现优雅的断点保存与恢复(除非你自己手动编写大量的状态持久化代码);
- Agent Executor的行为是黑盒的:LangChain的Agent Executor是一个封装好的组件——你无法完全控制Agent的每一步行为(比如Agent选择工具的逻辑、Agent执行工具的逻辑、Agent生成最终答案的逻辑),如果Agent的行为不符合你的预期,你很难调试和修改。
第三次变革:2024年初至今——「LangGraph + 显式状态 + 显式循环 + 显式节点行为 + 人类介入 + 断点保存」
2024年1月,LangChain官方推出了LangGraph的第一个预览版本;2024年3月,LangChain官方推出了LangGraph的第一个稳定版本;2024年5月,LangChain官方推出了LangGraph Studio——这是一个可视化的LangGraph编辑器,可以让开发者用拖拽的方式来搭建LangGraph图链,然后自动生成Python代码;2024年8月,LangChain官方推出了LangGraph Cloud——这是一个托管的LangGraph服务,可以让开发者快速部署和运行LangGraph图链,不需要自己搭建服务器。
这是LLM Agent开发的第三次重大变革——LangGraph解决了LangChain LCEL链的所有痛点,让开发者可以完全显式地控制Agent的每一步行为:
- 显式的结构化状态:可以用Python字典或者Pydantic模型来定义状态结构,完全控制哪些字段要更新、哪些字段要保留、哪些字段要删除;
- 显式的循环:可以显式定义从某个节点跳转到之前的任意节点,而且提供了循环次数限制的功能,防止无限循环;
- 显式的嵌套循环:可以用「子图(Subgraph)」来实现嵌套循环——比如外层循环是每一段论文,内层循环是每一段的翻译+术语一致性校验;
- 显式的节点行为:可以用Python函数来定义每个节点的具体行为,完全控制节点做什么、怎么做;
- 显式的边的跳转规则:可以用Python函数或者Lambda表达式来定义边的跳转规则,完全控制从哪个节点跳转到哪个节点;
- 优雅的人类介入:可以用「等待人类输入的节点」来实现人类介入,当图执行到这个节点时会自动暂停,保存当前的状态;
- 优雅的断点保存与恢复:可以用「检查点存储器(Checkpoint Sink)」来实现断点保存与恢复,LangGraph会自动在每个节点执行完成后保存当前的状态;
- 完全可调试的白盒行为:LangGraph的图链是完全可调试的——你可以打印每个节点执行前的状态、每个节点执行后的状态、每个节点的输出、边的跳转逻辑,非常容易调试和修改。
生产级LLM Agent的四大核心需求(只有LangGraph能完全满足)
通过前面的LLM Agent发展历程的回顾,我们可以总结出生产级LLM Agent的四大核心需求——而这四大核心需求,只有LangGraph能完全满足:
核心需求1:有结构化的、可持久化的、可显式更新的状态
真实世界的生产级LLM Agent,比如金融交易前风控规则逐条匹配+数据补充+重新匹配循环Agent,需要保存大量的结构化的中间变量:
- 用户的交易请求:比如交易金额、交易时间、交易对手、交易品种;
- 金融风控规则的列表:比如规则1(交易金额不能超过用户的单笔交易限额)、规则2(交易对手不能在黑名单里)、规则3(交易品种不能是高风险品种)、规则4(交易时间不能是在非交易时间);
- 每条规则的匹配结果:比如规则1的匹配结果是「不匹配,用户的单笔交易限额是100万,当前交易金额是200万」、规则2的匹配结果是「匹配,交易对手不在黑名单里」、规则3的匹配结果是「待匹配,需要补充交易品种的风险等级数据」、规则4的匹配结果是「匹配,交易时间是在交易时间」;
- 已补充的数据列表:比如已经补充了交易品种的风险等级数据(高风险);
- 已重新匹配的规则列表:比如已经重新匹配了规则3(不匹配,交易品种是高风险品种);
- 循环次数:比如已经循环了2次;
- 最终的风控决策:比如拒绝交易,原因是「交易金额超过单笔交易限额,且交易品种是高风险品种」。
这些结构化的中间变量,用LangChain的默认记忆(比如ConversationBufferMemory)是很难保存的——因为默认记忆只能保存字符串或者列表,不能保存结构化的Pydantic模型对象;而且默认记忆的状态更新是隐式的,你无法完全控制哪些字段要更新、哪些字段要保留;更重要的是,默认记忆的状态是保存在内存里的,无法持久化到磁盘、Redis、PostgreSQL等——如果程序突然崩溃了,所有的状态都会丢失。
而LangGraph可以完全满足这个核心需求:
- 你可以用Pydantic模型来定义结构化的状态结构——比如:
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
# 定义交易请求的Pydantic模型
class TradeRequest(BaseModel):
trade_id: str = Field(description="交易ID")
user_id: str = Field(description="用户ID")
trade_amount: float = Field(description="交易金额(元)")
trade_time: str = Field(description="交易时间(ISO格式)")
counterparty_id: str = Field(description="交易对手ID")
product_id: str = Field(description="交易品种ID")
product_risk_level: Optional[int] = Field(description="交易品种的风险等级(1-5,1最低,5最高)", default=None)
# 定义风控规则的Pydantic模型
class RiskRule(BaseModel):
rule_id: str = Field(description="规则ID")
rule_name: str = Field(description="规则名称")
rule_description: str = Field(description="规则描述")
match_result: Optional[str] = Field(description="匹配结果(匹配/不匹配/待匹配)", default=None)
match_reason: Optional[str] = Field(description="匹配原因", default=None)
needs_data: Optional[List[str]] = Field(description="需要补充的数据列表", default=None)
# 定义LangGraph的状态结构的Pydantic模型
class RiskControlState(BaseModel):
trade_request: TradeRequest = Field(description="交易请求")
risk_rules: List[RiskRule] = Field(description="风控规则的列表")
supplemented_data: Dict[str, any] = Field(description="已补充的数据列表", default_factory=dict)
rematched_rules: List[str] = Field(description="已重新匹配的规则ID列表", default_factory=list)
loop_count: int = Field(description="循环次数", default=0)
max_loop_count: int = Field(description="最大循环次数", default=10)
final_decision: Optional[str] = Field(description="最终的风控决策(通过/拒绝/需要人工审核)", default=None)
final_reason: Optional[str] = Field(description="最终的风控原因", default=None)
- 你可以用「检查点存储器(Checkpoint Sink)」来持久化状态——LangGraph内置了多种检查点存储器:
MemorySaver:内存中的字典,适合开发测试;FileSaver:本地文件系统,适合单机生产;RedisSaver:Redis,适合分布式生产;PostgresSaver:PostgreSQL,适合分布式生产,而且支持SQL查询历史检查点;
- 你可以在每个节点函数里显式更新状态——节点函数只需要返回新的状态字典的部分内容,LangGraph会自动合并到旧的状态字典里,你不需要手动合并;比如:
def increment_loop_count(state: RiskControlState) -> Dict[str, any]:
# 显式更新循环次数字段
return {"loop_count": state.loop_count + 1}
- 你可以完全控制哪些字段要更新、哪些字段要保留——如果你在节点函数里没有返回某个字段,LangGraph会自动保留旧的状态字典里的那个字段;如果你在节点函数里返回了某个字段,LangGraph会自动用新的值覆盖旧的状态字典里的那个字段;如果你想删除某个字段,可以返回
{"field_name": None}(但需要注意,如果你用的是Pydantic模型定义状态,删除字段可能会导致验证错误,所以推荐用可选字段Optional)。
核心需求2:有显式的、可控制的、可嵌套的循环
真实世界的生产级LLM Agent,比如论文全文翻译+术语一致性校验循环Agent,需要处理嵌套循环:
- 外层循环:遍历论文的每一段——比如论文有100段,外层循环需要执行100次;
- 内层循环:对每一段进行翻译+术语一致性校验——比如某一段的术语有3个错误,内层循环需要执行3次(每次修正一个错误,然后再检查术语一致性,直到所有错误都修正为止);
- 还有可能有第三层循环:如果某一段的翻译LLM不确定某个术语的正确翻译,需要调用外部的专业术语库API来查询——如果API调用失败,需要重试3次,这就是第三层循环。
这些显式的、可控制的、可嵌套的循环,用LangChain的LCEL链是几乎不可能实现的——因为LCEL链只能是有向无环图(DAG),不能有任何循环;如果要勉强实现,需要用全局变量、递归、或者非常混乱的RunnableLambda和RunnableBranch组合,代码会非常难以维护、难以调试、难以扩展。
而LangGraph可以完全满足这个核心需求:
- 你可以显式定义从某个节点跳转到之前的任意节点——比如从「术语一致性检查节点」跳转到「翻译修正节点」,再跳转到「术语一致性检查节点」,直到所有术语都一致为止;
- 你可以显式控制循环次数——比如在状态字典里定义「max_inner_loop_count」和「inner_loop_count」字段,在「术语一致性检查节点」里判断如果
inner_loop_count >= max_inner_loop_count,就强制终止内层循环,跳转到外层循环的下一段; - 你可以用「子图(Subgraph)」来实现嵌套循环——比如把「内层循环(翻译+术语一致性校验)」封装成一个子图,然后把这个子图作为外层循环的一个节点;
- LangGraph的子图可以有自己的状态、自己的节点、自己的边、自己的检查点存储器——非常灵活、非常容易维护、非常容易扩展。
核心需求3:有显式的、可自定义的、可调试的节点行为和边的跳转规则
真实世界的生产级LLM Agent,比如Bug复现报告生成+复现脚本编写+验证循环Agent,需要完全自定义节点行为和边的跳转规则:
- 节点行为的自定义:
- 「Bug复现报告生成节点」:需要调用LLM来生成Bug复现报告,报告里需要包含Bug的标题、Bug的描述、Bug的复现步骤、Bug的预期结果、Bug的实际结果、Bug的环境信息(操作系统、浏览器版本、软件版本);
- 「复现脚本编写节点」:需要调用LLM来编写复现脚本(比如Python脚本、Shell脚本、Selenium脚本);
- 「复现脚本验证节点」:需要调用Python的
subprocess模块来运行复现脚本,然后判断脚本的输出是否符合预期结果; - 「等待人类输入的节点」:如果复现脚本验证失败超过3次,需要暂停图,让人类介入,确认Bug的复现方向是否正确;
- 边的跳转规则的自定义:
- 从「Bug复现报告生成节点」:如果LLM生成的复现报告包含所有必要的信息(标题、描述、复现步骤、预期结果、实际结果、环境信息),就跳转到「复现脚本编写节点」;否则,跳转到「Bug复现报告修正节点」;
- 从「复现脚本编写节点」:如果LLM生成的复现脚本可以通过Python的语法检查(比如用
py_compile模块),就跳转到「复现脚本验证节点」;否则,跳转到「复现脚本修正节点」; - 从「复现脚本验证节点」:如果脚本的输出符合预期结果,就跳转到「终点」;否则,如果验证次数小于3次,就跳转到「复现脚本修正节点」;否则,跳转到「等待人类输入的节点」;
- 从「等待人类输入的节点」:如果人类确认Bug的复现方向正确,就跳转到「复现脚本重写节点」;否则,跳转到「Bug复现报告重写节点」。
这些显式的、可自定义的、可调试的节点行为和边的跳转规则,用LangChain的Agent Executor是几乎不可能实现的——因为Agent Executor的行为是黑盒的,你无法完全控制;而且Agent Executor的边的跳转规则是隐式的(由LLM的function calling决定),如果LLM的function calling出错,你很难调试和修改。
而LangGraph可以完全满足这个核心需求:
- 你可以用Python函数来定义每个节点的具体行为——节点函数可以做任意的事情:比如调用LLM、调用外部工具、调用Python的标准库、调用第三方库、更新状态字典、甚至根据逻辑动态生成下一个节点的名称;
- 你可以用Python函数或者Lambda表达式来定义边的跳转规则——边的跳转规则可以完全自定义:比如判断状态字典里的某个字段是否满足某个条件、判断节点函数的输出是否满足某个条件、甚至调用LLM来判断是否满足某个条件;
- LangGraph的图链是完全可调试的——你可以用
graph.get_graph().print_ascii()来打印图的结构、用graph.stream(state)来流式输出每个节点执行前的状态、每个节点执行后的状态、每个节点的输出、边的跳转逻辑,非常容易调试和修改。
核心需求4:有优雅的、可集成的、可自定义的人类介入和断点保存与恢复
真实世界的生产级LLM Agent,比如医疗诊断辅助Agent,需要人类介入——因为医疗诊断是一个非常严肃的事情,不能完全由LLM来决定,必须由医生来最终审核;而且医疗诊断辅助Agent的任务可能需要很长时间(比如需要分析大量的医疗影像、需要查询大量的医学文献),如果程序突然崩溃了,必须能够断点保存与恢复,否则医生之前的所有操作都会丢失。
这些优雅的、可集成的、可自定义的人类介入和断点保存与恢复,用LangChain的LCEL链是几乎不可能实现的——除非你自己手动编写大量的代码。
而LangGraph可以完全满足这个核心需求:
- 你可以用「条件边(Conditional Edge)」和「中断节点(Interrupt Node)」来实现人类介入——或者更简单的是,用LangGraph内置的「Human-in-the-Loop(HITL)」工具;
- 你可以用「检查点存储器(Checkpoint Sink)」来实现断点保存与恢复——LangGraph会自动在每个节点执行完成后保存当前的状态;
- 你可以自定义检查点的保存时机——比如你可以只在某些特定的节点执行完成后保存检查点,而不是每个节点都保存;
- 你可以恢复到任意一个历史检查点——比如你对刚才的诊断结果不满意,可以恢复到诊断前的检查点,重新上传医疗影像,再执行诊断;
- 你可以自定义人类介入的UI——比如你可以用LangGraph Studio的可视化UI,或者用Flask/Django/FastAPI搭建自己的Web UI,或者用Telegram/Discord/Slack搭建自己的聊天机器人UI。
问题描述
为了让大家更具体地理解「生产级LLM Agent的痛点」,我们现在提出三个具体的、真实的、生产级的问题——这三个问题,用LangChain的LCEL链是很难解决的,但用LangGraph可以非常优雅地解决:
问题1:论文摘要翻译+术语一致性校验循环
问题背景:你是一个学术编辑,需要把一篇中文计算机科学论文的摘要翻译成英文,而且必须保证翻译后的术语和用户提供的「计算机科学专业术语中英对照表」完全一致——比如用户提供的对照表里,「LangGraph」必须翻译成「LangGraph」(不要翻译),「有状态循环工作流」必须翻译成「stateful cyclic workflow」,「有限状态机」必须翻译成「finite state machine」。
问题具体描述:
1.
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)