X1-1:记忆的生命周期工程
Easy Data x AI 课程 · 扩展篇 · 上篇
给 Agent 加了记忆之后,新问题来了:不该忘的忘了,该忘的赖着不走,矛盾事实同时喂给 LLM。上篇解决这三个问题。
开篇
P3 从产品设计角度讲了 Agent 记忆系统:CoALA 框架怎么分类、艾宾浩斯曲线怎么规划衰减、冲突怎么裁决。D4 带你完成了基础接入,Agent 已经能把对话提炼成记忆条目。
但当你把记忆系统真正跑起来,条目上千、偏好频繁变化、检索开始返回互相矛盾的事实,三个工程问题会准时浮现:
- 衰减速度怎么定? 用户对花生过敏,两个月没被触发检索,保留率跌破阈值后被归档;而"今天喝了咖啡"这类临时信息反而长期占据检索结果。
- 矛盾记忆怎么自动发现? 用户上周说用 Java,这周说用 Go。两条记忆同时躺在库里,LLM 检索到两个答案,信哪个?
- 检索结果怎么保持稳定? 把矛盾记忆全塞进 Prompt 让 LLM 临场判断,同一个问题问两遍,答案可能不同。代码层能不能先做一轮聚合?
上篇把这三个问题串成一条完整的工程链路:写入时定层级 → 存续中算保留率 → 检索时叠加衰减 → 冲突时路由裁决 → 返回前确定性聚合。
目录
第一部分:从遗忘曲线到保留率公式
P3 讲过记、忘、想起三件套。上篇从"忘"入手:不是删掉数据,而是给每条记忆算一个保留率
1.1 为什么需要量化遗忘
如果没有衰减机制,记忆库只会越来越大。三个月前的临时偏好和今天的核心事实权重相同,Agent 会在 Prompt 里同时看到"喜欢详细解释"和"请简洁一点",回答风格飘忽不定。
艾宾浩斯遗忘曲线给出了直觉:记忆强度随时间下降,且下降速度可建模。工程上我们要做的,就是把这条曲线翻译成代码里可计算的分数。
1.2 从经典公式到工程形式
经典形式是指数衰减:
含义很直观:
| 符号 | 含义 |
|---|---|
| 时刻 | |
| 衰减速率常数,越大衰减越慢 | |
| 距上次巩固(或创建)经过的时间 |
在示例代码 x1_1_decay_score_demo.py 中,我们把它改写为更适合按小时计时的形式:
| 符号 | 含义 |
|---|---|
| 距记忆创建经过的小时数 | |
| 记忆的有效强度,由下文三个因子共同决定 | |
| 时间尺度换算:把按小时计的 |
为什么要引入
1.3 强度 的三个因子
示例代码中,强度由三个因子相乘得到:
公式与代码( calculate_strength 方法)逐项对应如下:
| 公式因子 | 符号 | 代码中的写法 | 说明 |
|---|---|---|---|
| 全局衰减速率 | decay_rate | 全库统一参数,默认 1.5; | |
| 层级倍率 | DEFAULT_DECAY_RATE_MULTIPLIERS[memory_type] | 由写入时的层级决定:working=1,short_term=7,long_term=60 | |
| 访问强化 | 1 + math.log1p(access_count) | access_count; |
代入代码就是:S = decay_rate × multiplier × (1 + reinforcement),其中 multiplier 来自层级字典,reinforcement = log1p(access_count)。
对应的具体实现(x1_1_decay_score_demo.py):
# λ:全局衰减速率,默认 1.5,全库统一
DEFAULT_DECAY_RATE = 1.5
# 层级倍率 M_type:写入时分层后,从这里取对应数值
DEFAULT_DECAY_RATE_MULTIPLIERS = {
"working": 1, # 临时层,S 基准最低
"short_term": 7, # 中期层
"long_term": 60, # 长期层,S 可达临时层的 60 倍
}
def calculate_strength(memory_type: str, access_count: int,
decay_rate: float = DEFAULT_DECAY_RATE) -> float:
"""计算有效强度 S = λ × M_type × (1 + log(1 + n))"""
# 因子 1:λ(全局基调)
# 因子 2:M_type(由 memory_type 查表)
multiplier = DEFAULT_DECAY_RATE_MULTIPLIERS[memory_type]
# 因子 3:1 + log(1 + n),n 即 access_count;从未被检索时 n=0,整项为 1
reinforcement = math.log1p(access_count)
return decay_rate * multiplier * (1 + reinforcement)
def calculate_retention(created_at, memory_type, access_count, current_time) -> float:
"""计算保留率 R = e^(-t / (24 × S)),返回值 ∈ [0, 1]"""
# t:距创建经过的小时数
hours_elapsed = (current_time - created_at).total_seconds() / 3600
strength = calculate_strength(memory_type, access_count)
# S 越大,分母 24×S 越大,R 衰减越慢
retention = math.exp(-hours_elapsed / (24 * strength))
return max(0.0, min(1.0, retention)) # 防止浮点误差越界用一个具体场景感受不同的强度 long_term;另一条「今天喝了咖啡」进了 working。一个月后再来检索,前者几乎还能完整召回,后者早已淡出排序。
先看长期层那条。它从未被检索过(
也就是说,一个月后仍保留约七成的「新鲜度」,检索排序里依然靠前。
如果同样内容误进了 working 层呢?
PowerMem 开源记忆系统即采用了类似的艾宾浩斯衰减实现,参数可通过配置文件覆盖。
动手实验 1:把公式跑起来看曲线
上面是纸面推演。脚本 x1_1_decay_score_demo.py 会把 working、short_term、long_term 三层记忆放在同一时间轴上,打印 1 小时到 90 天各节点的保留率,直观看到分层带来的差距;也会演示同一条记忆被多次检索后,
在项目根目录执行:
cd code/X1/part1_decay_conflict
python x1_1_decay_score_demo.py跑完可以对照输出思考两个问题:同样是 30 天后,长期层和临时层的
第二部分:记忆分层与参数调优
第一部分解决了"记忆在库里待久了会怎样衰减"。但衰减快慢在写入那一刻就已经定了:记忆进哪一层,直接决定初始强度
2.1 写入时的一次分流
P3 讲过,入库的不是对话原文,而是从对话里提炼出的事实,并打一个 0~1 的重要性分数。示例代码在写入时做被动巩固:分数一过阈值,层级就定了,后面很难回头改。
# 分层阈值:分数越高,进入越"慢衰减"的层级
LONG_TERM_THRESHOLD = 0.8
SHORT_TERM_THRESHOLD = 0.6
def classify_memory_type(importance_score: float) -> str:
"""根据重要性分数决定 memory_type,进而决定 M_type 取 1 / 7 / 60"""
if importance_score >= LONG_TERM_THRESHOLD:
return "long_term" # 对花生过敏 (0.95) → S 中 M_type=60
if importance_score >= SHORT_TERM_THRESHOLD:
return "short_term" # 正在学 Rust (0.72) → M_type=7
return "working" # 今天喝了咖啡 (0.20) → M_type=1| 重要性分数 | 层级 | 典型内容 | 30 天后大致命运 |
|---|---|---|---|
| ≥ 0.8 | long_term | 过敏、职业、核心偏好 | 保留率仍 > 0.7 |
| 0.6 ~ 0.8 | short_term | 进行中的项目、近期兴趣 | 中等衰减,频繁访问可升级 |
| < 0.6 | working | 临时状态、闲聊 | 数天内自然淘汰 |
示例代码里,importance_score 是事先写进的测试数据,真正上线时,这个 0~1 的分数要在写入前算出来:常见做法是让 LLM 读一遍已提炼的事实,判断它有多"值得长期记住",例如涉及安全(过敏)、稳定身份(职业、核心偏好)通常偏高,随口闲聊(今天喝了咖啡)偏低等。LLM 不可用时,可以回退到关键词(如"过敏""禁用")加内容长度等规则估分,避免分层这一步随 LLM 一起停摆。
被动巩固的局限也在这里:它只看单条记忆。用户连续三周搜 Rust 教程,每条单独打分都不高,合在一起却说明"正在系统学 Rust"。这种跨条目的模式,要留到下篇的 Reflection 机制来处理。
2.2 怎么选:用同一批数据做对比
decay_rate(
注意:这里的
是工程形式 中的 —— 越大,所有记忆衰减越慢。这与经典艾宾浩斯公式 中" 越大衰减越快"的方向相反,是因为我们把 从指数位置移到了强度的分母里。
没有万能默认值,可靠的做法是固定一批真实或模拟记忆,换几个 x1_2_decay_param_ablation.py 用的就是这套思路:
| 30 天后平均保留率 | 误忘风险 | 检索质量 | |
|---|---|---|---|
| 0.5 | 低 | 高 | 中(开始出现误忘) |
| 1.0 | 偏低 | 偏高 | 中 |
| 1.5(默认) | 适中 | 适中 | 好 |
| 3.0 | 高 | 低 | 中(无关旧记忆开始累积) |
| 5.0 | 极高 | 极低 | 差(噪声多,旧偏好赖着不走) |
可以结合具体的业务场景来进行调试:
- 医疗 / 法律:
,漏记比多记代价大,宁可旧信息停留也要保住安全关键事实 - 娱乐 / 社交:
,偏好迭代快,旧偏好早点退场反而干净 - 归档阈值(
低于此值视为遗忘):通用 0.3,保守 0.2,激进 0.4
动手实验 2:看 怎样误伤重要记忆
x1_2_decay_param_ablation.py 会加载 12 条不同层级、不同时间的模拟记忆,依次用
python x1_2_decay_param_ablation.py跑完重点看误忘计数:
第三部分:衰减如何参与检索
保留率
所以
3.1 两阶段检索:先语义,再衰减
最直觉的做法是:用户一问就从全库找语义最相关的记忆,同时把
所以工程上几乎都会拆成两阶段:先用便宜的方式圈一小批候选,再在这批候选上叠加衰减做精排。
阶段 1:粗筛(只看"跟问题有没有关系")
输入是用户 query,输出是一小份候选集。这一步只做语义匹配(向量相似度,或 D2/D3 里的混合检索),不问记忆新旧、也不问
阶段 2:精排(在候选里问"哪条更值得现在拿出来")
候选集已经很小,这时再对每条算
两阶段合在一起,最终得分是:
| 公式因子 | 代码变量 | 在流程中的位置 |
|---|---|---|
| 语义相关度 | semantic_score | 阶段 1 产出;决定能否进候选集 |
| 保留率 | 阶段 2 乘入;越旧、越少被访问,得分越低 | |
| 遗忘惩罚 | 阶段 2 额外乘 0.1; |
举个例子。用户问"推荐后端框架",阶段 1 可能同时圈进"用户主力语言 Go"和"用户三个月前说正在学 Rust"两条,因为都和后端、语言相关。进入阶段 2 后,前者若是 long_term 且
对应实现(x1_3_retrieval_with_decay.py):
ARCHIVE_THRESHOLD = 0.3 # R 低于此值视为"已遗忘"
FORGOTTEN_SCORE_MULTIPLIER = 0.1 # 已遗忘记忆在阶段 2 再 ×0.1
def two_stage_retrieval(memories, query, quality_threshold=0.15):
"""阶段 1 粗筛语义,阶段 2 在候选集上叠加衰减"""
candidates = []
for mem in memories:
# --- 阶段 1:只算 semantic_score,太弱的不进候选 ---
sem_score = embedding_similarity(mem["content"], query)
if sem_score < quality_threshold:
continue
# --- 阶段 2:只对候选算 R,合成 final_score ---
retention = calculate_retention(
mem["created_at"], mem["memory_type"], mem["access_count"]
)
forgotten_mult = FORGOTTEN_SCORE_MULTIPLIER if retention < ARCHIVE_THRESHOLD else 1.0
final_score = sem_score * retention * forgotten_mult
candidates.append({**mem, "final_score": final_score})
candidates.sort(key=lambda m: m["final_score"], reverse=True)
return candidates上面 Go / Rust 的例子,体现的正是"该忘的不走":旧 working 笔记语义仍相关,但
| 偏差 | 典型表现 | 根因 | 对应环节 | 缓解 |
|---|---|---|---|---|
| 沉默死亡 | 过敏信息两个月没被问到, | 高重要性 ≠ 高访问频率;long_term 也会随时间下沉 | 写入分层 + 存续 | 安全关键信息 importance ≥ 0.9,必要时调低 |
| 写入即巅峰 | 刚写入的临时笔记总在 Top-1 | 新记忆 | 检索精排 | 写入时分层把关 + 访问强化让高频记忆回血 |
| 与冲突脱节 | Java 旧偏好降权但仍 active,Prompt 里和 Go 并存 | 衰减只管"多旧",不管"是否已被新版本取代" | 写入裁决(第四部分) | 新事实写入时同步失效旧版 |
三者里,前两个仍发生在衰减链路内部,动手实验 3 的输出里看得见端倪;第三个则说明衰减退远旧信息,解决不了语义矛盾,必须靠第四部分的冲突管线。这也是为什么开篇三个工程问题不能只用一套公式打完。
动手实验 3:看 Top-5 怎样被衰减改写
x1_3_retrieval_with_decay.py 对同一 query 分别跑"只按语义排序"和"语义 × 衰减排序",并排打印 Top-5,你能直接看到哪些旧 working 记忆被
python x1_3_retrieval_with_decay.py对照输出可以逐项核对上表:有没有 working 层旧笔记因
跑通实验后,第三部分的衰减链路可以概括为:分层定
第四部分:冲突检测与裁决管线
第三部分末尾提到的与冲突脱节,正是从这里开始补:衰减只能让旧信息在排序里退远,不能告诉 Agent"Java 已被 Go 取代"。用户上周说主力语言 Java,这周说已全面转向 Go,两条记忆语义相近、内容矛盾,如果只等
第四部分回答开篇第二个工程问题:矛盾记忆怎么自动发现、怎么裁决。
4.1 从相似度到路由决策
P3 把冲突分成直接矛盾、时间演进、细化补充、并存偏好四种。写入管线里可以落成两步,示例代码对应 x1_4_conflict_detect.py:
步骤 1:相似度粗筛。新事实和存量记忆算 embedding 相似度,只有 ≥ 阈值的才进候选,避免每条新事实都触发 LLM。
| 参数 | 推荐值 | 设太高 | 设太低 |
|---|---|---|---|
similarity_threshold | 0.25 ~ 0.35 | 措辞差异大时漏检 | 无关记忆大量进候选,成本高 |
步骤 2:路由。对相似度最高的候选,决定 ADD / UPDATE / DELETE:
| 相似度区间 | 典型场景 | 动作 |
|---|---|---|
| ≥ 0.85 + 否定词 | "讨厌披萨" vs "喜欢披萨" | DELETE |
| ≥ 0.7 + 迁移词 | "已从 Java 转向 Go" | UPDATE |
| ≥ 0.7 无矛盾 | "学了 Rust" + "用 Axum 实践" | UPDATE(细化) |
| 0.5 ~ 0.7 | "喜欢意大利菜" + "最近在吃素" | ADD(并存) |
| < 0.5 | 弱相关 | ADD(独立新信息) |
示例代码用规则演示这套路由;线上通常把步骤 2 交给 LLM,因为"喜欢意大利菜"和"最近在吃素"并存,规则很容易误判成矛盾。
def detect_conflict(new_fact, existing_memories, similarity_threshold=0.3):
"""步骤 1 粗筛候选,步骤 2 对相似度最高的候选做路由"""
# --- 步骤 1:只保留 active 且相似度达标的记忆 ---
candidates = []
for mem in existing_memories:
if not mem["is_active"]:
continue
sim = embedding_similarity(new_fact, mem["content"])
if sim >= similarity_threshold:
candidates.append({**mem, "similarity": sim})
if not candidates:
return {"action": "ADD", "reason": "无相似记忆,独立新信息"}
top = max(candidates, key=lambda m: m["similarity"])
# --- 步骤 2:按相似度区间 + 关键词特征路由(见上表)---
if top["similarity"] >= 0.7:
has_negation = any(kw in new_fact for kw in ["不", "讨厌", "不喜欢"])
if has_negation and top["similarity"] >= 0.85:
return {"action": "DELETE", "target_id": top["id"],
"reason": "直接矛盾,旧记忆标记遗忘"}
# 迁移(Java→Go)或细化补充(学了 Rust + 用 Axum)均走 UPDATE
return {"action": "UPDATE", "target_id": top["id"],
"reason": "时间演进或细化补充,旧版下线新版写入"}
if top["similarity"] >= 0.5:
return {"action": "ADD", "reason": "相关但不矛盾,并存保留"}
return {"action": "ADD", "reason": "弱相关,作为独立新信息写入"}4.2 裁决执行:留版本链,不物理删除
检测到冲突后,x1_5_conflict_resolve.py 演示怎么改库。共同原则:数据不硬删,状态用字段标记,方便审计和回滚。
def resolve_conflict(memories_store, new_fact, action, target_id):
"""执行 ADD / UPDATE / DELETE,均保留历史记录"""
if action == "UPDATE" and target_id:
# 旧版下线,新版上线,replaces 串起版本链
for mem in memories_store:
if mem["id"] == target_id:
mem["is_active"] = False
break
memories_store.append({
"content": new_fact,
"is_active": True,
"replaces": target_id, # 可追溯:这条取代了谁
})
elif action == "DELETE" and target_id:
# 不删行,标记遗忘;检索时再 ×0.1,用户纠正后可恢复
for mem in memories_store:
if mem["id"] == target_id:
mem["is_active"] = False
mem["should_forget"] = True
breakJava → Go 的迁移走 UPDATE:旧版 is_active=False 留在库里,新版带 replaces 指向旧 id,日后可以回答"AI 为什么不推荐 Java 了"。直接否定(喜欢披萨 → 讨厌披萨)走 DELETE:旧版标记遗忘降权,而不是从磁盘抹掉,用户改口时还有恢复空间。
4.3 确定性聚合:检索之后、进 Prompt 之前
即便裁决正确,检索仍可能一次拉回多条记忆。全部塞进 Prompt 让 LLM 挑,同一问题问两遍,答案可能不同。x1_6_retrieval_aggregate.py 在代码层先做一轮合并:
def deterministic_aggregate(candidates):
"""同一话题只留 active 版本;多版本并存时才交给 LLM"""
topics = group_by_topic(candidates)
aggregated = []
for topic_name, group in topics.items():
active_mems = [m for m in group if m.get("is_active", True)]
if len(active_mems) == 1:
aggregated.append(active_mems[0]) # 单版本:代码直接定,结果可复现
elif len(active_mems) > 1:
aggregated.extend(active_mems) # 并存偏好:上下文完整交给 LLM
# is_active=False 的旧版不进入 Prompt,省 Token
return aggregated话题里只有一个 active 版本(比如当前主力语言 Go),代码层直接返回,省 Token,也保证两次查询结果一致。话题里多个 active 并存(意大利菜 + 近期吃素),才保留完整上下文,让 LLM 做综合判断。
动手实验 4~6:走一遍冲突管线
三个脚本建议按写入顺序跑,对应"检测 → 改库 → 检索后聚合":
python x1_4_conflict_detect.py
python x1_5_conflict_resolve.py
python x1_6_retrieval_aggregate.pyx1_4 会打印每条测试事实的路由动作,重点看"喜欢意大利菜 + 最近在吃素"是不是 ADD 而非 DELETE。x1_5 改库后查 is_active 和 replaces,确认 Java 旧版还在、只是不再返回。x1_6 对比聚合前后 Token 估算,体会代码层先合并能省多少上下文。
总结与参考资料
把四部分串起来,单 Agent 记忆的生命周期是这样运转的:
- 写入:重要性打分 →
classify_memory_type定层级 → 初始确定 - 存续:时间推移 + 访问次数 → 算
→ 低于阈值标记遗忘 - 检索:语义粗筛 →
参与精排 → 确定性聚合后再进 Prompt - 冲突:相似粗筛 → 路由裁决 → 版本链保留,旧版不硬删
开篇的三个坑,在这里各有一条落点:衰减靠
参考资料:
- P3 Agent 记忆系统设计
- D4 Agent 开发与记忆系统
- LOCOMO Benchmark — github.com/Shopify/locomo
