跳到主要内容

持久化与检查点

检查点系统​

检查点允许保存图的执行状态，以便稍后恢复。

检查点类型​

InMemorySaver（内存）​

from langgraph.checkpoint.memory import InMemorySaver

saver = InMemorySaver()

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

SQLiteSaver（SQLite数据库）​

from langgraph.checkpoint.sqlite import SqliteSaver

saver = SqliteSaver(db_path="checkpoints.db")

app = graph.compile(checkpointer=saver)

使用检查点​

基本用法​

from langgraph.checkpoint.memory import InMemorySaver

saver = InMemorySaver()
app = graph.compile(checkpointer=saver)

# 第一次执行
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(initial_state, config=config)

# 中断前的状态已保存

恢复执行​

# 稍后恢复执行
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(next_input, config=config)

与中断结合​

from langgraph.types import interrupt

@entrypoint(checkpointer=InMemorySaver())
def workflow_with_interrupt(data):
    step1_result = step1(data).result()
    
    # 中断等待人工输入
    approval = interrupt({
        "message": "需要批准",
        "data": step1_result
    })
    
    if approval:
        return step2(step1_result).result()
    else:
        return "已拒绝"

# 执行流程:
# 1. 调用 workflow_with_interrupt("data") -> 在interrupt处暂停
# 2. 用户审查
# 3. 继续执行 worflow -> step2执行

最佳实践​

  • 使用thread_id区分不同用户/会话
  • 定期清理旧的检查点
  • 根据应用选择合适的存储后端
  • 监控检查点存储空间

常见问题​

Q: 检查点会占用多少空间？ A: 取决于状态大小。优化状态可以减少占用。

Q: 可以跨多个图使用同一个检查点吗？ A: 不推荐。不同图的状态结构可能不同。

Q: 检查点会自动清理吗？ A: 需要手动实现清理逻辑。