AI Agent 长程记忆工程实战：从分层记忆模型到遗忘机制与生产级检索闭环

AI Agent 长程记忆工程实战：从分层记忆模型到遗忘机制与生产级检索闭环

如果你给一个 AI Agent 挂上 100 万 token 的上下文窗口，它依然会在第三天早上忘记用户说过"我对花生过敏"。这不是模型能力问题，而是架构问题：上下文窗口是易失的、按会话结清的工作区，而记忆是需要跨会话存活、可被证伪、允许被遗忘的基础设施。

很多团队把"上下文工程"（压缩、卸载、多智能体隔离）当成记忆方案，这是两个层次的事情。上下文工程解决的是当前窗口内如何摆放信息，记忆解决的是信息如何在系统里长期存活并被正确召回。前者是内存管理，后者是数据库。混淆二者最典型的后果是：Agent 每次都在昂贵的 summarizing 里烧毁 token，却仍然无法稳定记住三个月前的约定。

本文从工程落地视角拆解一套可生产的 Agent 记忆子系统：分层模型、写入链路、混合检索、遗忘与冲突消解、以及最重要的——如何评测它。

一、记忆的分类学：先定义清楚你要存什么

套用认知科学的划分，Agent 记忆可以映射到三类，每类的存储结构、召回策略、生命周期完全不同：

真正踩过坑的人会知道：绝大多数系统只需要把语义记忆做扎实。情景记忆容易退化成"聊天记录回放"，检索成本高、信噪比低；程序性记忆本质上是 prompt 模板和 workflow，不属于记忆系统，属于业务配置。

一个关键的工程判断是：语义记忆应当以结构化断言而非自由文本为主存。当你存的是 {"subject": "user:10086", "predicate": "allergy", "object": "peanut", "confidence": 0.97, "source": "conv_2026_08_12"} 而不是一整段对话时，你才拥有冲突检测、批量失效、审计追溯的能力。

二、写入链路：抽取 → 归一化 → 去重 → 固化

记忆写入不是把对话塞进向量库。生产级的写入链路至少包含四步，其中最容易被跳过的是去重合并——不做这一步，三个月后你的记忆库里会有 47 条"用户喜欢简洁的回复"，每一条都会被召回，然后互相打架。

下面是一个可运行的写入骨架，展示了抽取与记忆入库的核心逻辑：

from dataclasses import dataclass, field
from datetime import datetime, timezone
import hashlib, json

@dataclass(frozen=True)
class Fact:
    subject: str
    predicate: str
    object: str
    confidence: float = 1.0
    source: str = ""
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    @property
    def key(self) -> str:
        """同一 subject+predicate 下的事实视为可合并候选"""
        return hashlib.sha1(f"{self.subject}|{self.predicate}".encode()).hexdigest()[:16]

EXTRACT_PROMPT = """从对话中抽取对助手长期有用的事实断言。

要求：
1. 只抽取可长期成立的稳定事实，忽略一次性任务细节
2. 每条输出 JSON: {"subject","predicate","object","confidence"}
3. predicate 必须从受限词表选取: allergy, preference, constraint, role, goal, tech_stack
4. 无法确定的不要猜，宁可少抽

对话：
{dialogue}

输出 JSON 数组："""

async def write_memories(dialogue: str, store: "MemoryStore", llm) -> int:
    raw = await llm.complete(EXTRACT_PROMPT.format(dialogue=dialogue), json_mode=True)
    facts = [Fact(**f, source=dialogue[:64]) for f in json.loads(raw)]
    written = 0
    for fact in facts:
        if fact.confidence < 0.6:
            continue  # 低置信度直接丢弃，比存进去污染检索更划算
        await store.upsert(fact)
        written += 1
    return written

这里有三个刻意为之的设计：

1. 受限谓词词表。允许 LLM 自由生成 predicate，词表会在两周内膨胀到无法维护。把它限制在几十个枚举值内，后续的冲突检测、检索加权才有抓手。
2. 低置信度直接丢弃。记忆系统的成本不在写入，在于错误记忆被召回后的污染效应。宁缺毋滥。
3. source 溯源字段。当一条记忆导致错误行为时，你必须能回溯它是从哪次对话产生的，否则无法修复也无法做数据清洗。

三、检索闭环：向量检索不是万能的

纯粹用 embedding 相似度召回记忆，在生产环境会遇到三类典型失败：数值与实体精确匹配失效（"部署端口 8080"）、时间局部性丢失（用户昨天改了主意）、以及"相似但不相关"的语义漂移。

可靠的做法是 BM25 + 向量 + 元数据过滤 的三路混合检索，再统一打分：

import math

def reciprocal_rank_fusion(rank_lists, k=60):
    """RRF 融合多路召回，避免不同打分尺度归一化的麻烦"""
    scores = {}
    for ranked in rank_lists:
        for rank, item_id in enumerate(ranked, start=1):
            scores[item_id] = scores.get(item_id, 0.0) + 1.0 / (k + rank)
    return sorted(scores.items(), key=lambda kv: kv[1], reverse=True)

