我自己真正上手LangGraph之前一直以为它只是给LangChain套了层花哨的流程壳。直到我用它重写了一个多工具Agent被线上事故教育了一轮才意识到“有状态编排”和“中断恢复”这两个词到底有多值钱。简单说LangGraph把AI应用从“你调用大模型”变成“你定义一张图大模型和工具在图上跑”。这篇文章不讲官话从零开始带你搭线性工作流再一步步升级成能暂停、能恢复、能跨轮对话的有状态Agent全程带代码和踩坑记录。不管你是刚接触LangChain的小白还是被复杂Agent状态搞到头秃的开发者只要你懂一点Python能写最简单的Prompt函数这篇入门指南应该能让你少走我走过的弯路。1. 先别急着写代码搞懂LangGraph到底解决了什么问题1.1 为什么手写Agent编排容易翻车假设你要做一个能查天气、能算价格、能根据用户要求写推荐文案的Agent。用最朴素的做法你会在主循环里写一堆if-else比如“如果模型返回了工具调用就去执行工具把结果塞回消息列表再问一次模型”。这种模式早期能跑通但一旦任务复杂起来问题就暴露得很明显第一状态管理要自己盯。每一轮对话哪些是历史消息哪些是工具返回结果哪些是中间变量全靠手写维护。第二分支逻辑越堆越乱。如果某个节点需要人工审批或者某个工具失败了要重试你就要在Python代码里不断嵌套条件判断改一处崩三处。第三宕机、断网、用户取消任务时内存里的状态说丢就丢没法接着跑。我印象最深的一次事故是一个Agent跑了一个多小时的数据清洗跑到最后一步调外部接口超时整个进程崩溃所有中间数据全没了。后来复盘的时候我就想如果这个系统能把“当前跑到哪一步、中间结果是什么”持久化下来崩溃之后是不是能直接从断点恢复这就是LangGraph最核心的价值之一。1.2 核心抽象图、节点、边、状态LangGraph直白讲就是“用图来表达流程”。我不谈虚的直接把这几个概念当成乐高积木概念类比角色State一张公共台账所有节点共享的数据容器每个节点读它、写它Node一个小工真正干活的函数比如调用大模型、执行工具、做决策Edge一条传送带定义节点执行完后下一步去哪Conditional Edge岔路口根据当前State里的值决定走哪条路线Checkpointer存档点把当前State和运行进度持久化崩溃后可恢复Interrupt暂停键让Agent在中间停下等待用户输入常见于人工审批学LangGraph最需要扭转思路的点是你不写main()函数控制一切你写一堆小函数节点然后告诉图“谁先跑、谁后跑、什么条件下跑”。图帮你决定执行顺序它也负责把状态一路传递下去。刚上手时你可能会觉得这有点绕但等你真的搭过两个流程就会习惯这种“数据驱动流程”的心智模型。1.3 什么时候该选LangGraph并不是所有AI项目都非要上LangGraph。如果你只是单次调用大模型、做一次文本翻译用普通SDK就够了。我个人经验是出现下面三种情况之一就值得引入多个步骤之间有依赖和分工比如先生成、再审核、再发布。需要大模型自主决定调用哪些工具而且调用结果要参与后续决策。用户会和Agent多轮交互或者流程可能被打断、需要恢复。这篇入门指南的核心目标就是帮你搞定这三种情况。2. 从线性工作流入手你的第一个LangGraph应用2.1 环境准备与依赖安装不要想得太复杂一条pip install就能把基础依赖装齐。我建议你在虚拟环境里装避免污染系统Pythonpip install langgraph langchain-openai langchain-core如果你是国内网络模型接口可能不方便直连那就准备本地模型地址或者替换成任何一个兼容OpenAI接口的API端点。LangGraph本身不绑死模型厂商只要LLM能返回结构化消息就行。我还建议顺带装一个langgraph-checkpoint-sqlite后面讲中断恢复时用得上pip install langgraph-checkpoint-sqlite安装完之后我一般先做一个最简单的“一句话测试”确认环境没问题from langgraph.graph import StateGraph, START, END print(StateGraph imported OK)看到输出没报错说明基础环境已经就位。2.2 定义State让节点之间共享数据在LangGraph里State决定了每个节点的输入输出格式。最简单的定义方式是用TypedDictfrom typing import TypedDict, List class BlogState(TypedDict): topic: str # 输入主题 draft: str # 草稿内容 issues: List[str] # 审核发现的问题 final: str # 最终定稿这段代码看着不起眼但它规定了整张图的数据契约每个节点函数接收整个State返回一个字典只写自己想更新的字段。比如审核节点可以只返回{issues: [有错别字]}不会动其他字段。初学者最常见的误解是每个节点不就是接收上一个节点的输出吗不是。LangGraph里每个节点接收的是整个共享State返回值会增量更新到State。这种设计的好处是第三个节点可以同时读第一个节点产生的草稿和第二个节点产生的审核意见不用手动把“中间产物”当作参数层层传递。你可以把它理解成全公司共用一个共享文档每个部门只改自己那一栏。2.3 搭建“生成–审核–修改”三节点流程下面这段代码是这篇入门指南的第一个核心示例建议你亲手跑一遍。我们做一个博客文章流水线先生成文章再检查敏感词和错别字有问题就修改。from langgraph.graph import StateGraph, START, END def generate_node(state: BlogState) - dict: 生成初稿这里直接用模拟数据代替真实大模型调用 topic state[topic] # 真实场景中替换为 llm.invoke(...) draft f关于{topic}的一篇草稿内容包含基础介绍和操作建议。 return {draft: draft} def review_node(state: BlogState) - dict: 模拟审核发现一个明显问题触发修改流程 issues [缺少案例说明, 结尾过于仓促] return {issues: issues} def revise_node(state: BlogState) - dict: 根据审核意见修改草稿 old_draft state[draft] issues_text .join(state[issues]) final old_draft f。已根据以下意见修改{issues_text} return {final: final} # 构建图 workflow StateGraph(BlogState) workflow.add_node(generate, generate_node) workflow.add_node(review, review_node) workflow.add_node(revise, revise_node) workflow.add_edge(START, generate) workflow.add_edge(generate, review) workflow.add_edge(review, revise) workflow.add_edge(revise, END) app workflow.compile()编译好之后调用一次看看result app.invoke({topic: 家用咖啡机选购}) print(result[final])你会看到输出草稿加上修改意见说明三个节点按顺序执行了。这段代码就是完整的线性工作流START是图的入口END是出口。如果想再加一个节点只需要注册节点并连上边。注意节点函数不要修改传入的State本身而是返回要变更的字段。直接原地修改会导致LangGraph无法追踪状态变化在启用持久化后还会造成数据不一致。2.4 给线性流程加入条件分支上面的例子太顺利了现实中我们更希望审核没问题就直接发布有问题才去修改。这种“看情况走哪条路”的能力就是add_conditional_edgesdef route_after_review(state: BlogState) - str: if state.get(issues): return revise return publish workflow StateGraph(BlogState) workflow.add_node(generate, generate_node) workflow.add_node(review, review_node) workflow.add_node(revise, revise_node) workflow.add_node(publish, lambda state: {final: state[draft] [已发布]}) workflow.add_edge(START, generate) workflow.add_edge(generate, review) workflow.add_conditional_edges(review, route_after_review, {revise: revise, publish: publish}) workflow.add_edge(revise, END) workflow.add_edge(publish, END)这里route_after_review是一个普通函数输入当前State返回一个字符串表示下一站去哪。add_conditional_edges的第三个参数是路径映射表把函数返回值映射成图上真实节点的名字。这一步开始你已经拥有“Agent决策”的雏形了模型负责判断流程根据判断结果分叉。3. 深入State设计有状态编排的关键3.1 Reducer机制多节点如何合并同一个字段你在节点里返回{issues: [xxx]}时LangGraph默认会用新值覆盖旧的issues。这在有些场景下不对。比如多个节点分别发现不同问题你想把它们累积到一个列表里而不是后者覆盖前者。LangGraph提供了Reducer机制来解决这个问题。最常见的是把字段声明成可累加列表from typing import TypedDict, List, Annotated import operator class ReportState(TypedDict): content: str errors: Annotated[List[str], operator.add] scores: Annotated[List[int], operator.add]当你在两个不同节点里分别返回{errors: [A错误]}和{errors: [B错误]}最终State里的errors会变成[A错误, B错误]。operator.add就是预制的Reducer函数。你也可以写自己的Reducer比如取最大值、去重合并、只保留最近N条等。为什么这个细节重要因为它直接决定了你的Agent能不能在多轮、多节点协作中不丢数据。我把这理解为“多人协作时部门之间怎样合并报表”默认覆盖是“最新说了算”带Reducer是“大家的结果都算数”。设计State时先想清楚每个字段是要被覆盖还是被合并能省掉后面大量的数据bug。3.2 用MessagesState管理对话历史在上面例子中State里都是自定义字段。但做Agent对话时每个节点都要读取完整对话历史还要把模型输出追加进去。LangChain生态里有一个专用State类型叫MessagesStatefrom langgraph.graph import MessagesState class AgentState(MessagesState): 在对话消息基础上额外加一个用户ID字段 user_id: strMessagesState本质上是from typing import Annotated, TypedDict, List from langchain_core.messages import BaseMessage import operator class MessagesState(TypedDict): messages: Annotated[List[BaseMessage], operator.add]它自带一个messages字段使用operator.add合并策略。每次节点调用模型后把新生成的AIMessage追加到历史消息列表里。这样一来不同节点——比如“规划节点”“工具节点”“回答节点”——共享同一份对话历史谁需要都能读取完整上下文。我的建议是纯对话应用直接基于MessagesState扩展带状态机的业务流则自定义State。两者也可以混用不必非此即彼。3.3 一个能自我反思的Agent骨架既然有了条件边和消息状态我们就能做一个小而美的“反思型Agent”了。流程是先生成候选回答再让“评审”检查是否合格不合格就重新生成最多尝试三次。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) def generate_answer(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} def judge_answer(state: AgentState) - dict: last_message state[messages][-1].content judge_prompt f请判断以下回答是否完整{last_message}。如果合格回复PASS否则回复FAIL。 judgment llm.invoke([{role: user, content: judge_prompt}]).content print(评判结果:, judgment) return {judgment: judgment} # 需要先在State中定义judgment字段 def route_by_judgment(state: AgentState) - str: return generate if state.get(judgment, ).startswith(FAIL) else END graph StateGraph(AgentState) graph.add_node(generate, generate_answer) graph.add_node(judge, judge_answer) graph.add_edge(START, generate) graph.add_edge(generate, judge) graph.add_conditional_edges(judge, route_by_judgment, {generate: generate, END: END}) app graph.compile() result app.invoke({messages: [{role: user, content: 解释一下什么是依赖注入}]})这已经具备我们通常所说的“Agent循环”的基本结构模型输出、评估、条件判断、循环执行。在LangGraph里你不需要用while循环包裹模型调用图本身就是循环。4. Agent编排进阶工具调用与并行分支4.1 让模型真正“动手”接入工具节点只有当模型能调用外部工具时Agent才从“聊天机器人”进化成“干活的人”。LangGraph提供了ToolNode和tools_condition封装了“模型想调用工具就去执行工具执行完回到模型继续思考”的标准循环。from langchain_core.tools import tool from langgraph.prebuilt import ToolNode, tools_condition from langgraph.graph import StateGraph, START, END, MessagesState tool def get_weather(city: str) - str: 查询指定城市的当前天气 # 真实项目里这里接入天气API return f{city}今日晴温度26度 tool def multiply(a: float, b: float) - float: 计算两个数字的乘积 return a * b tools [get_weather, multiply] model_with_tools llm.bind_tools(tools) def call_model(state: MessagesState) - dict: response model_with_tools.invoke(state[messages]) return {messages: [response]} graph StateGraph(MessagesState) graph.add_node(model, call_model) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, model) graph.add_conditional_edges(model, tools_condition, {tools: tools, __end__: END}) graph.add_edge(tools, model) app graph.compile() result app.invoke({messages: [{role: user, content: 北京天气怎么样顺便算一下123乘以4}]}) print(result[messages][-1].content)这段代码里有两个重点第一个重点bind_tools把工具描述塞给模型。模型看到API表示“我想调用get_weather”返回一个tool_calls指令而不是直接输答案。第二个重点tools_condition是LangGraph预置的路由函数如果最新消息里有tool_calls就路由到tools节点执行真实函数执行完工具后边tools - model让模型看到工具结果继续回答用户。如果没有tool_calls就直接返回给用户走END。这就是最经典的ReAct循环而你不需要写任何while循环。注意tools_condition返回节点的字符串名称必须和我们注册的节点名一致。4.2 人工审核与高成本动作前的审批流工具调用加进来之后你很快会遇到一个现实问题Agent可以删除数据库记录、发邮件、下单你总不能让它自作主张。这时候让Agent在关键动作前“停下来等人工确认”就变得非常重要。LangGraph官方提供的做法是interrupt()。先记住一个前提interrupt()只有配合Checkpointer持久化才能用因为它需要把“在哪暂停、暂停时的状态是什么”写到存储里。from langgraph.types import interrupt, Command def place_order_node(state): 下单前需要人工审批 decision interrupt({ action: place_order, amount: state[amount], message: f确认给客户{state[customer]}下单金额{state[amount]} }) if decision approve: # 真正调用下单API return {status: approved, order_id: 202507001} else: return {status: rejected} graph StateGraph(MyState) graph.add_node(ask_llm, call_model) graph.add_node(place_order, place_order_node) graph.add_edge(START, ask_llm) graph.add_edge(ask_llm, place_order) graph.add_edge(place_order, END) app graph.compile(checkpointermemory_checkpointer) config {configurable: {thread_id: order-001}} result app.invoke({customer: 某公司, amount: 8999}, config)此时执行会停在下单节点。所有语言模型调用、之前节点的输出、工具结果全部被存档。恢复时你只需要用Command(resume...)传回人工决策final_result app.invoke(Command(resumeapprove), config)resume里的值会成为interrupt()的返回值。你可以传字符串也可以传结构化字典业务上更推荐后者Command(resume{action: approve, operator: 张三})。我实际项目里审批流还会配一个待办队列。前端把graph.get_state(config)返回的快照渲染给审批人看审批人点按钮后后台调用Command(resume...)。这套方案比自己在业务代码里做状态机要省力得多。4.3 并行分支与Map-Reduce加速批量任务有的流程天然可以并行。比如你让Agent写一篇行业综述它需要分别查市场数据、竞品动态、技术趋势三个任务互不依赖。LangGraph支持在一个节点后分叉成多条边多个节点可以并行跑最终汇合到一个合并节点。workflow.add_edge(START, research_market) workflow.add_edge(START, research_competitor) workflow.add_edge(START, research_tech) workflow.add_edge(research_market, aggregate) workflow.add_edge(research_competitor, aggregate) workflow.add_edge(research_tech, aggregate) workflow.add_edge(aggregate, END)在aggregate节点里你可以通过State拿到三个并行节点写回的结果字段拼接成完整报告。如果三个任务的数据量特别大更常用的做法是“Map-Reduce”把一个列表拆成N份每份跑同一个处理节点最后汇总。LangGraph官方文档里叫SendAPI适合微服务间批量调用场景这里先不展开。这里的底层逻辑是图是DAG有向无环图或带环图只要两个节点之间没有路径依赖LangGraph就会在可用资源内并行调度它们。这个特性并不是什么神秘魔法本质上是异步任务调度但它把你在业务代码里要写的并发逻辑统一收编到了图定义里。5. 中断恢复与持久化有状态Agent的生命线5.1 为什么持久化如此关键你现在应该理解了State在内存中就是Python对象进程重启就没了。但真实的Agent应用往往是常驻服务用户可能上午发起一个任务下午回来查看结果也可能中间网络断了几次。这时候你需要的不是简单的“把所有变量存数据库”而是把整个图的执行位置和完整上下文都保存下来以便从断点继续执行。持久化机制在LangGraph里叫Checkpointer它不只存State数据还会记录图内部执行到哪一步、哪些节点已完成、哪些节点在等待。这有点像游戏存档不只记录你还有多少血还记录你站在哪个地图哪个位置。5.2 三个Checkpointer怎么选Checkpointer存储位置适用场景MemorySaver内存本地调试、演示、临时测试SqliteSaverSQLite文件单机小项目、快速原型PostgresSaverPostgreSQL生产环境、多实例并发入门阶段首选MemorySaver一行代码就能唤起from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory)但如果服务进程重启MemorySaver里的存档会全部丢失。讲真它只适合演示。生产环境我建议直接用PostgresSaver因为多个服务实例可以共享同一个存档数据库用户请求被打到哪台机器都能读到同一份状态。SQLite版本的用法也很简单from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.sqlite) as saver: app graph.compile(checkpointersaver)注意MemorySaver不需要额外库但SqliteSaver需要langgraph-checkpoint-sqlite。如果你在Windows上遇到sqlite并发写锁问题优先换Postgres。5.3 thread_id按会话隔离状态同一个图编译出的应用可以被成千上万个用户同时使用。怎么区分“张三的存档”和“李四的存档”答案是config里的thread_id。每次启动/恢复一个任务时你传入config就相当于指定了用哪份存档。config1 {configurable: {thread_id: user-zhang-task-1}} config2 {configurable: {thread_id: user-li-task-1}} app.invoke({messages: [{role: user, content: 你好}]}, config1) app.invoke({messages: [{role: user, content: 帮我查天气}]}, config2)关键点在于同一个thread_id下的多次invoke是共享历史的。也就是说thread_id就是“跨轮对话的会话ID”。恢复中断时用的也是同一个thread_idLangGraph才能找到断点。我实际开发中会把thread_id设计成业务主键比如“订单编号”“客服工单ID”。这样天然就能按订单维度恢复Agent任务用户打电话进来客服输入订单号就能看到这个订单当前走到了哪一步。5.4 完整案例带人工审批和自动恢复的客服工单Agent把前面几个概念组装一下一个更接近生产环境的例子是这样的from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, START, END from langgraph.graph import MessagesState from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt, Command from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import ToolNode, tools_condition class TicketState(MessagesState): order_id: str status: str resolution: str tool def get_order_info(order_id: str) - str: 查询订单状态 return f订单{order_id}已发货物流单号SF123456 tools [get_order_info] model_with_tools llm.bind_tools(tools) def model_node(state: TicketState) - dict: response model_with_tools.invoke(state[messages]) return {messages: [response]} def escalate_node(state: TicketState) - dict: decision interrupt({ type: refund_approval, order_id: state[order_id], reason: 用户要求退款金额超过1000元 }) if decision approve: return {status: approved, resolution: 同意退款退款金额原路返回} return {status: rejected, resolution: 拒绝退款已与用户沟通优惠方案} workflow StateGraph(TicketState) workflow.add_node(model, model_node) workflow.add_node(tools, ToolNode(tools)) workflow.add_node(escalate, escalate_node) workflow.add_conditional_edges(model, tools_condition, {tools: tools, __end__: escalate}) workflow.add_edge(tools, model) workflow.add_edge(escalate, END) app workflow.compile(checkpointerMemorySaver()) config {configurable: {thread_id: ticket-10086}} # 第一次触发走到escalate就暂停 result1 app.invoke( {order_id: ticket-10086, messages: [{role: user, content: 我的订单还没到想申请退款}]}, config ) print(result1[status]) # 没有因为被中断了 # 人工同意退款 final app.invoke(Command(resumeapprove), config) print(final[resolution])这个案例把本节所有知识点串起来了模型自动决定是否调用工具必要时进入人工审批审批通过后继续执行。如果你在本地跑这段代码会看到第一次invoke触发中断后绝大多数字段其实已经持久化好了第二次Command(resumeapprove)会“接着”第一次的断点往下走而不是从头再来。6. 我踩过的坑与常见问题排查实录6.1 高频问题速查表问题现象最可能原因解决方案Interrupt requires a checkpointer使用interrupt()但没有compile(checkpointer...)给图加持久化存储如MemorySaver每次invoke都像第一次对话没有历史没有传thread_id或每次都换新thread_id固定thread_id作为会话标识两个节点返回同一字段后一个覆盖前一个字段没有配置Reducer用Annotated[字段类型, operator.add]或自定义Reducertools_condition路由报KeyError返回了未注册的节点名检查路径映射表确保映射目标真实存在恢复后节点重新执行了一遍副作用中断恢复机制会重新执行中断前的副作用节点将外部副作用放在interrupt()之后的节点中SQLite文件被锁SQLite并发写限制换PostgresSaver或使用连接池6.2 关于线程安全与并发状态的体会生产环境多用户并发时最容易被忽视的是同一个thread_id被两个请求同时操作。比如用户在Web界面点了一下“重试”前端又自动发起一次invoke两个请求就会竞争同一个存档。LangGraph本身不是分布式锁数据库层面可能会产生冲突。我的做法是在入口层面对thread_id做互斥单机用asyncio.Lock多实例用数据库行锁或分布式锁。这个问题不做防护线上必定出事故不是危言耸听。6.3 调试建议善用get_state和绘图如果你想知道“当前Agent到底卡在哪里”不要只打印日志直接查图的状态snapshot app.get_state(config) print(snapshot.values) print(snapshot.next)snapshot.next会告诉你下一个要执行哪个节点如果返回空列表说明流程已经进入END。配合graph.get_state_history(config)还能看到所有历史版本的状态快照这是排查中断恢复问题最有用的工具。如果需要可视化流程图LangGraph官方提供了Image方法Jupyter Notebook里可以渲染但我更推荐靠命名清晰的节点函数和条件路由来理解流程。图一旦复杂可视化图虽然酷维护时还是得靠代码。6.4 个人经验什么时候别硬上LangGraph最后说句实在话。如果你只是写一个一次性脚本、调几次大模型或者状态非常简单硬套LangGraph只会增加代码量。但如果是需要长期维护、多角色协作、要人工审批、要断点续跑的业务系统我强烈建议用LangGraph重新组织逻辑。我从第一次被线上事故教育之后把公司好几个核心流程都迁到了LangGraph上。回看代码最大的幸福感来自“不用自己写状态同步和异常恢复了”。每个节点都是纯函数输入State输出更新项图自动安排执行顺序。这是一种“结构化编程”的安心感。后续如果你要继续深入可以去看看Send API做动态并行了解Checkpointer下面的存储扩展模型以及LangGraph和LangSmith结合做链路追踪与可观测性。但先把这章里的线性工作流、条件路由、工具节点、中断恢复这几个概念玩熟LangGraph的骨架你就已经拿稳了。
