1. 失忆的根源先从一次线上故障说起我接手过不少商用 Agent 项目第一个要解决的永远不是意图识别也不是工具调用而是记忆。印象很深的一次故障是这样的用户和客服 Agent 连续对话了十几轮前面已经确认了我的订单尾号是8846帮我查一下物流结果后面只是追问了一句那它现在到哪了Agent 直接回答请问您的订单号是什么。从用户视角来看这个 Agent 就是健忘。但从技术视角看它倒不一定是忘而是根本没有记忆——每一次请求进来模型拿到的都是干干净净的上下文之前聊了什么一概不知。这种体验放在 demo 里无所谓放到商用环境里就是灾难。那为什么很多团队做 Agent 时记忆这一块总是最容易翻车我观察下来的结论是大多数人把记忆当成了一个功能而不是一套架构。他们以为往系统 prompt 里塞几句请记住用户之前说过的话模型就真的记住了。这里先戳破一个幻觉——大模型的记忆本质是上下文窗口里的 token窗口一关、请求一结束所有信息立刻归零。想要让 Agent记住必须靠外部存储把信息接住再在下一次请求到来时重新注入。这一进一出的链路才是记忆架构的全部。所以这篇文章我想用一套完整的代码实战把商用 Agent 的记忆体系从头到尾拆一遍会话记忆、短期记忆、长期记忆、遗忘机制每一层的作用、实现方式、容易踩的坑全部摊开来讲。适合正在做 Agent 落地、或者准备从 demo 往生产环境过渡的开发者参考也适合那些已经背过不少八股文、但没真正在工程里摸过记忆链路的朋友。2. 三类记忆的边界别再混为一谈了2.1 会话记忆的本质是上下文窗口的搬运工先说会话记忆。这是最容易被低估、却又最直接影响体验的一层它的定义是在一次完整对话中维持模型对前文的感知能力。早期的做法很粗暴直接把历史消息数组一股脑传给模型。但随着对话轮数增加token 消耗会急剧膨胀而且超出上下文长度之后前面的关键信息会被无情截断。更麻烦的是很多商用场景里对话不可能永远在一个连接里完成——用户可能中途刷新页面、切了设备、隔了十分钟才回复之前的会话在物理上已经断了。我们实验室里测过一组数据当上下文超过窗口的 60% 之后模型回答的准确率明显下降因为它要把大量注意力花在理解无关的早期对话上。所以会话记忆要做的不只是搬运还要做取舍。主流方案有两种滑动窗口和摘要压缩。滑动窗口是只保留最近 N 轮对话实现简单、成本低但缺点很直接——用户在十几轮前提过的关键信息会丢失摘要压缩则是每过几轮把前面若干轮的内容交给模型总结成一段摘要替换掉原始文本。后者本质上是在信息压缩率和保真度之间做权衡实际项目中往往是两者混用稍后我会在代码里展示一个组合方案。2.2 短期记忆跨会话的最小上下文单元如果说会话记忆管的是一次对话内那短期记忆管的就是跨会话、短时间。它的典型场景是什么用户访问你的 Agent 服务第一次问帮我查一下上海这周的天气第二次隔了半小时又问那下周呢。理想情况下Agent 应该知道那指的是上海、指代的是天气查询语义这就依赖短期记忆把用户的身份信息和最近行为上下文关联起来。但这里要特别强调整一点短期记忆不等于把上一次的 token 原样存下来。商用环境下我们更关心的是结构化信息——用户 ID、会话 ID、最近一次交互的操作类型、关键实体比如刚才提到的订单号、城市名、时间戳。它的实现载体通常是 Redis 这类 KV 存储配合 TTL过期时间来自动老化。为什么用 TTL因为短期记忆的特性就是过期即焚留存时间一般在 5 分钟到 24 小时之间。这个时间窗口的设计来源于产品需求如果 TTL 太短用户只是喝口水回来Agent 就忘了体验断裂如果太长又会留下大量无用信息干扰后续检索。我们线上常用的配置是用户级 key 24 小时、会话级 key 30 分钟不同数据类型分开管理。2.3 长期记忆从记住到记住且能想起来长期记忆是 Agent 表现聪明的关键。它的作用是把用户的历史偏好、事实性信息、交互模式沉淀下来在未来的某次对话中主动召回。比如一个理财助手用户在两个月前说过我目前基金仓位比较重想逐步调整成偏稳健的配置两个月后的某次咨询里Agent 应该能主动关联这个背景信息而不是每次都当新用户看待。长期记忆的实现有一个核心转变从精确匹配走向语义检索。因为用户的表达是千变万化的今天说稳健一点下周可能说我不想承受太大波动两者在字面上完全不同但在语义空间里距离很近。这就要求我们把历史信息切片、结构化存入向量数据库用户发起新对话时先用 embedding 模型把当前输入向量化在向量库里做 top-k 相似度检索把相关记忆片段取回来将取回的记忆注入到系统 prompt 中让模型感知商用项目里这个链路还要叠加一层记忆的准入机制——不是所有对话内容都值得写入长期记忆。我自己定过一个简单标准只有用户主动表达的偏好、明确的事实陈述、重复出现的实体才允许入库纯寒暄内容一律过滤。不设门槛的长期记忆存进去的全是垃圾检索出来更是灾难。2.4 遗忘机制记忆架构里最反直觉的一环很多团队做到长期记忆就停了因为他们觉得记住的越多越好。这是一个非常危险的认知。实际的商用场景里遗忘机制的重要性完全不亚于记忆本身核心原因有三。第一是成本。向量库不是免费的越多的记忆片段意味着越高的存储成本和检索延迟。第二是准确性。记忆库里堆了太多过时的、矛盾的、不同时间点的信息模型在召回时会精神分裂。今年 3 月的数据和 5 月的数据如果冲突到底信哪个第三是合规和隐私。随着各类数据保护法规的收紧用户有权要求删除个人数据而删除的实现如果不依赖遗忘机制你就只能自己造轮子。我把遗忘机制分成三类操作显式遗忘用户明确要求忘掉我之前说的关于信用卡的事情这是产品功能时间衰减信息在指定时间窗口后自动降低优先级超期后清除冲突覆盖新信息与旧信息矛盾时以新为准旧数据标记为过期。这三类操作在代码里都要落成独立的服务而不是混杂在业务逻辑里。3. 全链路代码实战从零搭一套可用的记忆系统3.1 会话记忆双层结构滑动窗口 摘要压缩这部分我直接用 Python 写一套精简但可以扩展的实现。核心思路是维护一个消息队列最近 N 轮原始消息完整保留超过 N 轮的消息通过摘要模型压缩成一段总结放在对话开头作为历史。from collections import deque from typing import List, Dict, Any import json class ConversationMemory: 滑动窗口 摘要压缩的会话记忆实现。 max_window: 原始消息最多保留多少条 summarize_every: 每隔多少轮触发一次摘要压缩 model: 负责摘要的 LLM 调用函数这里只留接口 def __init__(self, max_window: int 20, summarize_every: int 10, modelNone): self.max_window max_window self.summarize_every summarize_every self.model model # 原始消息队列存放最近的对话 self.messages: deque[Dict[str, Any]] deque(maxlenmax_window) # 存储压缩后的摘要 self.summary: str self.turns_since_summary 0 def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) self.turns_since_summary 1 if self.turns_since_summary self.summarize_every: self._summarize_old_messages() self.turns_since_summary 0 def _summarize_old_messages(self): 把队列前部即将被滑窗淘汰的消息压缩进摘要。 # 实际项目中这里应该调用 LLM 生成摘要 # 这里给出调用伪代码便于理解流程 if self.model is None: # 无模型时用截断兜底至少保证不报错 truncated .join(m[content][:50] for m in list(self.messages)[:-8]) self.summary (self.summary truncated)[-2000:] return old_parts list(self.messages)[:-8] # 保留最近8条不压缩 if not old_parts: return texts \n.join(f{m[role]}: {m[content]} for m in old_parts) prompt f请将以下对话压缩为100字以内的摘要保留关键事实用户名、订单号、偏好、时间等:\n{texts} new_summary self.model(prompt) # 这里的 truncate 是为了防止摘要无限膨胀实际项目中可以按 token 控制 self.summary (self.summary | new_summary)[-3000:] def build_context(self) - str: 把摘要和最近消息拼成送给 LLM 的上下文。 parts [] if self.summary: parts.append(f[历史摘要] {self.summary}) for m in self.messages: parts.append(f{m[role]}: {m[content]}) return \n.join(parts)注意一个细节deque(maxlenmax_window)在 Python 里天然支持滑动窗口但真正生产级实现里消息队列往往不在内存中而在 Redis 里用 List 类型做 LPUSH/RPUSH这样进程重启后会话还能恢复。这里的代码先把结构讲清楚后续如果你想上生产替换存储层即可。而且摘要压缩不要每轮都做否则会引入成本和时间延迟。每隔 5~10 轮做一次是比较合理的节奏。3.2 短期记忆基于 Redis 的 TTL 缓存实现短期记忆用 Redis 来做是性价比最高的选择。我给出一个封装了记忆读写、自动过期的类。记住短期记忆存的不是原始文本而是结构化的当时发生了什么。import json import time import redis class ShortTermMemory: 短期记忆基于 Redis TTL。Key 结构 stm:{user_id}:{session_id} - {last_intent, entities, last_time} def __init__(self, redis_url: str redis://localhost:6379/0): self.r redis.Redis.from_url(redis_url, decode_responsesTrue) def _key(self, user_id: str, session_id: str) - str: return fstm:{user_id}:{session_id} def save_context(self, user_id: str, session_id: str, intent: str , entities: dict None, ttl: int 1800): 写入短期记忆ttl单位秒默认30分钟。 payload { intent: intent, entities: entities or {}, last_time: int(time.time()), } key self._key(user_id, session_id) self.r.set(key, json.dumps(payload, ensure_asciiFalse), exttl) def load_context(self, user_id: str, session_id: str) - dict | None: 读取短期记忆不存在或已过期返回 None。 raw self.r.get(self._key(user_id, session_id)) if not raw: return None data json.loads(raw) # 这里可以做一次时间衰减超过一半有效期时降级为低置信度 # 实际项目中可以返回一个 confidence 字段 data[_age] int(time.time()) - data.get(last_time, 0) return data def update_entity(self, user_id: str, session_id: str, key: str, value: str, ttl: int 1800): 只更新某个实体字段不影响其他内容。 data self.load_context(user_id, session_id) if data is None: data {intent: , entities: {}} data[entities][key] value data[last_time] int(time.time()) self.r.set(self._key(user_id, session_id), json.dumps(data, ensure_asciiFalse), exttl)这里有一个非常实用的实践短期记忆的写入时机。不要每次都把整段对话塞进去而应该只抽取当前轮的关键信息——意图intent和实体entities。我通常在 Agent 的工具调用命中之后才写短期记忆因为命令已执行这个动作本身才是短期记忆最有价值的附着点。比如用户改了一次收货地址那这个地址就应该立刻写入短期记忆而不是等整段对话结束再批处理。3.3 长期记忆向量化存储 语义检索长期记忆的代码实现比前面两层复杂一些涉及 embedding 和向量检索。我选用轻量级的 Chroma 作为演示载体生产环境可以换成 Milvus、Qdrant 或 pgvector接口思路是通用的。import chromadb from chromadb.utils import embedding_functions class LongTermMemory: 长期记忆向量库 元数据过滤。 collection: 按用户分桶或按业务域分桶 metadata: 用户ID、时间戳、记忆类型、是否有效 def __init__(self, persist_dir: str ./ltm_store, model_name: str BAAI/bge-small-zh-v1.5): self.client chromadb.PersistentClient(pathpersist_dir) self.embed_fn embedding_functions.SentenceTransformerEmbeddingFunction( model_namemodel_name ) self.collections {} # 按业务域缓存 collection def _get_collection(self, domain: str default): if domain not in self.collections: self.collections[domain] self.client.get_or_create_collection( namefltm_{domain}, embedding_functionself.embed_fn ) return self.collections[domain] def add_memory(self, user_id: str, content: str, domain: str default, mem_type: str preference, ttl_ts: int 0): 写入长期记忆。ttl_ts 指定过期时间戳0表示永不过期。 id 用时间戳随机数就是为了后续能按 id 精准删除。 import uuid from datetime import datetime mem_id f{user_id}_{uuid.uuid4().hex[:12]} metadata { user_id: user_id, domain: domain, type: mem_type, created_at: datetime.now().isoformat(), expire_ts: ttl_ts, active: 1, } collection self._get_collection(domain) collection.upsert(ids[mem_id], documents[content], metadatas[metadata]) return mem_id def recall(self, user_id: str, query: str, domain: str default, top_k: int 5, min_score: float 0.3): 语义检索先按用户过滤再按相似度排序。 返回的每条记忆都带 score 和 metadata方便上层做衰减判断。 collection self._get_collection(domain) # where 条件用于过滤当前用户的记忆而不是全局搜 results collection.query( query_texts[query], n_resultstop_k, where{user_id: user_id}, include[documents, metadatas, distances], ) # chroma 返回的是距离距离越小越相似这里转成相似度分数 scored [] for doc, meta, dist in zip( results[documents][0], results[metadatas][0], results[distances][0], ): # 距离转分数score 1 - dist 取决于距离函数此处按 l2 近似 score round(1 - dist, 4) if score min_score: continue scored.append({content: doc, metadata: meta, score: score}) scored.sort(keylambda x: x[score], reverseTrue) return scored def delete_by_id(self, mem_id: str, domain: str default): 显式遗忘删除单条记忆。 collection self._get_collection(domain) collection.delete(ids[mem_id])这段代码里有几个细节值得反复强调。第一where{user_id: user_id}是必须的否则检索会跨用户泄露信息这是商用项目里最致命的隐私 bug。第二min_score阈值非常重要它决定了召回内容的准入条件阈值设低了鬼知道会把什么无关记忆拉进来设高了又可能召回不到东西。这个阈值需要根据实际 embedding 模型调通常要在测试集上跑一批数据找到准确率和召回率的最佳平衡点。我们线上用的 bge-small 系列0.3 这个初始值比较稳但要持续观测调整。3.4 记忆管线组装一次请求背后的调度顺序先把三个存储层的代码写好还不够真正的记忆架构在于一次请求进来时这三层如何协同工作。我画不出图直接用代码描述这个编排过程——这对刚接触记忆架构的同学比单独看任意一层都要有价值。class MemoryManager: 记忆编排层负责在 Agent 的每次请求前组装记忆请求后更新记忆。 def __init__(self, conv_mem: ConversationMemory, stm: ShortTermMemory, ltm: LongTermMemory): self.conv_mem conv_mem self.stm stm self.ltm ltm def build_prompt(self, user_id: str, session_id: str, user_input: str, domain: str default) - str: 请求前组装完整上下文。 优先级短期记忆 长期记忆召回 会话摘要避免上下文过度膨胀。 # 1. 短期记忆先看用户最近的操作上下文 stm_ctx self.stm.load_context(user_id, session_id) # 2. 长期记忆用当前用户输入做语义召回 ltm_recalls self.ltm.recall(user_id, user_input, domaindomain, top_k3) # 3. 会话记忆拼上当前会话的摘要和最近消息 conv_ctx self.conv_mem.build_context() sections [] if stm_ctx: sections.append([短期记忆] json.dumps(stm_ctx, ensure_asciiFalse)) if ltm_recalls: memories ; .join([r[content] for r in ltm_recalls]) sections.append(f[长期记忆] {memories}) if conv_ctx: sections.append([当前会话] conv_ctx) base_prompt f用户输入: {user_input} return \n.join(sections [base_prompt]) def update(self, user_id: str, session_id: str, user_input: str, agent_output: str, domain: str default): 请求后更新三层记忆。 是否写入长期记忆要看是否满足我们前面提到的准入规则。 # 会话记忆一定更新 self.conv_mem.add_message(user, user_input) self.conv_mem.add_message(assistant, agent_output) # 提取意图和实体此处简化实际应该接入 NER/意图识别 intent self._extract_intent(user_input, agent_output) entities self._extract_entities(user_input) # 短期记忆只要有意向或实体就更新 if intent or entities: self.stm.save_context(user_id, session_id, intentintent, entitiesentities) # 长期记忆准入规则判断 if self._should_store_long_term(user_input, intent): self.ltm.add_memory( user_id, contentf意图{intent}|实体{json.dumps(entities, ensure_asciiFalse)}|输入{user_input[:100]}, domaindomain ) def _should_store_long_term(self, user_input: str, intent: str) - bool: # 排除寒暄和无关指令 stopwords [你好, 谢谢, 再见, 测试, 在吗] if any(w in user_input for w in stopwords): return False # 只有支撑性意图才进入长期记忆 return intent in {preference, personal_info, task_goal}这个编排层是记忆架构的大脑。我自己的经验是update 的时机比 build_prompt 更难设计。因为什么值得写进长期记忆的判断直接决定了记忆库的质量。最初我们贪多求全把每一轮对话都往长期记忆库塞两周后整个向量库都要洗一遍。后来加了准入规则长期记忆量锐减 70%但有效召回率反而提升了一倍——因为检索到的内容信噪比高了。4. 商用场景绕过这些坑才算真正落地4.1 并发写入下的记忆一致性单用户的记忆读写很简单但商用 Agent 面临的是高并发。我在某个客服系统里面临的情景是同一用户同时打开两个浏览器标签页都会向 Agent 发消息。如果不做任何一致性控制短期记忆和会话记忆就会被后写入的数据覆盖造成精神分裂。解法思路是这样的。短期记忆以事件为单位做追加而不是整体覆盖。Redis 里不要直接 SET 一个 JSON 对象而是用 HSET 按字段更新每个字段记录更新时间戳读取时做多版本合并。会话记忆的deque在内存里只能承载单进程分布式环境需要把消息队列迁到 Redis Stream 或 Kafka并给每条消息分配全局递增 ID。还有一个更隐蔽的坑是长短期记忆的写入竞争。两个并发请求同时触发长期记忆写入可能会导致同一个用户的多条记忆在向量库里顺序颠倒。我们加了一层基于用户 ID 的分布式锁在LongTermMemory.add_memory之前先尝试获取锁拿到锁才允许写入。虽然会增加几十毫秒延迟但换来的是记忆序列的一致性这个代价完全值得。4.2 记忆污染检索到不该检索的东西记忆污染是我自己造的词但做过商用 Agent 的人一定深有体会。典型场景是Agent 从长期记忆里召回了一条两年前的偏好而用户在这两年里早就改变了习惯最终 Agent 给出一个非常不智能的回答。我在长期记忆的 metadata 里加了一个expire_ts字段就是为了应对这个问题。每次召回时不仅按相似度排序还要加一道新鲜度衰减如果记忆距今超过一定时间分数打折。比如 90 天以上的记忆分数乘以 0.7180 天以上的乘以 0.3过期时间戳已经小于当前时间的直接过滤掉。这套逻辑在代码里加进来很简单但效果极其显著。另外要留意的是元数据过滤条件的顺序Chroma 这类向量库在执行查询时是先执行where过滤再算相似度。如果把用户过滤放在相似度之后做性能会差很多而且容易把别的用户的高相似记忆误召回。这部分我在生产环境里通过实测对比过先过滤后检索的执行时间比反过来能低一个数量级。4.3 遗忘机制的工程实现从能删到安全地删遗忘机制听起来只是 DELETE 操作但在商用环境里有非常多的约束。以显式遗忘为例用户说忘掉我之前说过的手机号你需要做的不只是删除一条记忆而是从短期记忆Redis中删除用户的手机号实体从长期记忆向量库中标记或删除所有包含手机号的记忆片段从会话摘要如果有持久化中把相关 token 抹掉或重写摘要如果有日志系统还可能需要脱敏处理历史日志我们实现的方案是给长期记忆增加active字段。显式遗忘时不直接 delete因为向量库的 delete 可能会造成索引碎片而是把active置 0召回时用where{active: 1}过滤。每天凌晨统一做一次物理删除配合定期重建向量索引。这样既保证了用户已被遗忘的语义又避免了频繁删除对检索性能的影响。时间衰减和冲突覆盖的工程实现就藏在前面recall函数里那个score和expire_ts中。所谓时间衰减就是在排序时把时间因素乘进去所谓冲突覆盖是写入新记忆时先按用户和主题检索旧记忆如果存在同主题的活跃记忆就直接更新它而不是新增一条并存。这两条经验是我在踩了无数次坑之后总结出来的。4.4 成本控制记忆不是越贵越好最后说一个在商用项目里永远绕不开的问题成本。长期记忆的 embedding 调用、短期记忆的 Redis 内存、会话记忆的摘要模型调用都会产生费用和延迟。这块我的建议是分级治理。高频用户比如每天对话超过 20 轮的单独分配更精细的记忆策略低频用户用默认策略降级。会话摘要模型选择上不需要追求最强的大模型用中等规模的模型即可——摘要任务远不如创作任务那么依赖模型上限。向量库选型上如果业务规模不大直接用轻量级方案Chroma/SQLite 离线 embedding也能达到不错的效果只有 QPS 上来了才需要考虑 Milvus 这类分布式方案。另外embedding 模型的选择要结合语言场景。我们以中文业务为主一开始用了通用英文 embedding 模型召回效果很惨后来换成中文优化的 bge 系列效果提升很明显。这不是模型好坏的问题而是特征空间是否贴合目标语言的问题。如果你面向的是特殊领域比如法律、医疗还应该考虑要不要用领域微调的 embedding 模型。5. 实测排查两类高频怪问题的定位思路说实话记忆系统上线后你遇见的 Bug 大多数不是功能不可用而是表现很怪。这里我想把两类最高频的问题及排查思路完整走一遍。第一类是用户说没说过Agent 却像没见过。排查顺序是先看短期记忆有没有写入成功——用 Redis 直接查 key 是否存在TTL 是否被意外耗尽再看长期记忆的召回结果——是不是min_score阈值设得太高导致召回了空集最后检查会话摘要是不是在某个环节被意外截断了。这里有个经验把整个记忆链路的关键节点日志打出来按写入→存储→召回三个环节分别校验基本上一眼就能定位。第二类是不同用户之间记忆串了。这类问题性质严重得多通常不是功能 Bug而是隐私事故。排查重点放在检索条件上看recall请求里的where条件有没有正确拼上user_id。我们的代码里把user_id作为必传参数就是强制在 API 层面杜绝漏传。另一个被很多人忽略的是缓存如果短期记忆的 key 设计只用了 session_id 而没用 user_id那么 session 在不同用户间复用比如某些网关会导致 session_id 复用时就会串记忆。这一类 bug 在开发环境几乎测不出来因为开发环境的 session 是干净的但一上生产各种网关策略和 HTTP 客户端复用都会暴露问题。所以 key 设计上我坚持所有记忆 key 都必须同时包含 user_id 和 session_id缺一不可。最后是遗忘无效的排查。最常见的原因是物理删除和逻辑过滤没有双端同步。比如你把active置 0 了但召回条件里忘了加where{active: 1}再比如你以为自己做了 DELETE但删除的是错误集合——Chroma 的 collection 是分域的删错域等于没删。排查时建议把删除前后的集合数据量打出来对比不要只凭代码逻辑推断。6. 最后的实操心得这套记忆架构我在三个不同类型的商用项目里落地过总结下来最大的体会是记忆不是越强越好而是越适度越好。过度设计会让系统变得臃肿而缺失又会显得不智能。你要做的最重要的一步其实是给 Agent 定义清楚哪些该记、哪些该忘这不是纯技术活而是产品与技术的最强交集。再分享一个小技巧在调试记忆相关问题时我非常建议做一个记忆沙盒——把线上的真实对话脱敏后抽取一批样本在一个独立环境里回放用同一套记忆代码跑一遍边跑边检查每一层的写入与召回结果。这比在生产环境打日志调试高效得多尤其适合定位那些偶发性、只出现在长对话中的问题。如果你现在正从零把 Agent 往商用推我的建议是不需要一上来就追求大而全的架构。先把会话记忆做稳、再叠加短期记忆、最后再考虑长期记忆和遗忘机制。每一层都对应一种用户价值逐层叠加风险可控。先跑起来再优化这个节奏会比一步到位稳妥得多。