开始学习
课程 › 0基础Agent开发课 / LangGraph / LangGraph-Persistence持久化与断点续跑
— 20 min read

LangGraph-Persistence持久化与断点续跑

任何需要几分钟以上的 Agent 任务，如果不做持久化，都是在赌运气。服务器重启、容器 OOM、网络超时，任何意外都会让任务从头来过。生产环境没有"重跑"这个选项，或者说代价很高。

LangGraph Persistence：持久化与断点续跑

任何需要几分钟以上的 Agent 任务，如果不做持久化，都是在赌运气。服务器重启、容器 OOM、网络超时，任何意外都会让任务从头来过。生产环境没有"重跑"这个选项，或者说代价很高。

LangGraph 的 Checkpointer 机制就是为这个问题准备的。

1.1 为什么需要持久化

LangGraph 持久化架构图
LangGraph Checkpointer 工作流程 — 每个节点执行后自动保存状态快照，支持断点续跑和时间旅行

除了防止意外中断，持久化还解决了三类问题。

多轮对话的状态保存。用户今天和 Agent 聊了一半，明天继续聊，Agent 要记得上次聊到哪了。没有持久化，每次对话都是从零开始。

长时间任务的恢复。处理一批文档、分析一个大型代码库、生成一份长报告——这类任务可能跑几分钟甚至几十分钟。任何中断都应该能从最近的断点继续，而不是从头来过。

调试和回溯。出了问题，需要能看到 Agent 在每一步的状态是什么，才能定位问题。持久化的 State 快照就是完整的执行日志。

1.2 Checkpointer 的工作原理

LangGraph 在每个节点执行完之后，会自动把当前的完整 State 保存一个快照，这个快照叫 Checkpoint。

Checkpoint 的保存依赖 Checkpointer 对象，它负责实际的存储操作。LangGraph 内置了几种实现，也支持自定义实现。

每个 Checkpoint 有两个关键标识：

  • thread_id：这次任务/对话的唯一 ID。同一个 thread_id 的所有 Checkpoint 属于同一条执行链，串联起来就是完整的执行历史。
  • checkpoint_id：每个 Checkpoint 快照的唯一 ID，由 LangGraph 自动生成。通过 checkpoint_id 可以定位到任意一个历史快照，从那个状态重新执行。

使用时，每次调用 invoke 或 stream 都要通过 config 传入 thread_id：

python 
config = {"configurable": {"thread_id": "task_001"}}

# 第一次调用
result = app.invoke(initial_state, config=config)

# 如果中途崩了，第二次调用同一个 thread_id，LangGraph 会从最近的 Checkpoint 恢复
result = app.invoke(None, config=config)

传 None 作为输入，LangGraph 会找到这个 thread_id 最新的 Checkpoint，从那个节点的下一步继续执行，而不是从头开始。

1.3 三种 Checkpointer

1.3.1 MemorySaver

状态保存在内存里，进程重启后全部丢失。

python 
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

只用于开发阶段的调试。好处是零配置，坏处是生产不能用。

1.3.2 SqliteSaver

状态保存在 SQLite 文件里，进程重启后可以恢复。适合单机部署、本地工具、个人项目。

需要额外安装：

