搜索架构

robotmem 的 recall 工具使用混合搜索管道，将 BM25 全文搜索与向量相似度搜索通过 Reciprocal Rank Fusion (RRF) 进行融合。

管道概览

recall(query="how to grasp a cup", n=5, min_confidence=0.3)
         │
         ├── L1: 输入验证
         │   query 非空，top_k 限制在 [1, 100]
         │   fetch_mul = 有过滤器时 4x，否则 2x
         │
    ┌────┴────────────────────────────┐
    │                                 │
    ▼                                 ▼
 BM25 搜索                      向量搜索
 (FTS5 引擎)                    (vec0 引擎)
    │                                 │
    ├── tokenize_for_fts5()          ├── embedder.embed_one(query)
    │   CJK → jieba 分词             │   → float[384] 或 float[768]
    │   移除 FTS5 运算符              │
    │   过滤单字符非 CJK             ├── vec_search_memories()
    │                                │   WHERE embedding MATCH blob
    ├── fts_search_memories()        │   AND k = fetch_limit
    │   JOIN memories ON rowid       │   ORDER BY 余弦距离
    │   WHERE collection + active    │
    │   ORDER BY bm25() 分数         │
    │                                │
    ▼                                ▼
 排序列表 A                    排序列表 B
    │                                │
    └────────────┬───────────────────┘
                 │
          ┌──────▼───────┐
          │  RRF 融合    │
          │  k=60        │
          └──────┬───────┘
                 │
          ┌──────▼───────────────┐
          │  来源加权             │
          │  真实数据 × 1.5      │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  置信度过滤           │
          │  >= min_confidence   │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  会话过滤             │
          │  session_id 匹配     │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  上下文提取           │
          │  params/spatial/     │
          │  robot/task          │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  context_filter      │
          │  （结构化过滤）       │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  spatial_sort        │
          │  （最近邻排序）       │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  Top-K 截断          │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  MaxScore 归一化     │
          │  最佳 = 1.0          │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  Touch (L3)          │
          │  access_count++      │
          └──────────────────────┘

BM25 搜索 (FTS5)

BM25 是内置于 SQLite FTS5 扩展中的词频排序算法。robotmem 用它来进行关键词级别的匹配。

分词

在查询 FTS5 之前，查询文本会经过 tokenize_for_fts5() 处理：

1. CJK 检测：如果文本包含 CJK 字符（\u4e00-\u9fff），则应用 jieba 分词
2. 语法清理：移除 FTS5 运算符（AND、OR、NOT、NEAR）和特殊字符
3. Token 过滤：移除单字符非 CJK token（噪声太大）
4. 查询构造：剩余 token 用 OR 连接并加引号："token1" OR "token2"

# 示例: "how to grasp a cup"
# → tokens: ["how", "to", "grasp", "cup"]
# → FTS5: '"how" OR "to" OR "grasp" OR "cup"'

# 示例: "如何抓取杯子"
# → jieba: ["如何", "抓取", "杯子"]
# → FTS5: '"如何" OR "抓取" OR "杯子"'

FTS5 查询

SELECT m.id, m.content, ..., bm25(memories_fts) as bm25_score
FROM memories_fts
JOIN memories m ON m.id = memories_fts.rowid
WHERE memories_fts MATCH ?
  AND m.collection = ?
  AND m.status = 'active'
ORDER BY bm25(memories_fts)
LIMIT ?

bm25() 函数返回负分数（越低越相关），结果按此分数排序。

FTS5 索引字段

向量搜索 (vec0)

向量搜索使用 sqlite-vec 的 vec0 虚拟表，基于嵌入相似度进行近似最近邻 (ANN) 搜索。

嵌入流程

查询文本
    │
    ▼
embedder.embed_one(query)
    │
    ├── ONNX: FastEmbed 本地推理（~5ms）
    │   → float[384] (BAAI/bge-small-en-v1.5)
    │
    └── Ollama: HTTP API 调用（~20-50ms）
        → float[768] (nomic-embed-text)
    │
    ▼
floats_to_blob()
    │
    ▼
struct.pack('f' * dim, *embedding)
    → bytes（原始 float32 数组）

vec0 查询

