Chatbot 外部持久化记忆
SqliteSaver · PostgresSaver · Thread 多会话

1 回顾：我们已经做到了什么，还差什么

在 Module 2 的前几课中，我们逐步构建了一个功能越来越完善的对话型 Chatbot。核心成果包括：

• 自定义 State Schema：用 MessagesState 管理对话消息列表，加入 summary 字段保存摘要
• Reducer 机制：add_messages Reducer 确保消息是追加而非覆盖
• 消息修剪与过滤：避免 context window 爆炸
• 对话摘要节点：超过 6 条消息时自动触发摘要，只保留最近 2 条原始消息
• MemorySaver 检查点：支持同一次运行中跨轮对话的记忆

本课（Lesson 6）的目标：引入外部数据库检查点，让图的状态能够永久写入磁盘或云端数据库，彻底解决"断电即失忆"的问题。

2 MemorySaver vs 外部数据库：两种检查点的根本区别

检查点（Checkpointer）是什么？

LangGraph 的检查点机制是其状态持久化的基础设施层。每次图执行完一个节点后，LangGraph 会把当前完整的 State 序列化，交给检查点器保存。下次调用时，检查点器负责把上次的 State 还原出来，图就能继续上次的上下文。

MemorySaver 的内部机制

MemorySaver 用一个 Python 字典（dict）作为存储后端。它将序列化后的状态保存在 self.storage 这个内存字典中，key 是 (thread_id, checkpoint_id) 的组合。这解释了为什么它如此快速但又如此脆弱——数据从未落盘。

from langgraph.checkpoint.memory import MemorySaver

# MemorySaver：数据存在 Python 进程的内存里
memory = MemorySaver()

# 编译时挂上检查点器
graph = workflow.compile(checkpointer=memory)

# 问题：当这个 Python 进程结束，memory 对象销毁，所有数据消失

状态如何被序列化？

LangGraph 在写入检查点前，会把整个 State 对象序列化（默认使用 JSON 兼容的格式）。对于 messages 这类 LangChain 消息对象，它们实现了 __json__() 序列化接口，可以完整地被还原为原始类型（HumanMessage、AIMessage 等）。

3 SqliteSaver：轻量级持久化方案

SQLite 是世界上部署量最大的数据库引擎，被 Android、iOS、Firefox、Python 标准库等内置使用。它的特点是零配置、单文件、无服务进程，非常适合作为从内存存储升级到持久化的第一步。

安装依赖

# 安装 LangGraph 的 SQLite 检查点支持库
pip install langgraph-checkpoint-sqlite

方式一：内存模式的 SQLite（仍然不持久！）

SQLite 支持一种特殊的 ":memory:" 连接字符串，表示创建一个纯内存中的 SQLite 数据库。这和 MemorySaver 类似，进程退出数据消失，但它使用了 SQLite 的接口：

import sqlite3

# ":memory:" 表示内存 SQLite，进程退出即消失
conn = sqlite3.connect(":memory:", check_same_thread=False)

# check_same_thread=False 允许多线程访问同一连接
（LangGraph 的异步执行需要这个参数）

方式二：文件模式的 SQLite（真正持久化）

只需将 ":memory:" 换成一个文件路径，SQLite 就会在磁盘上创建（或打开）一个 .db 文件。所有数据写入这个文件，进程退出后依然存在：

import sqlite3
import os

# 创建存放数据库的目录
os.makedirs("state_db", exist_ok=True)

# 指定磁盘文件路径——数据将永久保存
db_path = "state_db/example.db"
conn = sqlite3.connect(db_path, check_same_thread=False)

# 用这个连接创建 SqliteSaver 检查点器
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver(conn)

SqliteSaver 内部如何工作？

4 重用消息摘要图：接上外部检查点

这是本课最重要的工程实践要点：图的定义代码完全不变，只需在 compile() 时换一个检查点器。我们把 Lesson 5 的完整消息摘要 Chatbot 图，直接接上 SqliteSaver。

