☰
AI Agent记忆架构实战:三层记忆分层设计与遗忘机制全解
2026/10/6 6:09:33 网站建设 项目流程

1. 从一次“失忆事故”说起:Agent 记忆到底缺在哪

做 Agent 开发的朋友,十有八九都经历过这种崩溃现场:用户上一轮刚告诉你“我叫老张,做跨境电商的,主攻欧美市场”,下一轮问“你记得我是做什么的吗”,Agent 一脸茫然地开始瞎编。更离谱的是,有些 Agent 在同一个会话里都能“翻脸”——前脚刚确认完订单信息,后脚就问你订单编号是多少。

这真的不是模型笨,也不是 prompt 写得不好,而是记忆架构压根没设计。我这两年帮好几家团队调过商用 Agent,几乎每次排查“健忘”问题,最后都落到同一个工程根源上:记忆被当成一整块聊天记录来对待,而没有按生命周期去拆分和管理。

先说清楚“健忘”发生在哪里。最普遍的一层,是模型的上下文窗口天然有限——哪怕现在 Claude、GPT 这类模型的上下文做到 200K token,你也不可能把所有历史对话全部塞进 prompt。塞不下是一层问题,塞进去之后互相干扰、关键信息被淹没,是更隐蔽的另一层问题。我见过有团队图省事,把全部历史消息拼进系统提示词,结果 token 费用按月暴涨,回答质量反而明显下降。原因很简单:信息装进上下文,不等于被模型“记住”,在超长上下文中提取关键信息的精度会肉眼可见地衰减。

商用场景里对记忆的真实需求,通常可以归纳成四类:会话连续性(同一个会话内上下文连贯)、用户画像沉淀(跨会话记住用户偏好和身份信息)、业务状态持久化(订单、任务、进度这类状态数据不能丢)、知识库接入(让 Agent 基于企业文档做问答)。如果只靠模型天然上下文,这四类需求一个都不可靠。所以正确做法,是把记忆拆成不同层级,每层管自己擅长的事,再用一套编排逻辑把它们串起来。这套东西,就是下文要展开的“会话记忆—短期记忆—长期记忆—遗忘机制”全链路。

2. 三层记忆架构怎么拆:会话、短期、长期各管一摊

2.1 为什么必须分层,而不是一把梭

很多人第一反应是“搞个 Redis 存聊天记录不就行了”,这就是典型的把记忆等同于存储。记忆系统真正难的,不是“存得下”,而是“取得到”和“取对了”。

分层设计背后的逻辑,来自一个核心矛盾:信息的时效性需求不同。用户刚说的“现在帮我查一下 A 订单”,这是秒级时效,必须立刻可用;用户三天前透露的“我偏好晚上处理邮件”,这是长期画像,要在合适的场景被重新唤醒;还有大量中间态信息——比如某个任务进行到一半的临时状态,既不能丢,又不能长期占着位置。时效性不同,存储格式、检索方式、过期策略就全都不一样。放在一个桶里,要么检索效率极低,要么互相污染。

我习惯把记忆分成三层。会话记忆(Conversation Memory)负责当前会话内的消息序列,保底保证“接得上话”;短期记忆(Short-term Memory)负责跨轮次但不过期的“工作记忆”,比如用户正在进行的任务流程、刚确认的临时偏好,通常自带 TTL;长期记忆(Long-term Memory)负责结构化沉淀的用户画像、业务事实和历史关键时刻,需要主动提取和写入。这三层在存储上天然分开,在读取时按优先级合并,遗忘策略也各有各的节奏。

2.2 每一层该存什么、不该存什么

这层我踩过最大的坑,就是“什么都往长期记忆里写”。看起来万无一失,实际上会把检索质量拖垮。分层不是简单按时间切,而是按信息类型切。

会话记忆只存原始消息和消息附带的元数据(时间戳、角色、消息 ID),原则上不改写内容,只是做截断或摘要。短期记忆存的是“流程状态”和“临时结论”,比如用户正在填的表单、当前选中的筛选条件、上一步操作产生的结果 ID。长期记忆存的是“事实”,包括用户身份画像(称呼、行业、偏好)、业务关键记录(订单历史、已确认的需求)、以及经过提炼的“关键结论”(比如用户明确说过的某条决策原因)。

