大模型长期记忆系统设计：记忆压缩、去重、动态遗忘、时间衰减与冲突处理应用实践24.0
【摘要】 大模型记忆系统设计与实现 摘要：本文针对大模型上下文窗口有限的痛点，提出了一套完整的记忆系统解决方案。系统采用三层架构设计（瞬时记忆、短期记忆、长期记忆），通过记忆压缩、去重、更新、衰减和冲突处理等机制实现全生命周期管理。关键技术包括：基于LLM的语义压缩算法、向量相似度去重策略、版本化更新机制、指数衰减遗忘模型，以及多维度冲突消解方案。文章详细阐述了各模块实现原理，并提供了包含数据模型、处理流水
一、前言
通用大模型都存在一个明显短板：上下文窗口有限。单次对话结束，超出窗口范围的历史信息就会丢失。想要实现长期对话、持续运行的AI智能体、自动化业务Agent，单纯依靠Prompt上下文远远不够。通常我们把全部历史对话直接塞进提示词，短期内尚可运行，随着交互次数增多，Token数量爆炸、推理成本飙升、大量冗余信息干扰模型判断，对话质量持续下滑。
要解决这个痛点，就需要一套独立于上下文窗口之外的记忆系统。一套完备的记忆系统不能只做到简单存储对话记录，必须拥有一整套生命周期管理机制：记忆写入、压缩精简、重复内容合并、信息迭代更新、基于时间的自然遗忘、多条记忆信息冲突时自动处理。
不少人会把记忆系统简单理解成数据库，仅仅做信息持久化。真正适配大模型的记忆引擎，核心价值是筛选有效信息，在合适时机召回合适内容，屏蔽噪声，模拟人类记忆规律。人类不会永久记住所有细节，旧记忆慢慢模糊，相似经历会合并，新旧认知矛盾时会修正想法。大模型记忆系统正是借鉴这一思路设计。
二、大模型记忆基础架构
1. 记忆分层理论
参考经典智能体记忆分层思路，我们将记忆划分为三层，各司其职，避免所有信息混杂在一起。
- 1. 瞬时记忆：对应大模型当前上下文窗口，生命周期最短。只保存当前一轮交互信息，对话结束或者达到长度阈值后，重要内容会被提炼写入短期记忆，无用信息直接丢弃。瞬时记忆不持久化，仅用于本轮推理。
- 2. 短期记忆： 最近一段时间产生的结构化记忆碎片，保留细节丰富。会持续接受压缩、去重运算，持续参与召回检索。随着时间推移，记忆权重持续衰减，满足条件后迁移至长期记忆。
- 3. 长期记忆：经过多层筛选、高度凝练的核心信息，保留关键结论，弱化细枝末节。衰减速度更慢，只有极高价值信息才能长期留存。检索成本更高，一般不会全部加载，按需召回。
三层记忆形成流转链路：瞬时记忆提取信息 → 存入短期记忆 → 持续优化治理（压缩、去重）→ 满足条件晋升长期记忆；长期记忆也可被召回，临时加载进短期记忆参与交互。
2. 记忆数据基础结构
任何记忆单元，都需要标准化字段，支撑后续所有运算。一份标准记忆对象基础字段设计如下：
memory_item = {
"memory_id": "uuid唯一标识",
"content": "记忆文本内容",
"embedding": "向量，用于相似度检索",
"create_time": "创建时间戳",
"last_access_time": "最后一次被召回时间戳",
"weight": "记忆权重（用于时间衰减）",
"tag": ["分类标签"],
"source": "对话/工具调用/外部输入",
"version": 1, # 记忆版本，用于更新、冲突处理
"is_valid": True
}
- weight：核心字段，时间衰减算法的运算载体；
- version：支持记忆迭代更新，保留变更轨迹；
- last_access_time：实现 “被频繁调取的记忆衰减变慢”，贴近人脑规律。
3. 完整数据流总览
- 1. 用户与模型交互，本轮对话存在瞬时记忆；
- 2. 记忆提取模块从对话中抽取关键事实、偏好、需求；
- 3. 新增记忆进入处理流水线：相似度检索去重 → 记忆压缩结构化；
- 4. 写入短期记忆库，初始化记忆权重；
- 5. 定时任务持续执行：时间衰减计算、低权重记忆清理；
- 6. 需要历史信息时执行向量检索召回记忆；
- 7. 召回多条记忆存在事实矛盾，启动冲突处理模块；
- 8. 用户产生新信息，可以触发原有记忆更新操作。
4. 基础Memory数据模型
该示例定义了一个结构化的长期记忆存储模块：
- MemoryItem数据类封装了记忆 ID、内容、向量、权重、标签、访问时间及版本等元信息；
- BaseMemoryStore提供基础的记忆增删查能力，为后续实现记忆检索、衰减淘汰及多轮对话上下文管理奠定基础。
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())
三、记忆压缩机制
1. 记忆压缩的必要性
原始对话文本存在大量冗余：客套话、重复语气、无效修饰、问答来回拉扯。如果直接原始文本存储，会带来三个严重问题：
- 向量存储占用空间大，检索速度下降；
- 召回时无效文本占用 Token，增加推理成本；
- 无关噪声干扰模型理解事实，降低回答准确性。
记忆压缩目标不是简单删减文字，而是保留事实主体，剔除无效修辞，实现信息密度最大化。压缩分为两大路线：规则化轻量压缩、大模型语义压缩。
2. 两类压缩实现方案
2.1 规则压缩（低成本，适合大批量预处理）
适用场景：实时高频产生记忆，不希望频繁调用LLM。 执行策略：
- 剔除语气词、重复助词；
- 合并连续同类描述；
- 去除无关情绪表达，只保留客观事实； 局限：无法深度理解语义，复杂长文本压缩效果有限。
2.2 LLM语义压缩（主流方案，优先推荐）
设计专用压缩Prompt，引导模型提炼核心事实。 示例Prompt模板：
请提取下面对话中的客观事实，精简表述。 要求：只保留关键信息，删除闲聊、语气、重复语句；不要主观推测；输出简洁陈述句。 对话内容：{{content}}
压缩分级策略：
- 轻度压缩：保留大部分细节，适合短期记忆；
- 深度压缩：只保留结论，去除过程描述，用于晋升长期记忆。
3. 压缩边界与风险控制
压缩存在信息丢失风险，必须设置约束：
- 1. 关键数据（数字、时间、用户偏好、限定条件）严禁丢失；
- 2. 多条事实不要强行合并为一句话，避免事实混淆；
- 3. 开启压缩日志，记录原始文本与压缩后文本，便于问题回溯；
- 4. 极短文本（低于阈值）直接跳过压缩，防止过度精简失真。
4. LLM记忆压缩封装
该示例实现了一个基于大模型的对话记忆压缩函数，支持轻度和深度两种压缩级别：轻度保留细节仅去闲聊，深度只提取关键结论。通过动态构造Prompt调用LLM对长文本进行摘要，有效降低记忆存储成本并控制上下文窗口。
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]}..."
四、记忆去重策略
1. 区分三类重复场景
很多系统只做完全字符串匹配去重，在语义场景下完全不够。我们把重复划分为三类：
- 完全重复：两段记忆文本几乎一模一样，最简单；
- 语义近似重复：表达方式不同，但描述同一事实（“我喜欢喝咖啡” 和 “咖啡是我的偏好饮品”）；
- 部分重叠：一条记忆包含另一条记忆的子集信息，存在信息覆盖。
针对不同类型，采取不同处理策略，不能一刀切删除。
2. 向量相似度去重流程
去重依托 Embedding 向量检索，标准执行流程：
- 1. 新记忆生成Embedding向量；
- 2. 在记忆库中检索Top-N相似记忆；
- 3. 设定相似度阈值：高于阈值判定为疑似重复；
- 4. 送入判别模块，区分完全重复、近似重复、部分重叠；
- 5. 执行对应合并/丢弃/标记操作。
阈值调优经验：
- 阈值过高：漏检大量语义重复；
- 阈值过低：误将不相关记忆判定重复。 生产环境建议阈值区间：0.80 ~ 0.88，可根据 Embedding 模型微调。
3. 重复记忆处理规则
- 完全重复：直接丢弃新增记忆，更新旧记忆last_access_time；
- 语义近似：两条记忆合并，触发一次记忆压缩，生成统一版本；
- 部分重叠：保留两条记忆，标记关联关系，等待后续更新模块处理；
4. 相似度检索去重实现
该示例实现了一个基于余弦相似度的记忆去重机制：计算新记忆与已有记忆的向量相似度，若超过阈值（0.85）则判定为重复，跳过入库并更新旧记忆的访问时间，避免语义重复内容堆积，保持记忆库的精简与高效。
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
五、记忆更新机制
1. 更新适用场景
随着交互持续进行，用户信息会发生变化：喜好变更、计划调整、纠正之前说错的信息。 典型场景：
- 用户之前说不喝茶，后续表示现在喜欢喝茶；
- 用户修改出行时间、地址；
- 用户纠正历史对话里的错误信息。
此时不能简单新增一条记忆，会造成信息冗余，埋下记忆冲突隐患。需要主动对旧记忆进行版本化更新。
2. 版本化更新设计
记忆对象内置version版本号，更新逻辑规范：
- 1. 检索找到相关历史记忆；
- 2. LLM判断新信息是否覆盖旧事实；
- 3. 若是有效更新：旧记忆标记历史版本（可选软失效），生成新版本记忆；
- 4. 保留历史记录，方便追溯，不直接物理删除；
- 5. 新旧记忆建立关联ID。
两种更新模式：
- 增量更新：旧事实依然成立，新增补充信息；
- 覆盖更新：旧事实失效，新信息替代原有内容。
3. 更新触发方式
- 主动触发：用户明确纠正信息；
- 被动触发：新增记忆去重阶段，模型识别信息更替。
4. 记忆版本更新函数实现
该示例实现了一个记忆的更新机制：不直接覆盖旧记忆，而是将其标记为失效，同时基于旧记忆的标签和来源创建版本号递增的新记忆入库，实现记忆的版本化管理，便于后续追溯历史变更。
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
六、时间衰减与动态遗忘
1. 时间衰减机制说明
人脑不会永久记住所有信息，久远记忆会慢慢淡化。映射到记忆系统：
- 很久没有访问的记忆，重要程度持续下降；
- 经常被检索调用的记忆，衰减放缓；
- 权重持续降低到阈值后，执行遗忘（删除/归档）。
如果没有衰减机制，记忆库会无限膨胀，越来越多过时信息持续参与检索，干扰模型判断。
2. 常用衰减算法模型
行业主流使用指数衰减模型，公式：
- W0：初始权重；
- λ：衰减系数，控制遗忘快慢；
- Δt：距离上次访问的时间间隔。
优化改进方案：每次记忆被召回，重置衰减计时（模拟 “回忆一次，记忆加深”）。 可配置参数：
- λ 越大，遗忘速度越快；短期记忆 λ > 长期记忆 λ；
- 设置最低权重阈值min_weight ，权重低于阈值执行遗忘。
3. 遗忘两种实现策略
- 软遗忘：记忆标记is_valid=False，不再参与检索，物理数据保留，用于数据分析、回溯。
- 硬遗忘：直接删除数据库记录，释放存储，无法恢复，适合存储资源紧张场景。
4. 定时任务调度方案
单独启动后台任务，周期性遍历短期记忆库批量更新权重。 执行间隔建议：5~30 分钟，根据业务并发调整，避免高频遍历造成性能压力。
5. 时间衰减计算实践
该示例实现了一个基于时间衰减的记忆权重衰减函数：利用指数衰减公式，根据距上次访问的时间间隔自动降低记忆权重，并设置最低权重下限防止权重归零，模拟记忆随时间自然遗忘的机制。
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
七、记忆冲突处理方案
1. 记忆冲突定义
当记忆库中存在两条或多条描述同一对象，但事实相互矛盾的信息，即为记忆冲突。 示例：
- 记忆 A：用户周末打算在家休息；
- 记忆 B：用户周末计划外出旅行。
如果不处理冲突，两条记忆同时被召回送入Prompt，模型收到矛盾信息，容易输出混乱答案。
冲突来源：信息更新不及时、用户前后说法变化、信息录入错误。
2. 冲突检测流程
- 1. 新增记忆写入前，检索相似主题记忆；
- 2. 将多条相关记忆送入LLM进行一致性判别；
- 3. 模型输出结论：无冲突/存在冲突/信息不确定。
冲突检测 Prompt 参考：
判断下面两条描述是否事实冲突。 A：{{mem_a}} B：{{mem_b}} 输出：冲突/无冲突/信息不足无法判断
3. 冲突消解策略
冲突消解策略的优先级依次从高到低：
- 时序优先策略：默认采信更新时间更近的记忆；
- 交互确认策略：严重冲突时，向用户发起询问，确认正确信息；
- 权重优先策略：权重更高（频繁访问）的记忆优先采信；
- 共存标注策略：无法自动判定时，两条记忆保留，打上冲突标签，召回时提示模型存在矛盾信息。
生产环境通用组合方案：
- 轻量冲突 → 时序优先自动消解；
- 重大事实冲突 → 标记冲突，条件允许时询问用户。
4. 冲突简易调度框架
该示例实现了一个记忆冲突检测函数：通过两两遍历记忆列表，调用大模型判断任意两条记忆内容是否存在语义冲突，返回冲突记忆对的 ID 组合，为后续冲突消解与记忆一致性维护提供依据。
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
八、记忆召回与实践要点
1. 记忆召回完整流程
记忆治理（压缩、去重、衰减、冲突处理）属于写入侧能力；召回是读取侧能力，二者配合才能生效。 标准召回链路：
- 1. 根据当前用户问题生成Query向量；
- 2. 向量检索获取Top-K相似记忆；
- 3. 过滤无效、低权重记忆；
- 4. 检查召回集合内部是否存在冲突；
- 5. 冲突记忆按规则处理后整理文本；
- 6. 组装进Prompt送入大模型。
不要盲目增加召回数量，过多记忆同样会造成Token浪费。一般Top 3～Top 8为常用区间。
2. 向量数据库选型建议
- 小规模测试：内存向量库、Chroma、FAISS；
- 线上生产：Milvus、Qdrant、PGVector； 所有向量库都需要配合业务代码实现衰减、版本、冲突逻辑，向量数据库仅负责相似度检索，原生不具备记忆生命周期管理。
3. 实践经验总结
- 只存原始对话，不做记忆压缩会导致Token持续膨胀；
- 使用字符串匹配去重，忽略语义重复；
- 缺少时间衰减，记忆库无限膨胀；
- 更新记忆直接新增，不做旧数据标记，持续产生冲突；
- 冲突不处理，矛盾信息同时输入模型；
- 遗忘采用直接删除，丢失调试与溯源数据。
九、总结
经过过才明白，刚开始搭建智能体记忆系统，以为只要搭建向量数据库、实现检索就算完成。真正落地后才会发现，检索只是整套体系中很小一环。一套可用、稳定、长期运行的记忆引擎，核心在于记忆全生命周期治理：从信息提取、压缩精简，到重复信息合并、内容迭代更新，再到模拟人类记忆规律的时间衰减与动态遗忘，最后解决多条信息之间的事实冲突。
记忆的核心机制，本质是让大模型学会筛选记忆、淡化过时信息、修正旧认知。脱离上下文窗口的长期智能体，离不开这套能力。同时，如何让模型自主判断记忆价值、实现自适应衰减系数、复杂多条记忆链式冲突自动推理消解，也是现在项目中遇到的实际问题，先实现现有的基础，在逐步解决项目中的难题。
附录：完整示例参考
整合提取项目中用到的所有模块，包括：数据模型、记忆压缩、相似度去重、版本更新、时间衰减、冲突检测、记忆召回。
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("未检测到冲突")
【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）