去重与合并

robotmem 通过写入时的多层去重管道防止冗余记忆，并在会话结束时合并相似的记忆。

去重管道

当调用 learn 时，新记忆在存储前会经过三层重复检测。

概览

learn(insight="grip_force=12.5N works best")
         │
         ├── 第 1 层：精确匹配
         │   SHA-256(content)[:16] → content_hash 查找
         │   O(1) — 即时拒绝
         │
         ├── 第 2 层：Jaccard 词元相似度
         │   FTS5 候选项 → 两两词元重叠度
         │   阈值：> 0.70 → 重复
         │
         └── 第 3 层：余弦向量相似度
         │   embed_one(content) → vec_search top-3
         │   阈值：> 0.85 → 重复
         │
         ▼
    is_dup=True → 返回 {status: "duplicate", method, existing_id, similarity}
    is_dup=False → 继续插入

第 1 层：精确匹配

最快的检查 — O(1) 哈希查找：

content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]

# 在数据库中检查
existing = conn.execute(
    "SELECT 1 FROM memories WHERE content_hash = ? AND collection = ? "
    "AND status = 'active' LIMIT 1",
    (content_hash, collection),
).fetchone()

• 使用 SHA-256 哈希的前 16 个字符
• 限定在相同 collection + active 状态
• 捕获精确重复（逐字符相同）
• 通过 idx_mem_hash 索引实现快速查找

第 2 层：Jaccard 词元相似度

用于检测措辞略有不同的近似重复：

def jaccard_similarity(a: str, b: str) -> float:
    tokens_a = set(a.lower().split()) - STOPWORDS
    tokens_b = set(b.lower().split()) - STOPWORDS
    intersection = tokens_a & tokens_b
    union = tokens_a | tokens_b
    return len(intersection) / len(union)

流程： 1. 使用 FTS5 找到与新内容匹配的 top-5 候选项 2. 计算每个候选项的 Jaccard 相似度 3. 阈值：> 0.70 → 标记为重复

停用词同时支持英文和中文过滤：

STOPWORDS = frozenset(
    "的 了 在 是 把 被 给 和 与 从 到 也 都 就 对 又 所 而 且 但 或 "
    "a an the is are was were be to of and in for on with "
    "preference constraint decision observation code config pattern "
    "architecture root_cause tradeoff revert".split()
)

分类相关术语也被包含在停用词中，以防止仅因共享分类关键词而产生的误匹配。

第 3 层：余弦向量相似度

用于检测使用不同词汇的语义重复：

# 仅在 embedder 可用时运行
embedding = await embedder.embed_one(assertion)
vec_results = db_cog.vec_search_memories(
    query_embedding=embedding, collection=collection, limit=3
)
for vr in vec_results:
    cosine_sim = 1.0 - vr["distance"]
    if cosine_sim >= 0.85:
        return DedupResult(is_dup=True, method="cosine", similarity=cosine_sim)

• 仅在嵌入后端可用时运行
• 优雅降级 — 如果嵌入失败，第 2 层仍能捕获大部分重复
• 无法在已运行的事件循环内执行（记录调试日志并跳过）

相似（非重复）追踪

中等相似度（Jaccard 0.40-0.70 或余弦 0.40-0.85）的记忆会作为 similar_facts 记录在结果中：

{
    "is_dup": False,
    "similar_facts": [
        {"id": 15, "assertion": "similar content...", "similarity": 0.55}
    ],
    "method": "none",
    "similarity": 0.55
}

这些信息返回给调用方供参考，但不会阻止插入。

会话内余弦去重

额外的去重层在当前会话范围内运作：

def check_session_cosine_dup(assertion, session_id, collection, db_cog, embedder):
    """在同一会话内，检查语义重复"""
    embedding = embedder.embed_one(assertion)
    results = db_cog.vec_search_memories(embedding, collection, limit=20)
    for r in results:
        if r["session_id"] != session_id:
            continue  # 仅检查同一会话
        cosine_sim = 1.0 - r["distance"]
        if cosine_sim >= 0.85:
            return DedupResult(is_dup=True, method="session_cosine")

这能捕获机器人在单个会话中多次记录相同观察的情况。

去重阈值

批量清理