def final_score(mem, rrf: float, now_ts: float, lambda_decay: float = 0.02) -> float:
    """最终排序 = 融合分 x 时间衰减 x 置信度 x 访问增益"""
    age_days = (now_ts - mem.created_at_ts) / 86400.0
    decay = math.exp(-lambda_decay * age_days)          # 时间衰减
    support = 1.0 + 0.1 * math.log1p(mem.recall_count)  # 被多次召回且被采纳的记忆加权
    return rrf * decay * float(mem.confidence) * support

几个实战要点：

• 时间衰减务必分层。用户的过敏信息几乎不该衰减（lambda ≈ 0），而"当前正在排查的故障"应当在一周内衰减到接近 0。给不同 predicate 配置不同半衰期。
• 不要一次性把 Top-20 记忆全塞进 prompt。更好的做法是两阶段：先用廉价召回拿候选，再让一个轻量模型对候选做一次相关性判定（rerank），只把通过的 3-5 条写进上下文。这一步能把上下文噪声降低一个数量级。
• 写入 path 要异步。把记忆抽取放进主链路会增加 TTFT（首 token 延迟）。用消息队列把对话推到后台 worker 处理，主链路只做检索。

四、遗忘机制：没有遗忘的记忆系统一定会崩

这是最反直觉、也最常被忽略的部分。一个只增不减的记忆库，六个月后会变成检索噪声源。遗忘不是可选优化，是系统存活的必要条件。

工程上通常组合三种策略：

1. 时间衰减 + 冷热分层。热数据（近 30 天活跃）常驻内存/向量库，冷数据落归档存储（Parquet + SQLite），只在深度检索时才捞取。这与数据库的冷热分离是同一套思路。
2. 冲突消解（Contradiction Resolution）。当用户说"我改用 VS Code 了"，旧的"用户使用 Vim"必须被标记失效而非删除——因为时间旅行类问题（"三个月前我在用什么编辑器"）需要历史版本。实践上给每条记忆加 valid_from / valid_to，做双时间轴管理。
3. 合并压缩（Consolidation）。定期把同一 key 下的多条相似事实交给 LLM 合并成一条精炼断言，并累加 support_count。这一步类似 LSM-Tree 的 compaction，是控制存储膨胀的核心手段。

async def consolidate(store, llm, batch_size: int = 200) -> None:
    for key, group in store.iter_facts_by_key(batch_size=batch_size):
        if len(group) < 3:
            continue
        merged = await llm.complete(
            MERGE_PROMPT.format(facts=json.dumps(
                [{"object": f.object, "confidence": f.confidence, "created_at": f.created_at}
                 for f in group], ensure_ascii=False))
        )
        new_obj = json.loads(merged)["object"]
        store.replace_group(key, new_obj, support=len(group))

一个务实的提醒：Consolidation 的 LLM 调用是有成本的，按 key 分组批量做，不要实时做。通常每日一次离线任务足够。

五、如何评测：记忆系统的三个硬指标

没有评测的记忆系统是玄学。建议至少监控这三个指标：

• 记忆召回率 (Memory Recall)：构造一批"标准答案明确"的提问（例如 200 条需要跨会话记忆才能答对的问题），统计 Agent 答对率。这是端到端指标，最诚实。
• 记忆污染率 (Hallucinated Memory)：从记忆库里抽样，人工或 LLM-judge 判定有多少条是错误或过时的。超过 5% 就应该报警并检查抽取环节。
• 无效召回率 (Irrelevant Retrieval)：被注入上下文但最终未被模型使用的记忆占比。这个指标高说明检索阈值或衰减策略有问题，白白浪费 token。

我的实战观点是：污染率比召回率更重要。召回不足表现为"Agent 记不住"，这是可感知但可容忍的退化；而一条错误记忆被高频召回，会让 Agent 在特定场景下稳定地犯错，用户却完全不知道为什么——这类 bug 的排查成本极高。

六、落地清单

如果你正在从零搭建，按这个顺序来：

1. 先做语义记忆 + SQLite + BM25，不要一上来就上向量数据库。SQLite 的 FTS5 足以支撑十万级事实断言，而且事务、备份、审计都是现成的。
2. 先定死受控谓词词表再开始收数据。schema 变更的成本远高于初期的过度设计。
3. 写入链路加人工审核队列，尤其是涉及用户偏好与安全约束的记忆，至少要保留可回滚的审计日志。
4. 遗忘机制从第一天就预留，哪怕先只做最简单的 valid_to 字段。
5. 评测集跟着第一版系统同步建立，200 条问题足够开始迭代。

小结

上下文工程让 Agent 在单次任务里表现聪明，记忆系统让 Agent 在三个月后仍然可靠。二者的技术栈完全不同：前者是压缩与调度，后者是数据库、检索与数据治理。

最容易被低估的是后半句里的"数据治理"。当你的 Agent 拥有十万条跨用户记忆时，你面对的问题已经不是 AI 问题了——它是 schema 演进、数据血缘、冷热分层、以及脏数据清洗。那些在 OLTP 时代踩过的坑，会在记忆系统里原封不动地再踩一遍。

一句话总结：把记忆当数据库做，不要把记忆当 prompt 做。