作为一名技术开发者你是否也遇到过这样的“抓狂”时刻——明明只是想让AI帮你做点稍微复杂的事情比如先查资料再分析或者做个多轮问答却发现自己写的代码满满都是状态标记、线程锁和无穷无尽的if-else。稍有不慎AI的思考路径就跑偏了整个状态陷入一片混乱最后不得不靠重启来“抢救”。这种依靠不确定性实现智能的体验让人既兴奋又焦虑。今天这篇爆款硬核技术文章我们不玩虚的直接带你彻底搞懂LangGraph。这是一个由LangChain团队推出的“图编排框架”能把你那些随时可能失控的AI智能体Agent变成一个状态稳定、逻辑清晰、有据可依的流程图。全文干货附完整可运行的代码建议收藏后仔细读。一、到底什么是LangGraph——给AI智能体的“路线规划师”要搞懂LangGraph我们要先明白它到底解决了什么问题。无论是做AI应用还是传统软件开发当任务非常简单时让程序按顺序执行第一步做完做第二步一切都很美好。但一旦涉及状态和循环决策比如“如果模型给的答案没把握就换个方式再试一次”问题就会变得极其复杂。我们不妨来做个对比传统链式调用例如LangChain的Chain模式像一个“传送带”不管三七二十一东西从A传到B再到C就结束了。它根本不在乎中途出了什么意外也不会停下来重新思考。LangGraph的图模式Graph通过状态机、节点和边来构建工作流。可以任意循环、分支、重试甚至可以随时暂停等待人工审核。这就像给AI画了一张详细地图让智能体在图上自己走但走的每一步都完全可控。LangGraph本质上是个低级别编排框架Low-level Orchestration Framework它把Agent的工作流抽象成一张图。三个关键要素如下节点Nodes这是图上的“处理单元”什么都能干——调用大语言模型LLM、执行工具函数、运行自定义Python逻辑。边Edges连接节点的“逻辑流”可以定义“这条路是必走的”普通边也可以定义“得分超过70分走哪条路”条件边。状态State在整张图里流转的“公共数据字典”。所有节点都能从这个“字典”里读数据、改数据状态跟着节点的执行一路更新传递。更底层来看LangGraph的运行时是基于Google的Pregel算法构建的。Pregel把计算分成“超步”来执行这种设计让LangGraph天然支持分布式并行执行这才是它能支撑企业级AI应用的根本原因。Pregel模式的执行分为三个阶段规划Plan——确定本轮要执行哪些节点执行Execution——并行运行所有被选中的节点更新Update——把节点的输出合并更新到共享状态中。一轮完成后系统自动进入下一轮直到没有节点需要执行为止。二、为什么非要用LangGraph传统LangChain路在何方很多开发者会问“我直接用LangChain的LCELLangChain Expression Language写链不就行了吗何必再学一个新框架”这里回答很简单因为LCEL适合做“简单”LangGraph适合做“强大”。传统的LangChain更像一个“功能工具箱”里面有提示词模板Prompt Template、文档加载器Document Loader、输出解析器Output Parser等一系列现成的零件。LCEL所做的就是把它们按照A→B→C的线性顺序串联起来。整个流程就像一条传送带启动传送带货物从流水线入口进入经过一道道工序最后从出口送出一次运行结束。这种流水线模式在简单任务中非常高效、简洁代码也易于阅读。但当智能体的分析结果不够确定比如置信度低于0.75需要回到上一步重新获取更多上下文再分析一遍时LCEL就完全无能为力了——你不得不在主流程外围手动写一堆if-else和状态变量拼命把状态拼回去。来看一个真实对比假设你需要构建一个基于Gemini 2.5 Flash、包含“上下文拉取—安全性分析—输出审核”三阶段的代码审查Agent。LangChain的写法是线性的——先去取上下文取完了做分析分析完了直接做审核执行路径完全固定无法根据中间结果动态调整。而LangGraph可以定义一个循环条件如果分析结果的置信度低于0.75且重试次数不超过3次就主动跳回上下文拉取步骤重来一轮。二者不是二选一的对立关系。实际上LangChain v1.0的Agent抽象层已经全面构建在LangGraph之上了。选LangChain相当于选择了开箱即用的组件库选LangGraph相当于选择了最底层、最灵活的编排能力和完全可控的执行流程。三、十分钟快速入门——从一次最简单的“Hello World”出发理论铺垫得差不多了我们现在动手写点能真正跑起来的代码。首先把环境配置好。在命令行中执行以下步骤# 创建并激活全新的conda环境 conda create -n langgraph python3.12 conda activate langgraph # 安装LangGraph核心库 pip install -U langgraph # 【可选】如果后续需要接入各种大模型建议同时安装 pip install langchain-openai # 用于OpenAI模型 pip install langchain-community # 用于社区支持的各种模型和工具 # 确认安装成功 pip show langgraph安装完成后让我们从一个最简单的图入手理解LangGraph的核心工作流程。# 1. 导入LangGraph的核心组件 from langgraph.graph import StateGraph, MessagesState, START, END # 2. 定义一个节点Node——模拟调用大语言模型LLM # MessagesState是一个内置状态类型包含messages字段专门用于存储对话消息列表 def mock_llm_node(state: MessagesState): 这个节点的功能很简单向共享状态中的消息列表追加一条AI的回复。 在实际项目中这里可以是真正的OpenAI调用、DeepSeek调用或其他任意逻辑。 # 从共享状态中读取现有消息虽然在这个简单示例中我们没用上但规范写法一定要保留state参数 # 返回一个字典LangGraph会自动将其中内容合并update到全局状态中 return {messages: [{role: ai, content: hello world}]} # 3. 定义一个图Graph # 通过StateGraph(MessagesState)初始化一张图并明确这张图的共享数据结构遵循MessagesState规范 graph StateGraph(MessagesState) # 4. 把前面定义好的节点添加到图中 # mock_llm是我们给这个节点起的名字后面添加边的时候要用这个名字来引用它 graph.add_node(mock_llm, mock_llm_node) # 5. 添加节点之间的有向边Edge # START是一个特殊常量代表图的入口END代表图的出口。 # 下面这行代码的意思是从 START 开始执行完mock_llm节点之后整个流程就结束了。 graph.add_edge(START, mock_llm) graph.add_edge(mock_llm, END) # 6. 编译图Compile # compile()这一步至关重要。LangGraph会把前面添加的所有节点和边的定义编译成一个真正可调用的运行时对象。 # 在编译的过程中LangGraph会验证图结构是否合法比如是否存在孤立节点、是否形成有效路径等。 graph graph.compile() # 7. 调用图并输出结果 # invoke()是执行图的核心方法。它接受一个包含初始状态的字典作为参数自动从START节点开始 # 按照定义的边所指定的路径依次访问各个节点直到抵达END。 response graph.invoke({messages: [{role: user, content: hi!}]}) # 8. 打印最终状态 # 你能看到最终的messages列表里既有用户原始的hi!也有mock_llm节点添加的hello world。 print(response)这个例子的执行过程是这样的invoke方法拿到初始状态后从START出发找到下一个连接的节点“mock_llm”执行这个节点函数向消息列表追加了一条新的AI消息然后按照定义好的边继续走到END整个图的执行就完成了。四、实战进阶一掌握“条件边”与“分支路由”如果流程图里只有一种直来直去的箭头那它和链式调用就没有本质区别了。LangGraph真正的杀手锏是条件边——根据程序运行时的状态动态决定下一步走向哪个节点。我们用一个充满不确定性的实用场景来说明AI对某个问题的回答“有多大的把握”。很多时候AI的第一次输出并不靠谱它只是强行给你一个回答自己也知道可能不对。此时我们需要在流程图里增加一个判断节点如果AI的置信度高于某个阈值就认为回答可靠直接输出如果置信度太低就调用一个“改写/重试”节点优化一下问题再让AI回答一次。import uuid from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver # 1. 定义比MessagesState更丰富的自定义状态结构 class EnhancedState(TypedDict): EnhancedState是我们自定义的图状态数据结构使用TypedDict可以让开发工具提供更好的类型提示和检查 # 对话消息列表用于存储多轮对话历史 messages: list[dict] # 存储用户原始问题没有经过任何改写 original_question: str # 存储经过改写后的优化问题如果有改写的话 rewritten_question: str | None # 存储置信度分数浮点数范围0~1越高代表AI对自己的回答越有把握 confidence: float # 当前已经尝试回答的次数每重试一次就1防止无限循环下去 attempt_count: int # 2. 定义各类节点的具体实现 def analyze_confidence_node(state: EnhancedState): 【模拟节点】分析当前AI回答的置信度。 在实际项目中这里应该根据某项具体的业务指标来判断。 比如检索到的文档与问题的语义相似度、AI模型输出的logprobs概率值等。 在此我们用当前“尝试次数”来模拟置信度的变化。 attempts state.get(attempt_count, 0) if attempts 0: # 初次尝试假设AI的回答置信度不高给一个比较低的分数0.5 confidence 0.5 elif attempts 1: # 第1次重试后置信度提升到0.75但仍不算非常理想 confidence 0.75 else: # 第2次重试后置信度达到0.95的高水平可以放心输出了 confidence 0.95 # 返回需要更新的状态LangGraph会自动合并到全局状态中 return {confidence: confidence, attempt_count: attempts 1} def rewrite_node(state: EnhancedState): 【模拟节点】将用户的问题改写得更清晰以便AI能给出更准确的回答。 在实际使用场景中你可以在这里调用一个专门的LLM来做“改写” 也可以调用一个模板函数来扩充问题上下文。 old_question state.get(original_question, ) # 简单的改写在原问题后面追加一段话让回答更聚焦“关键信息” rewritten f请你务必提取出以下问题中的关键信息: {old_question} return {rewritten_question: rewritten} def final_output_node(state: EnhancedState): 最终输出节点在这里生成最终的回答 # 打印最终携带的所有状态信息 print( 最终回答生成 ) print(f用户原始问题: {state.get(original_question)}) print(f是否经过了改写: {是 if state.get(rewritten_question) else 否}) print(f最终置信度: {state.get(confidence)}) print(f总尝试次数: {state.get(attempt_count)}) # 模拟生成一个最终回答 return {messages: [{role: ai, content: 这是经过多轮优化后的最终回答}]} def analyze_and_route(state: EnhancedState) - Literal[rewrite_question, end_process]: 【条件边函数】根据当前状态的置信度决定下一步要走哪条分支。 这是一个路由函数返回的字符串需要与图中其他节点的名称完全一致。 # 从当前全局状态中取出置信度 confidence state.get(confidence, 0.0) if confidence 0.8 and state.get(attempt_count, 0) 3: # 如果置信度低于0.8并且重试次数还不足3次就走“改写”分支 print(f决策中...置信度仅有{confidence}系统决策重新改写并重试。) return rewrite_question else: # 如果置信度足够高或者重试次数已经达到上限防御机制就走到最终输出节点 if confidence 0.8: print(f决策中...置信度{confidence}达到阈值系统决策直接输出最终结果。) else: print(f决策中...已重试{state.get(attempt_count, 0)}次但置信度{confidence}未达标系统决策强制输出。) return end_process # 3. 构建并编译图 builder StateGraph(EnhancedState) # 添加所有相关的节点 builder.add_node(analyze_confidence, analyze_confidence_node) builder.add_node(rewrite, rewrite_node) builder.add_node(final_output, final_output_node) builder.add_edge(START, analyze_confidence) # 设置条件边从analyze_confidence节点出发根据analyze_and_route函数的返回值决定走哪条路 builder.add_conditional_edges(analyze_confidence, analyze_and_route, { rewrite_question: rewrite, # 如果返回rewrite_question就到rewrite end_process: final_output # 如果返回end_process就直接到final_output }) # 为rewrite节点添加后续处理步骤重写后再次回到analyze_confidence节点进行新一轮分析和判断形成闭环 builder.add_edge(rewrite, analyze_confidence) builder.add_edge(final_output, END) # 编译图并加入一个内存版的检查点保存器InMemorySaver用来保存整个执行过程中的每一快照 memory_saver InMemorySaver() graph builder.compile(checkpointermemory_saver) # 4. 执行图并且传入唯一的线程ID以便后续能够随时恢复执行 config {configurable: {thread_id: str(uuid.uuid4())}} initial_state { messages: [{role: user, content: LangGraph是什么}], original_question: LangGraph是什么, rewritten_question: None, confidence: 0.0, attempt_count: 0 } print( 开始执行带有条件分支的LangGraph工作流 ) # 执行整个图 final_state graph.invoke(initial_state, configconfig) print(\n 执行完成最终状态 ) print(final_state)在这段示例代码里图的执行路径可能是“分析置信度→置信度低→改写→再分析→置信度达标→最终输出”。整个流程的控制权和路由逻辑完全掌握在你手里不再依赖LLM的黑盒输出。五、生产级必备技能状态持久化与人工审核机制5.1 状态持久化的核心思想LangGraph的持久化机制Checkpoint机制是它能媲美企业级框架的关键一环。默认情况下执行完一次invoke状态就随着Python进程的结束而消失了。持久化机制相当于在你开车的时候每到一个关键路口就给你拍一张快照如果后续车辆抛锚了可以回到最近的一张快照处重新发动发动机而不是从起点重来。LangGraph支持把状态快照保存到内存、磁盘文件、AWS DynamoDB、PostgreSQL等外部存储中。在编译图的时候只需要传入一个checkpointer对象即可开启这一能力。5.2 人工审核机制的核心思想LangGraph称之为Human-in-the-LoopHIL即“人在循环中”。在Agent的执行路径上设置一个中断点interrupt让AI系统在触碰到这些中断点时自动暂停等待人工输入决策批准、拒绝或修改后再继续执行。在企业级金融风控、重要数据删除、大额交易确认等场景中这个机制不可或缺。5.3 核心实现代码结合上述两个特性下面给出了一个完整可运行的示例。这个图在执行到最终发送之前会先中断等待用户输入“yes”或“no”来决定是否批准发送。import uuid from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import interrupt, Command # 1. 定义系统所需的状态 class EmailState(TypedDict): 邮件发送系统的状态结构 email_draft: str # AI生成的草稿内容 approved: bool # 是否经过人工批准 sent: bool # 是否已经发送 # 2. 节点函数 def draft_email_node(state: EmailState): AI生成邮件草稿的节点 print( [AI Agent]: 正在为您撰写邮件...) # 实际开发中调用LLM生成邮件内容 draft 尊敬的客户 感谢您购买我们的产品。您的订单已处理完毕预计3-5个工作日送达。 如有任何问题请随时联系我们。 祝您生活愉快 print(f [AI Agent]: 草稿已生成。\n{draft}) return {email_draft: draft, approved: False, sent: False} def human_approval_node(state: EmailState): 人工审核环节支持HITL机制的节点 print(\n⏸️ [System]: 系统已暂停正在等待您的审批...) # 使用LangGraph提供的interrupt()函数在此处将图的执行中断并返回用户输入的信息 user_input interrupt({ question: f\n请查看以下邮件内容是否批准发送(y/n)\n{state[email_draft]}\n, options: [y, n] }) if user_input.lower() y: print(✅ [Approval]: 邮件已获得批准将继续执行后续发送节点。) return {approved: True} else: print(❌ [Approval]: 邮件未获批准将终止发送。) return {approved: False} def send_email_node(state: EmailState): 发送邮件的节点仅在approved为True时执行 if state[approved]: print(f [System]: 正在发送邮件内容{state[email_draft]}) # 此处可接入真实SMTP服务 return {sent: True} else: print( [System]: 由于未获批准邮件发送已取消。) return {sent: False} # 3. 构建图 builder StateGraph(EmailState) builder.add_node(draft, draft_email_node) builder.add_node(human_approval, human_approval_node) builder.add_node(send_email, send_email_node) builder.add_edge(START, draft) builder.add_edge(draft, human_approval) builder.add_edge(human_approval, send_email) builder.add_edge(send_email, END) # 4. 启用持久化并编译图 memory InMemorySaver() graph builder.compile(checkpointermemory) # 5. 执行图并传入thread_id以便后续从中断点恢复 config {configurable: {thread_id: str(uuid.uuid4())}} print( 开始执行带有HITL机制的LangGraph ) try: # 第一次invoke会执行到human_approval节点处并抛出一个特殊中断 final_state graph.invoke({email_draft: , approved: False, sent: False}, configconfig) except Exception as e: # 捕获中断并处理 print(f\n [System]: 检测到中断事件请运行审批函数输入您的选择...) # 中断后利用Command(resumeuser_input)从中断点传入用户输入并恢复执行 user_decision input(\n请输入您的决定y/n) final_state graph.invoke(Command(resumeuser_decision), configconfig) print( 执行完成 ) print(f最终状态: {final_state})你不只是在构建一个简单的智能体你拥有了完全掌控它的权柄随时可以中断执行随时可以查询历史随时可以从某个快照重新开始。六、LangGraph vs LangChain一张对比图看清本质区别有了前面几个阶段的代码基础我们可以从六个维度来对比一下LangGraph和LangChain的本质区别特性维度LangGraphLangChain抽象级别低级编排框架提供细粒度控制高级工具包提供开箱即用的链式组件状态管理内置状态机和全局State自动传递需要开发者自行管理状态或依赖Memory模块手动实现执行模型基于有向图的并行执行支持循环、分支线性链式执行LCEL只支持A→B→C持久化能力原生自带Checkpoint机制支持多种存储后端需要额外开发或集成第三方存储人工干预原生支持Human-in-the-Loop中断与恢复不直接支持需要搭建复杂的回调体系适用场景复杂、有状态、无限循环或多分支交互的AI应用简单的单轮推理、标准问答、文档加载处理链本质上LangChain是为“快速构建标准Agent”而生的一套高层抽象它内部的Agent层在LangChain v1.0之后已经完全依托于LangGraph来运行而LangGraph则是为了“复杂工作流编排”而生的底层基础设施。二者是不可替代的互补关系而非二选一的竞争关系。七、总结LangGraph重构AI应用开发范式从最开始的不确定性焦虑到完全掌控AI的每一个决策节点通过这篇文章你从一个真实可运行的图开始逐步深入了解了LangGraph的节点、边、状态、条件分支、持久化和人工审核等核心机制。这些能力组合在一起彻底重构了我们构建AI智能体的模式图结构带来了无与伦比的掌控力告别线性链中的if-else地狱复杂业务中的“循环、重试、分支”通过StateGraph中的节点和边来单独编排每一处逻辑都能清晰地被追溯和调试。原生持久化机制让企业级部署变得可靠所有执行快照被系统自动保存一次运行途中掉电或崩溃可以从最新快照恢复内部数据和上下文完好无损。原生Human-in-the-loop将监管与效率完美融合在金融、医疗等高风险场景中既不过度依赖AI的黑盒决策也不完全依赖低效的人工审批而是在关键节点设置中断让人工精确参与每个关键判断让AI系统在全自动和半自动之间自由切换。生产友好的全链条支持LangGraph可以无缝集成LangSmith实现全链路可视化追踪无缝对接LangChain的模型和工具生态并支持LangGraph云平台的部署——这套组合拳让开发者在快速原型和生产上线之间几乎零切换成本。此时此刻AI智能体的发展已经迈入了“自主决策”和“复杂流程编排”的新阶段。无论你的项目是一个需要精细控制的金融客服Agent还是一个随时需要人工介入的医疗诊断系统或是需要多轮知识库检索的RAG应用LangGraph都为你提供了一套工业级的解决方案。你已经拿到了通往AI智能体高级编程的钥匙剩下的就是打开你的代码编辑器开始动手画你的第一张LangGraph流程图吧。