AI Agent记忆系统实战：分层记忆架构与长期记忆检索方案

AI Agent（智能体）在多轮对话和长期任务执行中最先撞上的墙不是推理能力，而是记忆。模型上下文窗口有限，会话结束状态即丢失，跨会话任务无法延续。构建AI Agent记忆系统，核心是把”上下文内记忆”升级为”持久化分层记忆”，让智能体具备跨会话的知识沉淀与经验复用能力。本文给出可直接落地的分层记忆架构设计与检索方案。

AI Agent记忆系统为什么需要分层设计

单一记忆池方案在实际工程中问题明显：所有历史全塞进上下文，token成本随对话轮次线性增长；无关细节淹没关键信息，检索精度下降。分层设计把记忆按生命周期与用途拆开，每层用不同的存储与检索策略。业界通行的分层结构为三层：

1. 工作记忆（Working Memory）：当前会话上下文，跟随请求存在，保存最近对话轮次与当前任务状态；

2. 情景记忆（Episodic Memory）：历史会话记录，按时间与任务归档，支撑”上次做到哪了”这类跨会话恢复；

3. 语义记忆（Semantic Memory）：从交互中提炼的稳定知识与用户偏好，类似智能体的”长期知识库”。

三层的数据流向是：工作记忆随会话沉淀为情景记忆，情景记忆经提炼进入语义记忆，检索时三层并行召回后统一重排。

记忆写入与提炼流水线实现

写入阶段不要把原始对话全量入库，先做提炼压缩。一条可行的流水线是：对话结束触发总结 → 抽取事实/偏好/待办 → 打标分类 → 写入对应层。用LangGraph实现的核心代码如下：

from langgraph.graph import StateGraph, MessagesState, START
from langchain_core.messages import SystemMessage
import json, datetime

MEMORY_LAYER_PROMPT = '''你是记忆提炼器。从对话中抽取三类信息并以JSON输出：
{"facts": ["稳定事实"], "preferences": ["用户偏好"], "todos": ["未完成任务"]}
没有则输出空数组，不要输出其他内容。'''

def extract_and_store(state: MessagesState, config):
    llm = state["llm"]
    # 只取本轮会话消息做提炼，避免全量重算
    summary = llm.invoke([
        SystemMessage(content=MEMORY_LAYER_PROMPT),
        *state["messages"][-20:]
    ]).content
    data = json.loads(summary)
    user_id = config["configurable"]["user_id"]
    now = datetime.datetime.now().isoformat()
    store = state["memory_store"]
    with store.transaction():
        # 情景层：原始会话摘要存档
        store.insert("episodic", user_id=user_id, ts=now,
                     summary=state.get("session_summary", ""))
        # 语义层：事实与偏好去重后写入
        for fact in data["facts"]:
            store.upsert("semantic", user_id=user_id,
                         content=fact, kind="fact")
        for pref in data["preferences"]:
            store.upsert("semantic", user_id=user_id,
                         content=pref, kind="preference")
        # 待办写入工作记忆层，供下次会话恢复
        for todo in data["todos"]:
            store.insert("working", user_id=user_id,
                         content=todo, done=False)
    return {"memorized": True}

提炼的关键在于”去重”。语义记忆反复写入同一事实会产生大量近重复记录，检索时互相干扰。upsert操作应以内容的语义哈希或归一化文本为唯一键，命中已有记录则更新时间戳而非新增。

长期记忆检索：向量召回与重排序结合

检索阶段决定智能体”记不记得准”。纯向量检索的通病是语义相近但情境不符，比如用户三个月前的项目偏好会干扰当前任务。可用的方案是混合检索加条件过滤：

1. 用当前任务描述与最近三轮对话生成检索查询，而非只用最后一条消息；

2. 向量召回top 20，同时用元数据过滤（用户ID、时间衰减、记忆类型）；

3. 用交叉编码器对候选重排，取top 5注入上下文。

import math, time

def time_decay(ts, half_life_days=30):
    '''半衰期时间衰减：30天前的记忆权重减半'''
    age_days = (time.time() - ts) / 86400
    return math.pow(0.5, age_days / half_life_days)

def recall(query, user_id, store, cross_encoder, k=5):
    # 混合召回：语义向量 + 关键词BM25 双路
    vec_hits = store.vector_search(query, user_id=user_id, top_k=20)
    bm25_hits = store.bm25_search(query, user_id=user_id, top_k=20)
    merged = dedup(vec_hits + bm25_hits)

    scored = []
    for hit in merged:
        decay = time_decay(hit.ts)
        # 交叉编码器对"任务+记忆"成对打分，比余弦相似度准
        relevance = cross_encoder.score(query, hit.content)
        score = relevance * decay * LAYER_WEIGHT[hit.layer]
        scored.append((score, hit))
    scored.sort(key=lambda x: -x[0])
    return [h for _, h in scored[:k]]

# 层级权重：当前任务待办 > 语义偏好 > 远期情景
LAYER_WEIGHT = {"working": 1.0, "semantic": 0.8, "episodic": 0.5}

时间衰减的半衰期要按业务调整：客服场景7天即可，助理类场景30天起步。待办类记忆不衰减，完成后显式标记done，避免同一任务被反复召回。

记忆冲突处理与遗忘机制

用户偏好会变化，旧记忆不改会持续污染检索结果。冲突处理的最小可行方案：新事实入库时，检索同类型近重复记忆，相似度超过阈值则触发”覆盖合并”，保留新记录并归档旧记录；相似度中等则两条并存，靠重排序按时间戳优先。遗忘机制上，设置容量上限与归档周期，episodic层保留最近90天，semantic层超过一年未被召回的记录转冷存储，控制在线库规模。

工程落地要点与验证指标

上线前用固定测试集验证三个指标：跨会话任务恢复成功率（新会话能否继续上次任务）、偏好命中率（注入的记忆是否被实际使用）、检索token占比（注入记忆占上下文比例建议控制在30%以内）。记忆系统不是越大越好，检索精度差的长记忆会主动降低智能体表现，宁可少召回也要保证召回的内容可用。

分层记忆架构的价值在于让智能体从”每次重来”变成”越用越懂你”。先用三层结构跑通最小闭环，再根据业务数据迭代提炼与重排策略，比上来就做复杂记忆图谱务实得多。