【第 5 篇：记忆系统——长短期记忆与混合记忆】

realllccc

171人浏览 · 2026-06-06 17:32:17

realllccc · 2026-06-06 17:32:17 发布

第 5 篇：记忆系统——长短期记忆与混合记忆

系列记录：《从零搭建企业级 LLM 应用》，这是第 5 篇
上一篇：RAG 知识库问答——检索只是第一步

追问时，是谁提供了上下文？

为了搞清楚记忆到底是怎么工作的，我追踪了一次追问的完整链路：

用户输入 "那最后一名呢？"
  │
  ├─ 代码从 SQLite 消息表读出历史对话：
  │   [用户:"最近排名怎么样？", 助手:"排名如下..."]
  │
  ├─ 拼成消息列表：
  │   [HumanMessage("最近排名怎么样？"),
  │    AIMessage("排名如下..."),
  │    HumanMessage("那最后一名呢？")]  ← 当前问题追加
  │
  ├─ 这个列表被传入 Agent 的 state_input
  │
  └─ Agent 看到完整上下文，理解"那"指代的是排名

关键发现：LLM 能理解"那"的指代，靠的是代码手动从 SQLite 消息表读出历史，拼成 messages 列表再传进去。不是 LangGraph 自动处理的。

这引出了一个问题：为什么不用 LangGraph 自己保存的状态？

两套 SQLite，各有各的活

事实上，项目里有两个 SQLite 文件都在存对话相关的东西：

第一套：SQLite messages 表（业务数据库）

conversations ── 对话标题、时间
messages      ── 角色、内容、步骤日志、Agent 名、question_vec（向量）

存的是业务层面的对话记录。由 chat_page.py 在用户发送和 Agent 回复时分别写入。UI 展示、对话列表、多轮追问的上下文注入，全靠它。

第二套：SqliteSaver（data/checkpoints.sqlite）

这是 LangGraph 框架内部用的。每个图节点（supervisor、rag_agent 等）执行完后，框架自动把整个图状态快照存下来——消息、中间变量、通道值。它是 Agent 执行流程的"存档"，不是给代码直接读的。

	messages 表	SqliteSaver
谁写入	自己的页面代码	LangGraph 框架自动
写几次/轮	2 次（用户+助手）	N 次（每经过一个图节点）
存什么	简洁的对话记录 + 业务字段	图状态全量快照
给谁用	LLM 上下文 + UI 展示	Agent 断点恢复
能去掉吗	不能，去了追问失忆	不能，去了流程中断

一个管"我们聊过什么"（给 LLM 看），一个管"图跑到哪了"（给 LangGraph 用）。

为什么要自己维护消息表

既然 LangGraph 的 SqliteSaver 已经存了消息，为什么还要自己再搞一套？

做了几个测试后找到三个理由：

checkpoint 数据不是给人读的。里面是 LangGraph 内部序列化格式，含中间状态（比如 Agent 调了工具但还没返回结果时的瞬时状态）。直接拿来做 UI 展示和上下文注入不可靠。
缺少业务字段。checkpoint 里没有 user_id、agent_used、steps_log、question_vec 这些字段。对话列表要按用户过滤、要显示用了哪个 Agent、要做向量检索，这些都得靠自己的表。
权限校验。自己的消息表可以通过 WHERE user_id = ? 做用户隔离，checkpoint 做不到。

所以最终方案是两套并存：自己维护的消息表是对话记录的"事实来源"，SqliteSaver 是框架内部的状态快照。

消息保存的完整时间线

一轮对话中，保存发生在哪些时刻：

用户发送 "最近排名怎么样？"
│
├── chat_page.py
│   └── INSERT INTO messages (role='user', content='最近排名怎么样？')
│         ← 消息表 写①
│
├── agent.py → _build_messages_with_history()
│   └── 从消息表读出历史 → 构建消息列表
│
├── agent.stream() 开始
│   ├── supervisor 决策完成 ──→ SqliteSaver checkpoint 写①
│   ├── rag_agent 检索完成 ──→ SqliteSaver checkpoint 写②
│   └── supervisor 汇总完成 ──→ SqliteSaver checkpoint 写③
│
├── chat_page.py
│   └── INSERT INTO messages (role='assistant', content=回复)
│         ← 消息表 写②
│
├── 经验记忆：异步嵌入用户问题向量
│   └── UPDATE messages SET question_vec = [...] WHERE id = ①
│         ← 消息表 写③（仅当 agent_used == 'rag_agent'）
│
└── st.session_state.messages.append()
      ← session_state 缓存（纯内存）

短期记忆（STM）：当前实现与瓶颈

实现方式

SQLite 消息注入（_build_messages_with_history()）：从消息表读出该对话的全部历史消息，构建为 [HumanMessage, AIMessage, ...] 列表，显式传入 state_input["messages"]。这是 LLM 感知上下文的唯一途径。

还有一个细节：系统会对历史助手消息做语义降权，防止旧数据结论污染当前回答：

# agent.py 第 134 行
tagged = f"[历史回复，其中的数据和结论为当时查询结果，不代表当前真实状态]{msg['content']}"
messages.append(AIMessage(content=tagged))

