第6章：Memory 操作管线 —— Extraction 到 Forgetting 全流程

把 Memory 当成一个完整的数据系统来看,它本质上是一条 ETL 管线:raw 数据 → 抽取事实 → 写入并去重 → 巩固成高层知识 → 周期性反思 → 衰减/遗忘 → 检索给 Agent。本章把这 6 个核心算子拆开讲透,串起 Mem0 / Letta / Zep 的真实 API 调用、代码实现和工程坑点。读完之后,你能自己设计一套 Memory 管线,而不是黑箱使用框架。

📑 目录

1. Memory 管线全景
2. Extraction:从 raw 到 fact
3. Writing & Update:去重、合并、冲突
4. Consolidation:Episodic → Semantic
5. Reflection:Agent 自我总结
6. Forgetting & Decay
7. Retrieval:多因子打分
8. AgeMem / RL 驱动的 memory tool calling(2025+)
自我检验清单
参考资料

1. Memory 管线全景

   ┌──────────────────────────────────────────────────────────┐
   │  Raw 输入                                                │
   │  - User 消息 / Assistant 回复                              │
   │  - Tool call results                                     │
   │  - 上传的文档 / 浏览的网页                                  │
   └──────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┴───────────────┐
              ▼                               ▼
   ┌──────────────────────┐      ┌────────────────────┐
   │ 1. Extraction         │      │ Direct Episodic    │
   │  LLM / Schema-driven  │      │ (原始事件,不抽取)    │
   └──────────────────────┘      └────────────────────┘
              │                               │
              ▼                               ▼
   ┌──────────────────────────────────────────────────┐
   │ 2. Writing & Update                              │
   │  Dedup / Merge / Conflict resolution             │
   └──────────────────────────────────────────────────┘
              │
              ├──────────────┐
              ▼              ▼
   ┌──────────────────┐  ┌────────────────────┐
   │ 3. Consolidation │  │ 4. Reflection       │
   │ Episodic→Semantic │  │ High-level summary  │
   └──────────────────┘  └────────────────────┘
              │              │
              └──────┬───────┘
                     ▼
         ┌──────────────────────┐
         │ 5. Decay / Forgetting │
         │ Importance + Recency  │
         └──────────────────────┘
                     │
                     ▼
         ┌──────────────────────┐
         │ 6. Retrieval          │
         │ 多因子打分 + filter    │
         └──────────────────────┘
                     │
                     ▼
              Agent's prompt

每一个算子都是独立的设计决策点——可以用规则、可以用 LLM、可以用统计模型,各有取舍。

2. Extraction:从 raw 到 fact

2.1 两种范式

范式	实现	优点	缺点
LLM 抽取	把对话喂给 LLM,让它输出 JSON facts	灵活、覆盖广	慢、贵、不稳定
Schema-driven	预定义 schema,LLM 填空 / 规则匹配	快、可控	漏掉 schema 外的信息

2.2 LLM 抽取示例(Mem0 风格)

EXTRACTION_PROMPT = """
你是一个事实抽取助手。从下面的对话中,抽取所有关于"用户"的事实。
返回 JSON,每条事实是一个对象:
  - fact: 事实陈述
  - category: preference / personal_info / event / opinion
  - confidence: 0.0-1.0

对话:
{conversation}
"""

def extract_facts(conversation: list[dict]) -> list[dict]:
    response = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": EXTRACTION_PROMPT.format(
                conversation=format_conv(conversation))},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)["facts"]

2.3 Schema-driven(LangMem 风格)

from pydantic import BaseModel
from langmem import create_memory_manager

class FoodPreference(BaseModel):
    food: str
    preference: Literal["love", "like", "neutral", "dislike", "allergic"]
    reason: Optional[str] = None

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet",
    instructions="抽取用户的食物偏好。",
    schemas=[FoodPreference],  # ← 强类型约束
)

result = manager.invoke({"messages": conversation_messages})
# result: List[FoodPreference],schema 严格保证

2.4 工程坑点

