持久化与会话恢复

第 12 章我们手写过会话落库。LangGraph 里同一件事由 checkpointer（检查点）完成：它把每次节点运行后的状态存下来，并绑定一个 thread_id。只要知道 thread_id，就能"续上"之前任何一次会话——多轮对话记忆、第 19 章的人工审批断点恢复，都依赖它。

什么是 checkpointer

图每跑完一个节点，checkpointer 都会保存一份状态快照。invoke 时带上 config={"configurable": {"thread_id": "..."}}，框架按 thread_id 区分不同会话：

• 同一个 thread_id：上次的状态自动作为本次起点（= 记住上文）。
• 不同的 thread_id：互不干扰（= 多用户/多会话隔离）。

MemorySaver：进程内内存持久化

最轻量的 checkpointer，数据存在当前进程内存里，适合开发与单进程服务：

from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
import os

model = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "deepseek-chat"),
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL"),
)

class State(TypedDict):
    messages: Annotated[list, add_messages]

def agent(state):
    reply = model.invoke(state["messages"])
    return {"messages": [reply]}

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph = builder.compile(checkpointer=MemorySaver())   # 关键：挂上检查点

多轮对话：同一 thread_id 记住上文

def chat(text, thread):
    cfg = {"configurable": {"thread_id": thread}}
    out = graph.invoke({"messages": [HumanMessage(text)]}, cfg)
    print(f"[{thread}] {out['messages'][-1].content}")
    print(f"[{thread}] 历史共 {len(out['messages'])} 条消息")

chat("我叫小明，记住我", "user-1")
# 输出：[user-1] 好的，小明！（历史共 2 条消息）
chat("我叫什么名字？", "user-1")
# 输出：[user-1] 你叫小明。（历史共 4 条消息，模型看到了前两轮）
chat("我叫什么名字？", "user-2")
# 输出：[user-2] 抱歉，我们还不太熟……（user-2 是新会话，没有上文）

第三轮换了一个 thread_id，历史是空的——这就是天然的会话隔离：每个用户/每次会话一个 thread_id，互不串台。多用户场景只需在入口处把 thread_id 映射到用户标识即可。上面模型回复的措辞仅为示意，实际以模型输出为准；请重点看两处确定性的表现：历史消息数随轮次递增、换 thread 后归零。

查看与改写历史：get_state / update_state

断点续跑、调试、人工修正都会用到：

cfg = {"configurable": {"thread_id": "user-1"}}
snap = graph.get_state(cfg)
print(snap.values["messages"][-1])       # 查看该会话最近一条消息
print(snap.next)                         # 下一次将执行哪个节点（空元组 () 表示已结束）

# 人工修正历史：插入一条"纠正消息"，下一次 invoke 就会基于它继续
graph.update_state(cfg, {"messages": [HumanMessage("我其实叫小红")]})
chat("我叫什么名字？", "user-1")
# 输出：[user-1] 小红。（框架用插入的消息续跑）

get_state 返回状态快照（values、next 等字段），update_state 直接写入或覆盖该线程的部分状态。生产环境建议用 update_state 做"人工纠错/回滚"类的后台管理功能。

生产环境：文件与数据库检查点

MemorySaver 进程一重启数据就没了。落盘用 SqliteSaver：

pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("chat.sqlite") as cp:
    graph = builder.compile(checkpointer=cp)   # 重启后同一 thread_id 仍可恢复

再往上，多实例共享一套会话数据通常用 PostgresSaver（langgraph-checkpoint-postgres），适合多进程/多机部署，写法与上面一致：compile(checkpointer=...)。具体安装与连接串以官方文档为准。

thread_id 管理约定

thread_id 只是普通字符串，怎么生成由应用决定：

import uuid

new_thread = str(uuid.uuid4())        # 匿名访客/新会话：每次生成一个
user_thread = f"user-{user_id}"       # 登录用户：固定 id，天然多用户隔离

建议：登录用户用稳定 id，匿名访客每次访问生成 uuid；前端把 thread_id 存进会话存储，后续请求原样带回即可续聊。

常见问题

• MemorySaver 适合生产吗？数据只在当前进程内存里，重启即失；单机开发没问题，多实例/高可用请换数据库检查点。
• 历史消息越来越长怎么办？业务上按需精简：保留最近 N 条或压成摘要后开新 thread 继续，避免单线程消息无限膨胀。
• get_state 里的 next 字段有什么用？它告诉你该线程下一次会执行哪个节点：配合第 19 章审批，next 指向审批节点时表示"正在等人批准"；为空元组 () 表示已运行结束。
• 不同进程能共享 SqliteSaver 吗？SQLite 文件支持本机多进程；跨机器共享请用 PostgresSaver 这类数据库检查点。

小结：checkpointer 按 thread_id 把每次运行后的状态落盘（MemorySaver/SqliteSaver/PostgresSaver），同一 thread_id 续跑即拥有历史，不同 thread_id 相互隔离；配 get_state/update_state 可查看与修正历史——这就是 LangGraph 会话恢复与审批断点的底座。