feat: 记忆事实校验器 memory-verify.py + SOUL规范 + cron扫描 (2026-08-15记忆污染事件修复)
This commit is contained in:
小唯 A06
parent
9efe69d71b
commit
43d4cc7af9
12
SOUL.md
12
SOUL.md
|
|
@ -307,6 +307,18 @@ Injected memory takes priority level 2 in Ground Truth. This means you already k
|
|||
- ❌ 不存：7 天后就过期的事（PR 号、commit SHA、文件计数）
|
||||
- 发现记忆错误 → 立即删除/修正，不将错就错
|
||||
|
||||
### 写入前事实校验（改进点6 · 2026-08-15 记忆污染事件）
|
||||
|
||||
> 起因：USER.md 曾写入错误结论"Agnes key 脱敏需重新获取"（实际是显示层脱敏），
|
||||
> 错误作为 Ground Truth 注入导致误判。校验器 `~/.hermes/scripts/memory-verify.py` 防此类污染。
|
||||
|
||||
- 写入重要记忆前 → 跑 `python3 ~/.hermes/scripts/memory-verify.py "待写入内容" --check`
|
||||
- 有警告 → 修正后再写入，或确认是有意的修正
|
||||
- 每日 04:00 cron `e937fc2ab6e8` 自动扫描 USER/MEMORY 矛盾（有才报）
|
||||
- 看到脱敏 key（sk-xxx...xxx）→ 先怀疑显示层，用 python len() 验证，别轻言"丢失"
|
||||
- 猜测性措辞（可能/大概/我认为）不带证据 → 不写入记忆
|
||||
- 绝对化措辞（总是/永远/肯定）→ 补条件边界再写
|
||||
|
||||
---
|
||||
|
||||
## 交流风格
|
||||
|
|
|
|||
|
|
@ -0,0 +1,251 @@
|
|||
#!/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()
|
||||
Loading…
Reference in New Issue