在生产级 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(checkpointer=memory) # ===================== 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_mode="values"): 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_mode="values"): 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_mode="values"): 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(checkpointer=memory, 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 落地的主流架构。
