252 lines
11 KiB
Python
252 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
memory-verify.py — 记忆写入前的事实校验器（v1.0.0）
|
||
|
||
用途：在向织忆/USER/MEMORY 写入新记忆前，检测与现有记忆的矛盾，
|
||
防止"记忆污染"（错误结论被当成 Ground Truth 注入）。
|
||
|
||
触发方式：
|
||
1. 手动：python3 memory-verify.py "要写入的记忆内容" --check
|
||
2. 批量扫描：python3 memory-verify.py --scan (检查全库矛盾)
|
||
3. 写入时自动：由 daemon.py 蒸馏前调用（预留 hook）
|
||
|
||
2026-08-15 起因：USER.md 曾写入错误结论"Agnes key 是脱敏的需重新获取"，
|
||
该错误作为 Ground Truth 注入导致误判 key 丢失。本工具防止此类污染。
|
||
|
||
确定性规则（不依赖 LLM）：
|
||
R1 脱敏误判检测：内容含 sk-.../ghp_.../nvapi- 等 key 模式 + "脱敏/丢失/需要重新获取"
|
||
→ 警告：可能是 Hermes 显示层脱敏，先验证文件真实长度
|
||
R2 绝对时间过期：内容含 2025/2026 日期但描述"当前/现在/最新"且已过 90 天
|
||
→ 警告：可能是过期结论
|
||
R3 猜测性措辞：内容含"可能/大概/我认为/应该是/猜测"但无证据标记
|
||
→ 警告：猜测不应写入记忆
|
||
R4 绝对化错误：内容含"总是/从不/永远/肯定/绝对"等绝对化词
|
||
→ 提示：记忆应保留条件边界
|
||
R5 语义矛盾：与现有记忆语义冲突（通过 recall API 检索对比）
|
||
→ 警告：列出冲突条目供人工裁决
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.request
|
||
from datetime import datetime, timedelta
|
||
|
||
# ── 配置 ──────────────────────────────────────────────────────────────────────
|
||
ZHIYI_URL = "http://localhost:7821/api/v1/recall"
|
||
ZHIYI_HEADERS = {
|
||
"X-API-Key": os.environ.get("ZHIYI_API_KEY", "zhiyi-dev-key-2026"),
|
||
"Content-Type": "application/json",
|
||
}
|
||
USER_MD = os.path.expanduser("~/.hermes/memories/USER.md")
|
||
MEMORY_MD = os.path.expanduser("~/.hermes/memories/MEMORY.md")
|
||
TOP_K = 5
|
||
|
||
# ── 确定性规则 ────────────────────────────────────────────────────────────────
|
||
|
||
KEY_PATTERNS = [
|
||
(r"sk-[A-Za-z0-9]{3}\.\.\.[A-Za-z0-9]{3,4}", "sk- 脱敏 key"),
|
||
(r"ghp_[A-Za-z0-9]{3}\.\.\.[A-Za-z0-9]{3,4}", "ghp_ 脱敏 key"),
|
||
(r"nvapi-[A-Za-z0-9]{3}\.\.\.[A-Za-z0-9]{3,4}", "nvapi 脱敏 key"),
|
||
(r"sk-[A-Za-z0-9]{5,}", "sk- key"),
|
||
]
|
||
KEY_LOSS_WORDS = ["脱敏", "丢失", "丢", "需要重新获取", "重新获取", "不完整", "不是完整"]
|
||
|
||
|
||
def rule_r1_key_misjudge(text: str) -> list[str]:
|
||
"""R1: key 脱敏误判检测
|
||
|
||
区分两种情况：
|
||
- 传播错误："key 是脱敏的，需要重新获取" → 报警
|
||
- 纠正错误："显示脱敏≠存储脱敏，文件是完整key" → 不报警（含纠正标记）
|
||
"""
|
||
warnings = []
|
||
# 纠正标记：说明该内容是在澄清"显示脱敏≠存储脱敏"的误解
|
||
correction_markers = ["≠", "不等于", "显示脱敏", "存储脱敏", "别误判", "不是存储", "完整key", "实际长度"]
|
||
is_correction = any(m in text for m in correction_markers)
|
||
if is_correction:
|
||
return warnings # 纠正性内容不报警
|
||
for pattern, label in KEY_PATTERNS:
|
||
if re.search(pattern, text):
|
||
for word in KEY_LOSS_WORDS:
|
||
if word in text:
|
||
warnings.append(
|
||
f"R1 检测到「{label}」+「{word}」：可能是 Hermes 显示层脱敏（显示 sk-7k9...2ikW ≠ 存储脱敏）。"
|
||
f"写入前必须用 python len() 验证 key.md 实际长度再下结论！"
|
||
)
|
||
break
|
||
break
|
||
return warnings
|
||
|
||
|
||
def rule_r2_stale_date(text: str) -> list[str]:
|
||
"""R2: 绝对时间过期检测"""
|
||
warnings = []
|
||
now = datetime.now()
|
||
# 匹配 2026-xx-xx 或 2026/xx/xx 日期
|
||
dates = re.findall(r"(20\d{2})[-/](\d{1,2})[-/](\d{1,2})", text)
|
||
for y, m, d in dates:
|
||
try:
|
||
dt = datetime(int(y), int(m), int(d))
|
||
if now - dt > timedelta(days=90):
|
||
if re.search(r"当前|现在|最新|目前", text):
|
||
warnings.append(
|
||
f"R2 检测到日期 {y}-{m}-{d}（已过 90 天）但描述为「当前/最新」："
|
||
f"可能是过期结论，确认是否仍有效。"
|
||
)
|
||
except ValueError:
|
||
pass
|
||
return warnings
|
||
|
||
|
||
def rule_r3_speculation(text: str) -> list[str]:
|
||
"""R3: 猜测性措辞检测"""
|
||
warnings = []
|
||
guess_words = ["可能", "大概", "我认为", "应该是", "我猜测", "也许", "似乎"]
|
||
evidence_markers = ["已验证", "实测", "证据", "确认", "测试通过"]
|
||
for word in guess_words:
|
||
if word in text:
|
||
# 如果整条内容没有任何证据标记，警告
|
||
if not any(ev in text for ev in evidence_markers):
|
||
warnings.append(
|
||
f"R3 检测到猜测性措辞「{word}」且无证据标记（已验证/实测/证据）："
|
||
f"猜测不应写入记忆，确认事实后再写。"
|
||
)
|
||
break
|
||
return warnings
|
||
|
||
|
||
def rule_r4_absolutism(text: str) -> list[str]:
|
||
"""R4: 绝对化措辞检测"""
|
||
warnings = []
|
||
abs_words = ["总是", "从不", "永远", "肯定", "绝对", "一定"]
|
||
for word in abs_words:
|
||
if word in text:
|
||
warnings.append(
|
||
f"R4 检测到绝对化措辞「{word}」：记忆应保留条件边界（在什么情况下成立），避免过度泛化。"
|
||
)
|
||
break
|
||
return warnings
|
||
|
||
|
||
def recall_zhiyi(query: str, top_k: int = TOP_K) -> list[dict]:
|
||
"""调用织忆 recall API 检索相关记忆"""
|
||
try:
|
||
req = urllib.request.Request(
|
||
ZHIYI_URL,
|
||
data=json.dumps({"query": query, "top_k": top_k}).encode("utf-8"),
|
||
headers=ZHIYI_HEADERS,
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
# 兼容不同返回结构
|
||
if isinstance(data, list):
|
||
return data
|
||
return data.get("results") or data.get("memories") or data.get("data") or []
|
||
except Exception as e:
|
||
return [{"error": str(e)}]
|
||
|
||
|
||
def rule_r5_semantic_conflict(text: str, existing: list[dict]) -> list[str]:
|
||
"""R5: 语义矛盾检测（与现有记忆对比）"""
|
||
warnings = []
|
||
for item in existing:
|
||
if isinstance(item, dict) and "content" in item:
|
||
old = item["content"]
|
||
# 简单矛盾启发式：新内容否定旧内容（出现"不/并非/错误/修正"等否定词）
|
||
negation = ["不是", "并非", "错误", "修正", "别", "不要", "禁止", "≠", "不等于"]
|
||
if any(neg in text for neg in negation):
|
||
# 提取共同主题词（简单交集）
|
||
old_words = set(re.findall(r"[\u4e00-\u9fff]{2,6}", old))
|
||
new_words = set(re.findall(r"[\u4e00-\u9fff]{2,6}", text))
|
||
overlap = old_words & new_words
|
||
if len(overlap) >= 2:
|
||
warnings.append(
|
||
f"R5 检测到与现有记忆语义冲突：\n"
|
||
f" 新内容: {text[:80]}...\n"
|
||
f" 旧记忆: {old[:80]}...\n"
|
||
f" 共同主题: {list(overlap)[:4]}\n"
|
||
f" → 若是修正，应更新旧条目而非新建；若是矛盾，需人工裁决。"
|
||
)
|
||
return warnings
|
||
|
||
|
||
# ── 主流程 ────────────────────────────────────────────────────────────────────
|
||
|
||
def check_text(text: str, do_recall: bool = True) -> list[str]:
|
||
"""对一段待写入文本执行全部规则，返回警告列表"""
|
||
warnings = []
|
||
warnings += rule_r1_key_misjudge(text)
|
||
warnings += rule_r2_stale_date(text)
|
||
warnings += rule_r3_speculation(text)
|
||
warnings += rule_r4_absolutism(text)
|
||
if do_recall:
|
||
existing = recall_zhiyi(text)
|
||
warnings += rule_r5_semantic_conflict(text, existing)
|
||
return warnings
|
||
|
||
|
||
def scan_existing_md(path: str, label: str) -> list[str]:
|
||
"""扫描已有记忆文件中的问题条目"""
|
||
warnings = []
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
# USER/MEMORY.md 用 § 分隔条目
|
||
entries = [e.strip() for e in content.split("§") if e.strip()]
|
||
for i, entry in enumerate(entries, 1):
|
||
for w in check_text(entry, do_recall=False):
|
||
warnings.append(f"[{label} #{i}] {w}")
|
||
except FileNotFoundError:
|
||
pass
|
||
return warnings
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="记忆写入前的事实校验器")
|
||
parser.add_argument("text", nargs="?", help="待校验的记忆内容")
|
||
parser.add_argument("--check", action="store_true", help="校验一段文本")
|
||
parser.add_argument("--scan", action="store_true", help="扫描现有记忆文件")
|
||
parser.add_argument("--json", action="store_true", help="JSON 输出")
|
||
args = parser.parse_args()
|
||
|
||
if args.scan or (not args.text and not args.check):
|
||
# 无参数时默认执行扫描（供 cron no_agent 调用）
|
||
warnings = []
|
||
warnings += scan_existing_md(USER_MD, "USER")
|
||
warnings += scan_existing_md(MEMORY_MD, "MEMORY")
|
||
if args.json:
|
||
print(json.dumps({"warnings": warnings, "count": len(warnings)}, ensure_ascii=False))
|
||
else:
|
||
if warnings:
|
||
print(f"⚠️ 发现 {len(warnings)} 个潜在问题：")
|
||
for w in warnings:
|
||
print(f" - {w}")
|
||
else:
|
||
# no_agent 模式：无输出 = 静默（看门狗模式）
|
||
pass
|
||
return
|
||
|
||
if args.text and args.check:
|
||
warnings = check_text(args.text)
|
||
if args.json:
|
||
print(json.dumps({"warnings": warnings, "count": len(warnings)}, ensure_ascii=False))
|
||
else:
|
||
if warnings:
|
||
print(f"⚠️ 写入前发现 {len(warnings)} 个警告：")
|
||
for w in warnings:
|
||
print(f" - {w}")
|
||
print("建议：修正后再写入，或确认这是有意的修正。")
|
||
else:
|
||
print("✅ 校验通过，可以写入")
|
||
return
|
||
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|