大模型长期记忆系统设计：记忆压缩、去重、动态遗忘、时间衰减与冲突处理应用实践24.0

memory_item = {
    "memory_id": "uuid唯一标识",
    "content": "记忆文本内容",
    "embedding": "向量，用于相似度检索",
    "create_time": "创建时间戳",
    "last_access_time": "最后一次被召回时间戳",
    "weight": "记忆权重（用于时间衰减）",
    "tag": ["分类标签"],
    "source": "对话/工具调用/外部输入",
    "version": 1,  # 记忆版本，用于更新、冲突处理
    "is_valid": True
}

import time
import uuid
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class MemoryItem:
    memory_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    content: str
    embedding: Optional[List[float]] = None
    create_time: float = field(default_factory=time.time)
    last_access_time: float = field(default_factory=time.time)
    weight: float = 1.0
    tags: List[str] = field(default_factory=list)
    source: str = "chat"
    version: int = 1
    is_valid: bool = True

class BaseMemoryStore:
    def __init__(self):
        self.memory_pool: dict[str, MemoryItem] = {}

    def add_memory(self, item: MemoryItem):
        self.memory_pool[item.memory_id] = item

    def get_all(self) -> List[MemoryItem]:
        return list(self.memory_pool.values())

def compress_memory_content(raw_text: str, level: str = "light") -> str:
    """
    level: light 轻度压缩 / deep 深度压缩
    """
    if len(raw_text) < 120:
        return raw_text

    if level == "light":
        prompt = f"""提取文本中的客观事实，精简语句，保留细节，删除闲聊话术：
{raw_text}"""
    else:
        prompt = f"""高度概括核心事实，去除全部过程描述，仅输出关键结论：
{raw_text}"""

    # 此处替换为真实LLM调用接口
    # result = llm.chat(prompt)
    # return result

    # 模拟返回
    return f"【{level}压缩结果】提取核心事实：{raw_text[:60]}..."

import numpy as np

def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
    a = np.array(vec1)
    b = np.array(vec2)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

def deduplicate_memory(new_mem: MemoryItem, store: BaseMemoryStore, threshold=0.85) -> bool:
    """
    返回True：存在重复，新记忆不入库；False：无重复，可以入库
    """
    if not new_mem.embedding:
        return False

    all_mem = store.get_all()
    for mem in all_mem:
        if not mem.embedding or not mem.is_valid:
            continue
        sim = cosine_similarity(new_mem.embedding, mem.embedding)
        if sim >= threshold:
            # 更新旧记忆访问时间
            mem.last_access_time = time.time()
            return True
    return False

def update_memory(store: BaseMemoryStore, target_id: str, new_content: str):
    if target_id not in store.memory_pool:
        return None
    old_mem = store.memory_pool[target_id]
    # 旧记忆保留，标记可选失效
    old_mem.is_valid = False
    # 创建新版本记忆
    new_mem = MemoryItem(
        content=new_content,
        tags=old_mem.tags.copy(),
        source=old_mem.source,
        version=old_mem.version + 1
    )
    store.add_memory(new_mem)
    return new_mem

import math

def decay_weight(mem: MemoryItem, decay_lambda=0.0001, min_weight=0.1):
    now = time.time()
    delta = now - mem.last_access_time
    new_weight = mem.weight * math.exp(-decay_lambda * delta)
    mem.weight = max(new_weight, min_weight)
    mem.last_access_time = now
    return mem.weight

def detect_memory_conflict(mem_list: List[MemoryItem]) -> List[tuple]:
    conflict_pairs = []
    # 两两对比送入模型判断冲突
    for i in range(len(mem_list)):
        for j in range(i+1, len(mem_list)):
            m1, m2 = mem_list[i], mem_list[j]
            # llm调用判断是否冲突
            # result = llm.check_conflict(m1.content, m2.content)
            # if result == "冲突":
            #     conflict_pairs.append((m1.memory_id, m2.memory_id))
    return conflict_pairs

import time
import uuid
import math
import numpy as np
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Tuple