一个简单判断标准:这条信息如果三条之后还用得上,放短期;如果下次会话还用得上,放长期;如果只是当前对话的铺垫,留在会话记忆里就行。按这个标准过滤,长期记忆的体量能压缩掉 70% 以上,检索精度完全不是一个量级。我还习惯给每层记忆加“写入审计”,记录是哪个环节、基于哪条原始消息写入的,出了问题好回溯——商用环境里,这条对排查“记忆污染”极其重要。

3. 全链路代码实战:从消息存取到主动遗忘

下面这部分我直接给出可运行的参考实现。为便于演示,我用 Python 写,存储层对 Redis 和向量库做了抽象,模型调用统一走一个llm_complete函数,实际项目里替换成 OpenAI SDK 或 Claude SDK 即可。

3.1 会话记忆:消息序列的组织与滑动窗口

会话记忆最朴素的实现,就是把消息追加进列表,拼 prompt 时取最近 N 轮。但商用环境有个细节:不能只按“轮”截断,要按 token 数截断。因为用户一句话可能写 2000 token,也可能一个字“好”。我见过按条数截断导致的上下文爆炸事故,所以老实上 tiktoken 或对应模型的 tokenizer 做预算。

from dataclasses import dataclass, field from typing import List, Dict, Optional import time, uuid, json @dataclass class Message: role: str # user / assistant / system / tool content: str msg_id: str = field(default_factory=lambda: uuid.uuid4().hex) ts: float = field(default_factory=time.time) class ConversationMemory: """会话记忆:维护消息序列,按 token 预算做滑动窗口截断""" def __init__(self, max_tokens: int = 8000, tokenizer=None): self.messages: List[Message] = [] self.max_tokens = max_tokens self._tokenize = tokenizer or (lambda s: len(s) // 2) # 退化方案:按字符估算 def append(self, role: str, content: str): self.messages.append(Message(role=role, content=content)) self._enforce_budget() def _enforce_budget(self): """从旧到新丢弃消息,直到总 token 数低于预算""" total = sum(self._tokenize(m.content) for m in self.messages) drop_count = 0 while total > self.max_tokens and len(self.messages) > 1: dropped = self.messages.pop(0) total -= self._tokenize(dropped.content) drop_count += 1 if drop_count: self._archive_to_short_term(drop_count) def _archive_to_short_term(self, drop_count: int): """被挤掉的消息不直接丢弃,交给短期记忆做摘要——这里先留个钩子""" # 实际实现见 3.3,会把最早的一批消息送给 LLM 做摘要 pass def to_prompt_messages(self) -> List[Dict[str, str]]: return [{"role": m.role, "content": m.content} for m in self.messages]

这里最关键的设计是_archive_to_short_term。很多人直接把旧消息删掉,导致用户上一句话“请基于我刚才发的需求文档回答”直接失效。正确做法是:被挤出的旧消息进入压缩通道,让 LLM 生成摘要后存入短期记忆,这样既控制 token 预算,又不丢关键信息。压缩摘要的触发点,正是截断发生的瞬间,而不是定时任务——因为只有这个时候我们才知道哪些信息即将“离开”上下文。

3.2 短期记忆:带 TTL 的工作记忆与 token 预算

短期记忆我常用 Redis 来存,字段结构非常简单:key 是short_term:{user_id}:{session_id},value 是一组带优先级的记忆条目。每个条目有四个关键属性:内容、优先级、写入时间、过期时间。优先级决定了上下文紧张时谁先被保留,也决定了“写入 token 预算”时谁先被选中。

class ShortTermMemory: """短期记忆:带 TTL 和优先级的工作记忆,用于流程状态与临时结论""" def __init__(self, redis_client, ttl_seconds: int = 1800, max_items: int = 20): self.r = redis_client self.ttl = ttl_seconds self.max_items = max_items def set(self, key: str, content: str, priority: int = 5): """写入一条短期记忆。priority 越高越优先保留。""" item = { "content": content, "priority": priority, "write_ts": time.time(), "expire_ts": time.time() + self.ttl, } self.r.hset(f"st:{key}", item["content"], json.dumps(item)) self._enforce_max_items(key) def get_active(self, key: str) -> List[Dict]: """获取未过期且未失效的短期记忆""" raw_items = self.r.hgetall(f"st:{key}") active = [] for _, raw in raw_items.items(): item = json.loads(raw) if item["expire_ts"] > time.time(): active.append(item) return sorted(active, key=lambda x: -x["priority"]) def _enforce_max_items(self, key: str): """条目超限时,优先淘汰低优先级且较旧的记忆""" all_items = self.get_active(key) if len(all_items) <= self.max_items: return # 排序:优先级升序、时间升序,先淘汰优先级低且老的 to_drop = sorted(all_items, key=lambda x: (x["priority"], x["write_ts"])) for item in to_drop[: len(all_items) - self.max_items]: self.r.hdel(f"st:{key}", item["content"])

