确定性优先记忆检索：用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 模型转成向量 - 存到向量数据库 - 查询时把问题转成向量算相似度。向量检索的优势很明显它能理解语义同义句也能召回。但它在生产环境中也有三个明显短板结果不稳定。Embedding 模型对不同文本的区分度不一样稍微改一下问题措辞TopK 结果可能就变了。不可解释。系统只知道“向量相似”但说不清到底是哪个关键词、哪条规则触发了这次召回。难以精确控制。当某些内容必须被准确命中时比如“2024 年 12 月 31 日的订单号”向量相似度并不能保证精确匹配。deterministic-first正好是对这三点的回应先用确定性规则锁定候选集再在候选集内做语义排序。这样可以保证关键业务规则永远优先于模糊语义匹配。1.3 CueMap 的核心思路CueMap这个名字看起来是CueMap。Cue 是“线索”Map 是“映射”。它的核心思路可以理解为先建立从“检索线索”到“记忆内容”的映射关系检索时优先根据明确的线索去定位而不是一开始就全量计算相似度。线索可以是用户查询中的强关键词实体名称业务规则比如时间范围、状态过滤历史轮次中的关键信息用户 ID、会话 ID 等上下文标识当你把“如何找”变成“先确定线索再展开检索”之后系统就从“大海捞针”变成了“按图索骥”。2. 环境准备与示例项目结构2.1 环境说明虽然 CueMap 作为一个具体项目可能有自己的技术栈但本文要演示的是它的核心思想所以选择一套大多数人电脑上都具备的技术组合操作系统Windows / macOS / Linux 均可Python 版本建议 3.9 及以上使用到类型注解和标准库特性数据库SQLitePython 内置无需安装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建议包含这些字段字段名类型说明memory_idstring记忆唯一标识contenttext记忆内容tagsstring标签多个用逗号分隔created_atdatetime创建时间updated_atdatetime更新时间sourcestring来源如笔记、对话、工单metadatatext额外上下文JSON 格式除了主表还需要一张 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它能做三件事向记忆库中添加笔记。根据查询内容先做确定性线索匹配再对候选集排序。在多轮查询中维持一个上下文状态实现 continuous recall。4.1 记忆存储模块文件路径cuemap-demo/memory_store.pyimport 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_okTrue) 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_asciiFalse)) ) # 为每个标签建立 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.pyimport 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( fSELECT * 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(keylambda x: x[0], reverseTrue) 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.pyfrom 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_factorylist) last_cues: List[tuple] field(default_factorylist) last_memory_ids: List[str] field(default_factorylist) 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.pyfrom 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(titleCueMap Demo, descriptiondeterministic-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( contentreq.content, tagsreq.tags, sourcereq.source, metadatareq.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_idreq.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 结果说明从运行结果可以看到确定性优先检索的优点没有安装任何向量模型系统的检索完全可复现。输入“低糖”能立刻基于cue_index命中饮食标签。输入“订单”能命中带“订单”标签的记忆。连续通话场景中短查询借助上下文状态也能命中上一轮相关的记忆。当然这个实现是简化版。真实 CueMap 项目如果要做生产级系统还需要在以下方面增强支持更多线索类型比如时间范围、数值范围。引入更完善的关键词抽取和实体识别。在候选集内集成轻量级向量排序模型。增加缓存和性能监控。5. 常见问题与排查思路在实现或使用这类“确定性优先记忆检索”系统时通常会遇到下面几个问题。5.1 检索结果为空问题现象常见原因解决思路检索结果为空线索提取失败cue_index中没有对应记录用inspect-cues接口查看解析出的线索检索结果为空记忆添加时没有成功生成关键词索引检查add_memory中extract_keywords是否返回了空列表检索结果为空查询词过于生僻放宽候选集加入模糊匹配或语义排序兜底建议先调用/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, timeout5)开启 WAL 模式PRAGMA journal_modeWAL生产规模超过单机写入能力后迁移到 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 性能优化方向如果数据量增长很快可以按下面方向优化把cue_index从 SQLite 迁移到 Redis 或者列存储。增加 Redis 缓存缓存高频查询的检索结果。将候选集内的排序从 Python 循环改为向量计算比如用numpy计算稀疏向量点积。检索接口做异步化避免慢查询阻塞主流程。但对大多数知识库和个人助手项目来说SQLite 加合理的 Cue 索引已经足够支撑几十万到几百万条记忆。7. 总结与下一步学习方向本文从 CueMap 项目的标题出发拆解了 deterministic-first 和 continuous recall 这两个核心概念。我带着你从零实现了一个简化版记忆检索系统它包含记忆存储、Cue 索引、确定性检索和多轮上下文状态管理。关键点回顾deterministic-first的核心是“先精确过滤再语义排序”。continuous recall的核心是“记住上一轮线索服务这一轮查询”。Cue 索引表是连接查询与记忆的桥梁。多轮状态管理必须考虑过期和隐私问题。可解释性是确定性检索相比纯向量检索的最大优势。如果你要继续深入可以按以下路线学习学习一门向量数据库的实战用法比如 SQLite-VSS、Chroma 或 Milvus理解在候选集内做向量排序的过程。研究 RAG 应用中的混合检索方案看业界如何把“BM25 向量检索 规则过滤”组合在一起。关于记忆机制的设计比如 Memory Bank、Agent Memory 之类的论文。尝试把本文的检索代码接入到大模型问答逻辑里实现一个带长期记忆的 AI 助手。最后有一个小建议无论用什么技术方案先明确你的业务里哪些检索是必须精确命中的哪些允许语义相似。只要能把这个边界画清楚deterministic-first 的设计思路在任何项目里都用得上。