在生产级 AI Agent 开发中长任务中断、接口超时崩溃、人工介入后重跑、重复调用付费 API 始终是核心痛点。LangGraph 原生提供的 Checkpoint 持久化与断点续跑能力彻底解决了这些问题 —— 通过节点级状态快照实现崩溃恢复、断点续执行、历史状态回溯、人机协同暂停续跑。一、为什么 Agent 必须做持久化绝大多数 AI Agent Demo 上线后都会遇到以下致命问题长任务执行中途程序崩溃、接口超时必须从头重跑重复消耗 API 算力与时间人工审核介入后无法从暂停点继续执行只能全流程重启多轮迭代任务出错后无法回溯到指定步骤重新调试排查成本极高多用户并发场景下会话状态混乱任务数据互相干扰。LangGraph 的Checkpoint 持久化机制就是针对这些问题的官方解决方案每执行完一个节点自动快照存储全量状态故障后精准从断点恢复无需重复执行已完成步骤是生产级 Agent 落地的必备基础能力。二、概念与原理2.1 术语定义Checkpoint检查点LangGraph 在每个节点执行完成后自动生成的全量状态快照包含当前所有状态数据、执行进度、节点历史断点续跑程序中断后基于已存储的 Checkpoint从最后一个完成的节点继续向下执行不重复运行已完成步骤持久化存储后端Checkpoint 的存储载体支持内存、SQLite、PostgreSQL、Redis 等多种方案Thread ID线程 / 会话 ID每个独立任务的唯一标识用于隔离不同会话的状态是多用户场景的核心参数。2.2 底层执行原理LangGraph 持久化的核心逻辑非常清晰工作流编译时绑定持久化存储组件每完成一个节点的执行自动将当前全局 State、执行链路、节点元数据写入存储后端生成 Checkpoint任务中断后使用相同 Thread ID 重新调用执行接口LangGraph 会自动读取最新 Checkpoint跳过已完成节点从断点处继续执行支持通过 Checkpoint ID 回溯到任意历史状态重新执行后续流程。2.3 主流存储后端对比存储方案适用场景优点缺点MemorySaver本地调试、单元测试零依赖、开箱即用、速度快程序重启数据丢失无持久化能力SQLiteSaver小型项目、单服务部署轻量、文件存储、无需额外服务并发性能弱不适合分布式部署PostgresSaver中大型生产环境、分布式部署高并发、稳定可靠、支持事务需要独立部署 PostgreSQL 服务RedisSaver高并发、短周期任务场景读写性能极高、支持过期自动清理数据持久化可靠性弱于关系型数据库三、环境准备基础环境要求 Python 3.8执行以下命令安装全套依赖# 安装LangGraph核心库 pip install langgraph # 安装SQLite持久化组件轻量首选无需额外服务 pip install langgraph-checkpoint-sqlite # 安装LangChain与OpenAI组件配合Agent使用 pip install langchain langchain-openai四、SQLite 持久化与断点续跑下面通过完整可运行的案例实现带 SQLite 持久化的工作流演示中断后断点续跑的完整流程。4.1 完整实现代码from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver import time # 1. 定义全局状态 class WorkflowState(TypedDict): task_name: str step: int result: str # 2. 定义工作流节点 def step_1(state: WorkflowState) - WorkflowState: print(执行节点1任务初始化) time.sleep(1) return {step: 1, result: 步骤1完成} def step_2(state: WorkflowState) - WorkflowState: print(执行节点2数据处理) time.sleep(1) return {step: 2, result: 步骤2完成} def step_3(state: WorkflowState) - WorkflowState: print(执行节点3结果输出) time.sleep(1) return {step: 3, result: 全部步骤完成} # 3. 构建工作流 builder StateGraph(WorkflowState) builder.add_node(step1, step_1) builder.add_node(step2, step_2) builder.add_node(step3, step_3) # 配置线性流程 builder.add_edge(START, step1) builder.add_edge(step1, step2) builder.add_edge(step2, step3) builder.add_edge(step3, END) # 4. 配置SQLite持久化 # checkpoints.db 为本地数据库文件自动创建 memory SqliteSaver.from_conn_string(checkpoints.db) # 编译时绑定持久化组件 graph builder.compile(checkpointermemory) # 5. 执行与断点续跑演示 if __name__ __main__: # 会话唯一ID区分不同任务/用户 config {configurable: {thread_id: task_001}} # 第一次执行模拟中途中断注释掉第二次执行即可测试中断场景 print( 第一次执行任务 ) initial_state {task_name: 测试持久化任务, step: 0, result: } for event in graph.stream(initial_state, config, stream_modevalues): print(f当前进度步骤{event[step]}状态{event[result]}) # 模拟执行到步骤2后程序崩溃中断 if event[step] 2: print(模拟程序崩溃中断执行) break # 第二次执行使用相同thread_id自动断点续跑 print(\n 断点恢复继续执行 ) # 无需传入初始状态自动从最新Checkpoint恢复 for event in graph.stream(None, config, stream_modevalues): print(f当前进度步骤{event[step]}状态{event[result]})4.2 执行结果说明运行代码后可以看到第一次执行完成节点 1、节点 2 后中断Checkpoint 自动写入 SQLite 数据库第二次执行时不会重复运行节点 1 和节点 2直接从节点 3 开始执行本地会自动生成checkpoints.db文件所有会话状态永久保存程序重启后依然可以续跑。五、实战5.1 查看历史 Checkpoint 与状态回溯支持查询指定会话的所有历史检查点回溯到任意节点重新执行# 获取指定会话所有Checkpoint列表 checkpoints list(memory.list(config)) print(f历史检查点数量{len(checkpoints)}) for cp in checkpoints: print(f检查点ID{cp[id]}对应步骤{cp[values][step]}) # 回溯到指定Checkpoint重新执行 # 取出步骤1对应的检查点 target_cp checkpoints[1] print(\n 回溯到步骤1重新执行 ) for event in graph.stream(None, {**config, configurable: {**config[configurable], checkpoint_id: target_cp[id]}}, stream_modevalues): print(f当前进度步骤{event[step]}状态{event[result]})5.2 配合 Human-in-the-loop 人工介入持久化是人机协同的核心基础关键节点暂停等待人工审核审核通过后从断点继续执行。from typing import Literal # 新增人工审核节点 def human_review(state: WorkflowState) - WorkflowState: print(进入人工审核环节流程已暂停) return state # 配置条件分支 def review_judge(state: WorkflowState) - Literal[step3, step2]: # 人工修改状态后判断是否通过 if state[result] 审核通过: return step3 return step2 # 重新构建带审核的工作流 builder StateGraph(WorkflowState) builder.add_node(step1, step_1) builder.add_node(step2, step_2) builder.add_node(human_review, human_review) builder.add_node(step3, step_3) builder.add_edge(START, step1) builder.add_edge(step1, step2) builder.add_edge(step2, human_review) builder.add_conditional_edges(human_review, review_judge) builder.add_edge(step3, END) # 编译时设置人工介入节点 graph builder.compile(checkpointermemory, interrupt_before[human_review]) if __name__ __main__: config {configurable: {thread_id: review_task_001}} # 执行到人工审核节点自动暂停 print( 启动任务等待人工审核 ) graph.invoke({task_name: 人工审核任务, step: 0, result: }, config) # 人工修改状态模拟审核通过 current_state graph.get_state(config) graph.update_state(config, {result: 审核通过}) # 从暂停点继续执行 print( 审核通过继续执行 ) result graph.invoke(None, config) print(f最终结果{result[result]})5.3 多会话隔离不同任务 / 用户使用不同的thread_id状态完全隔离互不干扰完美适配多用户生产场景# 用户1的任务 config1 {configurable: {thread_id: user_001_task_001}} graph.invoke({task_name: 用户1的任务, step: 0, result: }, config1) # 用户2的任务 config2 {configurable: {thread_id: user_002_task_001}} graph.invoke({task_name: 用户2的任务, step: 0, result: }, config2) # 两个任务状态独立存储互不影响六、最佳实践6.1 存储选型建议本地开发、单实例小型服务优先使用 SQLiteSaver零运维成本中大型分布式、高并发生产环境使用 PostgresSaver保证事务一致性与并发性能短周期、高频率临时任务使用 RedisSaver配合过期策略自动清理历史数据。6.2 状态优化建议不要在 State 中存储大文件、二进制数据大对象会大幅降低 Checkpoint 读写性能建议存储文件路径 / URL大文件单独存对象存储精简状态字段仅保留必要的业务字段冗余数据不存入全局 State敏感数据加密State 中包含用户隐私、密钥等敏感信息时写入存储前需做加密处理。6.3 运维与安全定期备份 Checkpoint 数据库避免数据丢失配置历史 Checkpoint 清理策略长期运行的服务定期清理过期会话数据控制存储体积生产环境禁止使用 MemorySaver程序重启会丢失所有状态仅可用于本地调试。6.4 性能优化节点粒度适中节点拆分过细会导致 Checkpoint 写入频繁影响性能粒度过大会增加断点重跑的重复成本高并发场景下使用连接池配置数据库连接避免频繁创建销毁连接。七、常见问题断点续跑重复执行节点原因节点执行过程中崩溃Checkpoint 仅在节点执行完成后写入崩溃节点不会生成快照续跑时会重新执行该节点解决保证节点逻辑幂等重复执行不会产生副作用是 Agent 开发的核心规范。SQLite 数据库锁定报错原因SQLite 是文件级锁高并发写入会触发锁冲突解决高并发场景切换为 PostgreSQL单实例场景控制并发数避免多线程同时写入同一个库文件。续跑时状态丢失原因未使用相同的thread_id或者编译工作流时未绑定对应的 checkpointer解决保证续跑时thread_id与首次执行完全一致工作流编译配置与首次保持相同。修改工作流代码后无法续跑原因工作流节点、边结构变更后旧的 Checkpoint 与新的工作流结构不兼容解决结构迭代后新任务使用新的会话 ID历史任务仅做查询不续跑。八、总结LangGraph 的 Checkpoint 持久化是生产级 Agent 的核心基础能力通过节点级状态快照实现断点续跑、人工介入、状态回溯、多会话隔离四大核心功能入门开发优先使用 SQLiteSaver 零成本接入生产环境根据并发规模选择 PostgreSQL 或 Redis 存储后端持久化开发核心规范保证节点逻辑幂等、精简状态字段、敏感数据加密、合理控制节点粒度结合 Human-in-the-loop 机制可实现「AI 自动执行 人工关键审核」的企业级工作流是当前 AI 落地的主流架构。
)