短期记忆的 TTL 我一般设 30 分钟,原因是商用 Agent 的典型会话间隔不会太长,30 分钟足够覆盖“用户离开一会再回来继续操作”的场景。如果用户超过这个时间没回来,这些临时状态本身就失去意义,直接过期反而是好事——避免旧状态干扰新会话。优先级字段是实战里加出来的,最初版本没有,结果发现任务流程中的“当前步骤状态”经常被“用户随口说的偏好”挤掉,加了优先级之后,任务状态永远 priority=9,临时闲聊一律 priority=3,稳定多了。

3.3 长期记忆:向量化存储与提取式检索

长期记忆是商用 Agent 记忆体系的压舱石。我这边用 ChromaDB 做向量库,embedding 用现成模型(比如 text-embedding-3-small),生产环境也可以换 Qdrant 或 Milvus。核心思路是:只在关键节点写入,在响应用户前主动检索相关记忆并注入上下文。

import chromadb from chromadb.utils import embedding_functions class LongTermMemory: """长期记忆:向量化存储 + 元数据过滤 + 提取式检索""" def __init__(self, collection_name: str = "agent_long_term"): self.client = chromadb.PersistentClient(path="./agent_memory_store") self.ef = embedding_functions.OpenAIEmbeddingFunction( api_key="YOUR_KEY", model_name="text-embedding-3-small" ) self.collection = self.client.get_or_create_collection( name=collection_name, embedding_function=self.ef ) def write_fact(self, user_id: str, fact_type: str, content: str, importance: float = 0.5): """写入一条长期记忆。fact_type 便于业务维度过滤,importance 控制检索权重。""" mem_id = f"{user_id}:{uuid.uuid4().hex}" self.collection.add( ids=[mem_id], documents=[content], metadatas=[{ "user_id": user_id, "fact_type": fact_type, "importance": importance, "write_ts": time.time(), }], ) def recall(self, query: str, user_id: str, top_k: int = 5, min_score: float = 0.15, fact_type: Optional[str] = None) -> List[str]: """按相关性检索,可叠加用户维度和业务类型过滤""" where = {"user_id": user_id} if fact_type: where["fact_type"] = fact_type results = self.collection.query( query_texts=[query], n_results=top_k, where=where, include=["documents", "metadatas", "distances"], ) summaries = [] for doc, meta, dist in zip( results["documents"][0], results["metadatas"][0], results["distances"][0], ): score = 1 - dist if score < min_score: continue # 用 importance 做加权,避免低价值信息挤占上下文 weighted_score = score * (0.5 + meta["importance"]) summaries.append((doc, weighted_score)) summaries.sort(key=lambda x: -x[1]) return [s[0] for s in summaries[:top_k]]

检索这步有个细节特别值得说:不要只按向量相似度排。我最初调试时经常发现,模型反而被“相似但无关”的记忆误导。比如用户问订单物流,检索出来的记忆全是“用户抱怨过某次物流慢”,这确实是语义相似,但根本不是当前需要的订单状态。后来加上fact_type过滤和importance加权之后,效果立竿见影。另一个细节是写入时机:我强烈建议只在“对话产生明确事实”时写入,而不是每轮都写。判断逻辑很简单——用户用了陈述性语气说出身份、偏好、决策,或者系统产生了任务结果,这两类才值得写长期记忆。

3.4 遗忘机制:显式清除、自动过期与检索弱化

遗忘是整套架构里最容易被忽略、却又最能拉开体验差距的部分。我见过太多 Agent“记太好”——用户三个月前一句玩笑话,也被当作画像反复引用,结果显得又蠢又冒犯。商用环境里,遗忘不仅是技术问题,还是合规和体验问题。“该忘的时候忘不掉”,比“该记的时候记不住”更致命。我的遗忘机制从三个方向做。