抽取频率:每条消息抽一次贵,通常每个 turn 结束抽一次或每 N 个 turn 批量抽
抽取范围:不要抽”今天天气真好”这种闲聊——用户标签先过滤
抽取者 ≠ 对话者:用一个便宜的小 LLM(GPT-4o-mini / Claude Haiku)做 extraction,贵 LLM 用于对话
错误抽取的代价:错的事实写进 memory,以后会污染所有相关检索——宁可漏抽,不要错抽

3. Writing & Update:去重、合并、冲突

新事实进来,有四种结果:

Action	何时	例子
ADD	新事实,无相关已存	第一次说”我住北京”
UPDATE	与已存事实冲突或补充	已存”住上海”,新说”搬到北京” → 更新
DELETE	已存事实被显式撤回	”我之前说错了,我不是素食主义者”
NOOP	已经存在等价事实	重复说”我喜欢咖啡” → 不动

3.1 Mem0 的实现(简化版)

def add_with_resolution(new_fact: str, user_id: str):
    # 1. 找相关已存 memory
    similar = vector_store.search(
        query=new_fact, user_id=user_id, limit=5,
        score_threshold=0.7,
    )
    
    if not similar:
        # 没有相关,直接 ADD
        return _write(new_fact, action="ADD")
    
    # 2. LLM 判断 action
    decision = llm.judge(
        prompt=f"""
        新事实: {new_fact}
        相关已存事实:
        {format(similar)}
        
        请判断对每条已存事实的 action:
        - ADD: 新事实独立保留
        - UPDATE id: 用新事实更新
        - DELETE id: 删除已存
        - NOOP: 等价不动
        
        返回 JSON: {{"action": ..., "target_id": ...}}
        """
    )
    
    # 3. 执行
    if decision["action"] == "ADD":
        _write(new_fact, action="ADD")
    elif decision["action"] == "UPDATE":
        _update(decision["target_id"], new_fact)
    elif decision["action"] == "DELETE":
        _delete(decision["target_id"])
    # NOOP 什么都不做

3.2 冲突解决的几条策略

策略	何时
Latest wins	用户偏好类(地址、口味)
Source confidence	来自 tool call 的 fact > 来自闲聊的 fact
Bi-temporal	保留所有版本,t_invalid 标记
LLM judge	复杂场景,让 LLM 决定保留哪条
保留多源,标记冲突	重要场景(医疗、金融)不擅自删除

3.3 工程要点

写入是异步的:用户消息回复要快(< 2 秒),memory 写入可以放后台 task queue
写入失败要可恢复:用 outbox pattern,raw 消息先入库,extraction job 失败可以 retry
批量 vs 实时:实时性不强的场景(每天总结),用 batch job 更便宜

4. Consolidation:Episodic → Semantic

4.1 什么是 Consolidation

把多条具体事件抽象成一条语义事实——类似人类睡眠时把短期记忆”巩固”为长期记忆。

例子:

Episodic:
  2026-04-15 用户搜索"无糖咖啡"
  2026-04-22 用户点了"美式不加糖"
  2026-04-29 用户问"哪家咖啡馆有 0 糖选项"
                ↓ Consolidation
Semantic:
  用户偏好:"咖啡不加糖"(置信度 0.92)

4.2 触发时机

数量阈值:相似 episodic 累积到 N 条
时间窗口:每天/每周离线 batch
显式触发:用户主动说”以后都按这个偏好来”
Reflection 触发(下一节):agent 自我反思时顺带 consolidate

4.3 实现

def consolidate_periodic(user_id: str, time_window_days: int = 7):
    # 1. 拿最近 N 天的 episodic memory
    recent_episodes = memory.get_episodic(
        user_id=user_id,
        since=datetime.now() - timedelta(days=time_window_days),
    )
    
    # 2. 按 cluster 分组(语义聚类)
    clusters = cluster_by_embedding(recent_episodes, n_clusters=10)
    
    # 3. 每个 cluster 让 LLM 抽 semantic
    for cluster in clusters:
        semantic = llm.summarize_to_semantic(
            episodes=cluster,
            instruction="如果这些事件能抽象成一条用户偏好/属性,请输出;否则输出 null"
        )
        if semantic:
            memory.add_semantic(semantic, source_episodes=[e.id for e in cluster])

