[智能体-224]：LangGraph的记忆载体State与Checkpointer机制详解，代码示例

LangGraph State + Checkpointer 完整详解

核心结论：

State = 运行时内存载体（单次会话上下文容器）；Checkpointer =持久化落地器（按 thread_id 把全量 State落地为检查点，实现会话记忆、断点续跑）

LCEL 靠RunnableWithMessageHistory外挂记忆；

LangGraph原生把记忆绑定在图状态 + 检查点，不需要额外包装器。因此，LangGraph需要大量存储单元（内存与外存）保存LangGraph在业务chain执行过程中，每个检查点的信息。正因为记录了每个检查点的状态信息，才使得LangGraph针对任意执行点进行状态恢复成为了可能！！！

一、基础概念拆分

1. State：记忆的内存容器

1. 本质：结构化数据模型，Graph 所有节点共享读写同一个 State 对象，所有对话消息、中间变量全部存在 State 字段里；
2. MessagesState：官方内置标准状态，自带messages: list[BaseMessage]，是对话记忆最常用载体；
3. 自定义 State：可新增任意字段（姓名、参数、工具返回值等）；
4. 更新规则：采用增量合并更新return {"字段": 新值}，不会全量覆盖整个状态。

python

运行

# 1. 内置标准对话状态 from langgraph.graph import MessagesState # 自带字段：messages = [] 存放用户/助手消息 # 2. 自定义扩展状态 from typing import Optional class CustomState(MessagesState): username: Optional[str] # 扩展自定义字段 count: int = 0

2. Checkpointer：State 的持久化存储器（会话记忆核心）

1. 绑定在graph.compile(checkpointer=xxx)，编译阶段注入；
2. key：thread_id（等价 LCEL session_id），不同 thread_id 隔离独立会话；
3. 执行逻辑：
   • 每一轮 invoke/stream启动前：根据 thread_id 从存储加载历史 State；
   • 每一轮执行完毕：自动把最新全量 State 落地存入检查点；
4. 主流实现：
   • MemorySaver：内存存储（调试）
   • SqliteSaver：本地文件持久
   • PostgresSaver/RedisSaver：线上分布式

3. State 与 Checkpointer 协作链路

plaintext

入参config(thread_id) → Checkpointer加载历史State → 节点修改State → Checkpointer自动落地新State

二、最简完整示例（MessagesState + MemorySaver）

python

运行

from langgraph.graph import StateGraph, START, MessagesState from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI # 1. 初始化LLM llm = ChatOpenAI(model="gpt-3.5-turbo") # 2. 定义节点：读取state.messages、生成回答、写入state def chat_node(state: MessagesState): resp = llm.invoke(state["messages"]) # 返回字典，自动增量合并进State["messages"] return {"messages": [resp]} # 3. 搭建图 builder = StateGraph(MessagesState) builder.add_node("chat", chat_node) builder.add_edge(START, "chat") # 4. 挂载检查点=开启会话记忆 checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) # 会话配置：thread_id区分会话 cfg = {"configurable": {"thread_id": "u001"}} # 第一轮 res1 = graph.invoke({"messages": [("user", "我叫王文兵")]}, config=cfg) # 第二轮自动加载上一轮State，记住姓名 res2 = graph.invoke({"messages": [("user", "我叫什么名字？")]}, config=cfg) print(res2["messages"][-1].content) # 输出：你叫王文兵

关键点：全程没有任何记忆组件、没有 RunnableWithMessageHistory，记忆由 State 存数据、Checkpointer 存快照自动实现。

三、自定义 State + Checkpointer 实战

扩展自定义字段，观察自定义数据随会话持久化

python

运行

from typing import Optional from langgraph.graph import StateGraph, START from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import BaseMessage # 自定义状态 class UserState: messages: list[BaseMessage] name: Optional[str] chat_times: int class UserStateDict(dict, UserState): pass # 节点1：提取用户名存入state def extract_name(state: UserStateDict): msg = state["messages"][-1].content if "我叫" in msg: name = msg.replace("我叫", "") return {"name": name, "chat_times": state.get("chat_times",0)+1} return {"chat_times": state.get("chat_times",0)+1} # 节点2：回复 def reply_node(state: UserStateDict): name = state.get("name","朋友") return {"messages": [("assistant",f"你好{name}，这是第{state['chat_times']}次对话")]} # 构图 builder = StateGraph(UserStateDict) builder.add_node("extract", extract_name) builder.add_node("reply", reply_node) builder.add_edge(START,"extract") builder.add_edge("extract","reply") # 绑定检查点 saver = MemorySaver() graph = builder.compile(checkpointer=saver) cfg = {"configurable":{"thread_id":"u002"}} # 第一轮 graph.invoke({"messages":[("user","我叫王文兵")]},cfg) # 第二轮自动读取state.name、chat_times out = graph.invoke({"messages":[("user","你还记得我吗")]},cfg) print(out["messages"][-1].content) # 输出：你好王文兵，这是第2次对话

四、Checkpointer 三大核心高级能力

1. 会话隔离：不同 thread_id 数据互不干扰

python

运行

cfg1 = {"configurable":{"thread_id":"t1"}} cfg2 = {"configurable":{"thread_id":"t2"}} graph.invoke({"messages":[("user","我是A")]},cfg1) graph.invoke({"messages":[("user","我是B")]},cfg2) # t1查名字是A，t2查名字是B，State完全隔离

2. 断点续跑

任务中途终止，复用同一个 thread_id 即可加载之前所有状态继续运行，Agent / 工具调用场景刚需。

3. 历史检查点回溯（回滚任意历史会话状态）

python

运行

# 获取该thread所有历史检查点 history = list(checkpointer.list(graph.get_config({"configurable":{"thread_id":"u001"}}))) # 选取上一个版本配置回滚 old_cfg = history[-2].config # 使用旧配置执行，回到历史对话状态 graph.invoke({"messages":[("user","继续")]},config=old_cfg)

五、持久化落地：SqliteSaver（落地磁盘，重启不丢记忆）

MemorySaver 进程销毁数据丢失，Sqlite 落地文件，生产测试首选：

python

运行

from langgraph.checkpoint.sqlite import SqliteSaver # 生成本地db文件 conn = SqliteSaver.from_conn_string("checkpoint.db") graph = builder.compile(checkpointer=conn) # 运行后关闭程序重启，再次用相同thread_id，历史状态依然存在

六、State+Checkpointer 和 LCEL 记忆横向对比

表格

七、关键总结

1. State = 运行时内存：活着的时候所有上下文、自定义参数全存在这里；
2. Checkpointer = 持久化磁盘：每次运行自动快照落地，靠thread_id实现多会话隔离记忆；
3. LangGraph 实现多轮对话不需要任何历史记忆包装类，是架构层面原生内置记忆；
4. 短期会话记忆只用这套组合，跨用户长期记忆用 Store 组件（另一个存储体系）。