SELECT m.id, m.content, ..., v.distance
FROM memories_vec v
JOIN memories m ON m.id = v.rowid
WHERE v.embedding MATCH ?    -- float32 数组的 blob
  AND m.collection = ?
  AND m.status = 'active'
  AND k = ?                  -- KNN 参数

distance 字段表示余弦距离（0 = 完全相同，2 = 完全相反）。

RRF 融合

Reciprocal Rank Fusion 将多个排序列表合并为一个统一排序，无需分数归一化。

公式

score(document) = Σ 1 / (k + rank_i + 1)

其中： - k = 60（可通过配置中的 rrf_k 设置） - rank_i = 每个列表中从 0 开始的位置 - 对包含该文档的所有列表求和

示例

doc_B 排名最高，因为它在两个列表中都有较好的位置。

为什么选择 RRF？

• 无需分数归一化：BM25 分数和余弦距离有不同的量纲
• 奖励一致性：被两个引擎同时找到的文档排名更高
• 鲁棒性：较高的 k 值更看重多列表同时出现而非单一排名
• 简单：除 k 外无需调整其他超参数

搜索模式

管道会根据可用结果自动选择最佳模式：

在降级模式（bm25_only）下，分配合成 RRF 分数：

score = 1.0 / (60 + rank + 1)  # 同样的公式，单个列表

来源加权

来自真实世界数据的记忆获得比仿真数据 1.5 倍的加成：

# 检测: context.env.sim_or_real == "real"
if parsed.get("env", {}).get("sim_or_real") == "real":
    m["_rrf_score"] = m["_rrf_score"] * 1.5

加权后，结果按 _rrf_score 降序重新排序。

融合后过滤

置信度过滤

if m.get("confidence", 0) < min_confidence:
    continue  # 跳过低置信度记忆

默认 min_confidence = 0.3。衰减到低于此阈值的记忆将被过滤掉。

会话过滤

if session_id and m.get("session_id") != session_id:
    continue  # 仅返回该 episode 的记忆

用于 episode 回放：recall(query="*", session_id="abc-123")。

上下文过滤（结构化）

在已解析的 context JSON 上进行点路径匹配：

# 等值匹配
{"task.success": True}

# 范围运算符
{"params.force.value": {"$lt": 15.0}}
{"params.force.value": {"$gte": 10.0, "$lte": 20.0}}

# 组合（AND 逻辑）
{"task.success": True, "robot.type": "UR5e"}

实现使用 _resolve_dotpath() 遍历嵌套字典：

# "task.success" → d["task"]["success"]
def _resolve_dotpath(d, path):
    current = d
    for key in path.split("."):
        if key not in current:
            return _MISSING
        current = current[key]
    return current

支持的运算符：

类型不匹配返回 False（不会崩溃）。

空间排序（最近邻）

基于坐标数组的欧几里得距离排序：

spatial_sort = {
    "field": "spatial.object_position",
    "target": [1.3, 0.7, 0.42],
    "max_distance": 0.1  # 可选截断距离
}

距离公式：

distance = sqrt(Σ (actual_i - target_i)²)

• 维度不匹配 → inf 距离（被过滤掉）
• 字段缺失 → inf 距离（被过滤掉）
• max_distance 截断在距离计算之后应用

MaxScore 归一化

所有过滤完成后，最高结果的 _rrf_score 被设为 1.0，其他所有分数相对于它进行归一化：

max_score = filtered[0]["_rrf_score"]
for m in filtered:
    m["_rrf_score"] = m["_rrf_score"] / max_score

这确保了最佳结果的分数始终为 1.0，使分数在不同查询之间具有可比性。

访问计数 (L3)

返回结果后，管道会更新所有返回记忆的访问统计：

batch_touch_memories(db.conn, hit_ids)
# 每条记忆: access_count++, return_count++, last_accessed = now

这有两个目的： 1. 时间衰减保护：被频繁召回的记忆维持更高的置信度 2. 分析：追踪哪些记忆最有用

获取倍数

当结构化过滤器或空间排序激活时，管道会获取更多候选记忆以补偿过滤损失：

这确保了有足够的候选记忆通过过滤来填充请求的 top_k 结果。

三层防御

搜索管道遵循 robotmem 一致的防御模式：

每个搜索引擎的独立故障不会导致另一个崩溃——管道优雅降级。