第3章：记忆与存档——Checkpointer 与状态持久化
本章目标
读完本章你会：
- 能解释"为什么没有 Checkpointer，Agent 每次调用都像第一次见面"
- 能用
MemorySaver让 Agent 在同一会话中记住之前的对话 - 能用
thread_id区分不同用户/会话的状态空间 - 能用
SqliteSaver实现重启不丢失的持久化存储
知识讲解
从一个生活例子开始
你去医院看病。假设这家医院没有病历系统：
你：医生，我上周来看过，咳嗽。
医生：你是新病人？哪里不舒服？
你：我上周说过了...算了，咳嗽，三天了。
医生：开点止咳药。（在便签上写了点什么，扔进抽屉）—— 一周后，你复诊 ——你：医生，我上周吃了止咳药，好了一点但还咳。
医生：你是新病人？哪里不舒服？
你：…（把整个过程再说一遍）
没有病历系统，每次就诊都是一次全新的对话——医生不记得你谁、不记得上次开了什么药、不知道你是加重了还是好转了。
这就是你当前的 LocalTrend Agent 的处境。 每次 graph.invoke() 都是一次全新的执行。没有任何"记忆"。用户问"继续分析上次的 AI 趋势"——Agent 一脸茫然："什么 AI 趋势？"
LangGraph 的 Checkpointer 就是 Agent 的病历系统。每次节点执行完毕后，它会自动保存一份 State 快照。下次同一个 thread_id 进来，Agent 从上次停下的地方继续——记得之前聊了什么、搜过什么、分析过什么。
| 没有 Checkpointer | MemorySaver | SqliteSaver |
|---|---|---|
| 每次都像新的 | 重启 Python 后就忘了 | 写入磁盘，永久保存 |
| 适合单次脚本 | 适合开发调试 | 适合生产环境 |
工作原理
Checkpointer 在什么时候存？
不是"你想存的时候手动存"。Checkpointer 自动在每个 super-step（超级步）完成后保存 State。一个 super-step 是"从一个节点执行到下一个节点开始之前"的边界。对于你的 ReAct 图：
super-step 1: agent_node 执行 → 快照保存
super-step 2: tool_node 执行 → 快照保存
super-step 3: agent_node 执行 → 快照保存
super-step 4: 条件边判断 → END → 快照保存
每次 invoke() 不带初始 State 时（或者带 config 复用 thread_id 时），图会从最后一个快照继续——直接拿上次的完整消息历史，而不是空白 State。
思考一下： 如果你在 ReAct 循环的中间（agent 刚发了 tool_calls，tools 还没执行）手动停止了程序，下次用同一个 thread_id invoke 会发生什么？它会从断点继续——tool_node 会接着执行那些待处理的工具调用。
thread_id：这个"病历"是谁的？
同一个图可能同时服务多个用户、多个对话。thread_id 就是用来区分它们的：
# 用户 A 的会话
config_a = {"configurable": {"thread_id": "user-a-chat-001"}}
graph.invoke({"messages": [HumanMessage(content="搜 AI 趋势")]}, config_a)# 用户 B 的会话——完全独立的状态空间
config_b = {"configurable": {"thread_id": "user-b-chat-002"}}
graph.invoke({"messages": [HumanMessage(content="搜职场趋势")]}, config_b)
A 和 B 的对话互不干扰——就像两个病人的病历本不会混在一起。
⚠️ 常见坑：如果你在
compile()时传了 checkpointer，但invoke()时忘了传thread_id，LangGraph 会报错。它需要知道把状态存到哪个"抽屉"里。
MemorySaver vs SqliteSaver：什么时候用哪个？
开发阶段（你现在） → MemorySaver （内存中，重启就没了，但零配置）
单机部署、小规模 → SqliteSaver （一个文件，重启还在，零运维）
多实例、高并发 → PostgresSaver （共享存储，生产级别）
本章先讲 MemorySaver 让你理解机制，再讲 SqliteSaver 给你持久化能力。
代码实战
环境准备
SqliteSaver 需要 langgraph-checkpoint-sqlite 包：
pip install -U langgraph-checkpoint-sqlite
MemorySaver 在
langgraph主包中自带，无需额外安装。
基础版：MemorySaver——让对话有"短期记忆"
新建文件 chapter03_memory.py。这段代码在上一章的 ReAct Agent 基础上加了 MemorySaver——改动只有几行，但效果是质变：
"""
第 3 章 基础演示：MemorySaver 让 Agent 记住对话
LocalTrend 主线增量——Agent 不再"失忆"
"""
import os
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver # from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI# ============================================================
# 配置 DeepSeek
# ============================================================
llm = ChatOpenAI(model="deepseek-chat",api_key=os.getenv("DEEPSEEK_API_KEY", "your-api-key-here"),base_url="https://api.deepseek.com",temperature=0.7,
)# ============================================================
# State、Tools、Nodes——和上一章结构一致
# ============================================================
class LocalTrendState(TypedDict):messages: Annotated[list, add_messages]iteration_count: int@tool
def search_trending(query: str) -> str:"""搜索当前热门话题趋势。参数 query: 搜索关键词"""print(f" [工具] 搜索: {query}")knowledge = {"AI": "AI 爆款趋势：Agent 开发成为 2026 最热方向，'零基础'、'实战'系列教程流量最高。","职场": "职场爆款：'反焦虑叙事' + '副业案例' 组合成为流量密码。",}for key, val in knowledge.items():if key in query:return valreturn f"关于'{query}'的趋势：实用教程和行业分析最受欢迎。"tools = [search_trending]
llm_with_tools = llm.bind_tools(tools)def agent_node(state: LocalTrendState) -> dict:"""Agent 思考节点"""response = llm_with_tools.invoke(state["messages"])iteration = state.get("iteration_count", 0)return {"messages": [response], "iteration_count": iteration + 1}def tool_node(state: LocalTrendState) -> dict:"""工具执行节点"""last_msg = state["messages"][-1]tool_map = {t.name: t for t in tools}results = []for tc in last_msg.tool_calls:result = tool_map[tc["name"]].invoke(tc["args"])results.append(ToolMessage(content=result, tool_call_id=tc["id"]))return {"messages": results}def should_continue(state: LocalTrendState) -> Literal["tools", "__end__"]:"""路由判断"""last_msg = state["messages"][-1]if state.get("iteration_count", 0) >= 8:return ENDif hasattr(last_msg, "tool_calls") and last_msg.tool_calls:return "tools"return END# ============================================================
# 构建图——和图结构完全相同，只是 compile 时多了 checkpointer
# ============================================================
def build_graph():builder = StateGraph(LocalTrendState)builder.add_node("agent", agent_node)builder.add_node("tools", tool_node)builder.add_edge(START, "agent")builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})builder.add_edge("tools", "agent")# 关键是这里：传入 MemorySavermemory = MemorySaver()return builder.compile(checkpointer=memory)# ============================================================
# 运行：对比"有记忆"和"没记忆"
# ============================================================
if __name__ == "__main__":graph = build_graph()# 用同一个 thread_id 做多轮对话config = {"configurable": {"thread_id": "local-trend-session-1"}}# --- 第一轮 ---print("=" * 60)print(" 第一轮对话")print("=" * 60)result1 = graph.invoke({"messages": [HumanMessage(content="帮我搜索 AI 领域的爆款趋势")]},config)print(f"\n 第一轮后消息数: {len(result1['messages'])}")print(f" Agent: {result1['messages'][-1].content[:150]}")# --- 第二轮：不传初始消息，依赖 checkpointer 恢复 ---print("\n" + "=" * 60)print(" 第二轮对话（使用同一个 thread_id）")print("=" * 60)result2 = graph.invoke(# 关键：第二轮不传全量初始化 State，只传新用户消息# State 中的 messages、iteration_count 都从 checkpoint 恢复{"messages": [HumanMessage(content="那职场领域呢？也帮我搜一下。")]},config)print(f"\n 第二轮后消息数: {len(result2['messages'])}")print(f" Agent: {result2['messages'][-1].content[:150]}")# --- 验证"有记忆" ---print("\n" + "=" * 60)print(" 验证：Agent 是否记得第一轮的内容？")print("=" * 60)# 消息历史包含两轮对话的所有消息all_messages = result2["messages"]print(f"总消息数: {len(all_messages)}（包含两轮对话的全部消息）")# 找到所有 HumanMessage（用户说的话）user_messages = [m for m in all_messages if isinstance(m, HumanMessage)]print(f"用户发言次数: {len(user_messages)}")# 换个 thread_id 试试——完全独立的状态print("\n" + "=" * 60)print(" 用新 thread_id——全新的对话")print("=" * 60)new_config = {"configurable": {"thread_id": "local-trend-session-2"}}result3 = graph.invoke({"messages": [HumanMessage(content="你好，今天天气怎么样？")]},new_config)print(f"新会话消息数: {len(result3['messages'])}（从头开始，不受 session-1 影响）")
运行这段代码，关键观察：
- 第一轮调用
invoke传入初始消息——正常执行 ReAct 循环 - 第二轮用同一个
thread_id，只传了新用户消息——Agent 从 memory 恢复了第一轮的完整消息历史，自动追加新消息 - 新 thread_id 是一个全新的会话——消息数从头开始
逐行解析新概念
MemorySaver() + compile(checkpointer=memory)
这是本章唯一的"结构级"改动。MemorySaver() 创建了一个内存中的状态存储。传给 compile() 后，LangGraph 在每次 super-step 完成后自动把 State 序列化并存入 MemorySaver。不需要你在节点里写任何保存逻辑——框架帮你做了。
config = {"configurable": {"thread_id": "..."}}
config 是 LangGraph 的配置字典，thread_id 是其最重要的字段。它告诉 LangGraph："这个执行属于哪个会话"。同一个 thread_id → 共享同一个状态存储。不同 thread_id → 完全隔离。
第二轮 invoke 不传全量 State
这是关键理解。第二轮的 invoke() 只传了 {"messages": [HumanMessage(...)]}——没有 iteration_count，也没有之前的消息。但图能正常运行，因为 LangGraph 在调用前做了两件事：
- 从 checkpointer 中恢复
thread_id="local-trend-session-1"的上一次最终 State - 把
invoke()的参数（新 HumanMessage）合并到恢复的 State 中
合并后的 State 包含完整的历史消息 + 新用户消息 + 之前的迭代计数——Agent 就像"没断过"一样继续。
⚠️ 常见坑：
thread_id一旦选定就不要变——如果你每次invoke都生成一个新的 thread_id（比如用uuid4()），那等于每次都是全新对话，checkpointer 等于白加了。
扩展版：SqliteSaver——持久化到磁盘
MemorySaver 的问题是重启就没。你的 LocalTrend 如果写到一半 Python 崩溃了，所有对话历史全丢。SqliteSaver 把状态写到 SQLite 数据库文件——重启、关机、搬机器，只要文件还在，记忆就在。
"""
第 3 章 扩展演示：SqliteSaver 持久化
Agent 的记忆写入磁盘，重启不丢失
"""
import os
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaverfrom chapter03_memory import build_graph, LocalTrendState
# ↑ 复用基础版的图构建逻辑（不包含 MemorySaver 的那部分）# ...（tool 定义、node 定义与基础版相同，此处省略）...def build_persistent_graph(db_path: str = "local_trend_memory.db"):"""构建带 SqliteSaver 的图。数据库文件不存在时会自动创建。"""builder = StateGraph(LocalTrendState)builder.add_node("agent", agent_node)builder.add_node("tools", tool_node)builder.add_edge(START, "agent")builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})builder.add_edge("tools", "agent")# SqliteSaver 需要一个数据库连接conn = sqlite3.connect(db_path, check_same_thread=False)checkpointer = SqliteSaver(conn)return builder.compile(checkpointer=checkpointer)if __name__ == "__main__":DB_PATH = "local_trend_memory.db"# --- 第一次运行：创建数据库并对话 ---print("=" * 60)print(" 首次运行——创建数据库")print("=" * 60)graph1 = build_persistent_graph(DB_PATH)config = {"configurable": {"thread_id": "persistent-session-1"}}result = graph1.invoke({"messages": [HumanMessage(content="搜索 AI 领域的爆款趋势")]},config)print(f"消息数: {len(result['messages'])}")print(f"Agent 最后回复: {result['messages'][-1].content[:150]}")# --- 模拟"重启"：重新创建 graph 对象，但用同一个数据库文件 ---print("\n" + "=" * 60)print(" 模拟重启——新的 Python 进程，同一个数据库")print("=" * 60)# 实际应用中这就是"程序重启后重新启动"graph2 = build_persistent_graph(DB_PATH)# 用同一个 thread_id——从数据库恢复之前的对话result2 = graph2.invoke({"messages": [HumanMessage(content="继续分析，这些趋势的写作风格有什么共同点？")]},config # 同一个 config!)print(f"消息数: {len(result2['messages'])}（包含重启前的历史）")print(f"Agent 最后回复: {result2['messages'][-1].content[:150]}")# --- 查看数据库中的状态 ---print("\n" + "=" * 60)print(" 查看 SQLite 中的 checkpoint 数据")print("=" * 60)conn = sqlite3.connect(DB_PATH)cursor = conn.execute("SELECT thread_id, checkpoint_id, parent_checkpoint_id FROM checkpoints ""WHERE thread_id = ?",("persistent-session-1",))rows = cursor.fetchall()print(f"数据库中该 thread 的 checkpoint 数: {len(rows)}")for row in rows[:5]: # 只看前 5 条print(f" thread={row[0]}, checkpoint={row[1][:20]}..., parent={str(row[2])[:20]}...")conn.close()print(f"\n 数据库文件位置: {DB_PATH}")print(" 删除此文件即可清空 Agent 的所有记忆。")
运行后检查你的工作目录——多了一个 local_trend_memory.db 文件。这就是 Agent 的"病历本"。删掉它，Agent 就失忆了；保留它，Agent 永远记得。
多会话管理
SqliteSaver 一个数据库文件可以存储无数个 thread 的状态。下面演示同时管理多个会话：
# 同一个 graph 对象服务多个线程
graph = build_persistent_graph("local_trend_memory.db")# 用户 A 在研究 AI 内容
graph.invoke({"messages": [HumanMessage(content="搜 AI 趋势")]},{"configurable": {"thread_id": "user-a"}}
)# 用户 B 在研究职场内容——完全独立，互不干扰
graph.invoke({"messages": [HumanMessage(content="搜职场趋势")]},{"configurable": {"thread_id": "user-b"}}
)# 切回用户 A——继续之前的对话
result = graph.invoke({"messages": [HumanMessage(content="继续，分析这些趋势的风格")]},{"configurable": {"thread_id": "user-a"}}
)
# result['messages'] 包含用户 A 三句话的全部上下文
这就是生产环境中"一个图服务 N 个用户"的基础架构。
本章小结
- 没有 Checkpointer 的图每次
invoke都从零开始——Agent 不记得之前的对话、搜索、分析结果。 - Checkpointer 在每个 super-step 后自动保存 State 快照——不需要在节点函数里手动存。
MemorySaver存在内存中，零配置，适合开发调试——但重启后消失。SqliteSaver写入 SQLite 文件，重启不丢失，适合单机部署和个人项目。thread_id是状态隔离的依据——同一个 thread_id 共享记忆，不同 thread_id 完全隔离。- 传入
compile()后，节点函数完全不用改——Checkpointer 对业务逻辑是透明的。 - checkpoint 形成链表——每个快照都有一个 parent，可以回溯到对话的任意历史时刻。
关键术语
| 术语 | 释义 |
|---|---|
| Checkpointer | LangGraph 的状态持久化组件，在每个 super-step 后自动保存 State |
| MemorySaver | 内存中的 Checkpointer 实现，重启后数据丢失，适合开发 |
| SqliteSaver | 基于 SQLite 的 Checkpointer，写入磁盘文件，重启不丢失 |
| thread_id | 会话隔离标识，同一个 thread_id 共享同一组 State 快照 |
| super-step | 图中一个节点执行完成的边界，每次 super-step 后触发 checkpoint |
| checkpoint | 某个 super-step 后的完整 State 快照，包含 metadata（来源节点、步数等） |
| config | invoke() 的配置参数，{"configurable": {"thread_id": "..."}} 是最常用的形式 |