半年前我接了一个某智能客服 Agent 的迭代任务,负责把它的多轮对话能力从“能聊两句”提升到“真的能干活”。临上线前发现一个特别尴尬的问题:用户第一天问过“你们支持导入 Excel 吗”,Agent 当时回答得很完整,结果第二天用户换个说法又问了一遍,它居然像第一次见面一样重新解释,甚至把导入格式说反了。翻日志时我看到了根因,不是模型变笨了,而是它根本没有记忆机制。后来我花了两三周时间,专门拆了一批头部 Agent 产品里公开的设计思路,又结合自己的项目从零搭了一套能落地的记忆系统。这篇文章就把那段实战里最核心的东西写出来:先拆清楚工业 Agent 里的“记忆机制”到底是什么,再一步步给出可以直接复制的存储模型、写入读取流程,以及我在真实环境里踩过的坑。适合想给 Agent 加记忆、但不想只停留在调 API 层面的开发者。
1. 先交底:Agent 记忆机制到底在解决什么
1.1 一次失败的多轮对话:症状与根因
先说那个让我下定决心重做记忆系统的具体场景。语音客服 Agent 接入的渠道里,用户经常前一天问“你们支持哪些文件格式”,第二天又追问“导入 Excel 的具体步骤是什么”。理想情况下,Agent 应该能回忆起前一天聊过“支持 Excel 和 CSV”,然后直接给出步骤。
但实际表现是:Agent 先反问“您指的是哪种格式”,用户又说了一遍之后,它依然没有联想起来。我看日志时发现了一个更难受的事实——第一轮对话其实还在上下文窗口里,只是因为系统提示词太长、知识库切片太多,首轮消息被截断了。也就是说,产品不是“不会记”,而是“什么都想塞进上下文,结果什么都没留住”。
这个场景让我意识到一件事:多轮对话能力的上限,往往不是模型推理能力,而是记忆机制的设计水平。把历史全部塞给大模型只能叫缓存,不叫记忆。缓存会溢出、会失效、会无差别占用 token,而记忆应该像人一样:重要的事长期记住,无关的事快速遗忘,需要的时候能准确想起来。
1.2 记忆不是一块缓存,而是一套分级体系
工业级 Agent 里谈“记忆”,不是单指某一块存储,而是一套分层的体系。我拆解过的几个头部产品,内部几乎都遵循类似的分层逻辑:
- 工作记忆:当前上下文窗口里的内容,对应人脑正在处理的信息,容量小但即时可用;
- 场景记忆:当前会话的摘要或关键节点,用于回答“这次对话聊到哪了”;
- 长期记忆:跨会话的用户偏好、明确事实、项目背景,对应“这个用户一直是这么要求的”;
- 程序记忆:Agent 自己总结出的执行偏好,比如“这个用户喜欢先听后问”。
工程上不需要完全复刻心理学分类,但分层的思想特别重要。不分层的后果就是:三天的聊天记录全部堆在 prompt 里,token 爆炸,关键信息反而难以被模型注意到。
我现在的做法是做一个明确的记忆分级,每一级承担不同职责,各自有独立的写入、读取和淘汰策略。这也直接影响后面的存储结构设计——如果从一开始就是一张大表存所有内容,后面做检索和遗忘会非常痛苦。
1.3 四个必须实现的记忆动作:写入、读取、更新、遗忘
任何记忆系统,往本质里拆就是四个动作。大厂产品里的记忆机制再复杂,也逃不出这个框架,只不过每个动作都做了针对性的优化。
| 动作 | 要回答的问题 | 工业产品里的常见做法 |
|---|---|---|
| 写入 | 什么值得记住? | 提取关键实体、压缩成摘要、按重要性打分 |
| 读取 | 什么应该现在想起来? | 向量检索 + 时间衰减 + 实体过滤,按预算组装 |
| 更新 | 旧记忆和现状冲突怎么办? | 按实体聚合,以时间较新者为主,标记待确认 |
| 遗忘 | 怎么控制存储和 token 成本? | 定期清理低价值条目,合并重复事实 |
这四个动作里最容易忽略的是“遗忘”。很多初版记忆系统只做写入和读取,结果跑了两周,库里堆了几万条碎片化记录,检索时召回结果越来越杂乱,用户画像反而失真。记住该记住的、忘掉该忘掉的,和“记得住”同等重要。
2. 从几类工业 Agent 里拆出来的共性设计
2.1 会话缓冲:大的窗口不等于好的记忆
早期的 Agent demo 喜欢把整个会话历史原封不动传给模型,窗口从 4k 换到 32k,以为窗口大了记忆就强了。实际跑下来并不是这么回事。窗口变大后,注意力被稀释,关键信息可能夹在大量寒暄里;更重要的是,成本成倍上涨,用户问一句,你背后传了整本《对话录》。
我观察到的工业做法是:给上下文窗口做预算管理。比如把 60% 的预算留给最近若干轮完整消息,30% 留给摘要压缩后的历史要点,剩下 10% 留给系统指令和检索到的长期记忆。窗口不是无限存的,而是“最近的全量 + 历史的浓缩”。
具体触发压缩的时机也很有讲究。我自己的规则是:当最近消息的 token 数超过窗口预算的 70% 时,触发一次摘要压缩,把最早的一部分对话合并成几条要点,而不是等塞满了才处理。处理得越晚,一次性压缩的文本越长,信息丢失率越高。
2.2 摘要记忆:用高密度文本换长期上下文
摘要记忆是同级产品里最常用也最有效的一种设计。
它的逻辑很简单:每个会话维护一个摘要条目,每处理若干轮对话,就把“上次的摘要 + 新增对话”一起喂给大模型,重新生成一份更完整但依然紧凑的摘要。摘要里保留的是决策、偏好、结论,而不是逐字逐句的聊天记录。
这个方案在信息密度上非常划算。一段 40 轮的长对话,原始文本可能有 10000 token,摘要之后能压到 600 token 左右,信息保留率在多数场景下能达到八成以上。代价是摘要本身是一个有损压缩过程,如果摘要触发的频率太低,一些中间过程的细节就会被丢掉。
我后来做的一个优化是:摘要里明确要求模型保留“用户明确表达的偏好”“Agent 承诺过的事项”“双方确认过的结论”这三类信息,因为它们是最容易被后续对话引用的。这个约束让摘要的可用性明显提升。
2.3 结构化记忆:实体、关系与用户画像
摘要记忆擅长描述“事情经过”,但在回答“这个用户是什么样的人”这种问题上力不从心。头部产品里通常还带一套结构化记忆,把用户的关键属性单独抽出来存成键值或知识图谱。
举例来说,用户说“我们团队 30 人,目前主要用 A 平台做数据看板”,系统应该抽取出结构化事实:团队规模=30,技术平台=A。后续无论过了多久,只要用户提到“我们团队”,Agent 就能从结构化记忆里直接读出规模信息,不需要重新问一遍。
这类结构化记忆的存储不复杂,一张实体表加一张关系表足够。难点在于抽取的准确性,以及冲突检测——用户第一次说团队 20 人,三个月后说 35 人,哪条是新的?这需要靠时间戳和会话归属来判断,也更适合用规则加模型结合的方式做,而不是无脑覆盖。
2.4 向量索引:把“曾经发生过的事”变成可检索的
摘要和结构化记忆适合处理清晰的信息,但大量细节是难以提前归类的。比如用户说过“我们有一次双十一活动把系统打崩了”,这句话很难拆成结构化字段,但它是一个具体事件,未来很可能被引用。
这种场景下,向量索引几乎是唯一的选择。把记忆条目向量化后,查询时用语义相似度召回,不必强制按关键词匹配,也支持用户用不同说法回忆同一件事。这就是很多大厂记忆系统底层的标配组件。
但我想强调的是,向量检索只是记忆系统里的一条通路,不是全部。它解决的是“语义相似”的问题,解决不了“该不该想起这条记忆”的问题。后者还需要叠加时间衰减、重要性权重、实体过滤这些规则,否则召回结果会变成“一堆看起来相关但其实没用的话”。
3. 自己搭建时的数据模型与存储选型
3.1 存储选型:嵌入式数据库加本地向量索引为什么够用
很多开发者一上来就引入分布式数据库和外部向量库,理由是“大厂都这么做”。但我的建议是:先搞清楚自己的业务规模。
以我负责的客服 Agent 为例,活跃用户几千人,每人每天产生几十条对话事件,一个月的记忆条目也就几十万条。这种量级下,一张设计良好的嵌入式数据库表配合落盘向量索引,单机延迟在毫秒级,运维成本几乎为零。等规模真的涨上去,再平滑迁移到外部存储也不迟。
选型时我坚持三条原则:
- 数据能落本地就落本地,避免为一个早期系统引入不必要的服务依赖;
- 数据库和向量索引用同一份 ID 关联,而不是复制两份数据;
- 所有字段都带时间戳,这是记忆系统做衰减和冲突判断的基础。
这套组合在数据量小的阶段比重型组件更快,也更稳定。
3.2 核心表结构设计
我最终的存储结构用了一张主表加两张辅助表。主表存记忆条目,辅助表存实体和实体关系。
记忆主表的核心字段如下:
CREATE TABLE memory_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, user_id TEXT NOT NULL, conversation_id TEXT NOT NULL, memory_type TEXT NOT NULL, -- event / summary / entity content TEXT NOT NULL, entities TEXT NOT NULL, -- JSON 数组,可参与检索 importance REAL NOT NULL DEFAULT 0.5, access_count INTEGER NOT NULL DEFAULT 0, last_accessed_at INTEGER NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, embedding_model TEXT NOT NULL, is_deleted INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX idx_memory_user_type ON memory_items(user_id, memory_type, is_deleted);字段设计里几个容易被忽略的点:
embedding_model必须记录向量化模型的版本,否则以后换模型时老向量和新向量混在一起,相似度计算直接失真,这一条是我踩过大坑后才加的;importance是遗忘机制的抓手,初始值由写入时打分决定,后续访问会适度提升;is_deleted用软删除而不是物理删除,方便排查线上问题,历史记录在出事故时能回放。
实体表单独存抽取出的关键实体,避免在文本里做依赖拼写的关键词匹配。实际效果是,当查询里出现“30 人”这种数字时,实体表的精确匹配比向量召回更可靠。
3.3 记忆项从哪来:事件、实体与摘要三条管线
记忆不能等着被调用时才想起去记,而是要在对话过程中主动落库。我从一开始就把记忆来源拆成三条管线:
- 事件管线:每一轮对话结束后,判断是否包含用户偏好、明确事实或承诺事项;有则写入
event类型记忆; - 实体管线:从事件中抽取实体,写入实体表,同时反填到事件记录的
entities字段里,作为结构化索引; - 摘要管线:当对话轮次达到压缩阈值,生成或更新当前会话的
summary类型记忆。
三条管线不是孤立跑的。事件管线会触发实体管线,摘要管线又依赖事件和实体的产出。实际执行顺序是先抽实体,再判断重要性,最后决定是否进入摘要压缩队列。这样既能保证关键事实第一时间落库,也能避免对每轮对话都做一次昂贵的摘要模型调用。
4. 核心流程落地:写入、读取与遗忘
4.1 写入管线:从对话事件到记忆条目
写入不是把每句话都存下来,而是“筛选要记的、抽取出要素、向量化、落库”。我在项目里基于规则打分,先筛一轮,再做向量化和入库,成本可控。
打分规则参考了多轮对话中真正重要的信号:
- 用户明确表达偏好或否定(“我不喜欢”“我想要”“别再”)→ 加 0.8;
- 包含具体数字、日期、版本号 → 加 0.4;
- 消息是用户主动提供的背景信息,而不是提问 → 加 0.3;
- Agent 承诺在某个时间点做某件事 → 加 0.6。
只有最终分数超过 0.7 的对话事件才进入记忆库。这样可以滤掉大量“好的呢”“谢谢”之类的填充内容。向量化在筛选通过之后再做,避免每轮都调用模型浪费成本。
def write_memory(user_id, conversation_id, role, content): importance = score_importance(role, content) if importance < 0.7: return entities = extract_entities(content) vector = embed_text(content) db.execute( "INSERT INTO memory_items " "(agent_id, user_id, conversation_id, memory_type, content," " entities, importance, created_at, updated_at, embedding_model)" " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ("agent_a", user_id, conversation_id, "event", content, json.dumps(entities), importance, int(time.time()), int(time.time()), EMB_MODEL) )写完后我会同步更新实体表的出现次数。这个次数后续会用到,同一个实体被反复提及,说明它与用户高度相关,检索时应该获得额外加分。
4.2 读取与上下文组装:先过滤,再召回,后重排
读取流程我经历过两个阶段的演化。第一版是直接向量检索,把 top 10 塞给模型,效果很差;第二版改成“先过滤、再召回、后重排”三段式,质量明显提升。
整个检索顺序是:
- 按
user_id和memory_type过滤,保证只查当前用户的记忆; - 过滤后如果候选集太大,选取重要性较高的前 200 条;
- 用查询向量做相似度召回,得到 top 20;
- 叠加时间衰减和实体命中加分的重排,最终取 top 5~8 条;
- 按预算把结果组装成“已知信息”区,放到 prompt 前面。
def retrieve(user_id, query, max_tokens=1200): candidates = db.fetch_filtered(user_id, limit=200) if not candidates: return "" query_vec = embed_text(query) scored = [] for item in candidates: sim = cosine_similarity(query_vec, item["vector"]) time_decay = 0.9 ** (days_since(item["last_accessed_at"]) / 7) entity_bonus = calculate_entity_bonus(query, item["entities"]) final_score = sim * 0.6 + item["importance"] * 0.3 + entity_bonus scored.append((final_score * time_decay, item)) scored.sort(key=lambda x: x[0], reverse=True) selected = scored[:8] return assemble_context(selected, max_tokens)时间衰减这块我调了很多轮。衰减太快,用户三个月前提过的事实就被遗忘了;衰减太慢,陈旧信息会持续挤占检索结果。目前用的半衰期是一周衰减 10%,加上last_accessed_at的机制:被用户再次提及时,会刷新访问时间,让“反复被提起的旧事”依然能保持高排序。
4.3 遗忘与更新:让记忆会瘦身
遗忘机制是我花时间最多的地方,因为它在线上系统里最容易出问题。线上跑了两周后,记忆库里的碎片条目开始堆积,同一件事被拆成了好几种说法,向量召回结果越来越散。
我实现的清理任务分三步执行:
- 软删除标记超过 30 天且
importance < 0.3的条目; - 对同一实体下的条目做聚合,保留重要性最高的前 3 条,其余合并成摘要;
- 当摘要记忆本身的 token 超过预算时,触发一次重压缩,用“旧摘要 + 新事件”生成新摘要。
def gc_memory(): cutoff = int(time.time()) - 30 * 86400 db.execute("UPDATE memory_items SET is_deleted=1 " "WHERE importance<0.3 AND last_accessed_at<?", (cutoff,)) groups = db.fetch_duplicate_groups() for group in groups: keep = sorted(group, key=lambda x: x["importance"], reverse=True)[:3] merged = summarize_items(keep) mark_deleted([x["id"] for x in group if x not in keep]) db.insert_summary(merged)冲突更新也在这里处理。用户先说“不用邮件通知”,一周后说“还是发邮件吧”。如果不做冲突检测,两条记忆会同时被召回,Agent 自己都会分裂。
我的规则是:同一实体下同一属性的记忆,以updated_at较新且重要性较高者为准;拿不准时,不强行删除旧记录,而是在新记录上标记“已更新”,检索时只返回新版本。这样做比无脑覆盖更安全,遇到用户反悔的场景还能保留轨迹。
5. 实测半年后,我记住的这些坑
5.1 向量化模型不一致会让记忆彻底“错位”
这个坑是我上线后第二周踩的。最初的记忆系统用某版向量模型生成向量,后来考虑到效果提升,我升级了向量模型,并在写入端换成了新版本。
问题在于旧记忆里的向量还是老模型生成的,新查询向量是由新模型生成的。如果两个模型输出的向量空间不一致,相似度计算就是鸡同鸭讲。当时最直观的表现是:用户明明反复提过某家竞品名字,检索结果却永远召不到那几条记忆。
排查了很久才发现是向量空间漂移。修复方案就是在写入时给每条记忆记录embedding_model字段,检索时只取与当前模型同版本的数据。更稳妥的做法是:切换模型前,用离线任务把全量历史记忆重新向量化,灰度验证后再切换。
5.2 摘要记忆的 token 成本比想象中高
设计摘要机制时,我最初是每满 5 轮对话就触发一次摘要,结果 token 账单直接翻倍。原因不难理解:摘要不是“只压新增”,而是把旧摘要加新增对话一起重新读一遍,这个输入长度其实不小。
后来把触发策略改成“最近消息超过窗口预算 70% 时才触发”,触发频率从每局一次降到了每几个会话一次,成本立刻降下来。另一个优化是摘要时只将新增事件里重要的部分带上,不再把整轮原始对话都重读一遍。
所以我的建议是:摘要的触发条件要结合 token 预算,而不是固定轮数。宁可摘要生成得少一点、触发时批量压缩,也不要频繁用大模型做重复文本压缩。
5.3 多轮会话里的记忆归属问题
刚开始做检索时,我犯过一个低级但概率很高的错误:没有把user_id放进过滤条件,结果用户 A 的记忆被用户 B 的查询召回了。这个属于低级失误,但因为线上用户量大、对话切得又快,问题在测试环境根本复现不出来。
更隐蔽的是会话归属问题。用户在一次新会话里说“我刚才提到的那个报表需求”,它指代的是上一场会话的内容。如果只做向量检索,模型很难把“刚才”和昨天的会话关联起来。我的解决方案是把最近一次会话的摘要单独标记为scene_summary,在读取时永远置顶保留,哪怕它的相似度不是最高的。这样至少能让 Agent 知道“当前上下文延续自哪段历史”。
这个坑提醒我:记忆系统不能只考虑语义相关性,还要考虑对话的时间连续性和指代关系。
6. 从能跑到能运维:三个必须做的工作
6.1 可观测性指标:记忆有没有真的工作
记忆系统上线后,最容易出现的情况是“Agent 看起来像失忆了,但又不知道是写入失败、检索失败,还是组装失败”。所以我把记忆链路的关键节点都加了可观测指标。
| 指标 | 观测点 | 出现问题时的症状 |
|---|---|---|
| 写入成功率 | 每条记忆是否成功入库 | 该记住的事实反复记不住 |
| 召回命中率 | 离线标注测试集的召回表现 | 用户重复提问时答案不一致 |
| 上下文预算占用率 | 检索结果实际占用的 token | 超过预算时关键记忆被裁掉 |
| 检索延迟 P95 | 从查询到返回记忆的耗时 | 用户明显感觉回复变慢 |
| 摘要触发频率 | 会话中触发压缩的次数 | 频率过高导致成本暴涨 |
这些指标在后台用一个看板就能展示。实际排查问题时,最有效的抓手是“回放日志”:把某次异常查询的输入、召回条目、最终组装后的 prompt 完整打出来,基本能一眼定位问题在哪一段。
6.2 评估一个记忆系统该看什么
很多人问记忆系统怎么评估,我觉得不要只看在线效果,离线测试更重要。我攒了一个 20 条场景的跨会话测试集,每个场景模拟一个用户在多场对话中提供的事实和偏好,然后检查 Agent 在后续对话里能否正确召回和运用。
评估时重点看三个维度:
- 相关召回率:应该被想起的记忆是否被选中;
- 噪声条目率:召回结果里有多少条对当前问题无用甚至有害;
- 答案一致性:同一事实被多次询问时,Agent 的回答是否前后一致。
第一版系统相关召回率只有 55%,噪声条目率却有 30%,所以线上表现就是“记了,但每次想起的东西都不一样”。优化重排策略后,相关召回率提升到 82%,噪声条目率降到 8% 左右,在线反馈明显改善。
6.3 给记忆加版本号:避免“召回突变”带来的线上事故
最后一条经验,来自一次发布事故。当时优化了实体抽取规则,上线后大量实体表发生变化,导致次日所有用户的记忆召回结果和前一天完全不同。用户发现 Agent“像换了一个人”,客服那边一天接到几十个投诉。
原因是实体抽取规则属于记忆系统的“上游口径”,它一变,下游的实体加分和检索结果全部跟着变。从那以后,我把记忆系统做成了版本化发布:任何影响写入或检索口径的变更,先在灰度环境对全量历史记忆做批量回放,比较新旧版本的召回差异,确认一致性指标达标后再切换线上流量。
实际操作中,我给记忆加了一个pipeline_version字段,和embedding_model一起组成系统指纹。发布前对比指纹不一致时的召回结果,能直观看到影响范围。
最后说一个我现在坚持的小习惯:所有记忆条目必须有importance和source两个字段。importance负责告诉我“这条记忆在遗忘时应该排第几”,source负责告诉我“这条记忆是从哪来的、出问题时去哪查”。在真实项目里,被问到记忆系统最难的地方,我的回答始终是“删除”和“定位”——能判断什么该忘、能追踪每一条记忆的来源,这套系统才算真正可用。先做到能写入、按需读取、敢删除,你的记忆系统已经能覆盖绝大多数工业场景了。