上一篇介绍了 LangGraph 如何通过 Conditional Edge 和 Command 控制执行路径，让 Graph 根据当前 State 决定下一步的执行。

实际情况中，Graph 往往不止一次调用。后续请求可能需要继续使用之前的状态，任务也可能暂停后再继续，甚至在进程重启后恢复执行。

这就需要状态持久化（Persistence）。

LangGraph 通过 Checkpointer 保存 Checkpoint，并用 Thread 关联同一任务的多次执行。

其中，Checkpoint 是某个执行时点的状态快照，Checkpointer 负责保存和读取这些状态，Thread 则用来关联同一任务产生的一系列 Checkpoint。

本文将结合具体示例，说明这三个概念如何配合完成状态的保存、读取和恢复。

一、为什么需要 Persistence

假设有一个订单处理 Graph。

第一次调用传入：

代码片段Text
订单：ORDER-1001
金额：299
状态：待确认

稍后订单确认，再次调用：

代码片段Text
状态：已确认

如果两次调用完全独立，第二次执行就无法获得第一次的订单号和金额，只能由业务代码重新传入完整状态。

流程简单这样处理可以接受，但如果是复杂的 Graph，则还需要知道：

  • 当前 State
  • 下一步执行哪个 Node
  • Interrupt 是否暂停
  • 之前经过哪些执行步骤
  • 故障恢复时从哪里继续

如果这些都由业务层维护，相当于还要自己实现一套工作流状态管理。

LangGraph 把这部分工作交给 Persistence。

Graph 配置 Checkpointer 后，执行过程中会持续生成 Checkpoint，并归到指定的 Thread 下。

Checkpointer 通常会在 Graph 的执行步骤之间保存状态，所以一次 invoke() 也可能产生多个 Checkpoint，而不是只在调用结束时保存一次。

代码片段Text
Thread
├── Checkpoint 1
├── Checkpoint 2
├── Checkpoint 3
└── ...

Checkpoint、Thread 和 Checkpointer 之间的关系如下：

代码片段Text
Thread
同一个任务的一组连续执行记录

Checkpoint
某个执行时点的状态快照

Checkpointer
负责保存和读取 Checkpoint

Interrupt、Durable Execution、Time Travel 等能力，都依赖这套持久化机制。

二、Checkpoint：可恢复的状态快照

Checkpoint 保存的不只是 State 中的字段值，还包括恢复执行需要的信息。

可通过 Graph 的 get_state 方法读取当前状态：

代码片段Python
snapshot = graph.get_state(config)

它的主要语法可以简化为：

代码片段Python
graph.get_state(
    config
) -> StateSnapshot

config 用于指定 Thread，返回值是 StateSnapshot。

StateSnapshot 中比较重要的字段包括：

代码片段Text
StateSnapshot
├── values
├── next
├── config
├── metadata
├── created_at
├── parent_config
├── tasks
└── interrupts

其中：

values 是当前 State。

next 表示下一步准备执行的 Node。Graph 已经结束时通常为空。

config 保存当前快照对应的配置，其中可以包含 thread_id 和 checkpoint_id。

metadata 保存执行步骤、状态来源等信息。

tasks 表示当前步骤相关的任务。

interrupts 保存尚未处理的 Interrupt。

因此，Checkpoint 主要包括这些内容：

代码片段Text
当前 State
    +
执行位置
    +
后续任务

而不是单纯把 State 序列化后写进数据库。

如果只指定 thread_id：

代码片段Python
config = {
    "configurable": {
        "thread_id": "order-1001"
    }
}

通常读取这个 Thread 当前的状态。

如果已经知道某个 checkpoint_id，也可以把它放到配置中：

代码片段Python
config = {
    "configurable": {
        "thread_id": "order-1001",
        "checkpoint_id": "..."
    }
}

这样就可以定位 Thread 中的具体 Checkpoint，这也是历史状态读取、Replay 和 Time Travel 的基础。

三、Thread：把多次执行连起来

启用 Checkpointer 后，调用 Graph 时需要指定 thread_id：

代码片段Python
config = {
    "configurable": {
        "thread_id": "order-1001"
    }
}

基本调用形式是：

代码片段Python
graph.invoke(
    input,
    config=config,
)

例如第一次处理订单：

代码片段Python
graph.invoke(
    {
        "order_id": "ORDER-1001",
        "amount": 299,
        "status": "待确认",
    },
    config=config,
)

后续仍然使用同一个 thread_id：

代码片段Python
graph.invoke(
    {
        "status": "已确认",
    },
    config=config,
)

LangGraph 会基于这个 Thread 已经保存的状态继续执行，因此第二次只需要传入新的状态变化，之前的 order_id 和 amount 仍然可以继续使用。启用 Checkpointer 后，thread_id 就是保存和查找 Checkpoint 的关键标识。

