从焦虑到掌控：一文讲透LangGraph，把AI智能体的决策链条变成一张清晰的流程图

作为一名技术开发者，你是否也遇到过这样的“抓狂”时刻——明明只是想让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的LCEL（LangChain 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 python==3.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(checkpointer=memory_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, config=config) print("\n========== 执行完成，最终状态 ==========") print(final_state)

在这段示例代码里，图的执行路径可能是“分析置信度→置信度低→改写→再分析→置信度达标→最终输出”。整个流程的控制权和路由逻辑完全掌握在你手里，不再依赖LLM的黑盒输出。

五、生产级必备技能：状态持久化与人工审核机制

5.1 状态持久化的核心思想

LangGraph的持久化机制（Checkpoint机制），是它能媲美企业级框架的关键一环。默认情况下，执行完一次invoke，状态就随着Python进程的结束而消失了。持久化机制相当于在你开车的时候，每到一个关键路口就给你拍一张快照，如果后续车辆抛锚了，可以回到最近的一张快照处重新发动发动机，而不是从起点重来。

LangGraph支持把状态快照保存到内存、磁盘文件、AWS DynamoDB、PostgreSQL等外部存储中。在编译图的时候，只需要传入一个checkpointer对象即可开启这一能力。

5.2 人工审核机制的核心思想

LangGraph称之为Human-in-the-Loop（HIL），即“人在循环中”。在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(checkpointer=memory) # 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}, config=config) except Exception as e: # 捕获中断并处理 print(f"\n🔄 [System]: 检测到中断事件，请运行审批函数输入您的选择...") # 中断后，利用Command(resume=user_input)从中断点传入用户输入并恢复执行 user_decision = input("\n请输入您的决定（y/n）：") final_state = graph.invoke(Command(resume=user_decision), config=config) print("========== 执行完成 ==========") print(f"最终状态: {final_state}")

你不只是在构建一个简单的智能体，你拥有了完全掌控它的权柄：随时可以中断执行，随时可以查询历史，随时可以从某个快照重新开始。

六、LangGraph vs LangChain：一张对比图看清本质区别

有了前面几个阶段的代码基础，我们可以从六个维度来对比一下LangGraph和LangChain的本质区别：

本质上，LangChain是为“快速构建标准Agent”而生的一套高层抽象，它内部的Agent层在LangChain v1.0之后已经完全依托于LangGraph来运行；而LangGraph则是为了“复杂工作流编排”而生的底层基础设施。二者是不可替代的互补关系，而非二选一的竞争关系。

七、总结：LangGraph重构AI应用开发范式

从最开始的不确定性焦虑，到完全掌控AI的每一个决策节点，通过这篇文章，你从一个真实可运行的图开始，逐步深入了解了LangGraph的节点、边、状态、条件分支、持久化和人工审核等核心机制。这些能力组合在一起，彻底重构了我们构建AI智能体的模式：

1. 图结构带来了无与伦比的掌控力

   ：告别线性链中的if-else地狱，复杂业务中的“循环、重试、分支”通过StateGraph中的节点和边来单独编排，每一处逻辑都能清晰地被追溯和调试。

2. 原生持久化机制让企业级部署变得可靠

   ：所有执行快照被系统自动保存，一次运行途中掉电或崩溃，可以从最新快照恢复，内部数据和上下文完好无损。

3. 原生Human-in-the-loop将监管与效率完美融合

   ：在金融、医疗等高风险场景中，既不过度依赖AI的黑盒决策，也不完全依赖低效的人工审批，而是在关键节点设置中断，让人工精确参与每个关键判断，让AI系统在全自动和半自动之间自由切换。

4. 生产友好的全链条支持

   ：LangGraph可以无缝集成LangSmith实现全链路可视化追踪，无缝对接LangChain的模型和工具生态，并支持LangGraph云平台的部署——这套组合拳让开发者在快速原型和生产上线之间几乎零切换成本。

此时此刻，AI智能体的发展已经迈入了“自主决策”和“复杂流程编排”的新阶段。无论你的项目是一个需要精细控制的金融客服Agent，还是一个随时需要人工介入的医疗诊断系统，或是需要多轮知识库检索的RAG应用，LangGraph都为你提供了一套工业级的解决方案。

你已经拿到了通往AI智能体高级编程的钥匙，剩下的就是打开你的代码编辑器，开始动手画你的第一张LangGraph流程图吧。