# ======================
# 1. 基础数据模型定义
# ======================
@dataclass
class MemoryItem:
    memory_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    content: str
    embedding: Optional[List[float]] = None
    create_time: float = field(default_factory=time.time)
    last_access_time: float = field(default_factory=time.time)
    weight: float = 1.0
    tags: List[str] = field(default_factory=list)
    source: str = "chat"
    version: int = 1
    is_valid: bool = True


class BaseMemoryStore:
    def __init__(self):
        self.memory_pool: Dict[str, MemoryItem] = {}

    def add_memory(self, item: MemoryItem):
        self.memory_pool[item.memory_id] = item

    def get_all(self) -> List[MemoryItem]:
        return list(self.memory_pool.values())

    def get_by_id(self, mid: str) -> Optional[MemoryItem]:
        return self.memory_pool.get(mid)

    def list_valid_memories(self) -> List[MemoryItem]:
        return [m for m in self.get_all() if m.is_valid]


# ======================
# 2. 工具函数：向量相似度
# ======================
def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
    a = np.array(vec1)
    b = np.array(vec2)
    dot = np.dot(a, b)
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return float(dot / (norm_a * norm_b))


# 模拟 Embedding 生成（真实场景替换成 embedding 模型调用）
def mock_embedding(text: str) -> List[float]:
    np.random.seed(hash(text) % 4294967295)
    return list(np.random.rand(32))


# ======================
# 3. 记忆压缩模块
# ======================
def compress_memory_content(raw_text: str, level: str = "light") -> str:
    """
    level: light 轻度压缩 / deep 深度压缩
    """
    if len(raw_text) < 120:
        return raw_text

    if level == "light":
        prompt = f"""提取文本中的客观事实，精简语句，保留细节，删除闲聊话术：
{raw_text}"""
    else:
        prompt = f"""高度概括核心事实，去除全部过程描述，仅输出关键结论：
{raw_text}"""

    # ========== 这里替换成真实LLM调用 ==========
    # resp = llm.chat(prompt)
    # return resp
    # 模拟压缩结果
    if level == "light":
        return f"[轻量压缩]{raw_text}"
    else:
        return f"[深度压缩]{raw_text[:70]}..."


# ======================
# 4. 记忆去重模块
# ======================
def deduplicate_memory(new_mem: MemoryItem, store: BaseMemoryStore, threshold=0.85) -> Tuple[bool, Optional[MemoryItem]]:
    """
    :return: (是否重复, 匹配到的旧记忆)
    """
    if not new_mem.embedding:
        return False, None

    valid_mems = store.list_valid_memories()
    for mem in valid_mems:
        if not mem.embedding:
            continue
        sim = cosine_similarity(new_mem.embedding, mem.embedding)
        if sim >= threshold:
            mem.last_access_time = time.time()
            return True, mem
    return False, None


# ======================
# 5. 记忆版本更新模块
# ======================
def update_memory(store: BaseMemoryStore, target_id: str, new_content: str) -> Optional[MemoryItem]:
    old_mem = store.get_by_id(target_id)
    if not old_mem:
        return None

    # 旧记忆软失效
    old_mem.is_valid = False
    # 构建新版本记忆
    compressed = compress_memory_content(new_content, level="light")
    new_emb = mock_embedding(compressed)
    new_mem = MemoryItem(
        content=compressed,
        embedding=new_emb,
        tags=old_mem.tags.copy(),
        source=old_mem.source,
        version=old_mem.version + 1
    )
    store.add_memory(new_mem)
    return new_mem


# ======================
# 6. 时间衰减 & 遗忘模块
# ======================
def decay_weight(mem: MemoryItem, decay_lambda=0.0001, min_weight=0.1) -> float:
    now = time.time()
    delta = now - mem.last_access_time
    new_weight = mem.weight * math.exp(-decay_lambda * delta)
    mem.weight = max(new_weight, min_weight)
    return mem.weight


def batch_decay_process(store: BaseMemoryStore, decay_lambda=0.0001, min_weight=0.1):
    """批量执行衰减 + 低权重记忆软遗忘"""
    all_valid = store.list_valid_memories()
    for m in all_valid:
        decay_weight(m, decay_lambda, min_weight)
        if m.weight <= min_weight:
            m.is_valid = False
            print(f"[遗忘触发] memory_id={m.memory_id}, weight={m.weight:.3f}")


