做完十几个 Agent demo 之后你会发现一个残酷的事实在 Jupyter Notebook 里跑得挺顺的智能体一上生产就原形毕露。用户刷新一下页面任务重跑一遍钱扣了两次凌晨三点模型调用超时整个流程从头再来前面写进数据库的状态全得重建更头疼的是人工审批环节——Agent 运行到一半要等领导点个按钮结果进程一重启图的状态没了。LangGraph 断点恢复和幂等执行就是专门治这三类毛病的。这篇文章我把这套东西从原理到落地一次讲透代码能直接抄。在正式动手之前有必要先说说 LangGraph 在整个技术栈里的位置。现在搜 LangGraph 教程十个里有八个会纠结它和 LangChain 的区别。我的理解很简单LangChain 是一把瑞士军刀里面全是工具函数、模型封装、Prompt 模板这些零件而 LangGraph 是装刀的战术背心它关心的是你身上这些装备怎么按顺序掏出来、掏到一半被打断能不能塞回去、事后能不能从某个位置继续掏。它通过状态图的方式组织 Agent 的执行流让每一步都有迹可循、可停可续而且把状态和检查点作为一等公民内置进了框架。本文会从断点恢复的底层机制讲起然后进入幂等执行这个工程化绕不开的命题最后用一个 FastAPI LangGraph 的实战项目把它们串起来。这个方案适合正在把 Agent 从 Demo 推向生产的开发者也适合被AI 下地干活折磨得怀疑人生的后端工程师。1. 先搞清楚一个前提LangGraph 到底在解决什么问题1.1 LangGraph 和 LangChain 的本质区别很多人问 LangChain 和 LangGraph 的面试题怎么答其实就是一行话的事LangChain 提供了与模型、工具、文档交互的抽象能力LangGraph 则是把这些能力组织成一张可执行、可暂停、可恢复的状态图。LangChain 的 Chain 也有顺序执行的逻辑但这种执行是一把梭的——链一旦启动要么跑完要么失败重来中间不保留可以被外部干预的状态。LangGraph 则把执行过程建模成由节点Node和边Edge构成的有向图每个节点就是一次计算或一个工具调用边定义了流转规则。关键差异在于LangGraph 里的图在每次节点运行前后都会生成一个状态快照也就是 checkpoint你可以通过它实现时间旅行、条件分支重放以及人工介入。这种差异带来的直接感受是用 LangChain 写 Agent像在流水线上干活从头到尾一气呵成用 LangGraph 写 Agent像在玩有存档的游戏任何时刻都能存一档、读一档、改一档再继续玩。建议学习路径也很直接先用pip install langgraph跑通官方手册中文版里的 ReAct 示例理解节点、边、状态这三个核心概念再看本篇文章涉及的状态持久化和中断机制最后再上手真实业务改造。1.2 断点恢复的本质把图执行变成可中断事务断点恢复听起来很高端本质上就是数据库事务里 Savepoint 和 Rollback 的思想搬到了 Agent 编排层。LangGraph 在执行一个节点前会检查是否有可恢复的 checkpoint如果有它不会从头跑而是直接恢复到上次中断后的节点继续执行。这里涉及三个核心概念thread_id、checkpointer、checkpoint_id。thread_id 是会话标识LangGraph 用它区分不同的对话和任务同一个 thread 的多次执行共享状态checkpointer 是存储后端决定 checkpoint 和状态持久化到哪里比如内存、SQLite、Postgrescheckpoint_id 则是每次图执行生成的状态版本号相当于游戏存档的时间戳。你可以这样理解thread_id 是游戏存档的文件名checkpointer 是硬盘checkpoint_id 是存档的时间点。一个 Agent 应用对应多个 thread_id每个 thread 有自己的一系列 checkpoint_id当你用同一个 thread_id 再次调用图时LangGraph 会找最新的 checkpoint 作为起点。这套机制天然适合 human-in-the-loop 场景Agent 执行到需要人工审批的节点先暂停把状态写好存档然后告诉外部我需要人来确认人类做出决定后应用再用同一个 thread_id 调用续跑接口LangGraph 会从断点接着走而不是把之前的工作推倒重来。2. 断点恢复的落地姿势从图级断点到节点级中断2.1 编译期断点 interrupt_before / interrupt_after用 LangGraph 实现断点最直观的方式是在编译图的时候指定断点位置。interrupt_before表示进入指定节点前暂停interrupt_after表示离开指定节点后暂停。这种方式适合流程固定的场景比如必须先审核、后下单。代码层面非常简单from langgraph.graph import StateGraph, START, END # 假设我们定义了一个简单的 Agent 图 graph StateGraph(MyState) graph.add_node(plan, plan_node) graph.add_node(execute, execute_node) graph.add_node(review, review_node) graph.add_edge(START, plan) graph.add_edge(plan, execute) graph.add_edge(execute, review) graph.add_edge(review, END) checkpointer SqliteSaver.from_conn_string(checkpoints.db) # 编译时指定断点 app graph.compile( checkpointercheckpointer, interrupt_before[review], # 进入 review 前停住 interrupt_after[execute], # execute 完成且写入状态后停住 )这样编译完之后你调用app.invoke(input, config{configurable: {thread_id: order_001}})图会一路执行到 execute 节点然后停在进入 review 之前。此时图的状态保存在 checkpoints.db 里进程挂了也不怕。要恢复只需要再次调用同一个图不需要重新传完整输入。很多人这里会踩坑以为恢复是重新调用一次原来的输入其实 LangGraph 的逻辑是只要 thread_id 一致且该 thread 存在未完成的执行调用就会尝试从最近的 checkpoint 续跑。可以想象成一个断点续传的过程恢复时只需要传你需要注入的外部决定比如审批结果result app.invoke( None, # 或传审批结果取决于业务逻辑 config{configurable: {thread_id: order_001}} )执行会从interrupt_before指定的节点重新开始先执行 review再继续后续边。编译期断点的缺点是死板一旦编译就固定了如果业务流程是动态的这种方式就不够灵活。2.2 节点内中断 interrupt() 与 Command(resume)LangGraph 提供了更灵活的动态中断方式在节点内部调用interrupt()函数。这个函数的作用相当于在任意指定的执行点举手暂停它会把一个 payload 暴露给外部系统比如前端页面等待外部通过Command(resume...)把决定塞回来。说一个真实业务Agent 在帮用户下单前需要确认价格是否可接受。你在确认价格这个节点里写from langgraph.types import interrupt, Command def confirm_price_node(state): # 准备好给用户看的报价信息 payload { order_id: state[order_id], total_price: state[total_price], items: state[items], } # 中断把 payload 暴露出去等待用户确认 decision interrupt(payload) if decision.get(approved): return {status: approved} else: return {status: rejected, reason: decision.get(reason)}interrupt()被调用后图的执行立刻暂停不会继续往下走。此时如果你用app.get_state(config)查看状态会发现图处于中断状态且可以拿到interrupt里的 payload。外部应用比如 FastAPI 接口把这个 payload 展示给用户用户点同意或拒绝应用把决定作为参数传回from langgraph.types import Command # 用户点了“同意” app.invoke( Command(resume{approved: True}), config{configurable: {thread_id: order_001}} )Command(resume...)会唤醒中断的图把值传给interrupt()的返回值。执行会接着 confirm_price_node 往下走。这种方式实现了真正意义上的动态暂停和外部介入LangGraph 官方叫它 human-in-the-loop是断点恢复最强的一个用法。2.3 状态检查、修改与跳过执行除了暂停和恢复我们经常需要在恢复前修改图的状态或者临时跳过某个节点。LangGraph 提供了get_state和update_state两个接口用途很像数据库里的查询和 UPDATE。config {configurable: {thread_id: order_001}} # 查看当前 state 和 checkpoint snapshot app.get_state(config) print(snapshot.values) # 当前状态字段 print(snapshot.next) # 下一步会执行哪些节点 # 手动修改 state app.update_state(config, {customer_name: 张三}, as_nodeplan)修改 state 的值会生成一个新的 checkpoint后续执行基于新的状态继续。这里有个细节update_state时你可以通过as_node指定以哪个节点的名义写入这会影响图的状态更新规则。如果你希望跳过某个节点可以在中断后手动 update_state把该节点要写入的状态直接写进去再恢复执行即可。日常开发里我通常把查看状态 人工修改 恢复执行三个动作做成三个独立 API方便前端根据业务场景自由组合。从实践来看这种设计比把所有逻辑都塞进一次 invoke 调用更可靠。3. 幂等执行让 Agent 重复跑也不会出事3.1 为什么 LangGraph 不保证幂等断点恢复解决了流程中断的问题但引出了下一个问题既然执行可以被暂停、恢复、甚至重放那么同一个节点被重复执行怎么办LangGraph 本身没有内置幂等机制。原因很直接框架层面无法判断你的工具调用是不是幂等的。比如一个节点里调用了发送短信的接口这个接口天然不是幂等的但另一个节点里只是做一次本地计算重复执行也没关系。框架不知道这些所以它选择把决定权交给你。但是 LangGraph 提供了实现幂等的关键信息。每次 invoke 都会生成一个 runtime run_id每次节点执行后生成 checkpoint_id。你可以把它们当作执行某段逻辑的流水号结合业务侧的幂等键就能判断某段副作用是否已经执行过。这里强调一句config里传入的thread_id是会话维度不是请求维度同一 session 里有多次 invokerun_id 每次都会变。所以不要拿 run_id 或 checkpoint_id 当业务幂等键它们只适合做内部调试和状态追踪。3.2 幂等设计三件套幂等键、副作用登记、唯一约束要让 Agent 的执行变得幂等我的实战经验是用三件套入口生成幂等键、副作用处先登记后执行、数据库加唯一约束。第一步在 Agent 任务创建时生成全局唯一的幂等键这个键要贯穿整个执行流通常叫run_id但注意是自己生成的业务 run_id不是 LangGraph 内部的 run。可以放在 state 最外层一路往下传。第二步在节点执行副作用发消息、扣积分、调用外部下单接口之前先往 operation_log 表插一条记录字段包括 run_id、操作类型、操作参数、状态。如果插入的时候抛唯一约束冲突说明同一 run 下这个操作已经执行过直接跳过副作用逻辑并复用上次的结果。第三步在数据库里给(run_id, op_key)建立唯一索引。这里的 op_key 可以是通知用户扣减库存这样的操作标识也可以是 action 名称加参数哈希。这样即使代码逻辑漏判了数据库也会拦住第二次执行。落地到 LangGraph 节点里基础设施可以做成一个装饰器def idempotent_node(op_key): def decorator(func): def wrapper(state): run_id state[run_id] log lookup_operation(run_id, op_key) if log: return log[result] # 已执行过直接返回缓存结果 result func(state) # 首次执行副作用 record_operation(run_id, op_key, result) # 写执行记录 return result return wrapper return decorator idempotent_node(deduct_inventory) def deduct_inventory_node(state): # 真正的扣减逻辑 return {inventory_left: state[inventory] - 1}这套方案的巧妙之处在于它把幂等从框架层下沉到了业务层。无论 LangGraph 怎么重放节点、怎么断点续跑只要 run_id 不变任何副作用都会被数据库的唯一约束挡住。3.3 工具调用层的幂等LangGraph 里最常见的副作用集中在工具调用上。函数调用tool calling模型会给每个工具调用生成一个唯一的tool_call_id这个 ID 在单轮对话内是唯一的。但问题是如果图执行重放同一个工具调用可能会被再次触发LangGraph 自带的 ToolNode 会再次执行该工具。好在 LangGraph 的 ToolNode 有内置的消息去重机制。当你用langgraph.prebuilt.ToolNode时如果某个tool_call_id已经出现在历史消息里ToolNode 会直接使用历史结果而不会再次执行工具。这个行为依赖于 checkpointer 保存的消息列表。所以只要你的图启用了 checkpointer并且工具节点用的是官方 ToolNode一定程度上已经具备按 tool_call_id 去重的能力。但如果你没用 ToolNode而是自己写节点处理工具调用那就必须自己实现幂等。我的做法是在工具执行前检查当前 tool_call_id 是否已经在状态里的 tool_results 字段中出现过出现过就直接拿结果没出现过才执行真正的调用。def smart_tool_executor(state): messages state[messages] last_ai_msg messages[-1] results [] for tool_call in last_ai_msg.tool_calls: existed state[tool_results].get(tool_call[id]) if existed is not None: results.append(existed) # 复用历史结果 continue result real_executor.invoke(tool_call) # 真正执行 state[tool_results][tool_call[id]] result results.append(result) return {tool_results: state[tool_results], messages: results}这套逻辑在你需要给工具调用加缓存、加审计、加限流的场景下更实用官方 ToolNode 的自动去重就不够用了。特别是当你对接的资金、积分系统有自己一套幂等凭证体系时用 tool_call_id 做联动往往比重新建一套更直接。4. 落地实战FastAPI LangGraph 的生产级断点恢复服务4.1 架构与存储选型从 SQLite 到 Postgres断点恢复依赖于 checkpointer所以选什么存储就决定了你能恢复到什么程度以及能扛多大并发。我把常用三个存储后端放在一起对比根据部署规模直接挑存储后端典型场景并发能力生产可用性备注MemorySaverDemo、本地调试单进程单线程低进程重启状态全丢慎用于生产SqliteSaver单机小规模服务受限于单机中注意线程锁和文件锁配套代码简单门槛低PostgresSaver多副本、高可用环境高高是生产推荐方案需要数据库连接池和迁移脚本单机场景直接上 SQLite 就够了连接字符串传一个路径即可。多副本部署、要做负载均衡的话建议直接上 Postgres。LangGraph 提供了 AsyncPostgresSaver异步接口配合 FastAPI 的 async 路由非常顺滑。需要强调一点postgres_saver 使用前必须创建表结构官方提供了checkpoint_ddl脚本不要用 SQLite 的表结构去 Postgres 里跑字段类型和索引都对不上踩一次坑少说浪费半小时。4.2 核心代码构建带审批的人类介入 Agent用 FastAPI LangChain LangGraph 组合做一个审批后通知的 Agent。业务路径是创建任务 - 模拟扣减积分 - 人工审批 - 审批通过后发通知。重点演示断点恢复与幂等抑制。# app.py —— 完整示例骨架 import os from typing import TypedDict, Annotated from fastapi import FastAPI from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.types import interrupt, Command import sqlite3 class AgentState(TypedDict): run_id: str user_id: str points_deducted: bool approval: str notified: bool conn sqlite3.connect(agent_state.db, check_same_threadFalse) conn.execute( CREATE TABLE IF NOT EXISTS operation_log ( run_id TEXT, op_key TEXT, result TEXT, PRIMARY KEY (run_id, op_key) ) ) conn.commit() checkpointer SqliteSaver(conn)注意连接池的check_same_threadFalse默认的 SQLite 连接不能跨线程用FastAPI 是并发模型不关掉会各种报 thread error。下面定义三个节点。第一个节点扣积分有工程量第二个节点做人工审批中断第三个节点发通知幂等保护。def deduct_points_node(state: AgentState): run_id state[run_id] # 幂等查操作日志 row conn.execute( SELECT result FROM operation_log WHERE run_id? AND op_key?, (run_id, deduct_points), ).fetchone() if row: return {points_deducted: True} # 已扣过不重复扣 # 真正扣减积分的代码这里用 print 模拟 print(f真实扣减积分user{state[user_id]}) conn.execute( INSERT INTO operation_log (run_id, op_key, result) VALUES (?, ?, ?), (run_id, deduct_points, ok), ) conn.commit() return {points_deducted: True} def approval_node(state: AgentState): decision interrupt({ message: 是否允许扣减积分并发送通知, user_id: state[user_id], run_id: state[run_id], }) return {approval: decision.get(decision, denied)} def notify_node(state: AgentState): if state[approval] ! approved: return {notified: False} run_id state[run_id] row conn.execute( SELECT result FROM operation_log WHERE run_id? AND op_key?, (run_id, send_notify), ).fetchone() if row: return {notified: True} print(f真实发送通知user{state[user_id]}) conn.execute( INSERT INTO operation_log (run_id, op_key, result) VALUES (?, ?, ?), (run_id, send_notify, ok), ) conn.commit() return {notified: True}构建图并编译builder StateGraph(AgentState) builder.add_node(deduct, deduct_points_node) builder.add_node(approval, approval_node) builder.add_node(notify, notify_node) builder.add_edge(START, deduct) builder.add_edge(deduct, approval) builder.add_edge(approval, notify) builder.add_edge(notify, END) app builder.compile(checkpointercheckpointer)FastAPI 路由部分提供三个接口from fastapi import FastAPI, HTTPException server FastAPI() server.post(/runs) def create_run(user_id: str): import uuid run_id str(uuid.uuid4()) # 启动图执行运行到 approval 节点会自动中断 app.invoke( {run_id: run_id, user_id: user_id}, config{configurable: {thread_id: run_id}}, ) return {run_id: run_id} server.get(/runs/{run_id}/state) def get_run_state(run_id: str): snap app.get_state({configurable: {thread_id: run_id}}) return { status: snap.next, values: snap.values, interrupts: [i.payload for i in snap.interrupts] if snap.interrupts else [], } server.post(/runs/{run_id}/resume) def resume_run(run_id: str, decision: str): snap app.get_state({configurable: {thread_id: run_id}}) if not snap.interrupts: raise HTTPException(status_code400, detail当前状态不可恢复) app.invoke( Command(resume{decision: decision}), config{configurable: {thread_id: run_id}}, ) return {status: resumed}这样前后端就能完成创建任务 - 查询审批信息 - 审批 - 恢复执行。整个生命周期里SQLite 里的 checkpoint 保存了每一步状态operation_log 确保了扣减积分和发通知这两个副作用不会被重复执行。4.3 完整调用流程演示用 curl 走一遍完整流程你会很直观看到断点恢复和幂等是怎么协同的。创建一次任务图会一路跑到 approval 节点然后自动暂停curl -X POST http://localhost:8000/runs?user_idu_100 # 返回 {run_id: abc-123}查询状态你会看到interrupts里有审批的 payloadnext指向 approval 之后要执行的下一个节点curl http://localhost:8000/runs/abc-123/state人工审批通过恢复执行curl -X POST http://localhost:8000/runs/abc-123/resume \ -H Content-Type: application/json \ -d {decision: approved}此刻的关键点来了如果你再次手动调用扣积分节点比如误操作operation_log 里已经存在(abc-123, deduct_points)这条记录节点会直接返回{points_deducted: True}不会真的扣第二次积分。同理send_notify也不会重复发通知。这个就是业务层的幂等兜底。如果把整条链路里的每一步都记录到数据库你甚至可以做重放审计——某个 run 到底扣了多少次积分、哪一步是幂等跳过的、哪一步真实执行了一查便知。5. 实战中的坑与排查经验5.1 断点恢复常见的报错与解决办法工程落地会遇到各种边界情况我把自己常遇到的几个问题整理成了速查表报错/现象根因处理方式Checkpointer required图里使用了 interrupt 但在 compile 时没传 checkpointercompile(checkpointer...)恢复调用不生效从头开始执行thread_id 不一致或使用了不同的 config key确认 token 不变检查 config 拼写Cannot update non-existent node...update_state 时 as_node 传了一个图里不存在的节点名查看图节点列表确保 as_node 合法resume 后状态不是预期值Command(resume) 传参结构不对检查 interrupt() 返回值的接收语义SQLitedatabase is locked多线程并发写同一个 SQLite 文件开启 WAL 模式或改用 PostgresSaver恢复后节点重复执行业务副作用没有做幂等保护用 3.2 节的三件套去拦截SQLite 的锁是最常见的坑尤其是 FastAPI 多 worker 部署时。建议在初始化连接后执行PRAGMA journal_modeWAL;能明显减少并发读写的锁冲突。但说到底多 worker 部署就别用 SQLite 了上 PostgresSaver 是正道。5.2 幂等没生效的几个隐蔽原因幂等逻辑看着简单实际有几种隐蔽情况会导致失效。最常见的是幂等键没有贯穿整个调用链。比如你的 run_id 在 create_run 接口生成了但某个工具节点内部又自己 new 了一个 uuid那节点侧查 operation_log 时主键永远对不上每次都会执行真实副作用。建议把幂等键放到 state 的顶层字段节点里只管 read不许 write。第二个隐蔽原因是副作用执行成功但记录写入失败。比如真实扣减积分接口调用成功了但 operation_log 的 INSERT 因为某种异常没提交此时 Agent 抛错重试业务逻辑发现日志里没有记录又扣了一次。解决办法是把副作用执行和副作用登记放在同一个事务边界里最好的方式是把扣减积分与记录日志放到同一个数据库事务要么都成功要么都失败。第三个隐蔽原因是参数变化导致的幂等绕过。比如 op_key 只取了操作名但操作里包含了金额参数第一次扣 100第二次改成扣 50唯一约束认为这是同一条记录直接跳过了扣 50 的请求。这就要求 op_key 的设计要包含关键的参数指纹一般是用action hashlib.md5(sorted_params)。5.3 生产部署检查清单最后给一份我每次上线 Agent 服务前都会过一遍的清单照着做能少走很多弯路生产环境 checkpointer 是否选择了 PostgresSaver并创建了正确的 checkpoint 表所有有外部副作用的节点是否都经过幂等装饰器或等效逻辑保护operation_log表是否建了(run_id, op_key)唯一索引关键工具调用是否用 tool_call_id 做了去重或缓存恢复接口是否做了并发控制避免同一个 thread_id 被两个人同时 resume是否对get_state、update_state做了操作审计方便排查人工干预的记录是否有兜底超时机制比如 Agent 长期处于中断状态或恢复失败时有守护任务负责清理或告警关于第 4 条里提到的并发控制简单做法是在恢复接口层加一把 Redis 锁锁的 key 是resume:{thread_id}保证同一时刻只有一个 resume 请求真正驱动图继续执行。否则两个请求同时Command(resume...)状态会乱成一锅粥。6. 断点状态设计与幂等键命名的一些心得这里讲一个容易被忽略的细节断点恢复时interrupt()的返回值本质上是从Command(resume)里透传过来的它不会自动做校验。如果你在审批节点里期望收到一个 dict但恢复端传了字符串节点代码可能直接报TypeError。所以我习惯在 interrupt 节点里加一层简单的 schema 校验比如用 Pydantic 解析解析失败就抛一个自定义异常FastAPI 层捕获后返回 400。这能避免生产环境出现匪夷所思的恢复错误。幂等键命名同样有讲究。业界惯例会区分request_id客户端发起请求时生成的 ID用于端到端全链路追踪。idempotency_key专门用于幂等控制的业务键同一业务动作多次重试时保持不变。run_idAgent 任务内部流转用的标识可以就是idempotency_key但注意它不是 LangGraph 框架的 runtime run_id。我通常直接让入口生成的 UUID 同时充当request_id、idempotency_key和thread_id三合一减少概念数量降低沟通成本。你可以在日志里同时打印这个值和 LangGraph 内部的 checkpoint_id方便追踪状态对应关系。# 日志示例 [RUN abc-123] checkpoint 9f2e... 到达人工审批 [RUN abc-123] checkpoint 9f2e... 恢复执行审批结果 approved [RUN abc-123] 幂等跳过 send_notify原因已执行最后再分享一个我在实战中坚持的习惯所有 LangGraph 节点都写成纯函数风格不直接操作外部全局状态只通过 state 的输入输出做数据流转。副作用统一收敛到独立的工具层或服务层节点只做编排。这样做的好处是当你要加断点、加幂等、加审计时改动面非常小每个节点就像积木一样可以自由插拔组合。这套理念配合 LangGraph 的 checkpointer基本能满足绝大多数生产级 Agent 服务的需求。