确定性优先记忆检索：用CueMap思路打造连续召回的知识库系统

从第一次看到 “Show HN: CueMap – deterministic-first memory retrieval for continuous recall” 这个标题开始，我就被其中两个关键词吸引了：deterministic-first和continuous recall。前者说的是“确定性优先”，后者是“连续召回”。这两个词放一起，等于直接回应了当前 AI 应用里最让人头疼的问题：知识库检索到底能不能稳定、实时、可追踪地返回结果？

本文不打算复述 CueMap 的源码，而是围绕这个项目提出的核心思路，拆解一套可以落到代码里的记忆检索系统。我们会从概念出发，聊清楚什么是 deterministic-first，什么是 continuous recall，然后用 Python + SQLite 实现一个最小可用版本。整个工程代码都可以直接复制运行，适合对 RAG（检索增强生成）、个人知识库、AI 助手记忆模块感兴趣的开发者。

读完本文你会掌握：连续召回的数据模型怎么设计、确定性优先检索如何实现、以及如何让检索结果在多次交互中保持连贯。

1. 背景与核心概念

1.1 Continuous Recall 是什么

continuous recall翻译过来是“连续召回”，它描述的能力是：系统不只是根据当前这一次输入做检索，而是在一段连续对话或连续任务中，不断利用之前的信息来帮助当前决策。

举个例子。你在和一个 AI 助手聊“低糖饮食计划”，第一轮它给你推荐了早餐，第二轮你问“那午餐呢”，如果系统只能看到第二轮这句话，它根本不知道午餐要围绕“低糖”展开。但如果它有 continuous recall，它会自动把第一轮提到的“低糖”作为上下文线索，继续检索相关笔记或知识片段。

这类能力在以下场景特别需要：

• 智能客服：用户分多轮描述问题，客服系统要持续召回历史工单和知识库条目。
• 个人知识库：笔记越来越多，检索要能结合用户最近查看的内容，给出连续相关的提示。
• AI 编程助手：多轮修改同一个文件时，要记住前面几轮的操作意图。
• 自动驾驶语音助手：连续语音指令需要结合上一条指令的语义。

连续召回的本质是：让检索过程拥有“短期记忆”，而不是把每一次查询当作孤立事件。

1.2 传统向量检索为什么不够

现在大多数知识库系统都在用向量检索。流程是：把文档切块 -> 用 Embedding 模型转成向量 -> 存到向量数据库 -> 查询时把问题转成向量，算相似度。

向量检索的优势很明显，它能理解语义，同义句也能召回。但它在生产环境中也有三个明显短板：

1. 结果不稳定。Embedding 模型对不同文本的区分度不一样，稍微改一下问题措辞，TopK 结果可能就变了。
2. 不可解释。系统只知道“向量相似”，但说不清到底是哪个关键词、哪条规则触发了这次召回。
3. 难以精确控制。当某些内容必须被准确命中时（比如“2024 年 12 月 31 日的订单号”），向量相似度并不能保证精确匹配。

deterministic-first正好是对这三点的回应：先用确定性规则锁定候选集，再在候选集内做语义排序。这样可以保证关键业务规则永远优先于模糊语义匹配。

1.3 CueMap 的核心思路

CueMap这个名字看起来是Cue+Map。Cue 是“线索”，Map 是“映射”。它的核心思路可以理解为：

先建立从“检索线索”到“记忆内容”的映射关系，检索时优先根据明确的线索去定位，而不是一开始就全量计算相似度。

线索可以是：

• 用户查询中的强关键词
• 实体名称
• 业务规则（比如时间范围、状态过滤）
• 历史轮次中的关键信息
• 用户 ID、会话 ID 等上下文标识

当你把“如何找”变成“先确定线索，再展开检索”之后，系统就从“大海捞针”变成了“按图索骥”。

2. 环境准备与示例项目结构

2.1 环境说明

虽然 CueMap 作为一个具体项目可能有自己的技术栈，但本文要演示的是它的核心思想，所以选择一套大多数人电脑上都具备的技术组合：