4.4 工程坑点

过度 consolidation 的危险:LLM 可能编造不存在的偏好——保留 source 链接 以便溯源和回滚
Consolidation ≠ Replacement:semantic 入库后,对应的 episodic 不应删除(可能以后需要 audit)
置信度阈值:置信度低的 semantic 不该直接进检索池,应该先标记”待验证”

5. Reflection:Agent 自我总结

5.1 什么是 Reflection

Reflection 是让 LLM 周期性回顾自己的 memory,产生更高层结论——Generative Agents 论文最经典的设计。

Memory Stream:
  - 用户每周三都问周末安排
  - 用户对"户外活动"反应积极
  - 用户对"室内活动"反应一般
        ↓ Reflection
  "用户重视周末规划,且偏好户外活动 —— 周二可以主动准备户外建议"

5.2 算法(Generative Agents 风格)

def reflect(memory_stream: list[Memory], agent_id: str):
    # 1. 取最近 importance 累计高的 memory
    recent = memory_stream.recent(n=100)
    if sum(m.importance for m in recent) < THRESHOLD:
        return  # 不到反思时刻
    
    # 2. LLM 自问 5 个 high-level 问题
    questions = llm.generate(
        prompt=f"""
        以下是关于 user 的最近 100 条 memory:
        {format(recent)}
        
        生成 5 个最高层的问题,这些问题的答案能从上述 memory 中推断出来。
        """,
        n=5,
    )
    
    # 3. 每个问题,LLM 回答并写回
    for q in questions:
        relevant_memories = retrieve(q, recent)
        insight = llm.answer(question=q, context=relevant_memories)
        memory.add(
            content=insight,
            type="reflection",
            source_memories=[m.id for m in relevant_memories],
            importance=8,  # reflection 默认高重要性
        )

5.3 Reflection vs Consolidation

很相似但有区别:

维度	Consolidation	Reflection
颗粒度	抽具体事实	抽高层模式
输入	同主题相似事件	任意 100+ 条 memory
输出	一条 semantic fact	一条 high-level insight
可被再 reflect	❌	✅(Reflection 可以反思 Reflection)

5.4 MemGPT / Letta 的 Self-edit

Letta 的 agent 在每次 thinking step 都可以自我决定:

LLM 输出:
  "我注意到用户最近多次提到'压力大',我应该把这条加入 core memory 的 human block"
  → 调用 core_memory_append("human", "用户最近压力大,需要更体贴的回复")

这是一种实时 reflection——不需要等离线 job,agent 自己决定何时 self-edit。

6. Forgetting & Decay

6.1 为什么需要遗忘

如果不遗忘:

存储无限膨胀
检索噪声变多——三年前的偏好和昨天的偏好被同等召回
计算成本上升

6.2 Ebbinghaus 衰减(MemoryBank 风格)

R(t) = e^{-(t - t_{\text{last access}}) / S}

其中 $S$ 是”巩固强度”:每次被检索时增大,体现”复习巩固”。

def calculate_decay(memory):
    delta = (now - memory.last_accessed).days
    return math.exp(-delta / memory.strength)

def retrieve_with_decay(query, user_id):
    candidates = vector_store.search(query, user_id, limit=50)
    for c in candidates:
        retention = calculate_decay(c)
        c.score *= retention   # 衰减后排序靠后
    return sorted(candidates, key=lambda c: -c.score)[:5]

def on_memory_accessed(memory):
    memory.last_accessed = now
    memory.strength *= 1.5  # 复习巩固

6.3 FadeMem 双层架构(2025)

更聪明的做法——双层 store,衰减率不同:

Layer	衰减率	内容
Long-term Memory Layer (LML)	慢(年级)	高 importance 的核心事实
Short-term Memory Layer (SML)	快(天/周)	低 importance 的偶发事件

