https://academy.langchain.com/courses/intro-to-langgraph

https://github.com/shangxiang0907/langchain-academy

Chatbot with message summarization & external DB memory 具备消息摘要与外部数据库记忆的聊天机器人

Review 回顾

We’ve covered how to customize graph state schema and reducer.

我们已介绍如何自定义图状态模式（state schema）与规约器（reducer）。

We’ve also shown a number of tricks for trimming or filtering messages in graph state.

我们还展示了多种用于修剪或过滤图状态中消息的技巧。

We’ve used these concepts in a Chatbot with memory that produces a running summary of the conversation.

我们已在具备记忆功能的聊天机器人中应用了这些概念，该机器人可生成对话的持续摘要。

Goals 目标

But, what if we want our Chatbot to have memory that persists indefinitely?

但如果我们希望聊天机器人拥有永久持久化的记忆，该怎么办？

Now, we’ll introduce some more advanced checkpointers that support external databases.

接下来，我们将介绍一些更高级的支持外部数据库的检查点器（checkpointer）。

Here, we’ll show how to use Sqlite as a checkpointer, but other checkpointers, such as Postgres are available!

此处，我们将演示如何使用 Sqlite 作为检查点器，但其他检查点器（例如 Postgres）也可用！

%%capture --no-stderr
%pip install --quiet -U langgraph-checkpoint-sqlite langchain_core langgraph langchain_openai

import os, getpass

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

from dotenv import find_dotenv, load_dotenv

load_dotenv(find_dotenv(usecwd=True))
_set_env("OPENAI_API_KEY")

Sqlite

A good starting point here is the SqliteSaver checkpointer.

一个良好的起点是 SqliteSaver 检查点器。

Sqlite is a small, fast, highly popular SQL database.

Sqlite 是一种 轻量、快速且广受欢迎 的 SQL 数据库。

If we supply ":memory:" it creates an in-memory Sqlite database.

若传入 ":memory:"，它将创建一个内存中的 Sqlite 数据库。

import sqlite3
# In memory
conn = sqlite3.connect(":memory:", check_same_thread = False)

But, if we supply a db path, then it will create a database for us!

但若提供数据库路径，则会自动为我们创建一个数据库！

# pull file if it doesn't exist and connect to local db
!mkdir -p state_db && [ ! -f state_db/example.db ] && wget -P state_db https://github.com/langchain-ai/langchain-academy/raw/main/module-2/state_db/example.db

db_path = "state_db/example.db"
conn = sqlite3.connect(db_path, check_same_thread=False)

# Here is our checkpointer 
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver(conn)

Let’s re-define our chatbot.

让我们重新定义聊天机器人。

from typing_extensions import Literal
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage

from langgraph.graph import END
from langgraph.graph import MessagesState


model = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "qwen-plus"), base_url=os.getenv("OPENAI_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"),temperature=0)

class State(MessagesState):
    summary: str

# Define the logic to call the model
def call_model(state: State):
    
    # Get summary if it exists
    summary = state.get("summary", "")

    # If there is summary, then we add it
    if summary:
        
        # Add summary to system message
        system_message = f"Summary of conversation earlier: {summary}"

        # Append summary to any newer messages
        messages = [SystemMessage(content=system_message)] + state["messages"]
    
    else:
        messages = state["messages"]
    
    response = model.invoke(messages)
    return {"messages": response}

def summarize_conversation(state: State):
    
    # First, we get any existing summary
    summary = state.get("summary", "")

    # Create our summarization prompt 
    if summary:
        
        # A summary already exists
        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:"

    # Add prompt to our history
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = model.invoke(messages)
    
    # Delete all but the 2 most recent messages
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": response.content, "messages": delete_messages}

# Determine whether to end or summarize the conversation
def should_continue(state: State)-> Literal ["summarize_conversation",END]:
    
    """Return the next node to execute."""
    
    messages = state["messages"]
    
    # If there are more than six messages, then we summarize the conversation
    if len(messages) > 6:
        return "summarize_conversation"
    
    # Otherwise we can just end
    return END

Now, we just re-compile with our sqlite checkpointer.

现在，只需使用我们的 sqlite 检查点器重新编译即可。

from IPython.display import Image, display
from langgraph.graph import StateGraph, START

# Define a new graph
workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)

# Set the entrypoint as conversation
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)

# Compile
graph = workflow.compile(checkpointer=memory)
display(Image(graph.get_graph().draw_mermaid_png()))

Now, we can invoke the graph several times.

现在，我们可以多次调用该图。

# Create a thread
config = {"configurable": {"thread_id": "1"}}

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

input_message = HumanMessage(content="what's my name?")
output = graph.invoke({"messages": [input_message]}, config) 
for m in output['messages'][-1:]:
    m.pretty_print()

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