class ForgetManager: """遗忘管理:主动遗忘 + 被动过期 + 检索弱化,三个层次配合""" def __init__(self, long_term: LongTermMemory, short_term: ShortTermMemory, redis_client): self.ltm = long_term self.stm = short_term self.r = redis_client # 1) 显式遗忘:用户说“忘掉”或业务要求清除时,直接删 def explicit_forget(self, user_id: str, keywords: Optional[List[str]] = None): """按用户维度或关键词维度删除长期记忆。关键词为空则全删。""" all_mem = self.ltm.collection.get(where={"user_id": user_id}) if keywords is None: ids = all_mem["ids"] else: # 简单实现:关键词命中即删。生产环境建议先检索再删,避免误伤。 ids = [ mem_id for mem_id, doc in zip(all_mem["ids"], all_mem["documents"]) if any(kw in doc for kw in keywords) ] if ids: self.ltm.collection.delete(ids=ids) # 2) 自动过期:长期记忆也设保鲜期,重要度高的更长寿 def collect_expired(self, importance_decay: float = 0.1): all_mem = self.ltm.collection.get() now = time.time() expired_ids = [] for mem_id, meta in zip(all_mem["ids"], all_mem["metadatas"]): age = now - meta.get("write_ts", now) lifespan = 30 * 24 * 3600 * (1 + meta.get("importance", 0.5)) if age > lifespan: expired_ids.append(mem_id) if expired_ids: self.ltm.collection.delete(ids=expired_ids) # 3) 检索弱化:不删,但让旧记忆的检索分数自然衰减 def apply_decay_to_recall(self, recall_result: List[tuple], now: float = time.time()): decayed = [] for doc, score, meta in recall_result: age_days = (now - meta.get("write_ts", now)) / 86400 decay_factor = max(0.5, 1 - age_days * 0.01) decayed.append((doc, score * decay_factor)) return decayed

显式遗忘要处理一个语义问题:用户说“忘掉我上次说的地址”,你不能直接把整个 user_id 的记忆全清了,那是一种“核弹式遗忘”。我实际开发时给 ForgetManager 加了“记忆条目级”的追溯能力——每条记忆写入时记录来源消息 ID,遗忘时定位到具体消息,再顺藤摸瓜删除由它派生的所有摘要。自动过期这层,我设的寿命基线是 30 天,importance 越高活得越久,这样“用户明确强调过的重要决策”能存半年,“随口一提的偏好”两周就淡出。检索弱化是第三道防线,不删除数据,只降低旧数据的命中概率,防止冷门旧记忆在特定场景下“诈尸”。

4. 记忆“翻车”现场:常见问题与排查实践

4.1 检索召回结果质量差,模型被记忆带偏

这是我在商用项目里遇到最多的问题。现象很典型:用户问“帮我看看上个月的报表”,Agent 检索回来的记忆却是“用户上次抱怨过报表格式不对”,然后开始道歉,完全不干活。排查路径我一般走三步。第一步查 embedding 的效果:把检索结果和 query 打印出来,人工看相似度排名是否合理,如果明显不合理,先怀疑 embedding 模型选型,换更适配领域数据的模型。第二步查过滤条件:确认是否按 user_id 和 fact_type 做了隔离,很多时候是测试时没区分用户,导致 A 用户的记忆被 B 用户检索到。第三步查加权逻辑:importance 权重是不是太高,导致语义匹配度一般但权重高的记忆压过了真正相关的记忆。我最后把加权上限压到 0.3,相关性排序稳定很多。

另一个容易踩的坑是检索结果直接拼 prompt 的顺序。我一开始按分数从高到低排,结果模型总是优先响应最靠前的记忆,哪怕它只是边缘相关。后来改成“最相关的放中间、次相关的放前后”的布局,配合 prompt 里“以下记忆仅供参考,以用户当前问题为准”的约束,准确率明显提升。这个经验未必普适,但值得在调试时花十分钟试一试。

4.2 记忆污染:会话里的一句话污染了长期画像

假设用户在某次会话里说“我今天想取消所有订阅”,这句话被写进了长期记忆,之后每次对话 Agent 都在劝用户退订——这就是典型的记忆污染。根本原因是写入逻辑太激进,把“临时情绪表达”当成了“稳定事实”。我在长期记忆写入前加了一道“事实性校验”钩子:用 LLM 判断该句话是否满足“陈述事实、面向未来、非情绪化”,只有三类可以通过——身份类(我是谁、我在哪、我的职业)、偏好类(我喜欢/我不喜欢/我需要)、业务类(已完成的任务、确认的决策)。情绪化表达、假设性提问、临时状态一律拦截。

