memoryweave/scripts/verify_7d_quality.py

#!/usr/bin/env python3
"""
7 维度记忆质量验证脚本
用法: python3 scripts/verify_7d_quality.py [--id <memory_id>]

检查 LanceDB 中记忆的 7 个质量维度:
1. recall_count    — 被召回次数（应随使用增加）
2. importance      — 重要性 = recency × (1+log(1+recall_count))
3. quality_score   — 综合质量评分（0-1）
4. useful_count    — positive 反馈总数
5. not_useful_count — negative 反馈总数
6. tier            — normal/core（core 永不衰减）
7. version         — 版本号（更新溯源）
"""
import sys
import argparse
import lancedb
from pathlib import Path

DB_PATH = "/var/lib/memoryweave"
API_KEY = "zhiyi-dev-key-2026"
API_URL = "http://127.0.0.1:7821"

def get_memories(limit=2000):
    db = lancedb.connect(DB_PATH)
    tbl = db.open_table("memories")
    return tbl.head(limit).to_pylist()

def check_recall_count(memories):
    """维度1: recall_count 应该 > 0 (被用过的记忆)"""
    zero = sum(1 for r in memories if r.get("recall_count", 0) == 0)
    nonzero = len(memories) - zero
    pct = nonzero / len(memories) * 100 if memories else 0
    status = "⚠️" if zero > len(memories) * 0.8 else "✅"
    print(f"  recall_count: {nonzero}/{len(memories)} ({pct:.1f}%) > 0  {status}")
    if zero > len(memories) * 0.8:
        print("    ⚠️ recall_count 几乎全为 0 → Go Update() 未正确处理 map[string]string $inc")
    return zero <= len(memories) * 0.8

def check_importance(memories):
    """维度2: importance 应有差异（recency × log(1+recall_count)）"""
    vals = [r.get("importance", 0) for r in memories]
    unique = len(set(vals))
    all_one = all(abs(v - 1.0) < 0.01 for v in vals)
    status = "⚠️" if all_one else "✅"
    print(f"  importance: {unique} unique values, all≈1.0: {all_one}  {status}")
    if all_one:
        print("    ⚠️ 所有 importance=1.0 → recall_count=0 导致公式退化")
    return not all_one

def check_quality_score(memories):
    """维度3: quality_score 应有分布（不是全 0 或全 1）"""
    vals = [r.get("quality_score", 0) for r in memories if r.get("quality_score", 0) > 0]
    if not vals:
        print("  quality_score: 全为 0  ⚠️")
        return False
    unique = len(set(vals))
    print(f"  quality_score: {unique} unique, range [{min(vals):.2f}, {max(vals):.2f}]  ✅")
    return True

def check_feedback(memories):
    """维度4+5: useful_count / not_useful_count"""
    useful = sum(1 for r in memories if r.get("useful_count", 0) > 0)
    not_useful = sum(1 for r in memories if r.get("not_useful_count", 0) > 0)
    print(f"  useful_count: {useful} memories > 0")
    print(f"  not_useful_count: {not_useful} memories > 0")
    return True

def check_tier(memories):
    """维度6: tier 分布"""
    tiers = {}
    for r in memories:
        t = r.get("tier", "normal")
        tiers[t] = tiers.get(t, 0) + 1
    print(f"  tier: {tiers}")
    return True

def check_version(memories):
    """维度7: version 应 >= 1"""
    v0 = sum(1 for r in memories if r.get("version", 0) < 1)
    print(f"  version: {v0}/{len(memories)} memories with version < 1  {'⚠️' if v0 else '✅'}")
    return v0 == 0

def check_timestamps(memories):
    """时间戳: created_at / updated_at / last_recalled_at"""
    zero_created = sum(1 for r in memories if r.get("created_at", "") == "" or "0001-01-01" in str(r.get("created_at", "")))
    zero_updated = sum(1 for r in memories if r.get("updated_at", "") == "" or "0001-01-01" in str(r.get("updated_at", "")))
    zero_recalled = sum(1 for r in memories if r.get("last_recalled_at", "") == "" or "0001-01-01" in str(r.get("last_recalled_at", "")))
    print(f"  timestamps: created_at zero={zero_created}, updated_at zero={zero_updated}, last_recalled_at zero={zero_recalled}")
    return zero_created == 0

def verify_specific_memory(mem_id):
    """验证指定记忆的 7 维度详细值"""
    memories = get_memories(5000)
    target = [r for r in memories if r.get("id") == mem_id]
    if not target:
        print(f"Memory {mem_id} not found in first 5000 records")
        return
    r = target[0]
    print(f"\n7维度详情 [{r.get('id', '?')[:20]}...]:")
    print(f"  recall_count:      {r.get('recall_count', 0)}")
    print(f"  importance:        {r.get('importance', 0):.4f}")
    print(f"  quality_score:     {r.get('quality_score', 0):.4f}")
    print(f"  useful_count:      {r.get('useful_count', 0)}")
    print(f"  not_useful_count:  {r.get('not_useful_count', 0)}")
    print(f"  tier:              {r.get('tier', 'normal')}")
    print(f"  version:           {r.get('version', 1)}")
    print(f"  created_at:         {r.get('created_at', '?')}")
    print(f"  updated_at:         {r.get('updated_at', '?')}")
    print(f"  last_recalled_at:   {r.get('last_recalled_at', '?')}")

def main():
    parser = argparse.ArgumentParser(description="7维度记忆质量验证")
    parser.add_argument("--id", help="检查特定记忆 ID")
    parser.add_argument("--limit", type=int, default=2000, help="采样数量")
    args = parser.parse_args()

    print("=" * 50)
    print("7维度记忆质量验证")
    print("=" * 50)

    if args.id:
        verify_specific_memory(args.id)
        return

    memories = get_memories(args.limit)
    print(f"\n采样 {len(memories)} 条记忆\n")

    checks = [
        ("维度1: recall_count", check_recall_count),
        ("维度2: importance", check_importance),
        ("维度3: quality_score", check_quality_score),
        ("维度4+5: feedback", check_feedback),
        ("维度6: tier", check_tier),
        ("维度7: version", check_version),
        ("时间戳", check_timestamps),
    ]

    all_ok = True
    for name, fn in checks:
        print(f"\n{name}:")
        if not fn(memories):
            all_ok = False

    print("\n" + "=" * 50)
    if all_ok:
        print("✅ 所有维度正常")
    else:
        print("⚠️ 存在维度异常，见上方详情")
    print("=" * 50)

if __name__ == "__main__":
    main()