Let’s confirm that our state is saved locally.

让我们确认状态已本地保存。

config = {"configurable": {"thread_id": "1"}}
graph_state = graph.get_state(config)
graph_state

Persisting state 持久化状态

Using database like Sqlite means state is persisted!

使用 Sqlite 等数据库意味着状态将被持久化！

For example, we can re-start the notebook kernel and see that we can still load from Sqlite DB on disk.

例如，我们可以重启笔记本内核，并仍能从磁盘上的 Sqlite 数据库加载状态。

# Create a thread
config = {"configurable": {"thread_id": "1"}}
graph_state = graph.get_state(config)
graph_state

Studio

⚠️ Notice

⚠️ 注意

Since filming these videos, we’ve updated Studio so that it can now be run locally and accessed through your browser.

自录制这些视频以来，我们已更新 Studio，使其现在可本地运行并通过浏览器访问。

This is the preferred way to run Studio instead of using the Desktop App shown in the video.

这是运行 Studio 的首选方式，而非视频中展示的桌面应用程序。

It is now called LangSmith Studio instead of LangGraph Studio.

它现在被称为 LangSmith Studio，而非 LangGraph Studio。

Detailed setup instructions are available in the “Getting Setup” guide at the start of the course.

详细的安装说明请参阅本课程开头的“入门设置”指南。

You can find a description of Studio here, and specific details for local deployment here.

您可在此处查看 Studio 的说明：链接，以及本地部署的具体细节：链接。

To start the local development server, run the following command in your terminal in the /studio directory in this module:

要在本地启动开发服务器，请在本模块的 /studio 目录下于终端中运行以下命令：

langgraph dev

You should see the following output:

您应看到如下输出：

- 🚀 API: http://127.0.0.1:2024
- 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
- 📚 API Docs: http://127.0.0.1:2024/docs

Open your browser and navigate to the Studio UI URL shown above.

打开您的浏览器并导航至上方显示的 Studio UI URL。

Load the chatbot in Studio, which uses module-2/studio/chatbot.py set in module-2/studio/langgraph.json.

在 Studio 中加载 chatbot，其对应 module-2/studio/chatbot.py 文件，并由 module-2/studio/langgraph.json 中指定。

主要讲了什么，能解决什么问题，实际开发有什么用

这篇主要讲的是：如何让 LangGraph 聊天机器人把对话状态保存到 SQLite 数据库，并通过“历史摘要”控制上下文长度。

它把两种机制结合起来：

1. Checkpointer 持久化：把消息、摘要等 Graph State 保存到外部数据库。
2. 消息摘要与删除：消息太多时，将旧对话压缩成摘要，只保留最近几条原始消息。

一、整体运行流程

每次调用 graph.invoke() 时，LangGraph 都会：

• 根据 thread_id 找到对应会话；
• 从 SQLite 中恢复之前的 messages 和 summary；
• 执行本轮对话；
• 将更新后的状态重新写入 SQLite。

因此，即使程序或 Notebook 重启，只要数据库文件还在，对话状态就能恢复。

---

二、代码里的核心部分

1. 使用 SQLite 保存状态

conn = sqlite3.connect(db_path, check_same_thread=False)
memory = SqliteSaver(conn)

graph = workflow.compile(checkpointer=memory)

这里的 SqliteSaver 是 LangGraph 的检查点器。

它会自动保存 Graph State，包括：

class State(MessagesState):
    summary: str

也就是：

• messages：近期对话消息；
• summary：较早对话的摘要；
• Graph 当前执行到的位置；
• 每个 thread_id 对应的状态版本。

需要注意：

sqlite3.connect(":memory:")

只存在于当前进程内，程序退出后就消失，适合测试。

而：

sqlite3.connect("state_db/example.db")

会写入磁盘，重启程序后仍然存在，才是真正的本地持久化。

2. 使用 thread_id 区分会话

config = {
    "configurable": {
        "thread_id": "1"
    }
}

thread_id 类似聊天系统中的 conversation_id。

例如：

thread_id = user_001_chat_001
thread_id = user_001_chat_002
thread_id = user_002_chat_001

不同 thread_id 拥有不同的对话状态。相同 thread_id 再次调用时，会接着之前的状态继续运行。

3. 把旧消息压缩成摘要

当消息超过 6 条：

if len(messages) > 6:
    return "summarize_conversation"

系统会让模型总结当前对话：

response = model.invoke(messages)

如果之前已经有摘要，就不是重新总结全部内容，而是在旧摘要基础上追加新信息：

"Extend the summary by taking into account the new messages above:"

这叫“滚动摘要”或“持续摘要”。

4. 删除旧消息

摘要生成以后：

delete_messages = [
    RemoveMessage(id=m.id)
    for m in state["messages"][:-2]
]