# ======================
# 7. 冲突检测模块
# ======================
def detect_memory_conflict(mem_list: List[MemoryItem]) -> List[Tuple[str, str]]:
    conflict_pairs = []
    # 两两对比
    for i in range(len(mem_list)):
        for j in range(i + 1, len(mem_list)):
            m1, m2 = mem_list[i], mem_list[j]
            # ========== 替换为真实LLM冲突判断 ==========
            # prompt = f"""判断两条事实是否冲突：
            # A:{m1.content}
            # B:{m2.content}
            # 只输出：冲突 / 无冲突"""
            # res = llm.chat(prompt).strip()
            # if res == "冲突":
            #     conflict_pairs.append((m1.memory_id, m2.memory_id))

            # 模拟规则测试：简单关键词模拟冲突
            if "周末在家休息" in m1.content and "周末外出旅行" in m2.content:
                conflict_pairs.append((m1.memory_id, m2.memory_id))
    return conflict_pairs


# ======================
# 8. 记忆写入流水线（完整链路：压缩→向量化→去重→入库）
# ======================
def memory_pipeline(raw_text: str, store: BaseMemoryStore, dedup_threshold=0.85) -> Optional[MemoryItem]:
    # 1. 压缩
    compressed_text = compress_memory_content(raw_text, level="light")
    # 2. 生成向量
    emb = mock_embedding(compressed_text)
    # 3. 构建记忆对象
    new_mem = MemoryItem(content=compressed_text, embedding=emb)
    # 4. 去重检测
    is_dup, old_mem = deduplicate_memory(new_mem, store, dedup_threshold)
    if is_dup:
        print(f"[去重] 识别相似记忆，不新增，更新旧记忆访问时间 id={old_mem.memory_id}")
        return None
    # 5. 无重复则入库
    store.add_memory(new_mem)
    print(f"[入库成功] memory_id={new_mem.memory_id}")
    return new_mem


# ======================
# 9. 记忆召回模块
# ======================
def recall_memory(query: str, store: BaseMemoryStore, top_k=5, threshold=0.6) -> List[MemoryItem]:
    query_emb = mock_embedding(query)
    candidates = []
    valid_mems = store.list_valid_memories()
    for m in valid_mems:
        if not m.embedding:
            continue
        sim = cosine_similarity(query_emb, m.embedding)
        if sim >= threshold:
            candidates.append((sim, m))
    # 相似度降序排序
    candidates.sort(key=lambda x: x[0], reverse=True)
    result = []
    for sim, mem in candidates[:top_k]:
        mem.last_access_time = time.time()
        result.append(mem)
    return result


# ======================
# 10. 主测试流程
# ======================
if __name__ == "__main__":
    memory_store = BaseMemoryStore()

    print("===== 步骤1：写入第一批记忆 =====")
    memory_pipeline("用户周末打算在家休息", memory_store)
    mid_test = memory_pipeline("用户喜欢喝美式咖啡", memory_store).memory_id

    print("\n===== 步骤2：更新记忆（用户修改喜好） =====")
    updated = update_memory(memory_store, mid_test, "用户现在更喜欢拿铁咖啡")
    print(f"新版本记忆ID: {updated.memory_id}, version={updated.version}")

    print("\n===== 步骤3：写入冲突记忆 =====")
    memory_pipeline("用户周末计划外出旅行", memory_store)

    print("\n===== 步骤4：批量衰减模拟 =====")
    batch_decay_process(memory_store, decay_lambda=0.00001)

    print("\n===== 步骤5：召回相关记忆 =====")
    recall_res = recall_memory("用户周末安排", memory_store, top_k=5)
    for item in recall_res:
        print(f"- {item.content} weight={item.weight:.3f}")

    print("\n===== 步骤6：冲突检测 =====")
    conflicts = detect_memory_conflict(recall_res)
    if conflicts:
        for a, b in conflicts:
            m1 = memory_store.get_by_id(a)
            m2 = memory_store.get_by_id(b)
            print(f"发现冲突:\n  A:{m1.content}\n  B:{m2.content}")
    else:
        print("未检测到冲突")