这个校验本身会消耗额外 token,但我算过账,值得。一台商用 Agent 每天处理几千次对话,如果没有校验,污染记忆带来的错误回答成本,远比那点校验 token 高。校验通过后,写入前还会做一次“去重合并”:检索已有记忆中是否已有等价条目,有则更新原条目的 write_ts 和 importance,而不是新增一条。不然三个月后同一个偏好存了 20 条,检索时全是它,也是另一种污染。

4.3 并发问题:同一个用户的多会话互相踩记忆

商用 Agent 经常一个用户同时开着多个会话——网页端一个、App 一个、企微机器人又一个。如果记忆系统不做会话隔离,很容易出现“App 里改的偏好,把网页端的行为记录覆盖了”。我在设计里给每条记忆加了scope字段:session级记忆只在本会话可见,user级记忆跨会话共享。写入时显式指定 scope,读取时根据当前上下文决定合并策略——用户级记忆永远加载,会话级记忆只在本会话内加载。

真正麻烦的是并发写同一份用户画像。比如用户在网页端说“改邮箱”,同时在 App 里说“改手机号”,两个会话同时更新 user 级记忆,后写的把先写的覆盖了。我在业务层处理:写入用户级记忆前,先读取旧值,用“合并而非覆盖”的策略,把不同字段分别更新。这个逻辑不复杂,但真出问题时极其隐蔽,尤其是企微机器人这种回调并发高的场景,不加锁或合并策略,用户画像三天两头莫名丢字段。参考实现里 Redis 的hset逐字段更新,天然规避了这个问题——这也是我推荐用哈希结构存画像的原因。

4.4 上下文膨胀与 token 费用失控

很多团队问我“记忆都接上了,为什么 prompt 还是那么大”。查到最后,几乎都是同一类问题:把“检索到的记忆”和“全部短期记忆”不加筛选地拼进上下文。三层记忆的设计初衷就是控制 token,但控制必须在读取侧完成。我建议的读取策略是:会话记忆控制在 60% 预算,短期记忆按优先级取前 3-5 条、约占 15%,长期记忆只取检索命中的前 2-3 条、约占 10%,剩余 15% 留作系统指令和输出预留。这个比例不是拍脑袋,是我在多个项目里调出来的经验值,核心原则是:记忆是参考信息,不是主角,主角永远是当前用户输入和相关工具结果。

5. 复盘:我在实际项目中沉淀的几条经验

这套记忆架构我在电商客服、销售线索管理、企业知识问答三个方向的 Agent 项目里跑过,说几条最实在的体会。

第一,记忆系统的迭代节奏,永远跟着“用户可感知的错误”走,而不是跟着“技术完备度”走。我最早做了很复杂的记忆图谱,但用户根本感知不到差别,反而增加排查成本。后来砍掉 80% 的功能,只保留“会话、短期、长期、遗忘”四件套,稳定性反而大幅提升。商用环境里,记忆系统最重要的指标不是“多智能”,而是“可预期”——用户说过的关键信息不丢,用户没说过的话不乱编,这就已经超越市面上九成 Agent。

第二,遗忘和写入同样重要,甚至更重要。我接手的一个客户项目,之前 Agent 被吐槽“越来越陌生”,排查发现是长期记忆里积累了太多过期业务状态,比如三个月前的库存数字、两个月前的促销价格,检索时反复命中,导致回答看起来永远在说旧数据。上了自动过期和重要性衰减之后,这类错误消失了大半。如果你的 Agent 也出现“越用越蠢”的迹象,先检查记忆系统是不是“只进不出”。

第三,所有记忆操作都要留审计日志。商用环境一旦出事,你需要在 10 分钟内回答“这条错误记忆是什么时候、基于什么消息、由哪个模块写入的”。没有审计,排查就是大海捞针;有审计,十分钟定位。这条我在多个项目里反复验证,属于“平时没用、出事救命”的设计。

最后分享一个调试技巧:我会在开发环境给每条注入 Agent 的记忆加一行不可见的注释标记,比如<!-- MEM:user_profile#20240901 -->,这样模型回答时如果引用了某条记忆,你能直接在输出里看到它引用了哪条、来自什么时候。这个技巧帮我发现过不少“模型脑补记忆”的问题——它明明没检索到任何记忆,却假装记得,有了标记就能一眼识破。希望这套链路和代码,能帮你把那个“健忘的 Agent”治好。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询