公司动态

Agent长期记忆架构设计:从分层存储到检索与遗忘机制

📅 2026/8/31 15:27:25
Agent长期记忆架构设计:从分层存储到检索与遗忘机制
很多人一开始接触 Agent都以为“给模型接个大模型 API 就能解决一切”。但真正把 Agent 推进到多轮任务、跨天会话、个性化服务之后第一个暴露的短板几乎都是同一个Agent 没有记忆。用户昨天交代过的偏好今天重新问一遍上一步已经算好的结果下一步又得从头说。模型本身的知识是固定的上下文窗口再大也装不下整个项目的历史于是 Agent 的“聪明”就变成了“每次都是第一次见到你”。长期记忆就是把 Agent 从“每轮对话重新开始”变成“越用越懂你”的那一层架构。它不是一个简单的 Redis 缓存也不是把聊天记录全塞进 Prompt。真正可用的长期记忆系统需要同时解决四件事记忆怎么分层、怎么写入、怎么检索、怎么遗忘。这篇文章会围绕这四个问题展开并且提供一个最小可运行的记忆模块示例供你在自己的 Agent 项目里直接改造。也许有人会问短期记忆都不够用为什么要一上来就折腾长期记忆我的判断是短期记忆决定 Agent 的响应质量长期记忆决定 Agent 的成长能力。前者解决“这一轮对话能不能接住”后者解决“这个 Agent 值不值得用户长期用”。本文适合已经在做 Agent 开发、正在被“多轮记忆丢失”“上下文越塞越长”“检索结果不准”困扰的开发者。1. 为什么长期记忆成了 Agent 开发的硬门槛先看一个真实的开发场景。你做了一个客服 Agent任务是根据用户历史订单和沟通记录回答售后问题。第一版很简单把用户最近的订单 JSON 塞进 Prompt 就行。但用两天就会发现用户的诉求散落在十几轮对话里有的关于退款有的关于发票有的关于物流全塞进 Prompt 会超过上下文窗口截断后关键信息反而丢了。于是你想把历史信息存进数据库每次先检索再回答。问题变成了怎么知道哪条历史记录对当前问题有用怎么避免检索出大量无关信息怎么让长期不用的旧记录不干扰新对话这套问题在传统 Web 开发里不存在因为程序的状态是由数据库查询驱动的但在 Agent 开发里“状态”变成了模型感知到的信息。模型一次只能读有限的上下文而历史记忆可能是无限增长的。长期记忆架构的核心任务就是在这两者之间做翻译把无限的历史压缩、索引、筛选成有限的高质量上下文。另一个容易被低估的问题是成本。LLM 接口按 Token 计费直接把聊天记录全量拼进 Prompt短期看不出问题Agent 跑到几百轮之后每一次请求的 Token 消耗都是天文数字。长期记忆不是“加分项”而是控制成本的基础设施。记忆分层做得好等于给模型配了一个“资料管理员”每次只把最相关的几份文件放到桌面上而不是把整个仓库搬过来。所以长期记忆并不是“把消息存起来”这么简单。它至少包含三个独立能力记忆的表示能力历史信息存成什么结构才能既支持关键词匹配又支持语义理解。记忆的调度能力每轮对话到来时如何用最少的检索次数、召回最相关的内容。记忆的维护能力旧记忆何时失效、何时压缩、何时删除避免记忆库变成垃圾堆。这三件事没有一件能靠“往 Prompt 里多塞几段历史”解决必须从架构层面设计。2. 记忆分层先把“记忆”拆成四种类型在动手写代码之前先建立一个统一的分层模型。认知科学里人类记忆被分为工作记忆、短期记忆和长期记忆长期记忆里又分出情景记忆、语义记忆和程序记忆。Agent 的记忆分层可以借鉴这个框架但不必照搬我把它简化成工程上可实现的四层记忆类型典型内容保存方式生命周期访问方式工作记忆当前任务状态、临时中间结果内存对象、状态变量当前任务结束即释放程序直读短期记忆本轮/近几轮对话上下文Token 窗口、局部缓存分钟级到小时级Prompt 注入或缓存读取情景记忆用户历史说过什么、做过什么数据库、向量库、文件天级到月级检索后注入语义记忆用户偏好、项目事实、知识库规则结构化存储、图谱长期有效查询或订阅更新这个分层的核心判断是不要把短期内会变的信息和长期稳定的信息混在一起存。一旦混存你会遇到两个麻烦一是短期信息过期后污染检索结果二是长期信息被短期噪声挤占永远召不回真正重要的内容。工作记忆和短期记忆的区别很微妙。工作记忆是 Agent 正在“思考”时用的草稿纸比如正在执行一个多步任务时当前步骤的结果、下一步计划。它不需要持久化进程崩了就该丢。短期记忆则是对话层面的上下文通常直接由框架的 message history 管理比如 LangGraph 里的 messages 列表、Chat 应用的 session 上下文。长期记忆里情景记忆和语义记忆是最值得花精力设计的。情景记忆记录的是“发生了什么”天然带有时间线和实体关系适合存入数据库检索时按时间、用户、会话等维度过滤。语义记忆记录的是“事实是什么”比如用户的偏好、项目的规范、领域知识这类信息稳定性高适合向量化存储支持语义检索。程序记忆在 Agent 场景里可以理解成“技能记忆”——Agent 学会的工具调用方式、任务流程模板。它的保存方式不是文本而是可复用的函数、Prompt 模板或编排流程。很多团队的 Agent 项目做着做着就发现真正提升效率的不是把更多背景塞进记忆而是沉淀了一套可复用的任务流程库。落地时这四层对应到存储选型上常见组合是工作记忆dict、dataclass或状态图节点里的局部变量。短期记忆Redis 或内存队列保存最近 N 轮消息。情景记忆PostgreSQL / MySQL JSONB按用户和会话建索引。语义记忆向量数据库如 Milvus、Qdrant、pgvector加知识库文档。3. 记忆写入与更新策略承载“记住”环节的细节很多人以为记忆的难点在检索实际上写入策略同样关键。写入什么、更新哪条、保留多久直接决定了检索时库里的数据是否干净。记忆写入的第一原则是结构化优先原文兜底。大模型生成的对话原文适合作为原始证据保存但检索时不宜直接对原文做匹配。更推荐的做法是每段对话结束后用一个提取流程从对话里抽取出结构化记忆比如用户 ID、实体、时间、事件类型、偏好标签、结论摘要连同原文的引用 ID 一起写入存储。这样既能支持精确的结构化过滤又能在必要时回看原始上下文。记忆更新有几个常见坑第一个坑是“只新增不合并”。用户第一次说“我喜欢简洁的回答”第二次说“以后回复请带步骤”两条记忆并存检索时会同时命中模型不知道听谁的。正确的做法是在写入前先做冲突检测根据同一实体、同一属性判断是否应该覆盖旧值。第二个坑是“把临时状态写成长期记忆”。用户说“今天帮我查一下天气”这只是一次性的短期请求不值得长期存储。判断依据很简单这条信息对未来的对话还有用吗如果答案是否定的它只进短期记忆不写长期库。第三个坑是“没有记忆来源追踪”。记忆条目必须记录来自哪次会话、哪条消息、哪个时间点否则一旦记忆错了根本没法定位和修正。Agents 行业里有一个很重要但容易忽略的做法所有长期记忆都保留 provenance来源信息并且允许用户或开发者手动删除。实现上记忆写入流程通常是对话结束或达到一定轮数后触发记忆提取。由 LLM 或者规则引擎从对话中抽取候选记忆。对候选记忆做去重、冲突检测、重要性打分。写入存储同时更新对应的索引和缓存。这里涉及一个重要概念记忆提取的触发时机不是每轮都做。每轮对话都调 LLM 提取记忆成本和延迟都不可接受。更常见的策略是对话空闲超过阈值、任务节点切换、或者明确收到用户指令时才执行一次记忆压缩与写入。4. 检索机制从“全库搜索”到“多路召回 重排”检索是记忆系统里最影响用户体验的环节。同样是“用户问上次说的那个方案”如果检索不到相关记忆Agent 就等于失忆。一个容易踩的误区是以为向量检索是银弹把所有记忆丢进向量库就完事。实际效果往往很差原因有三个第一用户的自然语言和记忆中存储的历史文本未必在语义上接近。比如用户今天问“发票怎么开”昨天的记录里写的是“需要报销凭证”向量相似度可能并不高但 BM25 可以通过关键词匹配到。第二向量检索对时间、用户、会话这类结构化条件不敏感检索“最近七天内该用户的退款问题”直接靠 embedding 很难精确过滤。第三向量库召回结果通常缺少解释性模型不知道这条记忆为什么被捞出来容易在推理时产生幻觉关联。因此主流的做法是多路召回 重排序。具体拆成三步第一步是召回。同时跑两路甚至三路检索关键词路BM25 或 Elasticsearch 的 match query擅长精确词匹配。向量路query embedding 与记忆库里的 embedding 算相似度擅长语义召回。元数据路按用户 ID、时间范围、实体类型等结构化条件直接过滤。第二步是融合。把多路召回的结果按分数加权合并。常见做法是 RRFReciprocal Rank Fusion它对各路排名的倒数求和能避免不同模型的分数尺度不一致问题。计算公式不复杂每篇文档的多路排名倒数相加得到融合分。第三步是重排。用一个更精细的模型或规则对融合后的候选集做精排。轻量方案是让 LLM 判断与当前问题的相关性工程上更稳定的是训练一个小模型或者用交叉编码器。如果不希望引入额外模型至少也要做基于规则的过滤过滤过期记忆、低置信度记忆、重复记忆。检索结果进入上下文时也要有格式约束。不要把原始数据库记录直接拼进 Prompt而是统一格式化成类似“记忆条目 时间 来源”的结构让模型知道每条记忆的可信度。这一步对回答质量的影响往往比用什么向量模型更大。对于有 LangGraph 开发经验的团队记忆检索可以写成一个独立的 node在对话主链路之外执行减少对生成步的延迟影响。检索结果还可以做缓存同一用户在短时间内的重复查询直接复用上次的检索结果避免反复访问向量库。5. 遗忘机制不是删数据而是让记忆保持有效记忆系统设计里最容易被忽略的是遗忘。很多人把遗忘理解为“定期删老数据”这只是第一步。真正有效的遗忘机制是为了让系统始终能适配用户的当前状态并防止数据过度囤积带来的管理和合规风险。遗忘机制可以分为两类主动遗忘和被动遗忘。主动遗忘是系统主动淘汰不再有价值的记忆。具体手段包括TTL 过期给记忆设置生命周期比如临时偏好 24 小时过期稳定偏好 90 天过期到期自动失效。LRU 淘汰当记忆条数超过容量上限时删除最久没有被访问的条目。重要性衰减每个记忆条目带一个重要度分数随时间衰减低到阈值就进入待删除状态。摘要合并多条相关的旧记忆定期被 LLM 压缩成一条概要记忆原始细节迁移到归档区。被动遗忘是指用户或开发者主动执行的删除操作包括“清除这段记忆”“重置 Agent 记忆”。这类操作必须要有而且必须真正生效不能只是把数据标记为不可见、还残留在存储里。在设计遗忘机制时有一个容易被忽略的情况记忆失效并不等于数据物理删除。法律和隐私合规场景下用户要求删除数据时你可能需要物理删除或至少保证不可恢复。所以遗忘机制在架构上要做区分逻辑失效标记为 inactive检索时过滤掉适用于日常淘汰。物理删除从存储中真正移除适用于用户隐私请求或敏感数据清理。归档把记忆移入冷存储不再参与在线检索但保留审计追溯能力。另一个关键点是“遗忘的连带关系”。删除一条情景记忆时如果它关联着若干语义记忆需要决定是否一并处理。比如用户删除了“上个月出差”的对话记录那么从这条记录里提取出的“客户 A 喜欢在南京出差”这条语义记忆到底保留还是删除实际项目中处理方式取决于场景如果语义记忆是独立总结出的结论可以保留如果语义记忆完全源自那条删除的对话最好删除或重新评估。遗忘机制的实现不需要复杂。下面是一个按 TTL 和访问热度结合的淘汰逻辑示例核心思路每条记忆记录last_access_at和importance_score定期扫描时计算综合得分低于阈值的进入待删除队列。为了避免遗忘本身成为性能负担遗忘任务可以放到离线定时器里执行不在对话主链路里同步做。6. 完整示例一个最小可运行的 Agent 记忆模块下面用一个完整的 Python 示例演示分层记忆、写入、检索、遗忘如何协同工作。这个示例不依赖特定 Agent 框架核心数据结构和逻辑可以直接移植进 LangGraph、AutoGen 或者自研框架。6.1 记忆数据结构定义# 文件路径memory/models.py from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional dataclass class MemoryItem: memory_id: str user_id: str session_id: str memory_type: str # episodic | semantic | short_term content: str keywords: list field(default_factorylist) embedding: Optional[list] field(defaultNone) importance_score: float 1.0 source_ref: str # 来源消息或会话 ID created_at: str field(default_factorylambda: datetime.now(timezone.utc).isoformat()) last_access_at: str field(default_factorylambda: datetime.now(timezone.utc).isoformat()) expire_at: Optional[str] None status: str active # active | inactive | archived这个结构解决的是“记忆长什么样”的问题。我特意加入了source_ref和memory_type字段前者用于来源追踪后者用于分层调度。importance_score是遗忘机制里做衰减和淘汰的基础。6.2 记忆存储与写入实现# 文件路径memory/storage.py import sqlite3 import json import uuid from datetime import datetime, timedelta, timezone from memory.models import MemoryItem class MemoryStore: def __init__(self, db_path: str agent_memory.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( memory_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, session_id TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, keywords TEXT, source_ref TEXT, importance_score REAL, created_at TEXT, last_access_at TEXT, expire_at TEXT, status TEXT ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_user_status ON memories(user_id, status) ) def upsert_memory(self, item: MemoryItem) - str: if not item.memory_id: item.memory_id str(uuid.uuid4()) item.last_access_at datetime.now(timezone.utc).isoformat() self.conn.execute( INSERT INTO memories ( memory_id, user_id, session_id, memory_type, content, keywords, source_ref, importance_score, created_at, last_access_at, expire_at, status ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(memory_id) DO UPDATE SET contentexcluded.content, keywordsexcluded.keywords, importance_scoreexcluded.importance_score, expire_atexcluded.expire_at, statusexcluded.status, last_access_atexcluded.last_access_at , ( item.memory_id, item.user_id, item.session_id, item.memory_type, item.content, json.dumps(item.keywords, ensure_asciiFalse), item.source_ref, item.importance_score, item.created_at, item.last_access_at, item.expire_at, item.status ) ) self.conn.commit() return item.memory_id def update_access_time(self, memory_id: str) - None: self.conn.execute( UPDATE memories SET last_access_at? WHERE memory_id?, (datetime.now(timezone.utc).isoformat(), memory_id) ) self.conn.commit()这一步的关键是upsert_memory方法。它同时完成了“新增”和“更新”两个能力同一条记忆多次提取时以memory_id为维度合并更新而不是无限插入新记录。check_same_threadFalse在示例里方便多线程调用生产环境建议用连接池。6.3 混合召回与重排# 文件路径memory/retrieval.py import json import math from memory.storage import MemoryStore class MemoryRetriever: def __init__(self, store: MemoryStore): self.store store def _bm25_score(self, query_terms, doc_terms, doc_len, avg_len, N, df): score 0.0 k1 1.5 b 0.75 for term in query_terms: tf doc_terms.count(term) if tf 0: continue idf math.log((N - df.get(term, 0) 0.5) / (df.get(term, 0) 0.5) 1) score idf * (tf * (k1 1)) / (tf k1 * (1 - b b * doc_len / avg_len)) return score def recall(self, user_id: str, query: str, top_k: int 5): rows self.store.conn.execute( SELECT * FROM memories WHERE user_id? AND statusactive , (user_id,) ).fetchall() # 简化处理逐条算分 # 生产环境应使用 ES/向量库的内置检索 candidates [] for row in rows: doc_text row[4] keywords json.loads(row[5] or []) # 关键词匹配分 kw_score sum(1 for kw in keywords if kw in query) # 简单共享词打分 query_terms set(query.lower().split()) doc_terms doc_text.lower().split() overlap len(query_terms set(doc_terms)) score kw_score * 2 overlap * 1.5 candidates.append((score, row)) candidates.sort(keylambda x: x[0], reverseTrue) results [] for score, row in candidates[:top_k]: mem self._row_to_memory(row) mem.importance_score min(5.0, mem.importance_score 0.1) results.append({memory: mem, score: score}) return results def _row_to_memory(self, row): from memory.models import MemoryItem return MemoryItem( memory_idrow[0], user_idrow[1], session_idrow[2], memory_typerow[3], contentrow[4], keywordsjson.loads(row[5] or []), source_refrow[6], importance_scorerow[7], created_atrow[8], last_access_atrow[9], expire_atrow[10], statusrow[11] )这里为了示例简洁用关键词和重叠词做打分但你应该能看清多路召回的结构一路来自关键词一路来自文本重叠生产环境可以把recall方法替换成“向量召回 BM25 元数据过滤”的组合。重要的是recall方法需要在返回结果的同时更新last_access_at这样遗忘机制才知道哪些记忆是最近被使用过的。如果接入真实向量检索可以把recall扩展成同时查询向量库和 SQLite然后做 RRF 融合。伪代码如下def hybrid_recall(self, query, user_id, top_k5): bm25_hits self._bm25_retrieve(query, user_id, top_k10) vector_hits self._vector_retrieve(query, user_id, top_k10) fused self._rrf_fuse([bm25_hits, vector_hits]) return fused[:top_k]6.4 遗忘与清理策略# 文件路径memory/forgetting.py from datetime import datetime, timedelta, timezone from memory.storage import MemoryStore class ForgettingScheduler: def __init__(self, store: MemoryStore): self.store store def expire_old_memories(self, max_age_days: int 90) - int: 按 TTL 逻辑过期超过 max_age_days 未访问的记忆置为 inactive cutoff datetime.now(timezone.utc) - timedelta(daysmax_age_days) cur self.store.conn.execute( UPDATE memories SET statusinactive WHERE statusactive AND last_access_at ? , (cutoff.isoformat(),) ) self.store.conn.commit() return cur.rowcount def decay_importance(self, factor: float 0.99) - int: 重要性衰减活跃但低于阈值的记忆进入待删除队列 self.store.conn.execute( UPDATE memories SET importance_score importance_score * ? WHERE status active, (factor,) ) cur self.store.conn.execute( UPDATE memories SET statusinactive WHERE statusactive AND importance_score 0.1 ) self.store.conn.commit() return cur.rowcount def summarize_and_compress(self, user_id: str, session_id: str) - None: 摘要合并的占位方法真实项目里调用 LLM 把同用户同会话的多条情景记忆压缩成一条语义记忆。 pass def hard_delete(self, memory_ids: list) - int: 物理删除用于用户主动请求或隐私清理 placeholders ,.join(? * len(memory_ids)) sql fDELETE FROM memories WHERE memory_id IN ({placeholders}) cur self.store.conn.execute(sql, memory_ids) self.store.conn.commit() return cur.rowcount遗忘调度器把“过期”“衰减”“摘要压缩”“物理删除”四种操作拆成了独立方法。设计上的主要考虑是不同操作的执行频率不同过期可以每天跑一次衰减可以每小时跑一次摘要压缩可以放在会话结束时触发物理删除则按需手动调用。这种拆分能让遗忘逻辑独立测试和观测。6.5 使用示例# 文件路径demo.py from memory.storage import MemoryStore from memory.models import MemoryItem from memory.retrieval import MemoryRetriever from memory.forgetting import ForgettingScheduler store MemoryStore(demo_memory.db) retriever MemoryRetriever(store) scheduler ForgettingScheduler(store) # 写入一条情景记忆 mem MemoryItem( memory_id, user_iduser_001, session_idsession_001, memory_typeepisodic, content用户反馈希望售后回复附带退款到账时间预估, keywords[退款, 售后, 到账时间], importance_score2.0, source_refsession_001:msg_12 ) store.upsert_memory(mem) # 写入一条语义记忆 sem MemoryItem( memory_id, user_iduser_001, session_idsession_000, memory_typesemantic, content用户偏好简洁回复重要步骤需要编号, keywords[偏好, 简洁], importance_score3.5, source_refsession_000:msg_3 ) store.upsert_memory(sem) # 检索用户的相关记忆 results retriever.recall(user_001, 退款什么时候到账) for r in results: print(f{r[score]:.2f} | {r[memory].content}) # 执行遗忘策略 print(expired:, scheduler.expire_old_memories(max_age_days30)) print(decayed:, scheduler.decay_importance(factor0.95))这个示例跑通之后你就拥有一个最简但可扩展的记忆系统骨架。7. 运行结果与效果验证在命令行里执行python demo.py预期输出大致是3.50 | 用户偏好简洁回复重要步骤需要编号 3.00 | 用户反馈希望售后回复附带退款到账时间预估 expired: 0 decayed: 0注意这里检索“退款什么时候到账”语义记忆“用户偏好简洁回复”也进入了候选原因是它包含了“用户”和“回复”等重叠词。这正是多路召回的典型现象召回结果不一定完全精确需要后续重排。真实项目中你可以通过调试日志观察每条命中的分数构成判断是关键词贡献大还是语义相似度贡献大。验证一个记忆系统是否合格不要只看“能不能搜到”要按下面的维度做效果测试验证维度测试方法通过标准召回准确性构造 20 条测试查询人工标注期望记忆Top5 命中率不低于 80%记忆更新同一事实两次写入内容变化后查询返回的是最新值而非旧值时效性写入过期记忆后查询过期记忆不出现在结果里用户隔离查询用户 A 的上下文不出现用户 B 的任何记忆来源可溯随机抽查 10 条记忆能定位到原会话和消息删除生效调用 hard_delete 后查询不再出现在召回结果中常见情况是召回准确率一开始只有 50% 左右。不要急着换 embedding 模型先检查是不是写入阶段没有做关键词抽取、没有给记忆打类型标签。多数召回问题其实出在“入库质量”上。8. 常见问题与排查思路问题现象可能原因排查方式解决方案检索总返回无关记忆记忆写入时没有提取关键词和类型标签查看记忆原始字段检查 keywords 是否为空在写入前增加关键词抽取和记忆分类步骤同一用户重复写入同一条记忆缺少去重或冲突检测检查 memory_id 是否每次重新生成基于内容哈希生成 memory_id或写入前做相似度匹配检索耗时过高每次请求都全表扫描开启 SQL 日志查看查询耗时引入向量数据库、Elasticsearch或增加 Redis 缓存用户要求删除后仍能搜到删除只做了逻辑标记但检索没有过滤检查查询语句是否包含 statusactive所有检索方法必须显式过滤 status 字段记忆越存越多Prompt 被撑爆没有做记忆数量上限控制查看每条记忆的 importance_score 分布设置 active 记忆条数上限超出后触发摘要合并多轮对话后历史被覆盖短期记忆和长期记忆共用存储键检查写入时 memory_type 是否正确短期记忆单独存 Redis长期记忆走数据库LLM 调用提取记忆失败提取提示词不明确或输出格式不固定单独测试提取 Prompt 的输出改用结构化输出模式用 Pydantic 或 JSON schema 约束排查的通用套路是先看写入日志确认数据存对了没有再看检索日志确认查询条件对不对最后看 Prompt 拼接确认最终喂给模型的记忆是否被正确格式化。记忆系统的问题七成发生在写入两成发生在检索只有一成发生在模型推理。9. 最佳实践与工程建议结合前面的架构设计和实际工程经验长期记忆系统有几个值得固化的工程原则。第一记忆库要和应用状态解耦。不要直接复用业务数据库的表结构也不要让 Agent 偶然的对话内容污染核心业务数据。记忆库独立建表、独立配置、独立权限方便单独清理和备份。第二所有记忆写入必须异步化。对话主链路的延迟瓶颈非常敏感记忆提取、向量化、写入数据库这些操作不应该阻塞用户响应。可以先把原始消息写入消息队列后续由消费者异步完成记忆提取和入库。第三记忆检索要做缓存。用户在同一时间段内的记忆集合变化不大完全可以用 Redis 缓存“用户级 Top-K 记忆列表”设置 5 到 10 分钟过期。这样可以省掉大量重复向量查询。第四记忆提取用稳定 Prompt 加结构化输出。不要依赖自由文本抽取否则字段一会儿叫preference一会儿叫hobby下游没法处理。定义好记忆数据结构让模型按 JSON 输出解析失败走重试。第五缺失来源追踪的记忆宁可舍弃。如果一条记忆无法定位到产生它的会话和消息那它在排查问题和处理用户删除请求时就是隐患。写入前必须携带source_ref。第六遗忘机制要可控、可观测。遗忘任务跑完后输出本次清理了多少条、释放了多少空间、哪些记忆被压缩合并。遗忘逻辑的日志同样要保留杜绝“偷偷把用户数据丢了”的体验。第七不同场景采用不同的记忆策略。客服机器人重点维护用户订单和偏好编程助手重点维护项目上下文和代码风格个人助理重点维护日程和关系图谱。不要设计一套通用记忆系统然后去套所有场景记忆字段的设计本质上是对业务模型的抽象。第八隐私与安全不能省。记忆库里可能有用户敏感信息存储时尽量脱敏至少加密敏感字段。检索接口必须带用户身份鉴权不能出现用户 A 的 ID 能查到用户 B 记忆的情况。涉及删除操作多做一步确认避免误删。10. 总结与后续学习方向本文主要把 Agent 长期记忆拆成四个核心环节分层、写入、检索、遗忘。分层上要区分工作记忆、短期记忆、情景记忆和语义记忆不同层用不同存储。写入上重点是结构化、去重、冲突合并和来源追踪。检索上不要迷信向量检索采用多路召回和重排能显著提升命中率。遗忘上用 TTL、重要性衰减、摘要合并、物理删除组合实现可控遗忘。文中的示例代码是一个可运行的最小骨架建议你在此基础上做二次开发先按自己的业务需求调整MemoryItem的字段再替换检索模块为向量库加 BM25 的真实实现最后补上异步写入和遗忘日志。如果你的项目已经基于 LangGraph可以从MemoryStore封装一个长期记忆中间件在每个对话节点结束后调用upsert_memory在节点执行前调用recall。这套思路和 LangGraph 的 state 管理天然兼容不需要改框架核心。下一步值得深入的方向有三个一是记忆摘要压缩的提示词设计这是减少记忆冗余的关键二是记忆冲突检测算法关系到同一用户多次表达相反偏好时系统怎么处理三是记忆评估体系如何用量化指标衡量一套记忆系统“记得准、忘得对、查得快”。把这三点补齐你的 Agent 长期记忆能力会明显超过大多数 Demo 级实现。建议先把本文的示例跑通再挑其中一个方向继续深入。