完整的图定义（与 Lesson 5 相同）

from typing_extensions import Literal
from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from langgraph.graph import END, MessagesState, StateGraph, START

model = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)

# State 继承 MessagesState，额外增加 summary 字段
class State(MessagesState):
    summary: str

# ── Node 1：调用 LLM ──
def call_model(state: State):
    summary = state.get("summary", "")
    if summary:
        # 把历史摘要注入为系统消息
        system_message = f"Summary of conversation earlier: {summary}"
        messages = [SystemMessage(content=system_message)] + state["messages"]
    else:
        messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": response}

# ── Node 2：生成/更新对话摘要，并删除旧消息 ──
def summarize_conversation(state: State):
    summary = state.get("summary", "")
    if summary:
        summary_message = (
            f"This is summary of the conversation to date: {summary}\n\n"
            "Extend the summary by taking into account the new messages above:"
        )
    else:
        summary_message = "Create a summary of the conversation above:"

    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = model.invoke(messages)

    # 用 RemoveMessage 删除除最近 2 条以外的所有消息
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": response.content, "messages": delete_messages}

# ── 路由函数：消息超过 6 条时触发摘要 ──
def should_continue(state: State) -> Literal["summarize_conversation", END]:
    if len(state["messages"]) > 6:
        return "summarize_conversation"
    return END

# ── 构建图 ──
workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)

关键一行：换掉检查点器

# 之前用内存检查点：
# graph = workflow.compile(checkpointer=MemorySaver())

# 现在用 SQLite 检查点——图的其他代码零改动！
graph = workflow.compile(checkpointer=memory)  # memory 是 SqliteSaver 实例

5 Thread（线程）与 thread_id：多用户多会话的核心机制

什么是 Thread？

在 LangGraph 中，Thread（线程）不是操作系统的线程，而是一个对话会话的逻辑隔离单元。每个 Thread 有一个唯一的 thread_id，LangGraph 用它来区分不同用户的对话、或同一用户的不同对话上下文。

你可以把 thread_id 理解为"对话房间号"：不同房间里的对话互不干扰，同一房间里的多次对话共享同一段历史记忆。

# 创建一个对话配置，指定 thread_id
config = {"configurable": {"thread_id": "1"}}

# 第一轮对话
input_message = HumanMessage(content="hi! I'm Lance")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
    m.pretty_print()

# 第二轮对话（同一 thread_id → 记得 Lance）
input_message = HumanMessage(content="what's my name?")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
    m.pretty_print()
# 输出：Your name is Lance!

# 第三轮对话
input_message = HumanMessage(content="i like the 49ers!")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
    m.pretty_print()

多用户场景：不同的 thread_id 完全隔离

# 用户 A 的对话（thread_id="user_alice"）
config_alice = {"configurable": {"thread_id": "user_alice"}}
graph.invoke({"messages": [HumanMessage(content="I'm Alice")]}, config_alice)

# 用户 B 的对话（thread_id="user_bob"）——与 Alice 完全隔离
config_bob = {"configurable": {"thread_id": "user_bob"}}
graph.invoke({"messages": [HumanMessage(content="I'm Bob")]}, config_bob)

# 用户 A 继续（仍然记得 Alice 的历史）
graph.invoke({"messages": [HumanMessage(content="what's my name?")]}, config_alice)
# → 输出：Your name is Alice!

查看当前 Thread 的状态

使用 graph.get_state(config) 可以随时查看某个 thread 当前的完整 State，包括 messages 列表和 summary 字段：

# 查看 thread "1" 的当前状态
config = {"configurable": {"thread_id": "1"}}
graph_state = graph.get_state(config)

# graph_state.values 包含完整的 State 字典
print(graph_state.values["messages"])   # 当前保留的消息列表
print(graph_state.values["summary"])    # 当前的对话摘要

6 状态持久化验证：重启内核后恢复会话

这是本课最有说服力的演示。我们来证明：即使 Jupyter Notebook 的 kernel 被完全重启（等同于服务器重启），保存在 SQLite 文件中的对话历史依然存在，可以无缝续接。