对于已积累重复数据的现有数据库，cleanup_exact_duplicates() 提供了批量清理工具：

ops = cleanup_exact_duplicates(db_cog, collection="default", dry_run=True)
# 返回：[{"old_id": 5, "keep_id": 3, "assertion_preview": "..."}]

# 执行清理
ops = cleanup_exact_duplicates(db_cog, collection="default", dry_run=False)

安全特性： - 试运行模式：在应用前预览更改 - 每次最多 200 个操作：防止失控清理 - 保留最高置信度：在重复项中，保留置信度最高（然后最新）的记忆 - 软删除：重复项标记为 superseded，而非物理删除

合并

在 end_session 时，robotmem 通过贪心 Jaccard 聚类合并会话中的冗余记忆。

算法

end_session(session_id="abc-123")
         │
    1. 查询可合并的记忆
       ├── 相同会话 + collection
       ├── status = 'active'
       ├── category 不在 (constraint, postmortem, gotcha) 中
       ├── confidence < 0.95
       └── perception_type 为 NULL
         │
    2. 如果少于 3 条记忆则跳过
         │
    3. 按 category 分组
         │
    4. 在每组内：两两计算 Jaccard 相似度
       ├── > 0.50 阈值 → 候选对
       └── 贪心聚类（簇内所有对必须超过阈值）
         │
    5. 每个簇：选择代表
       ├── 优先级：confidence DESC
       ├── 次优先：access_count DESC
       └── 次优先：created_at DESC
         │
    6. 非代表 → status = 'superseded'
       └── superseded_by = representative.id

受保护的记忆

以下记忆永远不会被合并，即使相似：

贪心聚类

聚类算法确保高质量的簇：

for i, a in enumerate(mems):
    cluster = [a]
    for j in range(i+1, len(mems)):
        b = mems[j]
        # 全对约束：b 必须与簇内每个成员相似
        sim_ok = all(
            jaccard_similarity(b["content"], c["content"]) > 0.50
            for c in cluster
        )
        if sim_ok:
            cluster.append(b)

这比单链接聚类更严格 — 簇内的每一对都必须超过阈值，防止"链式漂移"：即 A→B→C 有链接但 A 和 C 实际上不相似。

代表选择

在每个簇内，代表的选择依据：

1. 最高置信度 — 最可靠的记忆存活
2. 最高访问次数 — 最常被召回的 = 最有用的
3. 最新的（created_at DESC）— 最新的信息

合并响应

{
    "merged_groups": 2,
    "superseded_count": 3,
    "compression_ratio": 0.20,
    "avg_similarity": 0.65
}

示例

给定会话中的 5 条记忆：

#1: "grip_force=12.5N works for cups"        (category=observation, confidence=0.85)
#2: "12.5N grip force optimal for cylinders" (category=observation, confidence=0.80)
#3: "force 12.5N best for cup grasping"      (category=observation, confidence=0.90)
#4: "red objects need 15N force"             (category=observation, confidence=0.85)
#5: "always calibrate before grasping"       (category=constraint, confidence=0.95)

结果： - #5 受保护（constraint 分类 + 高置信度） - #1、#2、#3 聚为一簇（两两 Jaccard > 0.50） - #3 被选为代表（最高置信度：0.90） - #1、#2 → superseded，superseded_by = 3 - #4 保持独立（相似度不足以加入簇）

时间衰减

同样由 end_session 触发，时间衰减会降低近期未被访问的记忆的置信度：

公式

confidence_new = confidence × (1 - decay_rate) ^ days_since_last_access

参数

基准时间

衰减参考点是 last_accessed（由 recall 命中时更新），回退到 created_at：

julianday('now') - julianday(COALESCE(last_accessed, created_at))

这意味着： - 经常被召回的记忆保持高置信度（每次召回重置时钟） - 未使用的记忆逐渐淡化 - 从未被召回的记忆从创建日期开始衰减

衰减曲线

对于 confidence=0.90 和 decay_rate=0.01 的记忆：

在默认的 min_confidence=0.3 召回过滤器下，该记忆在约 93 天未被召回后将不再出现在搜索结果中。

数据库操作

所有合并和去重操作使用安全的数据库原语：

失败的事务会被记录到日志，但永远不会导致服务器崩溃。