scripts/dedupe_memory_fact_keys.py
Rama main · réplica de lectura del repositorio original.
"""记忆库 fact_key 归并清理（一次性 / 可重复执行）.
背景
----
2026-10-05审计发现：「用户要求禁止使用 file 工具和执行命令」这一条语义相同的
偏好，在 memory.db 中并存 4 条 **active** 记录，fact_key 各不相同
（excel_desktop_automation / tool_usage / desktop_tool / tool_restriction）。
根因是模型每次抽取现编新key，导致 store.add 的冲突失效与 _dedup 的语义去重
同时失效。
本脚本把**同义 fact_key 的多条 active 记忆归并为一条**：保留最新 + 重要度最高
的那条，其余标记为 ``deprecated``（不物理删除，符合"矛盾事实不要直接删除"的
既有原则，保留可追溯）。
用法::
# 干跑（只看会改什么，不落库）—— 默认
python scripts/dedupe_memory_fact_keys.py --db "D:/.scout/memory.db"
# 确认后真正执行
python scripts/dedupe_memory_fact_keys.py --db "D:/.scout/memory.db" --apply
# 顺带把所有旧式 key 重写为受控词表规范 key
python scripts/dedupe_memory_fact_keys.py --db "D:/.scout/memory.db" --rewrite-keys --apply
脚本对 db 只做写操作前会先备份到 ``<db>.bak-dedupe-<ts>``。
"""
from __future__ import annotations
import argparse
import logging
import os
import shutil
import sys
import time
from collections import defaultdict
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from scout.context.fact_key_registry import ( # noqa: E402
MERGE_SIM_THRESHOLD,
_key_similarity,
canonicalize_fact_key,
)
from scout.context.memory_extract import _jaccard, _tokenize # noqa: E402
logger = logging.getLogger("dedupe_fact_keys")
def _connect(db_path: str):
import sqlite3
return sqlite3.connect(db_path)
def _table_columns(conn) -> set[str]:
return {r[1] for r in conn.execute("PRAGMA table_info(memories)")}
def plan(conn, threshold: float, rewrite_keys: bool):
"""计算归并计划，不改库.
Returns:
``(clusters, rewrite_pairs)``：
- clusters: 同义 key 簇，每簇是一个 dict(keeper_id, dup_ids, canonical_key)
- rewrite_pairs: (id, old_key, new_key) —— 仅 --rewrite-keys 时非空
"""
cols = _table_columns(conn)
has_fk = "fact_key" in cols
has_status = "status" in cols
has_vt = "valid_to" in cols
if not has_fk:
logger.error("memories 表无 fact_key 列，无法归并。")
return [], []
sel = "id, content, importance, fact_key, created_at"
sel += ", status" if has_status else ", 'active' AS status"
rows = conn.execute(f"SELECT {sel} FROM memories WHERE status = 'active'").fetchall()
rewrite_pairs = []
if rewrite_keys:
for row in rows:
rid, fk = row[0], row[3]
if not fk:
continue
ck = canonicalize_fact_key(fk)
if ck != fk:
rewrite_pairs.append((rid, fk, ck))
# 按 fact_key 分组（非空）
by_key: dict[str, list[tuple]] = defaultdict(list)
for r in rows:
if r[3] and r[3].strip():
by_key[r[3].strip()].append(r)
# 找出语义等价的 key 簇
keys = list(by_key.keys())
parent: dict[str, str] = {k: k for k in keys}
def find(x: str) -> str:
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(a: str, b: str) -> None:
ra, rb = find(a), find(b)
if ra != rb:
# 保留字典序小的作为簇代表（确定性）
if rb < ra:
ra, rb = rb, ra
parent[rb] = ra
for i in range(len(keys)):
for j in range(i + 1, len(keys)):
a, b = keys[i], keys[j]
if find(a) == find(b):
continue
if _key_similarity(canonicalize_fact_key(a), canonicalize_fact_key(b)) >= threshold:
union(a, b)
clusters = []
groups: dict[str, list[str]] = defaultdict(list)
for k in keys:
groups[find(k)].append(k)
for _root, members in groups.items():
if len(members) < 2:
continue
# 收集簇内所有记忆
entries = []
for mk in members:
entries.extend(by_key[mk])
if len(entries) < 2:
continue
# 保留：active 优先（这里都是 active）→ 最新 created_at → 最高 importance → 最小 id
entries.sort(
key=lambda r: (r[4] or "", float(r[2] or 0), -r[0]),
reverse=True,
)
keeper = entries[0]
dups = [e[0] for e in entries[1:]]
clusters.append({
"keeper_id": keeper[0],
"dup_ids": dups,
"canonical_key": canonicalize_fact_key(members[0]),
"member_keys": members,
"n": len(entries),
})
clusters.sort(key=lambda c: c["n"], reverse=True)
_ = has_vt # valid_to 列在 apply 阶段按存在性处理
return clusters, rewrite_pairs
def apply_plan(conn, clusters, rewrite_pairs) -> int:
from datetime import datetime as _dt
cols = _table_columns(conn)
has_vt = "valid_to" in cols
now = _dt.now().isoformat()
n = 0
for rid, _old, new in rewrite_pairs:
conn.execute("UPDATE memories SET fact_key = ? WHERE id = ?", (new, rid))
n += 1
for c in clusters:
dup_sql = "UPDATE memories SET status = 'deprecated'"
params: list = []
if has_vt:
dup_sql += ", valid_to = ?"
params.append(now)
dup_sql += " WHERE id IN (%s)" % ",".join("?" * len(c["dup_ids"]))
params.extend(c["dup_ids"])
conn.execute(dup_sql, params)
n += len(c["dup_ids"])
conn.commit()
return n
def report_keyless(conn, clusters, threshold: float = 0.25) -> list[dict]:
"""报告「落在同义簇语义范围内、但没有 fact_key」的疑似重复条目.
为什么只报告不自动归并
----------------------
实测（2026-10-05）：id=1618「用户要求桌面自动化操作必须在真实窗口上点击和键入，
不接受用 file 工具或 shell/COM 等方式替代」与保留项 1666「用户要求禁止使用 file
工具和执行命令」**语义明显是同一条偏好**，但字符 Jaccard 仅 **0.178**。
这正好证明**内容相似度不能用作归并依据**——字符级 Jaccard 对"换一种说法表达
同一约束"几乎无能为力（此前实测同簇条目也只到 0.53~0.68）。若据此自动合并，
就会退化成"下调 dedup_threshold"那条已被验证会误杀合法事实更新的老路。
因此这类条目只做提示，由人工/LLM 复核后再决定是否回填 fact_key。
"""
if not clusters:
return []
cols = _table_columns(conn)
has_created = "created_at" in cols
sel = "id, content, importance, fact_key" + (", created_at" if has_created else "")
rows = conn.execute(f"SELECT {sel} FROM memories WHERE status = 'active'").fetchall()
flagged = []
for c in clusters:
keeper_row = conn.execute(
"SELECT content FROM memories WHERE id = ?", (c["keeper_id"],)
).fetchone()
if not keeper_row:
continue
keeper_tokens = _tokenize(keeper_row[0])
for r in rows:
rid, content = r[0], r[1]
if rid == c["keeper_id"] or rid in c["dup_ids"]:
continue
if r[3] and r[3].strip():
continue # 只看无 key 的
sim = _jaccard(keeper_tokens, _tokenize(content))
if sim >= threshold:
flagged.append({
"id": rid,
"suspect_cluster": c["canonical_key"],
"keeper_id": c["keeper_id"],
"jaccard": sim,
"content": content,
})
flagged.sort(key=lambda x: x["jaccard"], reverse=True)
return flagged
def main() -> int:
ap = argparse.ArgumentParser(description="归并同义 fact_key 的重复记忆")
ap.add_argument("--db", default=r"D:\.scout\memory.db", help="memory.db 路径")
ap.add_argument("--apply", action="store_true", help="真正写库（默认干跑）")
ap.add_argument("--rewrite-keys", action="store_true",
help="同时把旧式 key 重写为受控词表规范 key")
ap.add_argument("--threshold", type=float, default=MERGE_SIM_THRESHOLD,
help=f"key 语义归并阈值（默认 {MERGE_SIM_THRESHOLD}）")
ap.add_argument("-v", "--verbose", action="store_true")
args = ap.parse_args()
logging.basicConfig(
level=logging.INFO if args.verbose else logging.WARNING,
format="%(levelname)s %(message)s",
)
if not os.path.exists(args.db):
logger.error("找不到 db: %s", args.db)
return 2
conn = _connect(args.db)
clusters, rewrites = plan(conn, args.threshold, args.rewrite_keys)
print(f"db: {args.db}")
print(f"可归并簇: {len(clusters)} 涉及重复条目: {sum(len(c['dup_ids']) for c in clusters)}")
if args.rewrite_keys:
print(f"待重写 key 的条目: {len(rewrites)}")
print()
for c in clusters[:20]:
print(f" 簇({c['n']} 条) 规范键={c['canonical_key']}")
print(f" 成员键: {', '.join(c['member_keys'])}")
print(f" 保留 id={c['keeper_id']}，废弃 id={c['dup_ids']}")
if len(clusters) > 20:
print(f" ...另有 {len(clusters) - 20} 簇")
print()
# 无 key 疑似重复：只报告，不自动归并（理由见 report_keyless docstring）
try:
flagged = report_keyless(conn, clusters)
except Exception as exc: # noqa: BLE001 — 报告失败不影响主流程
logger.debug("keyless 报告跳过: %s", exc)
flagged = []
if flagged:
print(f"⚠ 另有 {len(flagged)} 条**无 fact_key** 的疑似重复（语义同簇，"
f"但字符相似度不足以自动判定，仅提示不处理）：")
for f in flagged[:10]:
print(f" id={f['id']} 疑似同簇={f['suspect_cluster']} "
f"(jaccard={f['jaccard']:.2f})")
print(f" {f['content'][:70]}")
if len(flagged) > 10:
print(f" ...另有 {len(flagged) - 10} 条")
print()
if not args.apply:
print("干跑模式，未写库。确认无误后加 --apply 执行。")
return 0
ts = datetime.now().strftime("%Y%m%d-%H%M%S")
backup = f"{args.db}.bak-dedupe-{ts}"
# 备份需连同 -wal/-shm 一起，否则 SQLite 打开备份会丢未 checkpoint 的数据
conn.commit()
conn.close()
for suffix in ("", "-wal", "-shm"):
src = args.db + suffix
if os.path.exists(src):
shutil.copy2(src, backup + suffix)
logger.info("已备份到 %s（含 -wal/-shm）", backup)
conn = _connect(args.db)
n = apply_plan(conn, clusters, rewrites)
conn.close()
print(f"完成：更新 {n} 条记录。备份 {backup}")
return 0
if __name__ == "__main__":
raise SystemExit(main())