• 操作系统：Windows / macOS / Linux 均可
• Python 版本：建议 3.9 及以上（使用到类型注解和标准库特性）
• 数据库：SQLite（Python 内置，无需安装）
• Web 框架：FastAPI（可选，用于把检索服务暴露成 HTTP 接口）
• 分词：jieba（用于中文关键词抽取，非必须）

版本不需要完全一致，重点是理解实现思路。

执行以下命令安装依赖：

pip install fastapi uvicorn jieba

如果只是跑核心检索逻辑，不启动 HTTP 服务，可以不装 FastAPI，下面大部分代码只依赖 Python 标准库。

2.2 项目结构

我们用一个清晰的结构来组织代码：

cuemap-demo/ ├── main.py # FastAPI 入口 ├── memory_store.py # 记忆存储与索引模块 ├── retriever.py # 确定性优先检索模块 ├── recall_state.py # 连续召回状态管理 ├── data/ │ └── cuemap.db # SQLite 数据库文件（运行后生成） └── README.md

本文会按模块讲解，每个模块都能独立测试。

2.3 数据初始化

运行任意模块前，先创建数据库目录：

mkdir -p cuemap-demo/data cd cuemap-demo

后续所有代码统一放在cuemap-demo目录下。

3. 核心原理拆解

3.1 Deterministic-first 的含义

deterministic-first直译是“确定性优先”。它强调在检索流程的最前面，先用可预测、可复现的规则缩小范围，再进入语义匹配阶段。

一个典型的检索管道如下：

查询语句 ↓ 1. 提取确定性线索（关键词、实体、规则） ↓ 2. 根据线索在索引中过滤候选集 ↓ 3. 如果候选集太小，扩充线索（同义词、历史上下文） ↓ 4. 在候选集上计算语义相似度 / 关键重叠度 ↓ 5. 返回排序后的记忆片段

这样做的好处是：

• 结果稳定：同样的查询在同样的数据上永远返回同样的候选集。
• 性能可控：语义相似度不需要在全库上计算，候选集通常很小。
• 可解释：检索结果可以带上“命中线索”，方便后续排查。

3.2 记忆检索的数据模型

记忆检索系统要存储的不仅仅是文本，还包括检索所需的结构化信息。一个记忆主体（Memory）建议包含这些字段：

除了主表，还需要一张 Cue 索引表，用来建立“线索 -> 记忆”的多对多映射。

3.3 Cue 的设计思路

这里所说的 Cue 就是检索线索，它比普通关键词更结构化。举个例子：

• 用户问：“上个月购买过 100 元以上的用户有哪些？”
• 可以拆出的 Cue：
  • 时间线索：month = 上个月
  • 金额线索：amount > 100
  • 实体线索：用户
  • 行为线索：购买

在 SQLite 中，我们可以把线索存储为cue_type + cue_value的组合。比如：

time:2024-05 amount:>100 entity:用户 behavior:购买

这样查询时就能用精确的 SQLWHERE条件过滤，这就是确定性检索。

4. 完整实战：实现一个 deterministic-first 记忆检索服务

下面我们直接写一个简化版 CueMap，它能做三件事：

1. 向记忆库中添加笔记。
2. 根据查询内容，先做确定性线索匹配，再对候选集排序。
3. 在多轮查询中维持一个上下文状态，实现 continuous recall。

4.1 记忆存储模块

文件路径：cuemap-demo/memory_store.py