新事实先入 SML,被高频访问 / importance 高的”晋升”到 LML。FadeMem 论文报告:比 Mem0 召回率 +16pp,存储省 45%。

6.4 显式删除(GDPR 友好)

用户主动要求”忘掉”某事时:

memory.delete(user_id="u123", filter={"category": "address"})
memory.delete(user_id="u123")  # 全部删除

为合规,所有 memory 系统都必须支持 hard delete + audit log。

7. Retrieval:多因子打分

7.1 经典三因子(Generative Agents)

\text{score}(m, q) = \alpha \cdot \text{recency}(m) + \beta \cdot \text{importance}(m) + \gamma \cdot \text{relevance}(m, q)

因子	计算
recency	$\exp(-\lambda \Delta t)$ ,衰减
importance	入库时 LLM 打的 0-10 分
relevance	$\cos(\text{embed}(m), \text{embed}(q))$

权重 α=β=γ=1/3 是默认起点,实际要按业务调。

7.2 现代 Hybrid Scoring

生产系统通常更复杂:

\text{score} = w_1 \cdot \cos\text{sim} + w_2 \cdot \text{BM25} + w_3 \cdot \text{recency} + w_4 \cdot \text{importance} + w_5 \cdot \text{KG distance}

可以用 learned weights(在 LongMemEval 上训练 Logistic Regression / lambdaMART)。

7.3 Filter 比 score 更重要

很多场景下 filter 排除非常关键:

results = memory.search(
    query="早餐推荐",
    user_id="u123",                    # 必须
    filter={
        "memory_type": "semantic",     # 只要语义事实,不要 episodic 事件
        "valid_at": now,               # bi-temporal:当前有效
        "importance__gte": 5,          # 不要太琐碎
    },
    limit=5,
)

7.4 Retrieval 的几个高级技巧

技巧	说明
Query rewriting	LLM 把用户原始 query 改写成更适合检索的形式
Multi-query	一个 query 生成 N 个改写,各自检索后融合
Hyde	LLM 先生成”假想答案”,用它的 embedding 检索
Rerank	召回 top-50,用 cross-encoder rerank 取 top-5
Time-aware rerank	召回后按时间窗口二次排序

8. AgeMem / RL 驱动的 memory tool calling(2025+)

8.1 核心思想

把 memory 的 5 个核心操作(store / retrieve / update / summarize / discard)当成工具暴露给 LLM,然后用 Reinforcement Learning 训练 LLM 何时调用哪个工具。

tools = [
    "memory.store(content)",
    "memory.retrieve(query, k=5)",
    "memory.update(id, new_content)",
    "memory.summarize(memory_ids)",
    "memory.discard(id)",
]

# 用 RL fine-tune,reward = 长会话准确率 - α·LLM 调用成本

8.2 学到的非显然策略

论文报告 RL 学出来的 policy 有以下”反直觉”行为:

主动 summarize 中间结果:发现”在长 trajectory 中,自己生成的中间总结后续被频繁检索 → 学会提前 summarize”
selective discard:语义高度相似的多条 memory,policy 学会主动 discard 低分的(免污染)
预取:对话开始时主动 retrieve 用户档案,避免后续频繁查

Generative Agents (Park et al., 2023):arXiv 2304.03442 —— Reflection、三因子打分原型
MemoryBank (Zhong et al., 2023) —— Ebbinghaus decay
A-MEM (Xu et al., NeurIPS 2025):arXiv 2502.12110 —— Update/Evolve 算子
FadeMem —— 双层 decay 架构
AgeMem / RL-trained memory tools —— 2025 年新兴方向

框架文档(对比实现)

Mem0 ADD / UPDATE / DELETE 决策:Mem0 docs
Letta Self-edit:Letta docs
LangMem create_memory_manager:LangChain 官方
Graphiti 自动 invalidation:Graphiti GitHub

工程实践博客

I built memory decay for AI agents using the Ebbinghaus forgetting curve:DEV
Architecture and Orchestration of Memory Systems in AI Agents:Analytics Vidhya
FadeMem: Why Teaching AI Agents to Forget Makes Them Remember Better:CO-RE