你是不是也遇到过这种情况昨天刚跟 Agent 对齐过的业务规则今天换个问法它就“失忆”了同一段上下文里明明给了明确指令多聊几轮之后它又开始自说自话。很多人第一反应是模型能力不行但我做了几个商用 Agent 项目之后得出一个更准确的结论大多数 Agent 的健忘根本不是模型理解力的问题而是记忆架构没设计好。会话记忆、短期记忆、长期记忆、遗忘机制这四层链路如果没理顺再强的模型也像一个没存档的游戏玩家每次开局都从零开始。这篇文章我会用一套可以落地的代码把商用 Agent 的记忆架构完整拆开从最基础的会话记忆管理到短期工作记忆、长期向量记忆再到大多数人会忽略的遗忘机制一条链路走完。适合正在做 Agent 应用、被“多轮对话失忆”“上下文塞爆”“知识无法沉淀”折磨的开发者参考。看完你至少能诊断出自己项目里的健忘到底发生在哪一层。1. 先搞懂Agent 为什么会“健忘”1.1 健忘的本质是状态丢失先说一个很多人没真正意识到的事实LLM 本身是无状态的。你每次调用模型接口它都是在独立地完成一次推理上一次调用留下的任何信息都不会自动保留。所谓 Agent 有记忆本质是我们把历史信息重新塞回上下文里让模型“看起来记得”。理解这一点之后“健忘”就有了非常具体的含义状态没有在正确的时间、以正确的方式被保存、传递和恢复。一个完整的 Agent 在真实业务里要维护的状态远不止对话文本。我梳理了一下至少有这几类对话历史用户和助手一来一回的消息序列包含角色标记和工具调用记录用户偏好比如“回复尽量简短”“不要用表格”“称呼我为李工”业务状态当前任务进行到哪一步哪些字段已确认哪些还没有工具结果上一步调用 API、数据库、文件系统返回的中间数据环境约束当前的时间、用户权限、可用渠道等常量信息这些状态如果在代码里只是零散地存在局部变量里一次请求结束就全部丢失了。就算你用了 Redis 或者数据库如果读写逻辑没有一个统一的架构照样会出现“这里写了那里没读”的漏档问题。我习惯用一个比喻来解释这个事人的记忆也不是一个文件夹而是分层的。你记不住三个月前某顿午饭吃了什么细节但你会记住这家店很难吃你记住今天上午和同事开会讨论的重点但会议逐字稿你懒得也记不住。Agent 的记忆也应该是分层设计而不是一股脑全塞进上下文。1.2 记忆分层从一次性对话到可沉淀知识商用 Agent 的记忆架构我建议至少要分四层记忆层级生命周期典型存储核心作用会话记忆一次会话内内存数组、Redis List保持多轮对话连贯性短期记忆一个任务周期内Redis、结构化对象保存当前任务的临时状态与用户偏好长期记忆跨会话、跨任务向量库、关系数据库沉淀用户画像、历史事实、业务经验遗忘机制持续运行定时任务 衰减计算清理噪声、防冲突、控成本为什么要这样分层核心原因是成本收益的取舍。LLM 的上下文窗口不是无限的就算有 128k 甚至 200k 的窗口你也不可能把所有历史都塞进去——token 贵推理慢而且信息一多模型反而抓不住重点。分层的基本逻辑是高频复用的信息放到读写快的地方低频但重要的信息放到持久可靠的地方没有价值的信息及时清理。我见过不少团队上来就搭向量库把所有对话记录全部向量化然后每次请求都检索。这样做的问题很明显向量库负责的是“语义相似度召回”它根本不适合做精确定序的多轮对话跟踪。你要让 Agent 记住上一句它说了什么消息数组就是最合适的工具上向量库反而是杀鸡用牛刀。所以整个架构的推进顺序应该是先把会话记忆做好再补短期工作记忆然后才是长期记忆和遗忘机制。别跳级。2. 会话记忆让 Agent 记得“刚才聊了什么”2.1 消息队列与上下文窗口会话记忆是最基础也最容易做崩的一层。表面上看就是维护一个消息数组每次调用 API 时把这个数组传进去。我用 OpenAI 兼容接口写一个最简实现from dataclasses import dataclass, field from typing import List, Dict, Any dataclass class ConversationBuffer: session_id: str messages: List[Dict[str, Any]] field(default_factorylist) def append(self, role: str, content: str, **kwargs): msg {role: role, content: content} msg.update(kwargs) self.messages.append(msg) return msg def to_openai_format(self) - List[Dict[str, Any]]: return [{role: m[role], content: m[content]} for m in self.messages]这段代码看着简单但真正商用之后你会遇到几个很现实的问题。第一消息里的角色不能丢。你不能只存文本内容role 字段必须保留。因为模型对角色非常敏感user、assistant、system、tool 各自承担不同的语义如果角色错乱模型的行为会变得不可预测。第二工具调用消息必须成对保存。如果你的 Agent 会调用工具那 function_call 的请求和 tool 的返回结果必须紧挨着放在一起。一旦顺序乱了模型就不知道某个工具结果对应哪次调用接着就会开始自说自话地编造工具输出。第三消息要带元数据。每条消息至少要记录时间戳、token 数、来源渠道。这不仅是排查问题的需要也是后面做窗口裁剪和记忆更新的计算基础。我自己的做法是给 append 加上 token 估算def append_with_tokens(self, role: str, content: str, **kwargs): msg self.append(role, content, **kwargs) msg[tokens] estimate_tokens(content) return msgestimate_tokens 可以用 tiktoken 做精确统计也可以用“字符数除以 4”这种粗估算。商用场景建议上 tiktoken因为窗口裁剪依赖这个数值算不准就会把不该裁的裁掉该裁的没裁掉。2.2 滑动窗口裁剪与摘要降载会话记忆的第二个痛点是上下文窗口有限。你不能无限地把消息往数组里堆否则迟早会超出模型的最大 token 限制。商用里最常见的两种策略是滑动窗口和摘要压缩。滑动窗口的思路很直接只保留最近 N 轮对话更早的消息直接丢弃。实现起来也很简单def trim_to_window( messages: List[Dict], max_tokens: int, estimate_func ) - List[Dict]: 从最新消息往前裁剪直到总 token 数不超过阈值的 80% trimmed [] total 0 # 倒序处理从最新消息往前累加 for msg in reversed(messages): tokens estimate_func(msg[content]) if total tokens int(max_tokens * 0.8): break trimmed.append(msg) total tokens return list(reversed(trimmed))这里有个细节必须注意阈值不要设成 max_tokens 的 100%要留出 20% 左右的余量。因为模型输出也要占 tokensystem prompt 和工具定义也要占窗口如果裁剪时把窗口塞满等到模型生成回复时就会报“上下文长度超限”的错。我刚开始做的时候就在这里吃过亏明明窗口设了 128k请求一发出就报错查了半天才发现是没留输出余量。滑动窗口的问题是它会无差别地丢弃早期信息。假设用户在第 3 轮告诉 Agent“我公司在上海发票要开增值税专用发票”第 20 轮再问“发票地址怎么写”此时第 3 轮的信息已经被裁掉了Agent 又变回一问三不知的状态。针对这种情况我推荐的做法是摘要压缩。思路是当消息数组超过阈值时把最早的一批消息交给 LLM 生成一段结构化摘要然后用摘要消息替换掉这批原始消息。相当于把零散的流水账合并成了一份浓缩笔记SUMMARY_PROMPT 请把下面的对话内容压缩成一份中文摘要要求保留 1. 用户已经确认过的关键需求和偏好 2. 双方做出的决策 3. 尚未完成的事项 4. 提到的具体实体时间、地点、人名、订单号等 对话内容 {transcript} def compress_messages(messages: List[Dict], llm_func) - Dict: transcript \n.join( f{m[role]}: {m[content]} for m in messages ) summary llm_func(SUMMARY_PROMPT.format(transcripttranscript)) return {role: system, content: f【此前对话摘要】{summary}}这个方案的额外成本是多一次模型调用但收益很大。我实测下来一个 15 轮以上的长会话用摘要压缩能把 token 消耗降 40% 左右而且关键信息的保留率比滑动窗口高得多。另外一个容易被忽略的点是摘要压缩要交给小模型或者快模型做不要每次都调用最强的旗舰模型。摘要任务对理解能力的要求没那么高用便宜快速的模型跑能把成本再降一档。3. 短期记忆给 Agent 一个“工作台”3.1 结构化工作记忆会话记忆解决的是“连贯性”但它是流式的、扁平的。用户说了十句话偏好和任务进度都淹没在消息流里。这时候就需要第二层短期记忆或者说工作记忆。短期记忆的思路是从对话流里抽取结构化信息存成一组明确的字段让 Agent 可以随时读取当前任务的状态。我举个例子一个售后客服 Agent 的短期记忆体可以是这样的dataclass class WorkingMemory: user_id: str preferences: Dict[str, Any] field(default_factorydict) task_state: Dict[str, Any] field(default_factorydict) extracted_entities: Dict[str, Any] field(default_factorydict) def update_preference(self, key: str, value: Any): self.preferences[key] value def update_task_state(self, key: str, value: Any): self.task_state[key] value为什么要用结构化字段而不是继续用自然语言文本因为 Agent 在推理时需要直接读取某个字段的值比如判断“订单号是否已经获取”。如果这个信息还埋在对话历史里Agent 就得多一步阅读理解既增加 token 消耗又可能理解错。这里的关键设计问题是谁来抽取这些结构化信息人工解析太机械正则也不可靠。我推荐一个看起来笨但实际很好用的方案在每个对话轮次结束后调用一次“记忆更新”提示词把最新发言里的变化合并进记忆 JSON。MEMORY_UPDATE_PROMPT 根据用户最新发言更新记忆 JSON。规则 1. 只更新变化字段原有字段尽量保留 2. 如果新信息和旧信息冲突以新信息为准 3. 不重要的寒暄内容不写入 4. 直接输出更新后的 JSON不要解释 当前记忆 {current_memory} 用户最新发言 {latest_user_msg} 你可以看到这个提示词里几条规则其实就是在定义短期记忆的写入策略只更新变化、冲突以新为准、过滤噪声。这也是我强调的“记忆写入门控”思路——不是每条消息都值得写入短期记忆。寒暄、确认、重复陈述这些都不应该占记忆空间否则短期记忆很快就会变成一堆没用的碎片。3.2 缓存与状态同步短期记忆虽然叫“短期”但它不能只存在进程内存里。商用环境里服务动不动就要重启、要扩容如果短期记忆全在内存变量里重启一次用户的进度就全没了。所以短期记忆必须落到 Redis 这样的缓存里。我用 Redis 存短期记忆的典型代码import json import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def save_working_memory(user_id: str, memory: WorkingMemory): key fworking_memory:{user_id} r.set(key, json.dumps(asdict(memory)), ex3600) def load_working_memory(user_id: str) - WorkingMemory: data r.get(fworking_memory:{user_id}) if not data: return WorkingMemory(user_iduser_id) return WorkingMemory(**json.loads(data))这里的ex3600不是随手写的它代表这个短期记忆的有效期是 1 小时。也就是说Redis 的 TTL 实际上已经在承担一部分“遗忘”的职责了——超过 1 小时没有活跃交互用户的临时偏好和任务进度就会被清空避免占用空间。这种设计是有意的短期记忆和长期记忆的重要区别就是“可丢失性”。短期记忆丢了用户重新说一遍就是了代价很低。所以短期记忆适合用 TTL 兜底而长期记忆里的用户画像和历史事实绝对不能随便丢。另外要提醒一个并发问题。同一个用户可能会同时发来多条消息或者异步回调同时触发记忆更新。如果读写 SD卡式的覆盖逻辑就可能出现 A 请求写入的偏好被 B 请求的旧数据覆盖的情况。简单做法是写操作走 Redis 的 Lua 脚本做原子更新复杂一点的做法是给每条记忆加版本号更新时做乐观锁。我建议项目初期用最简单的方案每次更新前先 load 再 save并发量上来之前这个方案基本够用。4. 长期记忆把经验写进“笔记本”4.1 向量检索与嵌入短期记忆解决的是“当前任务”长期记忆解决的是“跨会话沉淀”。这两者的技术选型差别巨大短期记忆适合用 Redis 存结构化字段长期记忆则要处理非结构化的、语义化的信息常见方案就是向量数据库配合语义检索。为什么不是关键词搜索我给你举个例子。用户第一次说“我需要开票信息是公司抬头的”第二次问的时候说的是“上次那个开票抬头你还记得吗”。这两句话表面没有任何重复关键词但语义上指的就是同一件事。只有把文本转成向量才能做语义层面的匹配。向量检索的完整链路包含三个环节先是 embedding 模型把文本变成向量然后写入向量库最后在需要时用查询文本的向量去检索相近的记忆。我用 Chroma 写一个最小实现import chromadb from chromadb.utils import embedding_functions client chromadb.Client() col client.get_or_create_collection( agent_long_term_memory, embedding_functionembedding_functions.DefaultEmbeddingFunction() ) def write_long_term(user_id: str, content: str, metadata: dict): col.add( ids[hash_content(user_id content)], documents[content], metadatas{**metadata, user_id: user_id} ) def recall_long_term(user_id: str, query: str, top_k: int 5): return col.query( query_texts[query], where{user_id: user_id}, n_resultstop_k )这段代码里有一个商用项目绝对不能省的细节where{user_id: user_id}。如果不把 user_id 作为过滤条件向量库里所有用户的记忆都会混在一起到时候用户 A 可能检索到用户 B 的订单信息这在真实业务里是重大事故。我在一个原型项目里就犯过这个错当时以为向量库会自动隔离跑起来才发现完全不是这么回事。选型方面我给一个比较稳健的建议本地开发、原型验证Chroma 或 FAISS零部署成本中小规模生产Qdrant 或 Milvus支持过滤、持久化、水平扩展已有 PostgreSQL 的场景pgvector 插件少维护一套基础设施4.2 记忆写入与检索的代码骨架有了长期记忆的读写能力之后真正复杂的其实是“什么时候写”和“用什么查”这两个问题。先说写入时机。我见过的错误做法是把所有对话记录一股脑全部向量化后写入。这种做法制造了大量噪声检索时召回的几乎全是废话。更合理的做法是每一轮对话结束后由 Agent 自己判断“这段对话里有没有值得长期记住的信息”。判断标准可以包括用户明确表达的偏好、涉及具体数据的事实、需要跨会话记住的承诺等。写入的内容也应该重写而不是原文照搬。比如用户说“以后给我发日报就行不要周报”改写后的记忆条目应该是“用户希望接收日报不要周报”而不是把原话完整存进去。这个改写工作也可以交给 LLM 做但这属于“后台维护任务”用便宜的小模型跑就行。再说是用什么去检索。很多人的第一反应是直接用用户的最新消息当查询词。实测下来这样效果不太行因为用户消息经常是碎片化的口语比如“那个东西呢”“上次的方案呢”。直接拿这种话去向量检索召回的准确率很低。我常用的优化是对查询词做一次改写把用户消息扩展成更完整的检索语义QUERY_EXPANSION_PROMPT 用户的原始提问是{user_msg} 请将其改写成一段适合检索用户历史记忆的查询文本需要补全指代词和省略的主语宾语。只输出改写后的文本。 这个操作看似多了一步但对召回准确率的影响非常直观。我做过对比同一批测试集里改写前 top-5 命中率只有不到 60%改写后能提升到 85% 左右。把以上这些串起来一个带长期记忆的 Agent 主流程大概是这样的def agent_with_long_term_memory(session_id: str, user_msg: str): user_id get_user_id(session_id) # 1. 改写查询词并召回长期记忆 expanded_query expand_query(user_msg) remembers recall_long_term(user_id, expanded_query, top_k3) # 2. 加载短期工作记忆 working load_working_memory(user_id) # 3. 把长期记忆和短期记忆注入 system prompt system_prompt build_system_prompt(remembers, working) # 4. 调用 LLM 生成回复 reply call_llm(system_prompt, get_history(session_id), user_msg) # 5. 对话结束后异步更新长期和短期记忆 async_update_memories(user_id, user_msg, reply) return reply这个骨架看着简单但每一步展开都有不少细节。重点是记忆不只是“存进去”和“取出来”它必须和组织业务的约束结合。比如注入 system prompt 时召回的 3 条记忆如果都跟当前问题无关那不如不注入因为多余的信息反而会干扰模型判断。我一般会给召回结果做一个相关性打分后置过滤分数低于阈值的直接丢弃。5. 遗忘机制不是所有记忆都该留着5.1 遗忘策略时效、重要性与冲突处理做了前三层之后很多项目会进入一个新的困境长期记忆越攒越多检索时噪声越来越大甚至出现两条互相矛盾的记忆同时被召回。这时候你就需要第五件事——遗忘。遗忘不是缺陷恰恰相反它是记忆系统必需的卫生机制。没有遗忘的记忆库会变成垃圾场什么都有什么都找不到。遗忘机制在工程上有几个可落地的策略。时效衰减是最基础的一种。每条长期记忆都带着一个“最后访问时间”和“访问次数”每隔一段时间计算当前权重太久没有被命中的记忆权重会越来越低直到低于阈值被归档。这种衰减方式可以用半衰期的思路来理解一个记忆如果超过 30 天没有被访问它的强度就减半再过 30 天再减半。跟放射性物质衰变一个道理。import datetime def compute_decay(item, half_life_days30): days (datetime.datetime.now() - item.last_accessed_at).days return item.importance * (0.5 ** (days / half_life_days)) def is_forgettable(item, threshold0.2): return compute_decay(item) threshold这里last_accessed_at的更新很重要。每次某条记忆被检索命中并成功注入上下文就应该更新这个时间戳。否则一个经常被用到的重要记忆也会因为“看起来很久没访问”而被误杀。重要性阈值是另一道防线。写入长期记忆的时候让 LLM 顺带为这条记忆的重要性打一个 0 到 1 的分数。超过 0.8 的属于核心用户画像基本不进遗忘流程低于 0.3 的属于边缘信息定期清理。这样可以在写入端做一次初筛减轻遗忘任务的负载。冲突处理是我发现很多人完全没有做的一个环节。用户今天说“我喜欢详细一点的报告”昨天说的却是“报告越简短越好”这两条记忆都躺在库里检索的时候都召回了模型就懵了不知道听哪个。处理冲突的原则其实很简单以新为准。新写入的记忆如果和旧记忆内容冲突旧的应该被标记为 deprecated 或者直接删除而不是继续留在库里参与召回。基于以上策略我通常会设计一张长期记忆表CREATE TABLE memory_items ( id BIGINT PRIMARY KEY, user_id VARCHAR(64) NOT NULL, content TEXT NOT NULL, importance FLOAT DEFAULT 0.5, access_count INT DEFAULT 0, last_accessed_at TIMESTAMP DEFAULT now(), status VARCHAR(16) DEFAULT active );status字段可以取 active、deprecated、archived 三个值。active 是正常可召回deprecated 是冲突废弃等待清理archived 是衰减后转入冷存储。这样既能保留追溯能力又不会让活跃记忆里混入脏数据。5.2 记忆巩固从短期到长期的结转逻辑如果你只做了“删除”那还不算完整的遗忘机制。遗忘的另一面是“巩固”——把短期记忆里经得起时间考验的信息转存为长期记忆。这就像人睡觉时大脑把白天的重要经历从海马体搬进大脑皮层形成长期记忆。Agent 也需要定期做类似的事。最简单的巩固逻辑是扫描短期工作记忆def consolidate_memories(user_id: str): # 1. 从短期记忆里挑出重要性高、反复出现的条目 candidates extract_high_value_working_items(user_id, score_threshold0.7) # 2. 写入长期记忆并设置来源标记 for item in candidates: write_long_term(user_id, item.content, {source: working_memory}) # 3. 清理短期记忆中的已巩固条目避免重复写入 clear_consolidated_items(user_id)这个定时任务跑多频繁我一般建议在低峰期跑比如每小时或者每天凌晨。巩固的时候需要注意的是去重一条短期记忆如果上一次已经写入过长期记忆这次就不要重复写。判断重复可以靠内容哈希也可以靠语义相似度。内容哈希简单但不够准用户换个说法就识别不出来了语义相似度准但成本高。我的折中方案是先用内容哈希精确去重再对相似度大于 0.95 的模糊去重。至此四层记忆就形成了一个闭环会话记忆保证单次对话连贯短期记忆承载当前任务状态长期记忆沉淀跨会话经验遗忘机制时刻清理噪声和控制存储成本。这个闭环本身也是个可以持续优化的系统。6. 实战中踩过的坑与排查技巧6.1 高频问题速查表我把这几个项目里高频出现的问题整理成了一张速查表运维和研发都可以直接拿去对照现象根因解决思路多轮对话后 Agent 开始胡言乱语窗口裁剪把 system 指令或早期关键上下文裁掉了裁剪时锁定 system prompt 和工具定义优先裁历史消息检索出的长期记忆全是旧的没有时效权重旧记忆和新记忆同等参与排序last_accessed_at 参与召回排序或者加 recency 加权用户 A 问到了用户 B 的信息向量检索没做用户隔离写入和检索都强制带 user_id 过滤条件后端再做一层权限校验记忆更新成本暴涨每轮对话都对大模型发起记忆更新请求记忆写入改成异步批量或用小模型处理记忆维护任务服务重启后用户任务进度丢失短期记忆只存在进程内存里所有短期记忆落到 Redis并设置合理的 TTL明知已解决的事项反复追问短期记忆里存了太多历史状态却未及时清理任务完成时主动清空 task_state 中对应字段表格里这几条都是真实发生过的案例其中“裁剪裁掉 system 指令”那条我记得最深。当时客服 Agent 上线后用户满意度一直上不去排查了半天才发现一旦对话超过 20 轮裁剪逻辑会把最开头的 system prompt 当成最旧的文本给裁掉。模型连自己的角色设定都丢了后面的对话自然越来越不可控。修复就一行代码裁剪时跳过 system 消息。但定位这个问题花了整整一个下午。6.2 排查方法与调优经验排查记忆链路问题我一般按这个顺序来做。先画一条完整的链路图从用户消息进来到系统回复出去所有记忆在哪里写入、在哪里读取、以什么格式流转全标注一遍。为什么非要画图因为很多健忘问题的根源不是存储方案不好而是记忆在某个环节根本没有被传递。比如短期记忆已经存到了 Redis但下一次请求加载的时候加载代码写在了另一条分支里压根没有执行。这类问题不看链路图很难发现。画完图之后给记忆相关的日志加上两个关键字段trace_id 和 time_cost。trace_id 帮你串起一次请求内所有记忆操作time_cost 帮你发现性能瓶颈。记忆链路慢通常发生在两个地方一是向量库慢二是 embedding 调用慢。如果是这两个地方优先考虑缓存策略而不是机器配置。调优方面我的建议是每一层分开调调好一层再动下一层。会话层先定窗口阈值。我的经验值是把窗口设成模型最大 token 的 60%~70%留足输出和 system 的空间。然后跑一批真实对话样本观察“关键信息丢失频率”哪里丢多了就把摘要压缩加到哪里。短期记忆层别急着加向量。先把结构化字段补全比如用户偏好、任务状态、实体信息。如果字段都已经够覆盖业务需求了其实短期记忆这层就够用了。向量检索是给长期记忆用的不要混用。长期记忆层上线前一定要做离线评测。拿一批真实用户问题人工标注每条问题的正确答案再跑检索统计 top-5 命中率。我见过最离谱的情况是某团队向量库上线一个月命中率只有 30%但一直没人发现因为没有人建立评估集。别省这个步骤。遗忘层的参数不要拍脑袋定。时效衰减、重要性阈值这些参数都会影响业务表现稳妥的办法是 AB 测试。比如把用户分成两组一组用旧的“只增不减”策略一组用新的遗忘策略对比记忆召回准确率和用户满意度用数据说话。最后再分享一个我自己的体会记忆架构的复杂度应该跟着业务需求渐进增加而不是一步到位。一个刚上线的 Agent会话记忆加滑动窗口就够覆盖 90% 的场景等用户开始反馈“它不记得我喜欢什么”的时候再补短期记忆真正需要长期记忆的是用户画像、跨会话偏好这类强需求场景。反过来如果你的业务根本不需要跨会话记忆强行上向量库只会增加维护成本还容易引入新的错误。我在实际项目里还有一个一直没有放弃的习惯给每条记忆都保留来源和写时间。这么做短期看只是多了两个字段长期看却是排查问题的救命稻草。你永远不知道哪天线上出现一条可疑记忆如果没有来源标记你连它是从哪里写进去的都不知道。这个习惯帮我省了很多查日志的时间强烈建议你也保留。