它会删除除最近两条以外的旧消息。

最终状态大致是：

summary:
  用户名叫 Lance，喜欢 49ers，之前询问过……

messages:
  最近一条用户消息
  最近一条 AI 回复

下次调用模型时，再把二者组合：

messages = [
    SystemMessage(content=f"Summary of conversation earlier: {summary}")
] + state["messages"]

这样模型既知道较早发生过什么，又不需要读取全部历史消息。

---

三、它解决了什么问题？

1. 程序重启后记忆丢失

如果只把状态放在 Python 变量或内存中：

sqlite3.connect(":memory:")

进程退出以后，对话就会消失。

换成磁盘 SQLite 后，程序重启仍然可以通过同一个 thread_id 找回状态。

2. 多轮对话越来越长

如果每次都把所有历史消息发送给模型，会产生：

• Token 消耗越来越大；
• 请求越来越慢；
• 成本越来越高；
• 最终超过模型上下文窗口；
• 大量无关历史可能干扰模型回答。

通过“摘要 + 最近消息”，上下文长度可以保持在相对稳定的范围。

3. 支持多个独立会话

借助 thread_id，同一套 Graph 可以同时管理很多对话：

用户 A 的会话 → thread_A
用户 B 的会话 → thread_B
用户 A 的另一个窗口 → thread_A_2

不会把不同用户或不同聊天窗口的消息混在一起。

4. Graph 执行被中断后可以恢复

Checkpointer 保存的不只是聊天内容，还保存 LangGraph 的执行状态。

这对复杂 Agent 很重要。例如 Agent 执行到“等待人工批准”时暂停，之后可以从检查点继续，而不是重新执行整个流程。

---

四、实际开发有什么用？

客服机器人

机器人可以记住当前工单的历史信息：

• 用户报告过什么问题；
• 已经尝试过哪些处理方式；
• 客服承诺了什么；
• 当前工单进展到哪里。

即使服务重启，也可以恢复这个会话。

AI 助手

例如你的 JobCopilot，可以按用户和任务保存状态：

thread_id = 用户ID + 求职任务ID

它可以记住：

• 用户的目标岗位；
• 不考虑哪些职位；
• 简历已经修改到哪个版本；
• 哪些职位已经分析或投递；
• 当前求职工作流执行到哪一步。

历史内容过长时，可以把旧过程压缩为：

用户目标是新加坡 AI Platform 岗位，需要 EP；
不考虑自动驾驶感知岗位；
已经分析 15 个职位，其中 4 个适合申请……

需要人工审批的 Agent

例如自动投递职位：

状态保存到数据库后，即使审批发生在几小时或几天后，也能继续原来的工作流。

长时间运行的工作流

适合：

• 深度研究 Agent；
• 代码生成与测试 Agent；
• 数据分析工作流；
• 多步骤审批流程；
• 需要失败重试的任务；
• 多 Agent 协作流程。

任务失败或服务重启后，可以从检查点恢复，减少重复调用模型和外部 API。

---

五、它和真正的“长期记忆”有什么区别？

这篇虽然称为“外部数据库记忆”，但严格来说主要讲的是：

把某个会话的 Graph State 持久化。

它更接近“会话记忆”，不完全等于跨会话的用户长期记忆。

能力	Checkpointer	真正的长期记忆
保存当前会话状态	是	可以
根据 thread_id 恢复聊天	是	可以
记住用户长期偏好	需要额外设计	是
跨多个聊天共享记忆	默认不行	是
按语义搜索相关记忆	默认不行	通常支持
典型存储	SQLite/Postgres	Store、Postgres、向量数据库

例如，用户在 thread_id="1" 中说“我喜欢 Python”，新建 thread_id="2" 后，默认不会自动知道这件事。

如果希望跨会话记住用户偏好，需要额外设计：

user_id → 长期记忆
thread_id → 当前会话状态

所以实际项目通常会同时使用：

• Checkpointer：保存某个工作流或会话的执行状态；
• 长期记忆 Store：保存用户偏好、事实和历史经验；
• 业务数据库：保存用户、职位、简历、申请记录等正式数据。

---

六、实际项目中的选择

SQLite 适合：

• 学习和实验；
• 单机应用；
• 本地开发；
• 个人项目；
• 并发量比较小的服务。

生产环境、多实例部署或高并发应用一般更适合 PostgreSQL，因为多个服务实例需要共享同一份状态，而且 SQLite 的并发写入能力有限。

一句话总结：

这篇教你用 thread_id + SQLite Checkpointer 保存每个会话的 LangGraph 状态，并通过“滚动摘要 + 删除旧消息”控制上下文长度，从而实现可重启、可恢复、成本可控的多轮聊天或 Agent 工作流。