⚠️ Alpha内测版本警告:此为早期内部构建版本,尚不完整且可能存在错误,欢迎大家提Issue反馈问题或建议。
跳转到正文

X1-1:记忆的生命周期工程

Easy Data x AI 课程 · 扩展篇 · 上篇

给 Agent 加了记忆之后,新问题来了:不该忘的忘了,该忘的赖着不走,矛盾事实同时喂给 LLM。上篇解决这三个问题。

开篇

P3 从产品设计角度讲了 Agent 记忆系统:CoALA 框架怎么分类、艾宾浩斯曲线怎么规划衰减、冲突怎么裁决。D4 带你完成了基础接入,Agent 已经能把对话提炼成记忆条目。

但当你把记忆系统真正跑起来,条目上千、偏好频繁变化、检索开始返回互相矛盾的事实,三个工程问题会准时浮现:

  1. 衰减速度怎么定? 用户对花生过敏,两个月没被触发检索,保留率跌破阈值后被归档;而"今天喝了咖啡"这类临时信息反而长期占据检索结果。
  2. 矛盾记忆怎么自动发现? 用户上周说用 Java,这周说用 Go。两条记忆同时躺在库里,LLM 检索到两个答案,信哪个?
  3. 检索结果怎么保持稳定? 把矛盾记忆全塞进 Prompt 让 LLM 临场判断,同一个问题问两遍,答案可能不同。代码层能不能先做一轮聚合?

上篇把这三个问题串成一条完整的工程链路:写入时定层级 → 存续中算保留率 → 检索时叠加衰减 → 冲突时路由裁决 → 返回前确定性聚合

目录

第一部分:从遗忘曲线到保留率公式

P3 讲过记、忘、想起三件套。上篇从"忘"入手:不是删掉数据,而是给每条记忆算一个保留率 R,让旧信息在检索排序中自然退居幕后。

1.1 为什么需要量化遗忘

如果没有衰减机制,记忆库只会越来越大。三个月前的临时偏好和今天的核心事实权重相同,Agent 会在 Prompt 里同时看到"喜欢详细解释"和"请简洁一点",回答风格飘忽不定。

艾宾浩斯遗忘曲线给出了直觉:记忆强度随时间下降,且下降速度可建模。工程上我们要做的,就是把这条曲线翻译成代码里可计算的分数。

1.2 从经典公式到工程形式

经典形式是指数衰减:

R(t)=eλt

含义很直观:

符号含义
R(t)时刻 t 的保留率,取值 [0,1],1 表示完全保留
λ衰减速率常数,越大衰减越慢
t距上次巩固(或创建)经过的时间

在示例代码 x1_1_decay_score_demo.py 中,我们把它改写为更适合按小时计时的形式:

R=et/(24×S)
符号含义
t距记忆创建经过的小时数
S记忆的有效强度,由下文三个因子共同决定
24时间尺度换算:把按小时计的 t 映射到按天衰减的直觉

为什么要引入 S 经典公式里只有一个全局 λ,所有记忆衰减速度相同。实际系统中,"对花生过敏"和"今天喝了咖啡"显然不该以同样速度被遗忘。S 就是给每条记忆单独调节衰减快慢的旋钮。

1.3 强度 S 的三个因子

示例代码中,强度由三个因子相乘得到:

S=λ×Mtype×(1+log(1+n))

公式与代码( calculate_strength 方法)逐项对应如下:

公式因子符号代码中的写法说明
全局衰减速率λdecay_rate全库统一参数,默认 1.5;λ 越大,所有记忆衰减越慢
层级倍率MtypeDEFAULT_DECAY_RATE_MULTIPLIERS[memory_type]由写入时的层级决定:working=1,short_term=7,long_term=60
访问强化1+log(1+n)1 + math.log1p(access_count)n 为历史检索次数 access_countn=0 时该项为 1(无强化),n 越大 S 越大,但增幅递减

代入代码就是:S = decay_rate × multiplier × (1 + reinforcement),其中 multiplier 来自层级字典,reinforcement = log1p(access_count)

对应的具体实现(x1_1_decay_score_demo.py):

python
# λ:全局衰减速率,默认 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))  # 防止浮点误差越界

用一个具体场景感受不同的强度 S 算出来的保留率 R 的差别。假设同一条事实「用户对花生过敏」写入了记忆库,写入时被判定为 long_term;另一条「今天喝了咖啡」进了 working。一个月后再来检索,前者几乎还能完整召回,后者早已淡出排序。