如果换一个新的 ID：

代码片段Python
config = {
    "configurable": {
        "thread_id": "order-1002"
    }
}

就是另一条独立的执行记录。

因此，Thread 更接近一个需要持续保存状态的会话或工作流实例。

实际项目中，thread_id 一般应由业务系统统一生成和维护。

如果每次请求都创建新的 thread_id，LangGraph 就无法找到之前保存的状态。

四、Checkpointer：状态保存在哪里

Persistence 在 Graph 编译阶段启用。

基本语法是：

代码片段Python
graph = builder.compile(
    checkpointer=checkpointer
)

StateGraph.compile() 接收一个 Checkpointer，返回可以执行的 CompiledStateGraph。启用后，Graph 才能在不同执行步骤之间保存和恢复状态。

开发和测试时，最简单的是 InMemorySaver：

代码片段Python
from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()
 
graph = builder.compile(
    checkpointer=checkpointer
)

它把 Checkpoint 保存在当前进程内存中。进程退出后，状态也会消失，因此一般仅用于开发和测试。

下面用一个完整例子看一下执行过程：

代码片段Python
from typing_extensions import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
 
 
class State(TypedDict, total=False):
    order_id: str
    amount: int
    status: str
 
 
def process_order(state: State):
    print(
        state["order_id"],
        state["amount"],
        state["status"],
    )
    return {}
 
 
builder = StateGraph(State)
 
builder.add_node("process_order", process_order)
builder.add_edge(START, "process_order")
builder.add_edge("process_order", END)
 
graph = builder.compile(
    checkpointer=InMemorySaver()
)
 
config = {
    "configurable": {
        "thread_id": "order-1001"
    }
}

第一次执行：

代码片段Python
graph.invoke(
    {
        "order_id": "ORDER-1001",
        "amount": 299,
        "status": "待确认",
    },
    config=config,
)

Node 可以读取：

代码片段Text
ORDER-1001 299 待确认

随后仍然使用同一个 Thread：

代码片段Python
graph.invoke(
    {
        "status": "已确认",
    },
    config=config,
)

第二次 Node 可以读取：

代码片段Text
ORDER-1001 299 已确认

虽然这次只传入了 status，之前保存的 order_id 和 amount 仍然存在。

这就是同一个 Thread 中 State 的延续。

Node 本身不需要操作 Checkpointer。它仍然只负责：

代码片段Text
读取 State
    ↓
完成处理
    ↓
返回 State 更新

Checkpoint 的生成和保存由 LangGraph Runtime 负责。

五、读取当前状态和历史状态

除了让 Runtime 自动恢复状态，还可以主动读取 Checkpoint。

读取当前状态

前面已经使用过：

代码片段Python
snapshot = graph.get_state(config)

例如：

代码片段Python
print(snapshot.values)
print(snapshot.next)

可能得到：

代码片段Text
{
    'order_id': 'ORDER-1001',
    'amount': 299,
    'status': '已确认'
}

()

values 是当前 State。

next 为空，表示当前没有等待执行的 Node。

读取状态历史

一条 Thread 通常包含多个 Checkpoint，可以通过：

代码片段Python
graph.get_state_history(config)

读取。

基本形式可以理解为：

代码片段Python
graph.get_state_history(
    config
) -> Iterator[StateSnapshot]

例如：

代码片段Python
history = list(
    graph.get_state_history(config)
)
 
for snapshot in history:
    print(
        snapshot.values,
        snapshot.next,
        snapshot.metadata,
    )

这样就可以观察同一 Thread 在不同执行阶段的状态。

例如：

代码片段Text
ORDER-1001
    ↓
待确认
    ↓
已确认

每次 Graph 执行过程中又可能包含多个执行步骤，因此实际的 Checkpoint 数量通常会比业务状态变化更多。

LangGraph 同时提供同步和异步的状态读取接口。

同步 API 常用：

代码片段Python
graph.invoke(...)
graph.get_state(...)
graph.get_state_history(...)

异步场景则有对应方法：

代码片段Python
await graph.ainvoke(...)
await graph.aget_state(...)
 
async for snapshot in graph.aget_state_history(...):
    ...

六、Persistence 和 Store 的区别

提到持久化，我们还需要区分 Checkpointer 和 Store。

两者保存的数据范围不同。

对比项CheckpointerStore
保存内容Graph State、Checkpoint应用数据
数据范围单个 Thread可跨 Thread
常见用途多轮执行、Interrupt、恢复用户偏好、长期事实
主要标识thread_idnamespace、key

官方 Persistence 也把二者作为两套互补机制：Checkpointer 保存 Thread 内的 Graph State，Store 保存需要跨 Thread 使用的数据。