比如上次查排名时张三在第一，但文件可能已更新，这个标签告诉 LLM"这是旧的，别当事实用"。

现状评估

特性	状态	说明
对话内上下文连续	✅	SQLite 全量加载 + SqliteSaver 状态持久化
服务重启后恢复	✅	文件持久化
多对话 / 用户隔离	✅	conversation_id + user_id
历史回复语义降权	✅	旧数据结论加标签，防止误导 LLM
历史条数限制	❌	全量加载，无 LIMIT
Token 计数	❌	无 token 统计
滑动窗口 / 截断	❌	旧消息不会自动丢弃
对话摘要	❌	超长对话无压缩

瓶颈

qwen-plus 上下文窗口约 131K tokens。每轮 RAG 检索返回大量文档片段时消耗更快。聊到 200+ 轮可能超限。

长期记忆（LTM）：两层架构

这个部分比最初预想的更完善。项目有两套长期记忆机制：

第一层：文档知识库（Dify RAG）

用户问题 → rag_search() → Dify 向量检索 → LLM 生成答案

内置查询改写、答案验证和至多重试 1 次。这是静态文档知识——公司制度、产品手册、操作规范等。

第二层：经验记忆缓存（Q&A 向量相似度匹配）🆕

这是最近加的功能，思路是：用户问过并被回答过的问题，不该每次从头检索。

实现流程：

用户提问
    │
    ├── Step 0: embed_text(question) → 用 Qwen text-embedding-v3 向量化
    │            search_cache(question_vec) → 扫描 messages 表中所有带向量的历史问题
    │            余弦相似度 ≥ 0.80 → 命中！
    │            命中的历史 Q&A 附加到 RAG 上下文（标注"仅供参考"）
    │
    └── Step 1-4: 正常 RAG 检索 → 生成答案

回答完毕后：
    └── 用户问题向量 → UPDATE messages SET question_vec = [...] WHERE id = ?

关键设计细节：

向量模型：Qwen text-embedding-v3，1024 维
相似度阈值：0.80（余弦相似度）
跨对话检索：扫描所有 conversation 的历史问答，不限于当前对话
噪音过滤：跳过"谢谢"“好的”“嗯”"然后呢"等短追问和纯闲聊，只缓存 6 字符以上的知识性问题
异常隔离：缓存逻辑包裹在 try/except 中，失败不影响主流程
仅对 RAG 问题生效：agent_used == "rag_agent" 时才写入向量，data_agent 的统计问题不缓存

两层 LTM 的对比

	Dify 知识库	经验缓存 (cache_service)
知识来源	人工上传的文档	自动从对话中沉淀
更新方式	手动上传	自动（每轮对话后）
检索方式	Dify API（语义检索）	本地 SQLite 向量扫描
覆盖范围	静态文档内容	历史问答经验
跨对话	✅（知识库全局共享）	✅（扫描所有对话记录）

现状评估

特性	状态	说明
文档知识检索	✅	Dify 向量库 + text-embedding-3-small
历史问答经验记忆	✅	cache_service 向量相似度匹配
混合检索 (hybrid_search)	❌	仅 semantic_search
重排序 (rerank)	❌	reranking_enable: False
用户画像存储	❌	无用户偏好、部门等画像
对话事实自动提取	❌	缓存的是"问题相似"，非"关键事实提取"

混合记忆（Hybrid Memory）：现状与差距

把上面的都拼起来，看整体状态：

用户输入
    │
    ├── [STM] _build_messages_with_history()
    │         → 从消息表加载当前对话全部历史
    │         → 历史回复加语义降权标签
    │         → 显式注入 state_input["messages"]
    │
    ├── [STM] SqliteSaver
    │         → LangGraph 内部状态持久化（透明运行）
    │
    ├── [LTM ①] rag_search() → Dify 知识库
    │         → 语义检索文档 → 答案验证 → 重试
    │
    └── [LTM ②] cache_service 经验缓存
              → 向量相似度匹配历史 Q&A
              → 命中后附加到 RAG 上下文

差距清单

组件	当前状态
Buffer (N 轮截断)	❌ 全量加载
Vector 对话记忆	⚠️ 有 Q&A 缓存，但只比对问题，非对话片段矢量库
Summary 对话摘要	❌ 未实现
User Profile 用户画像	❌ 未实现
Fact Extraction 事实提取	❌ 未实现
STM→LTM 自动合并	⚠️ 缓存是自动沉淀的一种形式，但缺少摘要和关键信息提取

总结：一张表看清全貌

问题	答案
追问时上下文从哪来？	SQLite messages 表手动注入（非 LangGraph 自处理）
为什么有两套 SQLite？	messages 管业务对话，SqliteSaver 管 Agent 执行快照
短期记忆怎么实现的？	全量历史消息 + 语义降权标签 + SqliteSaver 状态持久化
长期记忆怎么实现的？	Dify 文档 RAG + cache_service 历史问答经验匹配
经验缓存能跨对话吗？	能。扫描所有对话的消息记录
跨对话能记用户偏好吗？	不能。无用户画像存储
超长对话会怎样？	可能超出 LLM 上下文窗口导致截断或报错
算混合记忆吗？	雏形已有（STM→LTM 自动沉淀），但缺摘要、画像和事实提取