先看长期层那条。它从未被检索过(n=0),访问强化项为 1,于是 S=1.5×60×1=90。过了 30 天,t=720 小时,代入保留率公式:

R=e720/(24×90)0.72

也就是说,一个月后仍保留约七成的「新鲜度」,检索排序里依然靠前。

如果同样内容误进了 working 层呢?S 只有 1.5×1×1=1.5,同样是 30 天后,R0.0003,几乎已从检索结果中消失。写入时分层,本质上是在决定这条记忆能「活」多久;层级倍率 60 不是抽象参数,而是长期事实和临时闲聊之间数量级的差距。

PowerMem 开源记忆系统即采用了类似的艾宾浩斯衰减实现,参数可通过配置文件覆盖。

动手实验 1:把公式跑起来看曲线

上面是纸面推演。脚本 x1_1_decay_score_demo.py 会把 working、short_term、long_term 三层记忆放在同一时间轴上,打印 1 小时到 90 天各节点的保留率,直观看到分层带来的差距;也会演示同一条记忆被多次检索后,S 如何变大、曲线如何被「拉平」。

在项目根目录执行:

bash
cd code/X1/part1_decay_conflict
python x1_1_decay_score_demo.py

跑完可以对照输出思考两个问题:同样是 30 天后,长期层和临时层的 R 差了几个数量级?如果一条中期记忆被用户反复问到,它的保留率曲线和「写进长期层」相比,谁更划算?

第二部分:记忆分层与参数调优

第一部分解决了"记忆在库里待久了会怎样衰减"。但衰减快慢在写入那一刻就已经定了:记忆进哪一层,直接决定初始强度 S 有多大。回到开篇的例子,"对花生过敏"和"今天喝了咖啡"如果分层不同,一个月后的命运天差地别。分层就是在这条生命周期曲线的起点做分流。

2.1 写入时的一次分流

P3 讲过,入库的不是对话原文,而是从对话里提炼出的事实,并打一个 0~1 的重要性分数。示例代码在写入时做被动巩固:分数一过阈值,层级就定了,后面很难回头改。

python
# 分层阈值:分数越高,进入越"慢衰减"的层级
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.8long_term过敏、职业、核心偏好保留率仍 > 0.7
0.6 ~ 0.8short_term进行中的项目、近期兴趣中等衰减,频繁访问可升级
< 0.6working临时状态、闲聊数天内自然淘汰

示例代码里,importance_score 是事先写进的测试数据,真正上线时,这个 0~1 的分数要在写入前算出来:常见做法是让 LLM 读一遍已提炼的事实,判断它有多"值得长期记住",例如涉及安全(过敏)、稳定身份(职业、核心偏好)通常偏高,随口闲聊(今天喝了咖啡)偏低等。LLM 不可用时,可以回退到关键词(如"过敏""禁用")加内容长度等规则估分,避免分层这一步随 LLM 一起停摆。

被动巩固的局限也在这里:它只看单条记忆。用户连续三周搜 Rust 教程,每条单独打分都不高,合在一起却说明"正在系统学 Rust"。这种跨条目的模式,要留到下篇的 Reflection 机制来处理。

2.2 λ 怎么选:用同一批数据做对比

decay_rateλ)控制的是全库的衰减基调,参考默认值 1.5。设太大,旧偏好赖着不走;设太小,无关记忆占满检索结果;设得再小一点,还可能把该留的安全信息一并冲掉。

注意:这里的 λ 是工程形式 R=et/(24S), S=λMtype(1+log(1+n)) 中的 λ——λ 越大,所有记忆衰减越慢。这与经典艾宾浩斯公式 R=eλt 中"λ 越大衰减越快"的方向相反,是因为我们把 λ 从指数位置移到了强度的分母里。

没有万能默认值,可靠的做法是固定一批真实或模拟记忆,换几个 λ 跑一遍,看误忘和噪声哪边先失控。脚本 x1_2_decay_param_ablation.py 用的就是这套思路:

λ30 天后平均保留率误忘风险检索质量
0.5中(开始出现误忘)
1.0偏低偏高
1.5(默认)适中适中
3.0中(无关旧记忆开始累积)
5.0极高极低差(噪声多,旧偏好赖着不走)

可以结合具体的业务场景来进行调试:

  • 医疗 / 法律λ3.0,漏记比多记代价大,宁可旧信息停留也要保住安全关键事实
  • 娱乐 / 社交λ1.0,偏好迭代快,旧偏好早点退场反而干净
  • 归档阈值R 低于此值视为遗忘):通用 0.3,保守 0.2,激进 0.4

