第 5 章 · 检查点、线程与跨会话记忆
把“当前对话走到哪”与“用户长期偏好”分开存储。本章分别使用 checkpointer 和 store,并用三个调用证明它们的作用域不同。
工作机制与设计选择
Checkpointer 保存指定 thread_id 的图状态;Store 保存应用定义的命名空间键值数据,可被多个线程使用。thread_id 是会话定位符,不是身份认证。生产系统必须由可信服务端确定用户身份和命名空间。InMemorySaver 与 InMemoryStore 仅在当前进程中保留数据,不具备进程崩溃后的持久性。
示例
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
class State(TypedDict):
visits: int
greeting: str
store = InMemoryStore()
store.put(("demo-user", "preferences"), "language", {"value": "English"})
def greet(state: State):
preference = store.get(("demo-user", "preferences"), "language")
return {"visits": state.get("visits", 0) + 1, "greeting": preference.value["value"]}
builder = StateGraph(State)
builder.add_node("greet", greet)
builder.add_edge(START, "greet")
builder.add_edge("greet", END)
graph = builder.compile(checkpointer=InMemorySaver(), store=store)
a = {"configurable": {"thread_id": "conversation-a"}}
b = {"configurable": {"thread_id": "conversation-b"}}
assert graph.invoke({"visits": 0}, a)["visits"] == 1
assert graph.invoke({}, a)["visits"] == 2
assert graph.invoke({"visits": 0}, b)["visits"] == 1
assert graph.get_state(a).values["greeting"] == "English"
print(graph.get_state(a).values)运行与验收
无需模型。第二次在同一线程传入空更新,而不是 visits=0,因此累计到 2;另一线程从 1 开始,但读取同一个偏好。例中闭包持有 Store 以简化演示,服务化时改为运行时依赖注入。
容易踩的坑
不要把内存示例称为生产持久化。迁移数据库后还要测试备份恢复、租户隔离、数据保留期限和敏感信息清理。不要把 token 放进会被序列化的图状态。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 线程状态由谁保存?
2. 长期偏好适合放在哪里?
3. 内存实现能跨进程重启吗?
4. thread_id 能证明身份吗?
动手实践
- 增加不同用户命名空间,断言偏好互不串读。
- 重启程序,说明哪些数据消失以及原因。
- 输出 get_state 的 values 和 next,区分业务值与待执行节点。
官方参考
核验日期:2026-09-17。