Memory 记忆系统

from dataclasses import dataclass, field
from datetime import datetime
import hashlib
@dataclass
class Memory:
    key: str                       # 归属实体 + 主题，如 "user:42/pref/format"
    content: str
    tier: str                      # fact | preference | inference
    confidence: float              # 0-1，推断类必须低于 0.8
    source: str                    # 谁说的 / 从哪来的
    written_at: str = field(default_factory=lambda: datetime.now().isoformat(timespec="seconds"))
    superseded_at: str | None = None    # 被新版本取代的时间（软删除）
    evidence: list[str] = field(default_factory=list)   # 依据链
    @property
    def active(self) -> bool:
        return self.superseded_at is None
    def to_prompt_line(self) -> str:
        """回喂格式：内容 + 时间 + 层级 + 置信度，让模型自己能判断时效"""
        tag = f"[{self.tier}·{self.confidence:.1f}]"
        return (f"{tag} {self.content}\n"
                f"    记录于 {self.written_at[:10]}　来源：{self.source}")
class MemoryStore:
    def __init__(self):
        self._items: dict[str, list[Memory]] = {}   # key -> 版本链（保留历史）
    def write(self, m: Memory) -> str:
        """
        同 key 的旧版本不删除，只标记 superseded_at。
        这保证了「可解释」——任何时候都能回答「为什么系统会这么认为」。
        去重按内容指纹判断，而不是字符串相等：同一事实会有无数种表述，
        直接比字符串几乎永远为假（见误区 5.3）。
        """
        chain = self._items.setdefault(m.key, [])
        fp = memory_fingerprint(m.content)
        for old in chain:
            if old.active and memory_fingerprint(old.content) == fp:
                return "duplicate"
            if old.active:
                old.superseded_at = m.written_at     # 软删除，保留证据
        chain.append(m)
        return "written"
    def recall(self, keys: list[str], limit: int = 5) -> list[Memory]:
        """只返回当前有效版本，且带时间戳"""
        out = []
        for k in keys:
            actives = [m for m in self._items.get(k, []) if m.active]
            out.extend(actives)
        return out[:limit]
    def conflicts(self) -> list[tuple[Memory, Memory]]:
        """检测同 key 下无法用时序消解的并存冲突（理论上不该存在）"""
        res = []
        for chain in self._items.values():
            actives = [m for m in chain if m.active]
            if len(actives) > 1:
                res.append((actives[0], actives[1]))
        return res
    def forget(self, key: str, hard: bool = False) -> int:
        """
        删除能力是记忆系统的合规底线。
        hard=True 用于用户行使「删除权」，彻底清除包括历史版本。
        """
        if key not in self._items:
            return 0
        n = len(self._items[key])
        if hard:
            del self._items[key]
        else:
            for m in self._items[key]:
                m.superseded_at = m.superseded_at or datetime.now().isoformat(timespec="seconds")
        return n
    def explain(self, key: str) -> str:
        """可解释性：把一条记忆的完整演变史打印出来"""
        chain = self._items.get(key, [])
        if not chain:
            return f"{key}: 无记录"
        lines = []
        for m in chain:
            state = "生效" if m.active else f"已于 {m.superseded_at[:10]} 失效"
            lines.append(f"  {m.written_at[:10]}  {state}　[{m.tier}] {m.content}（来源 {m.source}）")
        return f"{key} 的演变：\n" + "\n".join(lines)
def memory_fingerprint(text: str) -> str:
    """内容指纹，用于跨库去重"""
    return hashlib.sha256(text.strip().encode()).hexdigest()[:16]