动手实验 2:看 λ 怎样误伤重要记忆

x1_2_decay_param_ablation.py 会加载 12 条不同层级、不同时间的模拟记忆,依次用 λ=0.55.0 计算保留率,并统计"高重要性却被判遗忘"的条数(误忘)。

bash
python x1_2_decay_param_ablation.py

跑完重点看误忘计数:λ 从 1.5 调到 0.5 时,有没有原本 importance ≥ 0.8 的记忆跌破 0.3?如果有,你的场景可能更适合保守一点的 λ(调大到 3.0),或者把归档阈值从 0.3 降到 0.2。

第三部分:衰减如何参与检索

保留率 R 如果只写在数据库里、检索排序完全不看它,衰减就停在"记账":库里明明标记某条旧偏好已接近遗忘,用户一问,Agent 仍按纯语义相似度把它捞进 Top-5,和刚写入时没什么两样。对用户来说,记忆系统并没有"越用越准、旧的不打扰",第一、二部分的工作等于白做。

所以 R 必须进入每次查询的排序公式,让"该淡出的旧记忆"在精排阶段自然沉底。开篇第一个工程问题"该忘的不走",主要靠下面这套两阶段检索来解决;但"不该忘的忘了"以及"新旧偏好打架",单靠衰减还不够,后面还会分别碰到。

3.1 两阶段检索:先语义,再衰减

最直觉的做法是:用户一问就从全库找语义最相关的记忆,同时把 R 乘进得分。记忆库只有几百条时没问题;一旦涨到百万级,每条都要算 R、再全量排序,向量索引本来毫秒级能完成的粗筛,会被拖成秒级扫描。

所以工程上几乎都会拆成两阶段:先用便宜的方式圈一小批候选,再在这批候选上叠加衰减做精排。

阶段 1:粗筛(只看"跟问题有没有关系")

输入是用户 query,输出是一小份候选集。这一步只做语义匹配(向量相似度,或 D2/D3 里的混合检索),不问记忆新旧、也不问 R 多少。语义太弱的直接丢掉,通常能从百万条压到几十或几百条。

阶段 2:精排(在候选里问"哪条更值得现在拿出来")

候选集已经很小,这时再对每条算 R,合成最终得分并排序。旧偏好、临时笔记因为 R 低,会在这一步自然沉底;长期事实、最近常被问到的记忆,因为 R 高或访问强化过,会排到前面。

两阶段合在一起,最终得分是:

final_score=semantic_score×R×Mforgotten
公式因子代码变量在流程中的位置
语义相关度semantic_score阶段 1 产出;决定能否进候选集
保留率R阶段 2 乘入;越旧、越少被访问,得分越低
遗忘惩罚Mforgotten阶段 2 额外乘 0.1;R 已低于归档阈值时使用,进一步降权但不删数据

举个例子。用户问"推荐后端框架",阶段 1 可能同时圈进"用户主力语言 Go"和"用户三个月前说正在学 Rust"两条,因为都和后端、语言相关。进入阶段 2 后,前者若是 long_term 且 R 仍高,得分靠前;后者若是 working 层、三个月未再被问到、R 已接近 0,乘进去后排到很后面,通常进不了 Top-5。Agent 不必在 Prompt 里同时看到两条再让 LLM 猜哪个更可信。

对应实现(x1_3_retrieval_with_decay.py):

python
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 笔记语义仍相关,但 R 把它压出 Top-5。不过两阶段检索搭好,不等于衰减链路就没有死角。写入分层、存续算 R、检索精排这三步如果各管一段、彼此不衔接,上线后常见下面三类偏差:

偏差典型表现根因对应环节缓解
沉默死亡过敏信息两个月没被问到,R 跌破 0.3高重要性 ≠ 高访问频率;long_term 也会随时间下沉写入分层 + 存续安全关键信息 importance ≥ 0.9,必要时调低 λ
写入即巅峰刚写入的临时笔记总在 Top-1新记忆 R=1.0,旧记忆已被衰减检索精排写入时分层把关 + 访问强化让高频记忆回血
与冲突脱节Java 旧偏好降权但仍 active,Prompt 里和 Go 并存衰减只管"多旧",不管"是否已被新版本取代"写入裁决(第四部分)新事实写入时同步失效旧版