import sqlite3 import uuid import json from datetime import datetime from typing import List, Dict, Optional DB_PATH = "data/cuemap.db" def get_connection() -> sqlite3.Connection: """获取数据库连接，自动创建目录和表结构""" import os os.makedirs("data", exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db() -> None: """初始化数据库表""" conn = get_connection() cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS memory ( memory_id TEXT PRIMARY KEY, content TEXT NOT NULL, tags TEXT DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, source TEXT DEFAULT 'note', metadata TEXT DEFAULT '{}' ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS cue_index ( memory_id TEXT NOT NULL, cue_type TEXT NOT NULL, cue_value TEXT NOT NULL, PRIMARY KEY (memory_id, cue_type, cue_value) ) """) conn.commit() conn.close() def add_memory(content: str, tags: List[str] = None, source: str = "note", metadata: Dict = None) -> str: """添加一条记忆，并建立 Cue 索引""" conn = get_connection() cur = conn.cursor() memory_id = str(uuid.uuid4()) now = datetime.now().isoformat() # 自动切分出的 Cue if tags is None: tags = [] # 默认 metadata if metadata is None: metadata = {} cur.execute( """ INSERT INTO memory (memory_id, content, tags, created_at, updated_at, source, metadata) VALUES (?, ?, ?, ?, ?, ?, ?) """, (memory_id, content, ",".join(tags), now, now, source, json.dumps(metadata, ensure_ascii=False)) ) # 为每个标签建立 Cue 索引 for tag in tags: cur.execute( "INSERT OR IGNORE INTO cue_index (memory_id, cue_type, cue_value) VALUES (?, 'tag', ?)", (memory_id, tag) ) # 为内容中的关键词建立 Cue 索引 keywords = extract_keywords(content) for kw in keywords: cur.execute( "INSERT OR IGNORE INTO cue_index (memory_id, cue_type, cue_value) VALUES (?, 'keyword', ?)", (memory_id, kw) ) conn.commit() conn.close() return memory_id def extract_keywords(text: str) -> List[str]: """提取文本中的关键词，这里做极简处理，实际可使用 jieba""" try: import jieba words = [w.strip() for w in jieba.cut(text) if len(w.strip()) >= 2] # 去重并限制数量 seen = set() result = [] for w in words: if w not in seen: seen.add(w) result.append(w) if len(result) >= 10: break return result except ImportError: # 不使用 jieba 时按简单规则分割 import re words = re.findall(r"[\w\u4e00-\u9fff]+", text) return list(set([w for w in words if len(w) >= 2]))[:10]

这段代码的核心是add_memory函数。它会在插入记忆正文的同时，把标签和关键词写入cue_index表，为后续确定性检索建立索引。

4.2 确定性优先检索模块

文件路径：cuemap-demo/retriever.py

import sqlite3 import json from typing import List, Dict from memory_store import get_connection def deterministic_search(query: str, limit: int = 5) -> List[Dict]: """ 确定性优先检索： 1. 从 query 中抽取确定性线索 2. 先用线索过滤候选集 3. 候选集不足时放宽条件 4. 在候选集内做简单排序 """ conn = get_connection() cur = conn.cursor() # 第一步：抽取线索 cues = extract_cues(query) if not cues: # 没有线索时退化为最近记录查询 rows = cur.execute( "SELECT * FROM memory ORDER BY updated_at DESC LIMIT ?", (limit,) ).fetchall() conn.close() return [dict(row) for row in rows] # 第二步：基于线索在 cue_index 中查询候选 memory_id candidate_ids = set() for cue_type, cue_value in cues: rows = cur.execute( """ SELECT memory_id FROM cue_index WHERE cue_type = ? AND cue_value = ? """, (cue_type, cue_value) ).fetchall() for row in rows: candidate_ids.add(row["memory_id"]) if not candidate_ids: # 第三a步：没有命中任何线索，采用 keyword 模糊匹配 like_pattern = f"%{query}%" rows = cur.execute( "SELECT * FROM memory WHERE content LIKE ? ORDER BY updated_at DESC LIMIT ?", (like_pattern, limit) ).fetchall() conn.close() return [dict(row) for row in rows] # 第三b步：查询候选记忆的完整内容 placeholders = ",".join(["?"] * len(candidate_ids)) rows = cur.execute( f"SELECT * FROM memory WHERE memory_id IN ({placeholders})", list(candidate_ids) ).fetchall() # 第四步：在候选集内排序，优先返回包含查询词数量多的记忆 scored = [] query_words = set(extract_keywords_simple(query)) for row in rows: content = row["content"] score = 0 for word in query_words: if word in content: score += 1 if row["tags"]: for tag in row["tags"].split(","): if tag in query: score += 2 scored.append((score, dict(row))) scored.sort(key=lambda x: x[0], reverse=True) conn.close() return [row for _, row in scored[:limit]] def extract_cues(query: str) -> List[tuple]: """ 从查询文本中提取确定性线索。 这里我们使用几个简单规则： - 标签：# 开头的内容当作标签线索 - 时间：包含“上一周”“上个月”“2024年”等时间词时，生成 time 线索 - 实体：包含“订单”“用户”“笔记”等词时，生成 entity 线索 """ cues = [] # 标签线索 if "#" in query: for part in query.split("#")[1:]: tag = part.strip().split()[0] if tag: cues.append(("tag", tag)) # 时间线索 time_words = ["上个月", "上一周", "昨天", "2024", "2025"] for tw in time_words: if tw in query: cues.append(("time", tw)) break # 实体线索 entity_words = ["订单", "用户", "笔记", "对话"] for ew in entity_words: if ew in query: cues.append(("entity", ew)) break if cues: return cues # 如果没有显式规则命中，退化为关键词线索 keywords = extract_keywords_simple(query) for kw in keywords[:3]: cues.append(("keyword", kw)) return cues def extract_keywords_simple(text: str) -> List[str]: """简单关键词抽取，不依赖外部库""" import re words = re.findall(r"[\w\u4e00-\u9fff]+", text) # 过滤纯数字和过短词 return [w for w in words if len(w) >= 2] def inspect_cues(query: str) -> Dict: """返回查询解析出的线索，方便调试""" return { "query": query, "cues": extract_cues(query) }

这里最关键的函数是deterministic_search。它先把用户问题转换成一组(cue_type, cue_value)，然后去cue_index表里精确定位候选记忆。整个过程没有用到向量计算，因此结果完全可复现。

4.3 连续召回状态管理

文件路径：cuemap-demo/recall_state.py

from typing import List, Dict from dataclasses import dataclass, field from retriever import deterministic_search, extract_cues @dataclass class RecallState: """管理多轮检索上下文""" session_id: str history: List[Dict] = field(default_factory=list) last_cues: List[tuple] = field(default_factory=list) last_memory_ids: List[str] = field(default_factory=list) max_history: int = 5 def update(self, query: str, results: List[Dict]) -> None: """更新检索状态""" cues = extract_cues(query) self.history.append({ "query": query, "cues": cues, "result_ids": [r["memory_id"] for r in results] }) # 只保留最近 max_history 条 if len(self.history) > self.max_history: self.history = self.history[-self.max_history:] # 缓存最近一条线索和命中的记忆 ID if cues: self.last_cues = cues if results: self.last_memory_ids = [r["memory_id"] for r in results] def build_context_query(self, query: str) -> str: """ 如果当前查询过短或过于模糊，结合上一轮的线索生成新的查询。 这是 continuous recall 的简化实现。 """ if len(query) >= 8: return query context_parts = [query] # 上一轮检索到的记忆正文前 50 字 if self.last_memory_ids: # 这里不实际查库，只做一个方法占位 context_parts.append("[上一轮线索]") for cue_type, cue_value in self.last_cues: context_parts.append(cue_value) return " ".join(context_parts)

RecallState类的核心是维护一个“最近几轮”的检索历史。当新查询太短或太模糊时，build_context_query会把上一轮的线索拼接到查询里，让检索器不至于失去上下文。

4.4 完整检索接口

文件路径：cuemap-demo/main.py

from fastapi import FastAPI from pydantic import BaseModel from typing import Optional from memory_store import init_db, add_memory from retriever import deterministic_search, inspect_cues from recall_state import RecallState app = FastAPI(title="CueMap Demo", description="deterministic-first memory retrieval demo") # 用字典保存不同会话的召回状态 recall_states = {} # 启动时初始化数据库 init_db() class MemoryCreate(BaseModel): content: str tags: list[str] = [] source: str = "note" metadata: dict = {} class QueryRequest(BaseModel): session_id: str = "default" query: str @app.post("/memory") def create_memory(req: MemoryCreate): """添加记忆""" memory_id = add_memory( content=req.content, tags=req.tags, source=req.source, metadata=req.metadata ) return {"memory_id": memory_id} @app.post("/retrieve") def retrieve(req: QueryRequest): """连续召回接口""" # 获取或创建会话状态 if req.session_id not in recall_states: recall_states[req.session_id] = RecallState(session_id=req.session_id) state = recall_states[req.session_id] # 构建带上下文的查询 context_query = state.build_context_query(req.query) results = deterministic_search(context_query) # 更新会话状态 state.update(req.query, results) return { "context_query": context_query, "results": results, "cues": inspect_cues(context_query), "session_id": req.session_id } @app.get("/inspect-cues") def inspect(query: str): """调试接口：查看某个查询解析出的线索""" return inspect_cues(query)

这里的/retrieve接口把状态管理和检索逻辑串起来了。每次查询都会先调用build_context_query，把上一轮的线索补充进来，再执行确定性优先检索。

4.5 运行与验证

先启动服务：

uvicorn main:app --reload --port 8000

然后添加几条测试记忆：

curl -X POST http://127.0.0.1:8000/memory \ -H "Content-Type: application/json" \ -d '{"content": "低糖早餐食谱：燕麦牛奶 + 水煮蛋", "tags": ["饮食", "低糖"]}' curl -X POST http://127.0.0.1:8000/memory \ -H "Content-Type: application/json" \ -d '{"content": "低糖午餐建议：鸡胸肉沙拉 + 糙米饭", "tags": ["饮食", "低糖"]}' curl -X POST http://127.0.0.1:8000/memory \ -H "Content-Type: application/json" \ -d '{"content": "2024-12-01 用户张三购买了 200 元课程", "tags": ["订单", "用户"]}'

接着测试检索：

curl -X POST http://127.0.0.1:8000/retrieve \ -H "Content-Type: application/json" \ -d '{"session_id": "test-1", "query": "低糖"}'

预期会返回两条饮食相关的记忆，并且返回结果中带有解析出的cue信息。这是第一天查询。

第二天查询时，如果用户只输入“午餐呢？”，状态管理器会判断查询过短，然后自动引入上一轮的线索“低糖”，最终仍然能召回午餐建议。

4.6 结果说明

从运行结果可以看到，确定性优先检索的优点：

1. 没有安装任何向量模型，系统的检索完全可复现。
2. 输入“低糖”能立刻基于cue_index命中饮食标签。
3. 输入“订单”能命中带“订单”标签的记忆。
4. 连续通话场景中，短查询借助上下文状态也能命中上一轮相关的记忆。

当然，这个实现是简化版。真实 CueMap 项目如果要做生产级系统，还需要在以下方面增强：

• 支持更多线索类型，比如时间范围、数值范围。
• 引入更完善的关键词抽取和实体识别。
• 在候选集内集成轻量级向量排序模型。
• 增加缓存和性能监控。

5. 常见问题与排查思路

在实现或使用这类“确定性优先记忆检索”系统时，通常会遇到下面几个问题。

5.1 检索结果为空

建议先调用/inspect-cues?query=你的问题，看看系统“看到”的线索是什么。如果线索为空，说明是线索抽取环节的问题。

5.2 连续召回没有生效

如果你发现第二轮查询没有带上第一轮的信息，优先检查recall_state.py中的build_context_query逻辑。

可能原因：

• 用户传了不同的session_id，导致状态不被复用。解决方案是确保前端会话 ID 一致。
• 第一轮检索结果为空，last_cues没有被更新。
• 查询长度大于阈值，没有触发上下文补充逻辑。

排查建议：

• 在/retrieve接口返回中打印context_query。
• 如果context_query等于原始query，说明没有拼接任何上下文线索。
• 检查RecallState.history是否积累了多轮数据。

5.3 中文分词不准

memory_store.py中的关键词抽取优先使用jieba，但jieba对专业领域词汇可能切得不理想。

解决思路：

• 把专业词汇添加到用户词典中：jieba.add_word("低糖饮食")
• 或者不强依赖分词，直接使用标签作为 Cue。
• 生产环境可以换成更专业的实体识别模型。

5.4 候选集仍然很大

当数据量达到百万级时，虽然cue_index能过滤掉大量无关内容，但某些高频标签（比如“用户”）可能导致候选集仍然很大。

解决思路：

• 使用组合线索过滤，多个线索同时命中才进入候选集。
• 给cue_index添加命中计数字段，高频线索在排序时降权。
• 使用更细粒度的 Cue 类型，比如“用户ID”而不是泛泛的“用户”。

5.5 SQLite 并发写入失败

SQLite 适合中小规模场景，但高并发写入时可能出现database is locked。

解决方案：

• 设置连接超时：sqlite3.connect(DB_PATH, timeout=5)
• 开启 WAL 模式：PRAGMA journal_mode=WAL
• 生产规模超过单机写入能力后，迁移到 PostgreSQL 或专业向量库。

6. 最佳实践与工程建议

6.1 让检索结果可解释

确定性优先的一个好处是“可解释”。在生产系统中，我强烈建议在检索结果里附带上命中的线索类型和线索值。比如：

{ "memory_id": "xxx", "content": "低糖午餐建议：鸡胸肉沙拉 + 糙米饭", "hit_cues": ["tag:饮食", "tag:低糖"] }

这样当业务方质疑为什么返回这条结果时，可以直接把hit_cues拿出来看。

6.2 合理设计 Cue 索引

Cue 不是越多越好。过多会导致索引膨胀，过少则起不到过滤作用。

我的建议是：

• 标签型 Cue：必须严格规范化，建立标签表统一管理。
• 关键词型 Cue：只保留长度 ≥ 2 且出现频率适中的词，去掉停用词。
• 实体型 Cue：通过实体识别生成，线上可以定期抽取。
• 数值范围型 Cue：不要存成普通字符串，单独建范围索引列。

6.3 连续召回要考虑隐私与安全

多轮检索意味着系统保存了用户的历史查询和浏览记录。在设计时，应该做到：

• 默认不记录完整对话，只保留必要的检索状态。
• 历史状态设置过期时间，比如 30 分钟后自动清理。
• 涉及用户 ID 的 Cue 需要做脱敏处理。
• 用户有权利清除会话状态。

在代码层面，可以给RecallState增加一个clear方法，并提供接口让业务方主动重置上下文。

6.4 清理与更新策略

记忆库不是一成不变的。当用户修改了笔记或删除了工单，对应记忆需要更新或移除。

建议在memory_table上维护updated_at，在cue_index表上建立级联删除外键。更新时先删除旧 Cue，再插入新 Cue。生产环境可以写一个后台任务，定期重算关键词索引。

6.5 性能优化方向

如果数据量增长很快，可以按下面方向优化：

1. 把cue_index从 SQLite 迁移到 Redis 或者列存储。
2. 增加 Redis 缓存，缓存高频查询的检索结果。
3. 将候选集内的排序从 Python 循环改为向量计算，比如用numpy计算稀疏向量点积。
4. 检索接口做异步化，避免慢查询阻塞主流程。

但对大多数知识库和个人助手项目来说，SQLite 加合理的 Cue 索引已经足够支撑几十万到几百万条记忆。

7. 总结与下一步学习方向

本文从 CueMap 项目的标题出发，拆解了 deterministic-first 和 continuous recall 这两个核心概念。我带着你从零实现了一个简化版记忆检索系统，它包含记忆存储、Cue 索引、确定性检索和多轮上下文状态管理。

关键点回顾：

• deterministic-first的核心是“先精确过滤，再语义排序”。
• continuous recall的核心是“记住上一轮线索，服务这一轮查询”。
• Cue 索引表是连接查询与记忆的桥梁。
• 多轮状态管理必须考虑过期和隐私问题。
• 可解释性是确定性检索相比纯向量检索的最大优势。

如果你要继续深入，可以按以下路线学习：

1. 学习一门向量数据库的实战用法，比如 SQLite-VSS、Chroma 或 Milvus，理解在候选集内做向量排序的过程。
2. 研究 RAG 应用中的混合检索方案，看业界如何把“BM25 + 向量检索 + 规则过滤”组合在一起。
3. 关于记忆机制的设计，比如 Memory Bank、Agent Memory 之类的论文。
4. 尝试把本文的检索代码接入到大模型问答逻辑里，实现一个带长期记忆的 AI 助手。

最后有一个小建议：无论用什么技术方案，先明确你的业务里哪些检索是必须精确命中的，哪些允许语义相似。只要能把这个边界画清楚，deterministic-first 的设计思路在任何项目里都用得上。