🧠 用 LangGraph 实现「会总结」的 AI 对话：对话记忆压缩 + SQLite 持久化

作者：AI工程笔记
标签：LangChain｜LangGraph｜对话系统｜记忆压缩｜OpenAI 兼容模型

在构建长期对话型 AI 应用时，我们常面临两个核心挑战：

上下文长度爆炸：随着对话轮次增加，消息列表越来越长，很快超出 LLM 的 token 限制；
状态无法持久化：用户刷新页面或重启服务后，AI 就“失忆”了。

今天，我们就用 LangGraph + SQLite 打造一个能自动总结历史对话并持久保存状态的智能聊天机器人。它不仅能记住你是“小明”，还能在后台默默压缩记忆，避免上下文膨胀！

🔧 技术栈一览

langgraph：用于构建状态机驱动的 AI 工作流
langchain_openai.ChatOpenAI：兼容 OpenAI 协议的模型接口（支持 DeepSeek、Moonshot、APIYi 等）
SqliteSaver：将对话状态持久化到 SQLite 数据库
自定义 State：引入 summary 字段实现记忆摘要

🧩 第一步：安全设置 API 密钥

为避免密钥硬编码，我们使用 getpass 动态输入：

import os, getpass

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

_set_env("OPENAI_API_KEY")
_set_env("OPENAI_BASE_URL")  # 例如：https://api.apiyi.com/v1

✅ 支持任意 OpenAI 兼容 API（DeepSeek、Qwen、Ollama 等均可）

💾 第二步：SQLite 持久化对话状态

LangGraph 提供 checkpointer 机制，我们用 SQLite 保存每个 thread_id 的完整状态：

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3, os

db_path = "state_db/chat.db"
os.makedirs(os.path.dirname(db_path), exist_ok=True)
conn = sqlite3.connect(db_path, check_same_thread=False)
memory = SqliteSaver(conn)

⚠️ 注意：check_same_thread=False 是为了支持多线程 FastAPI 场景（生产环境建议用连接池）

🧠 第三步：定义带「摘要」的状态

继承 MessagesState 并新增 summary 字段：

from langgraph.graph import MessagesState

class State(MessagesState):
    summary: str  # 存储压缩后的对话摘要

这样，我们的状态不仅包含消息列表，还自带一个“记忆快照”。

🤖 第四步：构建智能工作流

节点 1：调用模型（带摘要上下文）

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="deepseek-chat", temperature=0)

def call_model(state: State):
    messages = state["messages"]
    if summary := state.get("summary"):
        # 将摘要作为系统消息前置
        messages = [SystemMessage(content=f"此前对话摘要：{summary}")] + messages
    response = model.invoke(messages)
    return {"messages": [response]}

💡 摘要以 SystemMessage 形式注入，确保 LLM 优先关注

节点 2：自动总结对话（当消息 >6 条时触发）

def summarize_conversation(state: State):
    prompt = (
        f"目前为止的对话摘要：{state['summary']}\n\n请结合新消息扩展摘要："
        if state.get("summary")
        else "请对上方对话创建简洁摘要："
    )
    messages = state["messages"] + [HumanMessage(content=prompt)]
    summary = model.invoke(messages).content
    
    # 删除除最近 2 条外的所有消息（内存优化）
    to_delete = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": summary, "messages": to_delete}

✨ 关键技巧：用 RemoveMessage 原子操作清理历史，避免手动拼接列表出错

条件路由：何时需要总结？

1 2	def should_continue(state: State): return "summarize_conversation" if len(state["messages"]) > 6 else END

📏 阈值可调：6 条是经验值，可根据模型上下文窗口调整

🧪 第五步：编译图 & 测试对话

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(State)
workflow.add_edge(START, "conversation")
workflow.add_node("conversation", call_model)
workflow.add_node("summarize_conversation", summarize_conversation)
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)

graph = workflow.compile(checkpointer=memory)

# 同一 thread_id = 同一对话线程
config = {"configurable": {"thread_id": "user_123"}}

# 第一轮
graph.invoke({"messages": [HumanMessage("你好！我是小明")]}, config)

# 第二轮（AI 记得你！）
output = graph.invoke({"messages": [HumanMessage("我叫什么？")]}, config)
print(output["messages"][-1].content)  # → “你叫小明！”

✅ 效果验证：即使重启程序，只要 thread_id 相同，AI 依然记得你是谁！

🌟 为什么这个设计值得学习？

特性	价值
自动摘要压缩	解决上下文膨胀，延长对话寿命
SQLite 持久化	用户断连后仍可恢复上下文
状态机清晰	用图结构表达逻辑，易于扩展（如加工具调用）
兼容任意模型	只需换 `ChatOpenAI` 的 endpoint