LangGraph 的检查点(Checkpoint)机制允许在任意节点保存 Agent 状态,实现断点恢复和多轮对话的上下文保持。本文详细介绍检查点配置、状态持久化策略以及在生产环境中的应用。
LangGraph 的检查点机制是实现可靠 Agent 的关键。通过配置检查点,Agent 可以在任意时刻保存状态,并在需要时恢复继续执行。这对于长时运行的任务、错误恢复和会话保持至关重要。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
# 创建带检查点的图
checkpointer = MemorySaver()
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_edge("agent", END)
# 编译时指定检查点
app = graph.compile(checkpointer=checkpointer)
# 调用时会自动保存状态
config = {"configurable": {"thread_id": "session-123"}}
result = app.invoke({"messages": ["hello"]}, config)
# 恢复之前的会话
config = {"configurable": {"thread_id": "session-123"}}
# 获取历史状态
history = list(app.get_state(config))
print(f"Saved {len(history)} checkpoints")
# 更新状态后继续
app.update_state(config, {"messages": ["new message"]})
result = app.invoke(None, config) # None 表示使用当前状态
| 特性 | MemorySaver | SqliteSaver |
|---|---|---|
| 持久性 | 进程内存 | SQLite 文件 |
| 适用场景 | 开发/测试 | 生产环境 |
| 并发支持 | 单进程 | 多进程 |
| 状态大小 | 受内存限制 | 受磁盘限制 |
from langgraph.checkpoint.postgres import PostgresSaver
# PostgreSQL 检查点(生产推荐)
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@localhost:5432/langgraph"
)
checkpointer.setup() # 初始化数据库
app = graph.compile(checkpointer=checkpointer)
Q1: 检查点会影响性能吗?
Q2: 如何管理历史状态?
app.get_state_history() 获取完整历史app.update_state() 修改状态app.delete_state() 删除不需要的状态状态保存和恢复验证通过
检查点机制工作正常