例如当前订单中的：

代码片段Text
订单号
金额
当前状态
处理结果

都属于当前工作流继续执行需要的数据，适合放在 State 中，由 Checkpointer 保存。

而类似：

代码片段Text
用户默认收货方式
常用配送地址 ID
应用长期配置

如果多个 Thread 都需要使用，就更适合 Store 或业务数据库。

Store 的基本键值操作是：

代码片段Python
store.put(
    namespace,
    key,
    value,
)
 
item = store.get(
    namespace,
    key,
)

其中 namespace 是一个字符串元组，key 是该 namespace 下的唯一标识。

例如：

代码片段Python
from langgraph.store.memory import InMemoryStore
 
store = InMemoryStore()
 
store.put(
    ("users", "user-1001"),
    "preferences",
    {
        "delivery": "express"
    },
)
 
item = store.get(
    ("users", "user-1001"),
    "preferences",
)

Store 和 Checkpointer 也可以同时传给 Graph：

代码片段Python
graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
)

Checkpointer 和 Store 的使用场景区分：

代码片段Text
当前 Thread 继续执行需要的数据
→ State + Checkpointer

多个 Thread 都要使用的数据
→ Store / 业务数据库

七、SQLite 和 PostgreSQL

InMemorySaver 无法跨进程保存状态。

如果应用重启后还要继续使用之前的 Thread，就需要数据库型 Checkpointer。

SQLite

安装：

代码片段Bash
pip install -U langgraph-checkpoint-sqlite

同步版本：

代码片段Python
from langgraph.checkpoint.sqlite import SqliteSaver

可以直接使用文件保存 Checkpoint：

代码片段Python
with SqliteSaver.from_conn_string(
    "checkpoints.sqlite"
) as checkpointer:
 
    graph = builder.compile(
        checkpointer=checkpointer
    )

SqliteSaver 适合本地开发、Demo 和轻量单机应用。异步场景可以使用：

代码片段Python
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

并配合：

代码片段Python
async with AsyncSqliteSaver.from_conn_string(
    "checkpoints.sqlite"
) as checkpointer:
    ...

SQLite Checkpointer 提供同步和异步实现，不过它的定位仍以轻量场景为主。

PostgreSQL

生产环境通常更适合使用 PostgreSQL Checkpointer。

安装：

代码片段Bash
pip install -U \
    "psycopg[binary,pool]" \
    langgraph \
    langgraph-checkpoint-postgres

同步版本：

代码片段Python
from langgraph.checkpoint.postgres import PostgresSaver

基本用法：

代码片段Python
DB_URI = (
    "postgresql://postgres:postgres"
    "@localhost:5432/postgres?sslmode=disable"
)
 
with PostgresSaver.from_conn_string(
    DB_URI
) as checkpointer:
 
    checkpointer.setup()
 
    graph = builder.compile(
        checkpointer=checkpointer
    )

setup() 用于初始化所需的数据表和索引，首次使用 PostgreSQL Checkpointer 时需要执行。

异步版本对应：

代码片段Python
from langgraph.checkpoint.postgres.aio import (
    AsyncPostgresSaver
)

初始化时使用：

代码片段Python
async with AsyncPostgresSaver.from_conn_string(
    DB_URI
) as checkpointer:
 
    await checkpointer.setup()

八、生产环境需要注意事项

启用 Persistence 后，State 会进入持久化存储，因此 State 的字段需要控制范围。

例如 RAG 检索一次返回几十篇完整文档，如果后续流程只需要：

代码片段Text
document_id
score
摘要

就没有必要把全部原始内容长期保存在 State 中。

否则随着 Checkpoint 增加，存储空间、序列化开销和读取成本都会增加。长生命周期 Thread 也需要根据恢复、审计和调试需求设计 Checkpoint 保留策略。官方文档同样提醒，Checkpoint 长期累积会增加延迟和存储成本。

另一个需要注意的是敏感数据。

只要数据进入 State，就可能进入 Checkpoint，包括用户输入、Tool 返回值和业务字段。

因此数据库权限、密钥管理、数据保留时间等问题仍然需要由应用负责。LangGraph 的 Checkpointer 支持自定义序列化，也提供 EncryptedSerializer 对 Checkpoint 内容进行加密。

总结

LangGraph 通过 Persistence 将 Graph 的执行状态持久化，使任务能够跨调用、跨进程保存和恢复，并为 Interrupt、Time Travel 等能力提供基础。

本文重点介绍了 Checkpoint、Thread 和 Checkpointer 的关系，以及状态的保存、读取和历史记录，并说明了 Store、SQLite、PostgreSQL 在不同场景下的使用边界。