Agent长期记忆架构设计：分层存储、混合检索与遗忘机制

在开发 Agent 应用时很多团队会先遇到一个尴尬的瓶颈模型能力不差工具调用也通但对话一长就“失忆”用户之前说过的偏好、项目中已经确认过的结论下次一问就全忘了。这篇文章我会基于项目实战经验完整拆解 Agent 长期记忆架构的设计思路讲清楚短时记忆和长期记忆的边界、分层存储方案、检索策略以及遗忘机制并提供可直接落地的实现思路与代码片段。1. 背景与核心概念1.1 Agent 记忆是做什么的Agent 记忆简单理解就是让 AI Agent 具备“记住信息”的能力。在早期的 LLM 应用中模型每次请求都是无状态的。用户问一句模型答一句请求结束之后上下文窗口里的内容就随之清空。后来出现了多轮对话通过把历史消息重新拼进 Prompt 来实现“会话内记忆”但这本质上只是把聊天记录当成文本前缀拼接并不算真正的记忆系统。Agent 长期记忆架构要解决的是更复杂的问题用户几周前提到过自己对某个领域很熟悉后续回答应该引用这个背景。用户在多个对话中反复强调同一类偏好Agent 应该形成稳定的画像记忆。已经执行完成的任务在后续新的任务中需要作为背景知识复用。记忆会过时旧结论应该被新结论替换而不是永远占据优先位置。一句话概括长期记忆让 Agent 从“每次重新认识用户”进化成“越用越懂用户”。1.2 为什么长期记忆是 Agent 工程的瓶颈目前限制 Agent 走向生产环境的核心问题除了模型本身的推理能力就是记忆能力。上下文窗口再大也不可能无限拼接历史。而且把大量原始历史直接塞进 Prompt会带来几个问题推理质量下降。无关内容太多会干扰模型的注意力。Token 成本飙升。每个请求都要重新计算历史记录。信息时效性无法处理。旧信息没有过期新信息没有优先级。跨会话复用能力缺失。换个会话记忆就断了。所以长期记忆不是“把历史消息存下来”这么简单它需要一套完整的存取架构包括写入策略、表示方式、检索机制、更新与遗忘规则。这也是本文重点讲解的内容。1.3 理解本文需要哪些基础本文会涉及向量检索、Prompt 工程、Python 开发但不会依赖某一个特定向量数据库。建议读者满足以下条件有大模型 API 调用经验。基本了解词向量Embedding的概念。了解 FastAPI 或 Flask 这类 Python Web 框架。如果你刚开始接触 Agent建议先跑通一个最简单的 ReAct Agent再来读记忆模块会更顺畅。2. 记忆分层从短时到长期2.1 短时记忆与会话上下文短时记忆在 Agent 系统中通常指的是“当前会话内的上下文信息”。它距离模型最近实时性最强。当用户发起一个新的请求时Agent 会将系统提示词System Prompt最近几轮对话内容当前工具调用返回的结果临时计算出的中间状态拼接到当前请求的上下文中。短时记忆的特点是访问快、生命周期短、容量受限。它受限于模型的上下文窗口长度超出部分会被截断或压缩。很多团队做上下文压缩Context Compression本质上就是在优化短时记忆的使用效率。一个容易踩的坑是把所有历史消息都当作短时记忆。这样做的后果前面已经说过——质量下降、成本升高。正确做法是把短时记忆和长期记忆解耦短时只保留“当前任务直接相关”的信息其余沉淀到长期记忆库中。2.2 长期记忆的定义与特点长期记忆是与具体会话解耦的持久化记忆。它的特征是生命周期长达数天、数周甚至数月。存储介质通常是向量数据库、关系型数据库或对象存储。不随对话结束而消失。需要配合检索才能发挥作用。长期记忆解决的问题是跨会话信息复用。比如用户第一次使用 Agent 时说“我习惯看简洁的回答”如果没有长期记忆第二次对话 Agent 就会忘记有了长期记忆Agent 在新会话中可以通过检索把这个偏好重新拉取进 Prompt。2.3 记忆分层的常见方案实际项目中长期记忆可以进一步细分。常见的分层方案如下记忆层生命周期存储方式典型内容短期记忆会话内内存缓存 / 上下文窗口多轮对话、工具中间结果工作记忆任务执行期间内存对象当前目标、步骤、临时状态情景记忆跨会话向量数据库用户具体说过的话、做过的任务语义记忆跨会话向量数据库 结构化存储用户偏好、知识结论、画像信息程序记忆长期代码 / 配置文件Agent 的技能、流程、工具调用模式短时和工作记忆服务于“当下任务”情景和语义记忆服务于“跨会话理解”程序记忆服务于“能力沉淀”。2.4 短时与长期记忆的转换时机记忆不是简单地“全存入”或“全不存”。需要考虑写入时机任务完成时。一个任务执行完毕提取关键结论写入长期记忆。对话出现明确偏好时。用户说“以后都按这个格式来”需要触发记忆写入。重要事实出现时。比如用户透露了自己的职位、行业、工作背景。周期性同步。例如每天定时把短期缓存中的高频信息同步到长期存储。实践中不要每条消息都写入长期记忆否则会造成存储膨胀和检索噪声。更推荐的方式是设置“记忆提炼”环节由模型对会话内容做二次加工再决定落库哪些信息。3. 记忆存储与表示3.1 语义记忆与情景记忆的区分在 Agent 记忆系统中我们通常把长期记忆分为两类情景记忆Episodic Memory记录的是“发生过的事”。例如用户昨天下午请求生成了一份产品需求文档主题是智能客服。语义记忆Semantic Memory记录的是“抽象出来的规律”。例如用户从事软件产品经理工作关注智能客服方向喜欢结构化文档输出。两者可以分开存储也可以统一存储后打标签区分。在检索时语义记忆的优先级通常更高因为它已经完成了信息抽象直接可用。3.2 记忆条目的核心字段设计记忆条目时建议包含以下字段字段类型说明memory_idstring记忆唯一 IDuser_idstring归属用户或会话主体contenttext记忆内容本体memory_typestring情景记忆 / 语义记忆 / 偏好importancefloat重要性评分0 到 1created_atdatetime创建时间updated_atdatetime更新时间last_access_atdatetime最近一次被检索到的时间access_countint被访问次数sourcestring来源会话或任务 IDmetadatajson扩展字段可存情绪、场景、实体等importance、last_access_at、access_count这三个字段是遗忘机制的基础。后面会详细讲。3.3 向量嵌入与结构化抽取原始文本不能直接参与相似度计算需要先转换为向量。代码如下# 文件路径memory/embedding.py from openai import OpenAI client OpenAI() def get_embedding(text: str, model: str text-embedding-3-small) - list: 将文本转换为向量 response client.embeddings.create( modelmodel, inputtext, ) return response.data[0].embedding在写入记忆库时不只是存content还要存content对应的向量。这样检索阶段才能用向量相似度召回。同时建议对记忆内容做结构化抽取。例如提取实体、意图、情感倾向等信息存进metadata字段# 文件路径memory/extractor.py from pydantic import BaseModel class MemoryMeta(BaseModel): entities: list[str] [] topic: str emotion: str neutral is_preference: bool False def extract_metadata(content: str): 调用 LLM 抽取记忆条目的结构化信息 prompt f 请分析以下用户记忆内容返回 JSON 格式 - entities: 实体列表 - topic: 主题 - emotion: 情感 - is_preference: 是否为用户偏好 内容{content} # 这里接入你使用的 LLM做 JSON 输出解析 # 返回 MemoryMeta 实例 pass抽取结构化信息不是为了炫技而是为了之后能按主题、实体做精确过滤弥补纯向量检索的不足。4. 记忆检索机制4.1 单纯向量检索的局限向量检索是目前最主流的记忆召回方案。核心思路是把用户当前的问题向量化然后到记忆库中计算相似度返回 Top K 条最接近的记忆。在短期对话场景中向量检索效果并不差。但长期使用后会暴露几个问题关键词不匹配。用户当前问题里提到“接口文档”而记忆中写的是“API 说明”字面差异大向量表示可能不够接近。时间敏感信息失效。一条半年前的记忆向量相似度可能仍然很高但它已经过期了。重要度未体现。高价值的核心偏好可能因为表达方式差异导致召回不到。冷启动问题。新用户的记忆库内容很少向量检索基本无结果。所以生产级系统一般不会只用向量检索而是采用混合检索策略。4.2 BM25 关键词检索BM25 是一种经典的关键词匹配算法适合解决“字面差异小但向量表达不够接近”的情况。其优点是实现成熟、无需训练、可解释性强尤其适合检索实体名称、专有名词、代码名这类内容。在 Python 中可以使用rank_bm25库pip install rank-bm25核心代码# 文件路径memory/bm25_search.py from rank_bm25 import BM25Okapi import jieba def segment(text: str) - list[str]: 中文分词 return list(jieba.cut(text)) def build_bm25_index(corpus: list[str]): 构建 BM25 索引 tokenized_corpus [segment(doc) for doc in corpus] return BM25Okapi(tokenized_corpus) def search_bm25(bm25_index, query: str, top_k: int 5): BM25 检索 tokenized_query segment(query) scores bm25_index.get_scores(tokenized_query) top_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue)[:top_k] return top_indices需要注意的是中文场景下 BM25 需要分词器英文场景直接用空格切分即可。4.3 混合检索与多路召回混合检索的目的是把不同检索方式的优势结合起来提高召回率。常见组合方式向量检索解决语义相似但字面差异大的问题。BM25 检索解决专有名词精确匹配问题。规则过滤根据时间范围、记忆类型、实体过滤候选集。流程如下对用户当前问题做规则解析提取过滤条件。并行执行向量检索和 BM25 检索。合并两路召回结果去除重复记录。按融合分数重新排序。取 Top K 条放入 Prompt。代码示意# 文件路径memory/retriever.py import numpy as np from embedding import get_embedding from vector_db import query_vector_db from bm25_search import search_bm25 def hybrid_search(query: str, memory_collection, bm25_index, top_k: int 5): # 1. 向量检索 query_vec get_embedding(query) vector_hits query_vector_db(memory_collection, query_vec, top_ktop_k * 2) # 2. BM25 检索 bm25_hits search_bm25(bm25_index, query, top_ktop_k * 2) # 3. 合并与加权 score_map {} for idx, score in vector_hits: score_map[idx] score_map.get(idx, 0) 0.6 * score for idx, score in bm25_hits: score_map[idx] score_map.get(idx, 0) 0.4 * score # 4. 排序 sorted_hits sorted(score_map.items(), keylambda x: x[1], reverseTrue) return sorted_hits[:top_k]多路召回的思想可以继续扩展。例如加一路“核心人物/事实召回”从当前问题和记忆库中分别抽取实体如果实体匹配则直接加权。这种方案在生产环境中效果很好能显著提升重要记忆的召回稳定性。4.4 重排序混合检索虽然提升了召回率但召回结果的精确度还需要重排序来提高。简单方案是做加权融合像上面代码一样。效果更好的是用 Cross-Encoder 模型或 LLM 做重排。思路第一步先多路召回 20 条候选记忆。第二步用重排模型逐条计算与当前问题的相关度。第三步取前 5 条进入最终 Prompt。重排虽然增加一次模型调用但能明显提升最终回答质量。尤其是在记忆库较大、噪声较多的情况下。5. 记忆遗忘与更新5.1 为什么需要遗忘机制大脑会遗忘Agent 也需要。如果记忆只增不减系统会面临检索噪声越来越大相似度分数被无关记忆干扰。存储成本持续上升。过期信息占据召回位置导致 Agent 给出过时答案。用户修改后的偏好无法替代旧偏好。遗忘机制的目的不是无脑删除而是让记忆库保持在“高价值、低冗余”的状态。5.2 时间衰减打分一种常见的遗忘机制是基于时间衰减的打分模型。核心公式score importance * decay_factor(age) * access_boost(access_count)其中importance是写入时评估的重要性。decay_factor随时间下降越久没被访问分数越低。access_boost被频繁使用的记忆会获得一定保护。实现如下# 文件路径memory/forgetting.py from datetime import datetime, timedelta def calculate_memory_score( importance: float, created_at: datetime, last_access_at: datetime, access_count: int, now: datetime None, ): now now or datetime.utcnow() # 时间衰减超过30天分数减半 age_days (now - created_at).days decay_factor 0.5 ** (age_days / 30) # 访问衰减超过7天未访问访问加成降低 days_since_access (now - last_access_at).days if days_since_access 7: access_boost 1 min(access_count, 10) * 0.05 else: access_boost 1.0 return importance * decay_factor * access_boost当分数低于某个阈值时系统将这条记忆标记为“待归档”或直接清理。要注意不要只按一条分数来删记忆。建议设计一个定时任务批量扫描低分记忆人工或模型确认后再删除。5.3 记忆合并与去重遗忘之外还应该做记忆合并。举个例子第一天用户说“我喜欢简洁的回答方式”。第二天用户说“回答不要超过 200 字”。第三天用户说“输出格式帮我精简一点”。这三条内容存在向量库里语义高度相似。如果每次都把三条全部召回既浪费空间也干扰模型。更好的做法是定期合并成一条用户偏好简洁输出回答尽量控制在 200 字以内。合并策略可以使用模型批量处理把一组相似度超过阈值的记忆合并成一条高重要性的新记忆删除旧条目。5.4 记忆更新流程当新记忆和旧记忆冲突时需要更新而非追加。建议采用版本化更新新记忆写入时检索现有记忆库。如果找到语义相似度超过阈值且类型相同的旧记忆。将旧记忆标记为superseded已替换。写入新记忆并关联旧记忆 ID。这样可以保留历史的可追踪性同时保证 Agent 默认使用最新的记忆。6. 完整实战案例长期记忆仓库模块下面我们把前面讲到的概念整合成一个可落地的长期记忆仓库模块。这个模块可以嵌入到现有的 Agent 项目中。6.1 项目结构agent-memory/ ├── config.py # 配置 ├── main.py # 入口示例 ├── memory/ │ ├── __init__.py │ ├── embedding.py # 向量化 │ ├── extractor.py # 结构化抽取 │ ├── storage.py # 存储 │ ├── retriever.py # 检索 │ ├── forgetting.py # 遗忘 │ └── models.py # 数据模型 └── requirements.txt6.2 数据模型定义# 文件路径memory/models.py from datetime import datetime from pydantic import BaseModel class MemoryItem(BaseModel): memory_id: str user_id: str content: str memory_type: str semantic # semantic / episodic / preference importance: float 0.5 embedding: list [] metadata: dict {} created_at: datetime datetime.utcnow() updated_at: datetime datetime.utcnow() last_access_at: datetime datetime.utcnow() access_count: int 0 status: str active # active / archived / superseded6.3 记忆写入流程写入记忆的完整流程接收原始文本。生成向量。调用 LLM 抽取结构化信息。计算重要性评分。检查与旧记忆的冲突。存入向量数据库。# 文件路径memory/storage.py from datetime import datetime from models import MemoryItem from embedding import get_embedding def save_memory(user_id: str, content: str, memory_type: str semantic) - MemoryItem: # 1. 生成向量 embedding get_embedding(content) # 2. 计算重要性简化版 importance estimate_importance(content) item MemoryItem( user_iduser_id, contentcontent, memory_typememory_type, importanceimportance, embeddingembedding, ) # 3. 检查冲突是否已有相似记忆 conflict find_similar_memory(item) if conflict: # 旧记忆标记为 superseded supersede_memory(conflict.memory_id) item.memory_id conflict.memory_id # 复用原来的 ID # 4. 写入向量库 upsert_to_vector_db(item) return item def estimate_importance(content: str) - float: 根据关键词和内容长度粗略估计重要性 high_importance_keywords [偏好, 以后, 总是, 不要, 禁止, 记住, 我是] score 0.5 for word in high_importance_keywords: if word in content: score 0.1 return min(score, 1.0)6.4 读取与检索# 文件路径memory/retriever.py from datetime import datetime from embedding import get_embedding def retrieve_relevant_memories(user_id: str, query: str, top_k: int 3) - list[str]: # 1. 向量检索 query_vec get_embedding(query) vector_results query_by_vector(user_id, query_vec, top_ktop_k * 2) # 2. 常见相关记忆加权 # 例如用户基本信息、最近常用主题 profile_results query_user_profile(user_id, query) # 3. 合并按分数排序 merged merge_results(vector_results, profile_results) # 4. 更新访问记录 for item in merged[:top_k]: update_access_time(item) return [item.content for item in merged[:top_k]]在 Agent 中使用的方式# 在构建 Prompt 时插入记忆 memories retrieve_relevant_memories(user_id, user_query) prompt f 你是一个有帮助的 AI 助手。以下是你对用户的记忆 {chr(10).join(memories)} --- 当前问题{user_query} 6.5 遗忘调度# 文件路径memory/forgetting.py from datetime import datetime def run_forgetting_schedule(): 定时任务每天执行一次。 找出低分记忆归档删除。 all_memories get_all_active_memories() now datetime.utcnow() for item in all_memories: score calculate_memory_score( importanceitem.importance, created_atitem.created_at, last_access_atitem.last_access_at, access_countitem.access_count, nownow, ) if score 0.2: archive_memory(item.memory_id)集成到 FastAPI 中# 文件路径main.py from fastapi import FastAPI from memory.retriever import retrieve_relevant_memories from memory.storage import save_memory app FastAPI() app.post(/memory) def create_memory(user_id: str, content: str, memory_type: str semantic): item save_memory(user_id, content, memory_type) return {memory_id: item.memory_id, status: saved} app.post(/retrieve) def retrieve_memory(user_id: str, query: str): memories retrieve_relevant_memories(user_id, query, top_k3) return {memories: memories}定时遗忘任务建议使用 Celery Beat 或 APScheduler不要在主线程里阻塞运行。7. 常见问题与排查思路实战中我遇到过不少记忆模块的坑下面整理成表问题现象常见原因解决思路检索结果和当前问题无关只用了向量检索专有名词匹配不到改用混合检索加入 BM25记忆库越来越大响应变慢没有任何遗忘机制设置时间衰减打分和定时清理用户改口后 Agent 仍用旧偏好新旧记忆冲突时未做替换写入时先查重冲突则标记 superseded记忆召回太碎上下文装不下Top K 设置过大缩减为 3 到 5 条先做重排中文场景下 BM25 效果差未做中文分词引入 jieba 或等价分词器用户的敏感信息被长期保存未设置安全过滤写入前过滤身份证号、密码等敏感字段嵌入模型换了历史向量失效向量空间不一致嵌入模型必须固定版本升级时重新向量化排查检核清单先确认问题出在写入环节还是检索环节。在写入环节检查是否有向量、是否有结构化标签是否通过冲突检测。在检索环节打印召回结果的分数看是分数普遍偏低还是分数高但语义不相关。检查 Top K 是否过大导致噪声混入。检查遗忘任务是否在正常运行是否误删了重要记忆。8. 最佳实践与工程建议8.1 写入侧不要什么都存长期记忆的质量取决于写入侧的把关。建议所有写入内容经过一道模型提炼接口而不是直接把用户原始消息原封不动存入记忆库。一个实用的提炼模板从以下对话中提取值得长期记忆的信息 1. 用户的个人背景信息。 2. 用户明确表达的偏好。 3. 当前任务的关键结论。 4. 用户经常关注的主题。 不要提取一次性的、与长期无关的对话细节。这样能大幅减少记忆库里的无效条目。8.2 检索侧把记忆做成多级过滤先做粗召回再做精排。不要把 20 条记忆全部塞进 Prompt。我建议粗召回 20 条候选。精排取 3 到 5 条。按记忆类型加权偏好 语义 情景。设置最低相关度阈值低于阈值的直接不返回。8.3 存储侧向量库加关系库双写纯向量库无法高效完成部分过滤和统计操作。生产环境建议向量数据库负责语义相似度召回。关系型数据库负责记录记忆元数据、状态、时间字段。这样在遗忘调度时直接用 SQL 筛选低分记忆效率远高于全量扫向量库。8.4 安全与隐私边界长期记忆保存的是用户信息需要格外注意敏感信息脱敏后再入库不要保存明文密码、身份证号、银行卡信息。设置记忆可见范围区分个人空间和共享空间。提供“删除记忆”功能用户有权清除自己的历史记忆记录。遗忘任务要有审计日志便于追踪记忆的写入、更新和删除。对记忆库访问做最小权限控制内部系统和外部系统使用不同的密钥。8.5 可观测性与评估记忆模块上线后需要持续评估效果。常见的评估指标记忆命中率检索出来的记忆中对最终回答有帮助的比例。用户反馈满意度用户对回答中利用记忆的比例感知。记忆错误率错误记忆进入上下文的比例。建议在关键节点添加日志记忆写入日志、记忆召回日志、记忆进入 Prompt 的完整记录。这样出了问题可以回溯。9. 总结与学习路线本文梳理了 Agent 长期记忆架构的核心环节。首先界定了短时记忆和长期记忆的边界。短时记忆服务于当前会话长期记忆服务于跨会话理解。二者要解耦不能把上下文窗口当作长期存储。接着介绍了记忆分层方案包括情景记忆、语义记忆和偏好记忆并给出了记忆条目的核心字段设计。在存储与表示环节讲解了如何将原始文本转为向量并抽取结构化元数据。检索环节重点对比了向量检索和 BM25 的优劣给出了混合检索与多路召回的实现方案。遗忘机制部分介绍了基于时间衰减的打分模型、记忆合并与更新策略并给出了完整的 Python 代码实现。如果你要动手搭建自己的 Agent 记忆系统建议按下面顺序推进先做一个最简单的“写入-读取-拼接 Prompt”记忆模块验证链路通顺。接入混合检索提升召回质量。加入遗忘调度防止记忆库无限膨胀。再考虑多用户隔离、敏感信息过滤等生产级能力。长期记忆是 Agent 工程中最值得投入的模块之一。框架和模型会持续迭代但记忆架构的设计思想是通用的。希望这篇文章能给你一个清晰的实现起点后续在项目中遇到新的坑和方案也欢迎回来继续交流。