Chatbot external persistent memory
SqliteSaver · PostgresSaver · Thread multi-session

1Review: What we have accomplished and what remains to be done

In the first few lessons of Module 2, we gradually built a conversational Chatbot with increasingly complete features. Core results include:

• Custom State Schema:useMessagesStateManage conversation message list, joinsummaryField save summary
• reducer mechanism：add_messagesreducer ensures that messages are appended rather than overwritten
• Message pruning and filtering: Avoid context window explosion
• Conversation summary node: When there are more than 6 messages, the summary is automatically triggered and only the most recent 2 original messages are retained.
• MemorySaver checkpoint: Supports the memory of cross-turn conversations in the same run

The goal of this lesson (Lesson 6): introduceExternal database checkpoint, so that the status of the graph can be permanently written to the disk or cloud database, completely solving the problem of "amnesia when power is cut off".

2MemorySaver vs External Database: The Fundamental Differences Between the Two Types of Checkpoints

What is a checkpointer?

LangGraph’s checkpointing mechanism is itsState persistence infrastructure layer. Each time the graph executes a node, LangGraph will serialize the current complete State and hand it over to the checkpointer for storage. The next time it is called, the checkpointer is responsible for restoring the last State, and the graph can continue with the last context.

Internal mechanism of MemorySaver

MemorySaverUse a Python dictionary (dict) as the storage backend. It saves the serialized state inself.storageIn this memory dictionary, the key is(thread_id, checkpoint_id)combination. This explains why it's so fast yet so fragile - the data never hits the disk.

from langgraph.checkpoint.memory import MemorySaver

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

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

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

How is state serialized?

LangGraph will serialize the entire State object (using JSON-compatible format by default) before writing the checkpoint. formessagesThis class of LangChain message objects implements__json__()Serialization interface, can be completely restored to the original type (HumanMessage、AIMessageetc.).

3SqliteSaver: lightweight persistence solution

SQLite is the most deployed database engine in the world and is used built-in by Android, iOS, Firefox, Python standard library, etc. Its characteristics areZero configuration, single file, serverless process, ideal as the first step in upgrading from in-memory storage to persistence.

Install dependencies

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

Way 1: SQLite in memory mode (still not durable!)

SQLite supports a special":memory:"Connection string, indicating the creation of a purely in-memory SQLite database. This andMemorySaverSimilarly, the data disappears when the process exits, but it uses the SQLite interface:

import sqlite3

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

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

Method 2: SQLite in file mode (real persistence)

Just change":memory:"Replace it with a file path, and SQLite will create (or open) a.dbfile. All data is written to this file and remains after the process exits:

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)

How does SqliteSaver work internally?

4Reuse message summary graph: connected to external checkpoints

These are the most important engineering practice points of this course:The definition code of the graph remains completely unchanged, just incompile()Change a checkpointer from time to time. We directly connect the complete message summary Chatbot graph of Lesson 5 to SqliteSaver.

Complete graph definition (same as 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)

The key line: replace the checkpointer

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

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

5Thread (thread) and thread_id: the core mechanism of multi-user multi-session

What is Thread?

In LangGraph,ThreadNot an operating system thread, but aA logically isolated unit of a conversation session. Each Thread has a uniquethread_id, LangGraph uses it to distinguish conversations between different users, or different conversation contexts for the same user.

you can putthread_idUnderstood as "conversation room number": conversations in different rooms do not interfere with each other, and multiple conversations in the same room share the same historical memory.

# 创建一个对话配置，指定 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()

Multi-user scenario: different thread_ids are completely isolated

# 用户 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!

Check the current status of Thread

usegraph.get_state(config)You can view the current complete State of a thread at any time, including the messages list and summary field:

# 查看 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"])    # 当前的对话摘要

6State persistence verification: session restoration after kernel restart

This is the most convincing demonstration of the course. Let's prove that even if the Jupyter Notebook kernel is completely restarted (equivalent to a server restart), the conversation history saved in the SQLite file still exists and can be seamlessly continued.

Verification steps

• 1

  Step 1: Conduct several rounds of dialogue (as shown in the previous section)

  Chatted with Chatbot for several rounds in thread "1", status writtenstate_db/example.db

• 2

  Step 2: Restart Notebook Kernel

  Click the Jupyter "Restart Kernel" button. The Python process is completely restarted, and all variables in memory (includingMemorySavercontent) disappears.

• 3

  Step 3: Reconnect to SQLite and rebuild the graph

  Rerun the initialization code and reconnect tosame one.dbFile, recompile the graph (the code is exactly the same)

• 4

  Step 4: Use the same thread_id to query the status

  callgraph.get_state(config), you will find that all historical messages and summaries are completely restored from the database

# ── 重启 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", "（暂无摘要）"))
# 输出：上次对话的完整历史！

Further verification: Continue this conversation

# 重启后，继续上次的对话——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!

7Production-level solution: PostgresSaver and deployment architecture

SQLite is suitable for stand-alone scenarios, but when your Chatbot needs servicesLarge number of concurrent users, or needScale out across multiple servers, you need to upgrade to an enterprise-level database like PostgreSQL.

How to use 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)

Production deployment architecture comparison

Integration in LangGraph Studio

The Studio configuration corresponding to this lesson is inmodule-2/studio/chatbot.pyin. Uselanggraph devAfter the command starts the local development server, you can visually view the complete state history of each Thread in LangSmith Studio's UI interface, replay the conversation at any checkpoint, and even edit the State directly. This is a powerful tool for debugging external memory problems.

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

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

8Complete process review and best practices

Knowledge map of this course

Complete migration path

• 1

  Development Phase: Using MemorySaver

  Zero configuration, fast iteration, focus on graph logic rather than storage details

• 2

  Local testing/small deployments: Upgrading to SqliteSaver

  Just change one linecompile()Call, verify persistence logic, test restart recovery, test multi-user isolation

• 3

  Production: Migrate to PostgresSaver

  Just change the samecompile()Get complete production-level concurrency, transactions, and backup capabilities in one line

• 4

  Special requirements: Implementing custom checkpointer

  inheritBaseCheckpointSaver, realizeget()、put()and other interfaces, you can access any backend (Redis, DynamoDB, self-developed storage, etc.)

Position on the learning route