三者里,前两个仍发生在衰减链路内部,动手实验 3 的输出里看得见端倪;第三个则说明衰减退远旧信息,解决不了语义矛盾,必须靠第四部分的冲突管线。这也是为什么开篇三个工程问题不能只用一套公式打完。

动手实验 3:看 Top-5 怎样被衰减改写

x1_3_retrieval_with_decay.py 对同一 query 分别跑"只按语义排序"和"语义 × 衰减排序",并排打印 Top-5,你能直接看到哪些旧 working 记忆被 R 挤出榜单、哪些因访问强化留在前面。

bash
python x1_3_retrieval_with_decay.py

对照输出可以逐项核对上表:有没有 working 层旧笔记因 R 低被挤出 Top-5(说明精排在工作)?有没有 long_term 过敏信息久未触发也掉出榜单(沉默死亡)?如果语义排序和衰减排序的 Top-1 换了人,是不是新写入的记忆 R=1.0 在抢位置(写入即巅峰)?

跑通实验后,第三部分的衰减链路可以概括为:分层定 S → 时间推移算 R → 两阶段检索把 R 乘进得分。剩下"新旧矛盾同时进 Prompt"的问题,交给第四部分。

第四部分:冲突检测与裁决管线

第三部分末尾提到的与冲突脱节,正是从这里开始补:衰减只能让旧信息在排序里退远,不能告诉 Agent"Java 已被 Go 取代"。用户上周说主力语言 Java,这周说已全面转向 Go,两条记忆语义相近、内容矛盾,如果只等 R 慢慢降权,Prompt 里会长期并存两个答案。

第四部分回答开篇第二个工程问题:矛盾记忆怎么自动发现、怎么裁决。

4.1 从相似度到路由决策

P3 把冲突分成直接矛盾、时间演进、细化补充、并存偏好四种。写入管线里可以落成两步,示例代码对应 x1_4_conflict_detect.py

步骤 1:相似度粗筛。新事实和存量记忆算 embedding 相似度,只有 ≥ 阈值的才进候选,避免每条新事实都触发 LLM。

参数推荐值设太高设太低
similarity_threshold0.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,因为"喜欢意大利菜"和"最近在吃素"并存,规则很容易误判成矛盾。

python
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 演示怎么改库。共同原则:数据不硬删,状态用字段标记,方便审计和回滚。

python
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
                break

Java → Go 的迁移走 UPDATE:旧版 is_active=False 留在库里,新版带 replaces 指向旧 id,日后可以回答"AI 为什么不推荐 Java 了"。直接否定(喜欢披萨 → 讨厌披萨)走 DELETE:旧版标记遗忘降权,而不是从磁盘抹掉,用户改口时还有恢复空间。

4.3 确定性聚合:检索之后、进 Prompt 之前

即便裁决正确,检索仍可能一次拉回多条记忆。全部塞进 Prompt 让 LLM 挑,同一问题问两遍,答案可能不同。x1_6_retrieval_aggregate.py 在代码层先做一轮合并:

python
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:走一遍冲突管线

三个脚本建议按写入顺序跑,对应"检测 → 改库 → 检索后聚合":

bash
python x1_4_conflict_detect.py
python x1_5_conflict_resolve.py
python x1_6_retrieval_aggregate.py

x1_4 会打印每条测试事实的路由动作,重点看"喜欢意大利菜 + 最近在吃素"是不是 ADD 而非 DELETE。x1_5 改库后查 is_activereplaces,确认 Java 旧版还在、只是不再返回。x1_6 对比聚合前后 Token 估算,体会代码层先合并能省多少上下文。


总结与参考资料

把四部分串起来,单 Agent 记忆的生命周期是这样运转的:

  1. 写入:重要性打分 → classify_memory_type 定层级 → 初始 S 确定
  2. 存续:时间推移 + 访问次数 → 算 R → 低于阈值标记遗忘
  3. 检索:语义粗筛 → R 参与精排 → 确定性聚合后再进 Prompt
  4. 冲突:相似粗筛 → 路由裁决 → 版本链保留,旧版不硬删

开篇的三个坑,在这里各有一条落点:衰减靠 Sλ 调优,矛盾靠写入时裁决,检索稳定性靠两阶段排序和聚合。中篇会把视角拉到多 Agent:同一个人的记忆,在不同工具之间该共享还是隔离。

参考资料:

  1. P3 Agent 记忆系统设计
  2. D4 Agent 开发与记忆系统
  3. LOCOMO Benchmark — github.com/Shopify/locomo

Built with VitePress | GitHub 仓库