Checkpointer：持久化检查点

5.1 Checkpointer 简介

Checkpointer 是 LangGraph 的状态持久化机制，让你的 Agent 可以：

• 暂停和恢复执行
• 维护多轮对话上下文
• 实现线程隔离

python
from langgraph.checkpoint.memory import MemorySaver

# 内存存储（适合开发和测试）
checkpointer = MemorySaver()

# 编译时传入 checkpointer
app = graph.compile(checkpointer=checkpointer)

5.2 线程与会话

每个对话会话对应一个thread_id，状态在不同线程间完全隔离。

python
# 线程 A 的对话
config_a = {"configurable": {"thread_id": "user_123_session_1"}}
result_a = app.invoke({"messages": ["你好"]}, config_a)

# 线程 B 的对话（完全独立）
config_b = {"configurable": {"thread_id": "user_456_session_1"}}
result_b = app.invoke({"messages": ["你好"]}, config_b)

# 恢复线程 A 的对话（从上次暂停的地方继续）
result_a2 = app.invoke({"messages": ["继续刚才的话题"]}, config_a)

5.3 持久化后端对比

三种 Checkpointer 后端，适用于不同场景：

5.3.1 内存存储（开发用）

python
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()  # 进程重启后丢失

5.3.2 SQLite（生产入门）

python
from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("messages.db")

5.3.3 PostgreSQL（生产级）

python
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/db"
)

5.4 中断与恢复

使用 interrupt 暂停执行，等待外部处理后再恢复。

python
from langgraph.errors import NodeInterrupt

def approval_node(state: AgentState) -> dict:
    if state["amount"] > 10000:
        # 暂停执行，等待审批
        raise NodeInterrupt(f"需要审批：金额 {state['amount']}")
    return {"approved": True}

# 恢复执行（外部审批后）
app.invoke(
    None,  # 不传新输入
    config={"configurable": {"thread_id": "tx_123"}}
)

5.5 本章小结

• Checkpointer 实现状态持久化，让 Agent 具备记忆能力
• 每个会话对应一个 thread_id，状态完全隔离
• 支持多种后端：MemorySaver（开发）、SqliteSaver（单机生产）、PostgresSaver（分布式生产）
• NodeInterrupt 实现人工审批流程，常用于大额订单、敏感操作
• 使用 config 参数控制线程、checkpoint 和配置