验证步骤

• 1

  步骤一：进行若干轮对话（如上节所示）

  在 thread "1" 中与 Chatbot 聊了几轮，状态已写入 state_db/example.db

• 2

  步骤二：重启 Notebook Kernel

  点击 Jupyter 的 "Restart Kernel" 按钮。Python 进程完全重启，所有内存中的变量（包括 MemorySaver 的内容）全部消失。

• 3

  步骤三：重新连接 SQLite 并重建图

  重新运行初始化代码，重新连接到同一个 .db 文件，重新编译图（代码完全相同）

• 4

  步骤四：用相同的 thread_id 查询状态

  调用 graph.get_state(config)，你会发现所有历史消息和摘要都完整地从数据库还原出来了

# ── 重启 Kernel 后，重新运行以下代码 ──

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

# 重新连接同一个 .db 文件
db_path = "state_db/example.db"
conn = sqlite3.connect(db_path, check_same_thread=False)
memory = SqliteSaver(conn)

# 重新编译图（节点/边定义完全不变）
graph = workflow.compile(checkpointer=memory)

# 使用相同的 thread_id——神奇的事情发生了
config = {"configurable": {"thread_id": "1"}}
graph_state = graph.get_state(config)

print("历史消息数量：", len(graph_state.values["messages"]))
print("对话摘要：", graph_state.values.get("summary", "（暂无摘要）"))
# 输出：上次对话的完整历史！

进一步验证：继续这个对话

# 重启后，继续上次的对话——LLM 仍然记得 Lance 喜欢 49ers
input_message = HumanMessage(content="which nfl team do I like?")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
    m.pretty_print()
# 即使 Kernel 重启过，仍然输出：You like the San Francisco 49ers!

7 生产级方案：PostgresSaver 与部署架构

SQLite 适合单机场景，但当你的 Chatbot 需要服务大量并发用户，或者需要多台服务器横向扩展时，就需要升级到 PostgreSQL 这样的企业级数据库。

PostgresSaver 的使用方式

# 安装 PostgreSQL 检查点支持库
# pip install langgraph-checkpoint-postgres psycopg

from psycopg import connect
from langgraph.checkpoint.postgres import PostgresSaver

# 连接到 PostgreSQL 数据库
conn_string = "postgresql://user:password@localhost:5432/chatbot_db"
conn = connect(conn_string, autocommit=True)

# 初始化检查点表（首次使用时调用）
checkpointer = PostgresSaver(conn)
checkpointer.setup()  # 自动创建所需的数据库表

# 编译图时挂上 PostgreSQL 检查点——其余代码零改动！
graph = workflow.compile(checkpointer=checkpointer)

生产部署架构对比

LangGraph Studio 中的集成

本课对应的 Studio 配置在 module-2/studio/chatbot.py 中。使用 langgraph dev 命令启动本地开发服务器后，可以在 LangSmith Studio 的 UI 界面中可视化地查看每个 Thread 的完整状态历史、在任意检查点重放对话、甚至直接编辑 State。这是调试外部记忆问题的强大工具。

# 在 module-2/studio 目录下运行：
langgraph dev

# 访问 Studio UI：
# https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

8 完整流程回顾与最佳实践

本课的知识图谱

完整的迁移路径

• 1

  开发阶段：使用 MemorySaver

  零配置，迭代快，专注图逻辑而非存储细节

• 2

  本地测试 / 小型部署：升级到 SqliteSaver

  只需改一行 compile() 调用，验证持久化逻辑、测试重启恢复、测试多用户隔离

• 3

  生产环境：迁移到 PostgresSaver

  同样只改 compile() 一行，获得完整的生产级并发、事务、备份能力

• 4

  特殊需求：实现自定义 Checkpointer

  继承 BaseCheckpointSaver，实现 get()、put() 等接口，即可接入任意后端（Redis、DynamoDB、自研存储等）

学习路线上的位置