bash 
pip install langgraph-checkpoint-sqlite
python 
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("agent_checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

check_same_thread=False 是 SQLite 的参数，允许多线程访问同一个连接，LangGraph 异步调用时需要这个。

1.3.3 PostgresSaver

状态保存在 PostgreSQL 里，支持多实例并发访问。适合生产环境、需要水平扩展的部署场景。

需要额外安装：

bash 
pip install langgraph-checkpoint-postgres psycopg[binary]
python 
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/agent_db"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    # 在 with 块内使用 app

生产环境里，PostgreSQL 是最稳妥的选择。多个服务实例可以共享同一个 PostgreSQL，thread_id 保证了不同任务之间的数据隔离。

1.4 完整示例：文档分析 Agent 的持久化与恢复

下面是一个完整示例，演示三件事：用 SqliteSaver 持久化、模拟中途崩溃后恢复、查看历史状态。

python 
import sqlite3
import time
from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver


# ============================================================
# 1. 定义 State
# ============================================================

class DocumentAnalysisState(TypedDict):
    documents: List[str]          # 待分析的文档列表
    processed_docs: List[str]     # 已处理的文档摘要
    current_index: int            # 当前处理到第几个文档
    comparison_result: str        # 横向比对结果
    final_report: str             # 最终报告


# ============================================================
# 2. 定义节点
# ============================================================

def process_documents_node(state: DocumentAnalysisState) -> dict:
    """逐篇处理文档，每次处理一篇"""
    docs = state["documents"]
    current_index = state["current_index"]
    processed = list(state["processed_docs"])  # 复制，避免引用问题

    total = len(docs)
    print(f"\n[处理节点] 开始处理文档 {current_index + 1}/{total}")

    if current_index >= total:
        print("[处理节点] 所有文档已处理完毕")
        return {}

    # 处理当前这篇文档
    doc = docs[current_index]
    # 实际项目里这里调用 LLM 提取关键信息
    summary = f"文档{current_index + 1}摘要：{doc[:30]}... 的核心观点"
    processed.append(summary)

    print(f"[处理节点] 文档 {current_index + 1} 处理完成")

    return {
        "processed_docs": processed,
        "current_index": current_index + 1,
    }


def should_continue_processing(state: DocumentAnalysisState) -> str:
    """判断是否还有文档需要处理"""
    if state["current_index"] >= len(state["documents"]):
        return "done"
    return "continue"


def compare_documents_node(state: DocumentAnalysisState) -> dict:
    """横向比对所有文档的摘要"""
    print(f"\n[比对节点] 对 {len(state['processed_docs'])} 篇文档进行横向比对")

    # 实际项目里调用 LLM 做比对分析
    comparison = "横向比对结果：所有文档均涉及核心主题，存在以下共同点和差异..."

    return {"comparison_result": comparison}


def generate_report_node(state: DocumentAnalysisState) -> dict:
    """生成最终分析报告"""
    print(f"\n[报告节点] 生成最终报告")

    # 实际项目里调用 LLM 综合生成报告
    report = (
        f"文档分析报告\n"
        f"共分析 {len(state['processed_docs'])} 篇文档\n\n"
        f"各文档摘要：\n" +
        "\n".join(f"  - {s}" for s in state["processed_docs"]) +
        f"\n\n比对分析：\n{state['comparison_result']}"
    )

    return {"final_report": report}


# ============================================================
# 3. 构建图
# ============================================================

def build_analysis_agent(checkpointer):
    graph = StateGraph(DocumentAnalysisState)

    graph.add_node("process", process_documents_node)
    graph.add_node("compare", compare_documents_node)
    graph.add_node("report", generate_report_node)

    graph.add_edge(START, "process")

    # 处理节点循环，直到所有文档处理完
    graph.add_conditional_edges(
        "process",
        should_continue_processing,
        {
            "continue": "process",
            "done": "compare",
        }
    )

    graph.add_edge("compare", "report")
    graph.add_edge("report", END)

    return graph.compile(checkpointer=checkpointer)


# ============================================================
# 4. 演示：持久化 + 断点续跑
# ============================================================

DB_PATH = "/tmp/demo_checkpoints.db"

def demo_persistence():
    """演示持久化和断点续跑"""

    documents = [
        "这是第一篇文档，讨论了 LangGraph 的基础架构设计...",
        "这是第二篇文档，分析了 Agent 状态管理的最佳实践...",
        "这是第三篇文档，介绍了生产环境 Agent 的部署方案...",
    ]

    thread_id = "analysis_task_001"
    config = {"configurable": {"thread_id": thread_id}}

    initial_state = {
        "documents": documents,
        "processed_docs": [],
        "current_index": 0,
        "comparison_result": "",
        "final_report": "",
    }

    # -------------------------------------------------------
    # 第一阶段：正常启动，跑到一半"崩溃"
    # -------------------------------------------------------
    print("=" * 60)
    print("第一阶段：启动 Agent，处理部分文档后模拟崩溃")
    print("=" * 60)

    conn1 = sqlite3.connect(DB_PATH, check_same_thread=False)
    checkpointer1 = SqliteSaver(conn1)
    app1 = build_analysis_agent(checkpointer1)

    # 只执行两步，模拟中途崩溃（用 stream 逐步执行）
    step_count = 0
    for event in app1.stream(initial_state, config=config):
        step_count += 1
        node_name = list(event.keys())[0]
        print(f"[事件] 节点 '{node_name}' 执行完成，已保存 Checkpoint")

        if step_count >= 2:
            print("\n[模拟] 进程崩溃！任务中断。")
            break  # 模拟崩溃，中止执行

    conn1.close()

    # -------------------------------------------------------
    # 第二阶段：重启后，从断点恢复
    # -------------------------------------------------------
    print("\n" + "=" * 60)
    print("第二阶段：服务重启，从断点恢复执行")
    print("=" * 60)

    # 重新创建连接（模拟服务重启）
    conn2 = sqlite3.connect(DB_PATH, check_same_thread=False)
    checkpointer2 = SqliteSaver(conn2)
    app2 = build_analysis_agent(checkpointer2)

    # 检查当前状态
    current_state = app2.get_state(config)
    print(f"\n[恢复] 从 Checkpoint 读取状态：")
    print(f"  已处理文档数：{current_state.values.get('current_index', 0)}")
    print(f"  已有摘要数：{len(current_state.values.get('processed_docs', []))}")
    print(f"  下一步将执行：{current_state.next}")

    # 传入 None，LangGraph 从最近的 Checkpoint 继续执行
    print(f"\n[恢复] 继续执行剩余任务...")
    final_result = app2.invoke(None, config=config)

    print(f"\n[完成] 任务执行完毕")
    print(f"最终报告预览：\n{final_result['final_report'][:200]}...")

    conn2.close()


# ============================================================
# 5. 演示：查看历史状态（get_state_history）
# ============================================================

def demo_state_history():
    """演示如何查看完整的历史状态"""
    print("\n" + "=" * 60)
    print("查看历史状态")
    print("=" * 60)

    thread_id = "analysis_task_001"
    config = {"configurable": {"thread_id": thread_id}}

    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    checkpointer = SqliteSaver(conn)
    app = build_analysis_agent(checkpointer)

    # 获取这个 thread_id 的所有历史 Checkpoint
    history = list(app.get_state_history(config))

    print(f"\n共找到 {len(history)} 个历史快照：")
    for i, snapshot in enumerate(history):
        checkpoint_id = snapshot.config["configurable"].get("checkpoint_id", "unknown")
        current_index = snapshot.values.get("current_index", 0)
        doc_count = len(snapshot.values.get("processed_docs", []))
        next_nodes = snapshot.next

        print(f"\n  快照 {i + 1}：")
        print(f"    checkpoint_id: {checkpoint_id[:16]}...")
        print(f"    已处理文档数: {current_index}")
        print(f"    已有摘要数: {doc_count}")
        print(f"    下一步: {next_nodes}")

    conn.close()


if __name__ == "__main__":
    demo_persistence()
    demo_state_history()

1.5 时间旅行（Time Travel）

Checkpointer 除了用于恢复，还有一个非常有用的能力：回滚到任意历史状态重新执行。

这个特性被 LangGraph 文档称为 Time Travel（时间旅行：能够回到过去某个执行节点的状态，重新执行，用于调试和复现问题）。

假设 Agent 在第三步做出了一个错误决策，可以回滚到第二步的快照，修改一下 State 里的某个字段，然后从第二步重新执行，观察不同输入下 Agent 的行为有什么不同。

python 
def demo_time_travel():
    """演示时间旅行：回滚到历史状态重新执行"""
    thread_id = "analysis_task_001"
    config = {"configurable": {"thread_id": thread_id}}

    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    checkpointer = SqliteSaver(conn)
    app = build_analysis_agent(checkpointer)

    # 获取历史快照列表
    history = list(app.get_state_history(config))

    if len(history) < 2:
        print("历史快照不足，无法演示时间旅行")
        return

    # 取倒数第二个快照（早一步的状态）
    target_snapshot = history[-2]
    target_checkpoint_id = target_snapshot.config["configurable"]["checkpoint_id"]

    print(f"\n[时间旅行] 回滚到快照：{target_checkpoint_id[:16]}...")
    print(f"[时间旅行] 该快照时已处理文档数：{target_snapshot.values.get('current_index', 0)}")

    # 从指定 Checkpoint 重新执行
    replay_config = {
        "configurable": {
            "thread_id": f"{thread_id}_replay",  # 用新的 thread_id 避免污染原记录
            "checkpoint_id": target_checkpoint_id,
        }
    }

    # 也可以在重新执行前修改 State
    # app.update_state(replay_config, {"some_field": "new_value"})

    result = app.invoke(None, config=replay_config)
    print(f"[时间旅行] 从历史断点重新执行完成")

    conn.close()

这个能力在调试复杂 Agent 时非常实用。不需要每次都从头跑，直接从出问题的那一步之前开始复现。

1.6 Checkpointer 的工作流程

自动保存
Checkpoint-1

自动保存
Checkpoint-2

自动保存
Checkpoint-3

时间旅行
从 cp_002 重新执行

START

节点 1
执行

checkpoint_id: cp_001
state: 对象
thread_id: task_001

节点 2
执行

checkpoint_id: cp_002
state: 对象
thread_id: task_001

节点 3
执行

checkpoint_id: cp_003
state: 对象
thread_id: task_001

END

节点 3
重新执行

每个节点执行后都有一个对应的 Checkpoint。时间旅行可以从任意 Checkpoint 重新分叉执行。

1.7 几个实践细节

thread_id 的设计要考虑业务含义。不要用随机 UUID，用能体现业务语义的 ID，比如 user_123_session_456、doc_analysis_batch_20240315。这样出问题了，光看 thread_id 就知道是哪个任务。

生产环境不要用 MemorySaver。持久化方案从一开始就该定下来，后期改代价不大，但会需要数据迁移。

SqliteSaver 有并发限制。SQLite 不支持高并发写入，多个线程同时写同一个文件会有锁竞争。如果 Agent 需要同时处理多个任务，要么用 PostgresSaver，要么给每个任务用独立的 SQLite 文件。

定期清理历史 Checkpoint。一个长期运行的系统，历史 Checkpoint 会越积越多。建议按时间或按数量设置保留策略，避免存储无限膨胀。LangGraph 目前没有内置的清理机制，需要自己写定时任务直接操作数据库。

1.8 小结

持久化是生产级 Agent 的基础能力，不是可选项。

开发阶段用 MemorySaver 够用，但部署前必须切换到持久化方案。单机用 SqliteSaver，生产用 PostgresSaver。换一个 Checkpointer 对象，其他代码不用动。

时间旅行不只是调试用。出了问题，直接从出问题的那步之前开始复现，比从头重跑要快得多。

下一篇讲 Stream 流式输出，解决 Agent 跑几分钟用户只能盯着空白屏幕等的体验问题。

←上一节LangGraph-Subgraph子图与模块化设计下一节LangGraph-Stream流式输出与实时反馈→
本页目录