持久化

原文：Persistence

一句话

LangGraph 内置持久化层通过检查点(checkpoint)保存图状态，支持人机交互、对话记忆、时间旅行调试和容错执行。

什么时候翻这页

需要实现 Agent 记忆功能时
需要实现人机交互工作流时
需要调试 Agent 执行过程时
需要实现容错机制时
需要跨线程共享状态时

核心概念

检查点(Checkpoint)：图在特定时间点的状态快照，保存在每个超级步骤(super-step)边界
线程(Thread)：每个检查点分配的唯一ID，包含一系列运行的累积状态
超级步骤(Super-step)：图的单个"滴答"，所有计划在该步骤执行的节点执行（可能并行）
持久化模式(Durability modes)：平衡性能和数据一致性的三种模式："exit"、"async"、"sync"
内存存储(Store)：用于跨线程共享信息的接口

怎么做

基本持久化设置

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

# 定义图和工作流
workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

# 添加检查点
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# 执行图时必须指定 thread_id
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar":[]}, config)

获取和更新状态

# 获取最新状态
config = {"configurable": {"thread_id": "1"}}
graph.get_state(config)

# 获取特定检查点的状态
config = {"configurable": {"thread_id": "1", "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"}}
graph.get_state(config)

# 获取状态历史
list(graph.get_state_history(config))

# 更新状态
graph.update_state(config, {"foo": "new_value"})

# 从特定节点更新状态
graph.update_state(config, {"foo": "new_value"}, as_node="node_b")

跨线程状态共享

from langgraph.store.memory import InMemoryStore
from langgraph.runtime import Runtime

# 创建存储
store = InMemoryStore()

# 在节点中使用存储
async def update_memory(state: MessagesState, runtime: Runtime[Context]):
    user_id = runtime.context.user_id
    namespace = (user_id, "memories")
    memory_id = str(uuid.uuid4())
    await runtime.store.aput(namespace, memory_id, {"memory": memory})

# 搜索记忆
memories = await runtime.store.asearch(
    namespace,
    query=state["messages"][-1].content,
    limit=3
)

设置持久性模式

# 指定持久性模式
graph.stream(
    {"input": "test"},
    durability="sync"  # 可选 "exit", "async", "sync"
)

命令 / API 速查

graph.compile(checkpointer=checkpointer) - 编译图时添加检查点
graph.get_state(config) - 获取最新状态
graph.get_state_history(config) - 获取状态历史
graph.update_state(config, values, as_node=...) - 更新状态
graph.stream(..., durability=...) - 指定持久性模式
store.put(namespace, key, value) - 存储值
store.search(namespace, query, limit) - 搜索值
InMemorySaver() - 内存检查点保存器
SqliteSaver() - SQLite 检查点保存器
PostgresSaver() - PostgreSQL 检查点保存器

与 Hello-Agents / Claude Code 的联系

在 Hello-Agents 中，我们介绍了 LangGraph 的基本概念和工作流。持久化功能扩展了这些基础概念，使 Agent 能够记住过去的交互、支持人机协作，并允许调试和恢复执行。与 Hello-Agents 中的简单示例不同，持久化功能使 Agent 能够在长时间运行和多会话场景中保持状态连续性。

初学者易错点

忘记在执行图时指定 thread_id，会导致无法保存或恢复状态
混淆检查点(checkpoint)和超级步骤(super-step)的概念
误认为 update_state 会修改原始检查点，实际上会创建新检查点
忽略持久性模式的选择，可能导致性能或一致性问题
尝试在没有检查器的情况下使用需要持久化的功能（如中断）

语义检索

持久化

一句话