LangGraph State + Checkpointer 通俗详解

一句话概括:

State = 单次聊天临时记事本(内存里);

Checkpointer = 记事本存档柜(存硬盘 / 数据库,靠 thread_id 分文件夹

LCEL 需要额外套一层RunnableWithMessageHistory外挂记忆;

LangGraph 天生自带记忆,不用额外包装。

一、State:对话运行时的「记事本」

1. 通俗理解

整个智能体流程图里,所有节点共用同一个笔记本 State

  • 用户提问、AI 回答、中间计算数据、姓名、计数全都写在这本本子上;
  • 每个节点只能往本子新增内容,不会整本擦掉(增量更新)。

2. 两种 State

  1. MessagesState(官方通用记事本) 自带固定一栏:messages,专门存一轮轮问答消息,做聊天首选。

python

运行

from langgraph.graph import MessagesState
# 内置:state["messages"] = [用户消息,助手消息...]
  1. 自定义 State(加厚记事本,多加栏目) 自己新增:姓名、年龄、对话次数等栏目。

python

运行

from typing import Optional
class MyState(MessagesState):
    username: Optional[str] # 姓名栏
    chat_count: int = 0     # 聊天次数栏

3. 规则:改本子只返回改动内容

节点想改记事本,只写变化字段,剩下原有内容自动保留:

python

运行

# 只更新名字和次数,历史消息不动
return {"username":"王文兵","chat_count":1}

二、Checkpointer:记事本的「存档柜子」+ thread_id 文件夹

1. 通俗理解

Checkpointer是柜子,thread_id是柜子里独立文件夹:

  1. 每次一轮对话结束,自动把整本 State 记事本拍照存档放进对应thread_id文件夹;
  2. 下次同一个thread_id聊天,先从文件夹取出上次存档→还原成 State 记事本→接着聊天;
  3. 不同thread_id=不同文件夹,聊天记录互不串门。

2. 三类常用柜子

  1. MemorySaver:内存临时柜子,程序一关存档全丢(调试用)
  2. SqliteSaver本地文件柜子,存.db文件,重启软件记录还在
  3. Postgres/RedisSaver:线上服务器柜子,多台程序共用存档(生产环境)

3. 挂载时机

编译图的时候绑定柜子:

python

运行

graph = builder.compile(checkpointer=存档柜子)

三、完整运行流程(生活化举例)

用户:我叫王文兵 → 后续:我叫什么名字?

  1. 传入thread_id="u001",Checkpointer 去 u001 文件夹找存档;
  2. 第一次无存档,新建空白 State 记事本;
  3. 聊天节点把「我叫王文兵」写进 State;
  4. 本轮结束,Checkpointer 把整本记事本存入u001文件夹;

下次提问:我叫什么名字

  1. 再次传入thread_id="u001"
  2. Checkpointer 读取 u001 存档,还原上次带名字的 State;
  3. AI 读取记事本里姓名,正确回答;
  4. 再次自动存档新记录。

thread_id="u002" = 新开空文件夹,完全不记得王文兵。

四、极简可运行代码

python

运行

from langgraph.graph import StateGraph,START,MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

llm=ChatOpenAI(model="gpt-3.5-turbo")

# 聊天节点:读写State记事本
def chat(state:MessagesState):
    res=llm.invoke(state["messages"])
    return {"messages":[res]}

# 搭图
graph_builder=StateGraph(MessagesState)
graph_builder.add_node("chat",chat)
graph_builder.add_edge(START,"chat")

# 配置存档柜
saver=MemorySaver()
graph=graph_builder.compile(checkpointer=saver)

# thread_id=文件夹编号
config={"configurable":{"thread_id":"user_01"}}

# 第一轮存名字
graph.invoke({"messages":[("user","我叫王文兵")]},config)
# 第二轮自动读取存档
ans=graph.invoke({"messages":[("user","我叫啥?")]},config)
print(ans["messages"][-1].content)
# 输出:你叫王文兵

五、Checkpointer 三大实用能力(大白话)

  1. 多会话隔离 thread_id1thread_id2两个文件夹,两个用户聊天记录互不干扰。
  2. 断点续聊 聊到一半程序关掉,再次打开用同一个thread_id,从上次记录继续聊。
  3. 时光回溯 调取文件夹里历史存档,回到之前某一轮对话状态重新聊天。

六、Sqlite 持久化(关掉程序记录不消失)

python

运行

from langgraph.checkpoint.sqlite import SqliteSaver
# 存档存入本地check.db文件
sqlite_saver=SqliteSaver.from_conn_string("check.db")
graph=graph_builder.compile(checkpointer=sqlite_saver)

七、LCEL VS LangGraph 记忆一句话区分

  • LCEL:本子(Chain)本身不带收纳,需要外接收纳袋RunnableWithMessageHistory装历史;
  • LangGraph:自带本子 State + 配套储物柜 Checkpointer,原生存聊天记录。

总结

  1. State = 运行中临时记录本(内存),承载当前轮所有上下文;
  2. Checkpointer = 持久化储物柜thread_id划分独立会话存档;
  3. LangGraph 实现多轮记忆不需要额外记忆包装类,架构原生支持。
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