如果你是从之前那几篇一路看过来的,应该还记得我们在Demo Agent上折腾过工具调用、琢磨过多轮对话。但不管做到哪一版,都有一个特别扎眼的毛病——它是个“金鱼脑”。用户昨天刚告诉它“我每周三下午都在某某健身房练背”,今天问一句“那我明天下班后该去哪儿”,它就彻底懵了。这种体验放在聊天玩具里还能忍,真要做成客服、助手、陪练这类产品级Agent,根本不行。
记忆管理就是来解决这个问题的。说白了,Agent的记忆管理要管的不只是“把聊天记录存下来”,而是要让Agent知道该记什么、存在哪儿、什么时候想起来、什么时候忘掉。这篇教程我尽量不堆概念,直接按我实际做项目的思路来写:先给记忆分类,再讲存储选型,然后拆写入、检索、遗忘这三个环节,最后附一个能跑的行程助手核心代码。想给Agent装上真正能用的记忆,照着这套思路落地基本不会跑偏。
1. 记忆缺失的代价:没有记忆的Agent只能算高级问答机
先别急着写代码。我见过不少人给Agent加记忆的方式就是在 messages 数组里越堆越长,堆到报错就问“我该怎么处理长上下文”,然后跑去调窗口大小。方向从一开始就错了。记忆管理不是一个存储问题,它是一套围绕“信息生命周期”的机制设计。你首先要搞明白Agent需要哪几种记忆,它们各自解决什么问题。
1.1 三种记忆到底指什么:工作记忆、情景记忆、语义记忆
我在设计记忆模块时,会把Agent的记忆分成三层,和人的记忆做一个粗粒度对照:
- 工作记忆:指当前这轮任务里临时要用的信息,对应到工程上就是上下文窗口里的消息列表。它该被快速读取、快速丢弃,不需要落地持久化。
- 情景记忆:指用户和历史交互中发生的具体事件,比如“昨天下午3点用户问过机票退改规则”“上周二订了去某地的酒店”。这类记忆带时间线特征,适合按时间存储和回放。
- 语义记忆:指从历史交互中提炼出来的事实和偏好,比如“用户偏好靠窗座位”“用户出差多、常驻某城市”。这类记忆没有强时间属性,是用户画像的构建基础。
这个划分不是学院派名词,它直接决定存储和检索方案。情景记忆要求你能回答“什么时候发生了什么”,所以要有时间戳和事件结构;语义记忆要求你能回答“用户是个什么样的人”,所以要能合并、去重、更新。把这两类混在一张表里存,后面检索一定乱。
1.2 Agent失忆的三个典型现场
我实际调试过几个Agent项目,“失忆”的表现基本可以归成三类:
第一类是跨会话失忆。用户昨天在客服Agent上报了工单号,今天再来问处理进度,Agent问“请问您的工单号是多少”。这不是技术问题,是产品体验问题——用户会觉得你在装傻。
第二类是事实冲突。用户在闲聊里说过“我平时不吃辣”,结果第二天Agent推荐了一堆川菜馆。它不是找不到信息,而是相关记忆没有被检索出来,或者检索出来了但权重不够,被无关上下文淹没了。
第三类是上下文窗口的“虚假记忆”。很多人的记忆就是无脑往messages里塞对话,结果上下文越来越长,模型开始“记住”一些被裁剪掉的中间内容,或者因为太长的关系把早期关键指令给稀释了。这本质上还是没有做记忆管理造成的。
这三类问题的共同根源,是你把“存储”当成了“记忆”。存下来的只是字节,能被合适时机调取、能被更新和覆盖的才是记忆。后面几节我会逐个说清楚怎么设计。
2. 存储设计:别一上来就上向量数据库
很多人一说给Agent加记忆,脑子里第一个蹦出来的就是向量数据库。我的建议是先忍住。存储选型应该跟着记忆类型走,不同类型的记忆有完全不同的读写模式,一股脑全做向量化纯属给自己挖坑。
2.1 短期记忆:一段受token预算控制的对话窗口
短期记忆的实现听上去简单,就是维护最近几轮对话。但有个细节很容易被忽视:这个窗口的大小不是“我记N轮”,而是“我的token预算允许我装多少轮”。轮数是不可控的,某用户一句废话可能顶别人五句,token才是真正约束模型能力的单位。
我在项目里一般这样做:
- 维护一个消息列表,每进来一轮就追加。
- 每次追加后统计当前总token数,超出预算就从最旧的消息开始淘汰。
- 预算分配上给短期记忆留一个上限,比如总窗口的40%,剩下的留给系统指令、检索到的长期记忆和当前用户输入。
这样做的核心思路是:短期记忆是“最贵”的记忆,因为每次请求它都要随上下文发送给模型,是成本最敏感的部分。所以它只在当前会话内存活,一旦会话结束就该被压缩、沉淀或者直接丢弃。
2.2 长期记忆:从KV到向量库的分级方案
长期记忆才有资格谈持久化。但“持久化”三个字底下,至少有两种完全不同的存储需求:
一种是结构化事实型记忆,比如“用户姓名”“会员等级”“偏好口味”。这类数据字段明确、写一次要读很多次,而且经常要精确更新。这种场景用一个简单的键值存储或者关系表就能搞定,把它向量化反而是负优化。
另一种是语义片段型记忆,比如一段对话里用户描述过的某种复杂需求、一段你希望Agent以后能关联起来回忆的内容。这类信息没有固定字段,只能靠语义相似度去检索,这才是向量数据库的用武之地。
所以我的经验是:长期记忆不要只依赖单一存储,按信息形态分级。你可以先画一张表做选型参考:
| 记忆类型 | 典型内容 | 推荐存储 | 检索方式 | 更新频率 |
|---|---|---|---|---|
| 用户画像型 | 姓名、偏好、常驻地 | KV或关系表 | 精确匹配 | 低频 |
| 事件记录型 | 做过什么、买过什么 | 时序日志表 | 按时间范围筛选 | 高频追加 |
| 语义片段型 | 对话中的隐含需求 | 向量索引 | 相似度检索 | 中频 |
| 会话摘要型 | 某次长聊的压缩结果 | 文档存储 | 结构化+向量混合 | 会话结束时写入 |
如果你刚开始做,别急着搭一堆组件。我通常的建议是:先用KV存事实,用文件或者一张表存事件,用向量库存语义片段。等量级上来、检索效果出现瓶颈了,再考虑引入专门的向量库服务。
2.3 我比较推荐的混合存储结构
跑过一个完整的客服Agent之后,我沉淀出来的存储结构大概是这个形态:
每个用户有一个独立的记忆空间,里面有“用户资料表”存结构化属性;“事件表”按时间戳记流水;“语义记忆索引”存从聊天中抽取出来的说明性内容。三张表通过 user_id 关联,检索时先查精确字段,再走语义搜索,最后把结果按重要性和时间排序拼进提示词。
有些平台喜欢用向量库一统天下,把用户资料也编码成向量。我试过,后果是更新某个字段时需要重新生成向量,否则就会出现“资料明明改了,Agent还按旧字段回答”的诡异现象。结构化数据就该被精确更新,别硬塞给向量库。
3. 记忆的生命周期:写入、检索、遗忘三道闸门
存储方案定了,记忆管理真正核心的部分才开始。我给记忆设计了三个处理环节:写入、检索、遗忘。你可以把它们理解成三道闸门,每一道都要有明确的策略,否则记忆系统就是一锅粥。
3.1 写入闸门:不是所有对话都值得记
很多人的第一版长期记忆是这样的:每轮对话结束后,把整段对话文字塞进向量库。结果很快就会发现检索出来的记忆全是垃圾,因为日常对话里90%的句子不具备长期记忆价值。
写入策略的核心是“筛选”。我在实际项目中会用两种方式配合:
第一种是规则抽取。针对明确的字段,比如用户报了自己名字、说了某个偏好、提到某个具体日期,直接用规则从对话里抽出来写进结构化表。优点是精确,缺点是无法覆盖复杂表述。
第二种是总结沉淀。会话结束后,让大模型基于整段对话生成几条记忆摘要,例如“用户提到在公司午休时间喜欢到楼下咖啡店坐坐”。每次生成摘要时,我还会额外要求模型标注“可遗忘项”,也就是那些只对单次任务有用、不值得长期保存的临时信息。这一步能大幅降低噪声。
写入的触发时机也很讲究。我的做法是:会话进行中只写确定的、规则能识别的事实;会话结束后才做全局总结沉淀。实时写入容易把用户中途反悔的话也记进去,之后又要做一遍纠错,非常不划算。
3.2 检索闸门:相似度之外还要有过滤和重排
写完记忆,Agent碰到新输入时要从记忆里挑相关内容,也就是检索。检索如果只做“向量相似度TopK”,那基本等于盲人摸象。我会在检索链路里加三道护栏:
第一道是用户隔离。所有检索必须带 user_id 过滤,千万不能做成全局检索,否则用户A的记忆会串到用户B那里。这种错误在产品环境是事故级的。
第二道是时间衰减。同样相似度的两条记忆,一条是昨天的、一条是半年前的,它们的可信度完全不同。我习惯在打分公式里叠加一个时间衰减因子,让近期记忆获得更高的排序权重。
第三道是上下文重排。向量检索结果先各给一个相似度分,汇集到提示词时会再做一轮重排,把那些“语义相似但实际是另一个话题”的结果降权,把和当前任务真正相关的记忆往前放。这一步效果很明显,尤其当你的向量模型质量一般的时候。
3.3 遗忘闸门:让记忆既能新增也能过期
这一条是被很多人忽略的。记忆只增不减的系统最后一定会变质。用户三个月前的偏好可能已经变了,而遗忘机制就是防止Agent抱着过期数据不放。
我做的至少有三层:
- 短期遗忘:工作记忆随会话结束清空,不残留。
- 时间过期:给记忆打上 ttl(有效期),比如“用户在出差”这类信息两周后就失效,再检索到就自动丢弃。
- 合并覆盖:用户说“我现在不吃辣了”之后,新的“不吃辣”记忆要覆盖旧记录。我一般用时间戳加置信度来判断谁覆盖谁,新记忆的置信度权重更高。
还有一个容易被忽视的问题:大模型在总结记忆的时候会“脑补”,写出来一些对话里没有的细节。我在长期记忆写入时会做一个一致性校验,让模型输出必须附带上原始对话段落编号,如果哪条总结找不到对应出处,宁可丢弃也不写入。这个坑我踩过一次,修起来非常痛苦,因为错误记忆会像滚雪球一样影响后续所有对话。
4. 手把手做一个带长期记忆的行程助手
理论环节说完,下面进入实操。我用一个“行程助手”来演示,它需要记住用户提到的地点偏好、时间安排等长期信息,并在后续对话里正确调用。这个Demo刻意不依赖特定云服务,代码结构让你可以轻松替换成自己手头的向量库和模型接口。
4.1 整体链路与最小依赖
整个链路是这样的:
- 用户输入一句话。
- Agent先检索该用户的长期记忆。
- 短期记忆窗口加上检索结果,一起组装成提示词。
- 调用模型生成回复。
- 回复之后对当前对话做一次提取,把确定的偏好写入长期记忆。
代码层面只需要 Python 3 和一个能调用大模型API的环境。我这里自己实现了一个极简向量索引(基于余弦相似度),只用来演示语义检索的原理。你在实际项目里完全可以换成任何一个你想用的开源向量库,接口保持一致就行。
import time import math def estimate_tokens(text: str) -> int: # 中英文混合场景下的粗略token估算 return max(1, int(len(text) * 1.3)) def cosine_similarity(a, b): dot = sum(x * y for x, y in zip(a, b)) norm_a = math.sqrt(sum(x * x for x in a)) norm_b = math.sqrt(sum(x * x for x in b)) if norm_a == 0 or norm_b == 0: return 0.0 return dot / (norm_a * norm_b)这里我把 token 估算简化成等长计费,真实项目里建议用对应的分词器接口。
4.2 核心代码拆解:短期记忆与长期记忆模块
先看短期记忆模块。它的职责就是维护一个受预算控制的滚动窗口:
class ShortTermMemory: def __init__(self, max_tokens=1500): self.messages = [] self.max_tokens = max_tokens def add_message(self, role: str, content: str): self.messages.append({ "role": role, "content": content, "ts": time.time(), }) self._trim() def _trim(self): while len(self.messages) > 1: total = sum(estimate_tokens(m["content"]) for m in self.messages) if total <= self.max_tokens: break # 从最旧的消息开始淘汰 self.messages.pop(0) def to_text(self) -> str: return "\n".join(f"{m['role']}: {m['content']}" for m in self.messages)下面是自己实现的迷你向量索引,只保留核心逻辑,方便你理解检索原理:
class SimpleVectorIndex: def __init__(self, embed_fn): self.embed_fn = embed_fn self.items = [] # 每个元素是 (metadata, vector) def add_item(self, metadata, text): vector = self.embed_fn(text) self.items.append(({**metadata, "text": text}, vector)) def query(self, query_text, top_k=3, user_id=None, max_age_seconds=None): query_vec = self.embed_fn(query_text) scored = [] now = time.time() for metadata, vector in self.items: if user_id is not None and metadata.get("user_id") != user_id: continue if max_age_seconds is not None: score_age = 1.0 if metadata.get("ts", 0) - now > 0 else max(0.0, 1.0 - (now - metadata.get("ts", now)) / max_age_seconds) else: score_age = 1.0 score = cosine_similarity(query_vec, vector) * score_age scored.append((score, metadata)) scored.sort(key=lambda x: x[0], reverse=True) return scored[:top_k] def search_all(self, keyword, user_id=None): results = [] for metadata, _ in self.items: if user_id is not None and metadata.get("user_id") != user_id: continue if keyword in metadata.get("text", ""): results.append(metadata.copy()) return results注意上面我用了一个关键词搜索方法 search_all,它是给精确更新用的。用户说“我改主意了”的时候,你需要先找到旧记忆再覆盖,而不是直接再插一条,否则会出现两条互相矛盾的记忆同时被检索出来。
长期记忆模块封装了“写事实”和“读记忆”两个能力:
class LongTermMemory: def __init__(self, embed_fn): self.embed_fn = embed_fn self.facts = {} # user_id -> {属性: 值} self.index = SimpleVectorIndex(embed_fn) def record_fact(self, user_id, attribute, value): # 结构化事实,直接精确覆盖 if user_id not in self.facts: self.facts[user_id] = {} self.facts[user_id][attribute] = value def get_fact(self, user_id, attribute): return self.facts.get(user_id, {}).get(attribute) def memorize(self, user_id, text, importance=0.5): self.index.add_item({ "user_id": user_id, "ts": time.time(), "type": "episodic", "importance": importance, }, text) def recall(self, user_id, query_text, top_k=3): semantic = self.index.query(query_text, top_k=top_k, user_id=user_id, max_age_seconds=86400) results = [] for score, metadata in semantic: results.append({"src": "semantic", "score": score, "content": metadata["text"], "ts": metadata["ts"]}) facts = self.facts.get(user_id, {}) for attr, value in facts.items(): if attr and value: # 结构化事实转成自然语言参与排序,维持统一输出格式 results.append({"src": "fact", "score": 0.8, "content": f"{attr}:{value}", "ts": 0}) results.sort(key=lambda x: x["score"], reverse=True) return results[:top_k]这一步就有意识地把结构化事实和语义片段统一成带 score 的列表,在组装提示词时一视同仁。实际项目里很多人把两条检索链路分开做,结果提示词里一会儿出现句子、一会儿出现字段,模型经常搞混。
4.3 跑通一次完整对话的验证过程
组合起来你就能模拟一次完整交互了。我习惯把每次交互画成五步:查记忆、更新短期记忆、调模型、取回复、沉淀记忆。代码如下:
class MemoryAgent: def __init__(self, embed_fn, llm_fn): self.stm = ShortTermMemory() self.ltm = LongTermMemory(embed_fn) self.llm_fn = llm_fn # 大模型调用函数 self.user_id = "demo_user" def process_input(self, user_text): # 1. 检索长期记忆 recalled = self.ltm.recall(self.user_id, user_text, top_k=3) memory_text = "\n".join(f"[记忆] {item['content']}" for item in recalled) # 2. 短期记忆写入用户输入 self.stm.add_message("user", user_text) # 3. 组装提示词 system_prompt = ( "你是一个行程助手。请结合已知记忆回答," "如果记忆中没有,可以明确说不知道。\n" ) messages = [ {"role": "system", "content": system_prompt + memory_text}, ] messages.extend( {"role": m["role"], "content": m["content"]} for m in self.stm.messages ) # 4. 调用模型 reply = self.llm_fn(messages) # 5. 简单提取并沉淀长期记忆 self.stm.add_message("assistant", reply) if "我喜欢" in user_text: self.ltm.record_fact(self.user_id, "偏好", user_text.split("我喜欢")[-1].strip()) else: self.ltm.memorize(self.user_id, user_text, importance=0.3) return reply跑一轮验证是这样的:
def llm_fn(messages): # 接入你的大模型API服务商 # response = your_llm_client.chat(messages) last = messages[-1]["content"] return f"我已收到约束,也参考了记忆中的信息。你说的是:{last}。" embed_fn = lambda text: [float(ord(c) % 100) / 99.0 for c in text[:32]] agent = MemoryAgent(embed_fn=embed_fn, llm_fn=llm_fn) print(agent.process_input("我喜欢靠窗的座位")) print(agent.process_input("帮我订明天下午去某地的车票")) print(agent.ltm.recall("demo_user", "坐车偏好", top_k=5))上面这个简易 embedding 函数工业上没什么用,纯粹让Demo能独立跑通。但后缀的 recall 结果能看到“靠窗”相关记忆被检索出来,说明链路是通的。换成真实embedding模型后,语义召回的效果会直接上一个台阶。
5. 工程化路上最常踩的五个坑
代码能跑只是开始。把这套东西放到真实数据上,你大概率会遇到下面这些问题。我把它们按杀伤力从大到小排一遍,每个都附上我的处理经验。
5.1 看似能搜到,其实召回全是噪声
最迷惑人的情况是:你搜索“靠窗”结果也能搜出来,但搜索“用户喜欢哪种座位”或者“这次的偏好是啥”,搜出来的东西就开始乱套。原因基本出在三个地方:
一是嵌入模型太弱,中文长句、近义词辨识度不够。解决办法是换更强的大模型向量接口,或者对长文本做分块再向量化,块与块之间保留重叠,避免语义跨块断掉。
二是没有做查询改写。用户问“附近吃饭推荐”,但记忆里存的是“用户喜欢清淡口味”,这句原文和“吃饭”在语义上有关联但向量距离未必近。我现在的做法是检索前先用模型把用户问题改写成若干个候选查询句,分别检索再合并结果,召回率会明显提升。
三是缺少过滤。比如你存了十条“用户喜欢靠窗”,然后检索出来十条全是靠窗。不是你记性差,是你没有做去重。写入前先检索一次,如果已经有相同语义的记忆,就做合并和更新时间戳,不要每次都插新。
5.2 记忆漂移:用户改主意之后旧记忆还在
这是“甜蜜的负担”——系统记住太多,忘记太少。用户说“我不喜欢靠窗了”,结果你下次还是推荐靠窗座位,因为他原来喜欢靠窗的句子还静静躺在向量库里。
要解决记忆漂移,我的方案是给记忆加置信度和生命周期两层属性。置信度低的记忆不被输出;生命周期过了的记忆即使相关也要被筛掉。用户的最新说法写入时,对同一主题的旧记忆做主动降权和过期,而不是等到检索时才发现问题。
实际操作上,我在“新增记忆”之外总会跑一条“更新记忆”流程:拿新文本去检索自己用户的旧记忆,相似度超过一定阈值就把旧记忆标记为失效,并建立指向新记忆的跳转。这种方式比较笨,但稳定可靠。
5.3 记忆膨胀带来的成本失控
长期记忆臃肿最直接的后果是每次对话都带着一大段历史摘要,token费用直线上升。我在一个上线项目里遇到过提示词被顶到十几万 token 的情况,一个月的接口账单涨了四成。
解决思路是分档:先把最常命中的记忆放在本地KV里,每次必读;把大段的语义记忆做成摘要索引,只有命中关键词才加载全文;再定一个规则,超过多少天没有被检索过的记忆转入冷存储,不再参与实时检索。
另外,提醒你一个很多人忽略的隐藏成本:每次把记忆写进向量库也要花一次embedding调用。如果每轮对话都写,成本一样不低。我的策略是批量写入——会话结束后统一处理,而不是对话过程中每条都调一次embedding接口。
5.4 用户隔离与数据边界
这个问题做ToB项目尤其要命。多个用户的数据存在同一个索引里,如果查询时没有严格带 user_id 做过滤,一旦出了串号事故,在合规上基本无法解释。这不是技术细节问题,是一票否决项。我从一开始就规定:所有写入和读取的操作函数强制要求传 user_id,不允许出现不带隔离条件的公开查询接口。宁可代码冗余一点,也不能靠“程序员记得加过滤”来保证数据边界。
5.5 把记忆和工具调用混在一起
最后一坑比较隐蔽:Agent如果同时有工具调用和记忆系统,你很容易把工具返回的结果也当成记忆存下来。比如查天气工具返回了“今天晴”,这是实时数据,明天就失效了,不该进入长期记忆。我的经验是给记忆打上来源标签:用户陈述、模型总结、工具结果。不同来源使用不同的有效期,工具结果一般只留在短期记忆里,不沉淀到长期记忆。
最后补充一个偏实战的小技巧
聊到这儿,我想把记忆系统的可观测性单独拎出来强调一下。线上跑Agent的时候,最难受的不是功能不对,而是你没法判断它为什么这么回答。是不是记忆检索错了?还是根本没检索到?还是检索到了但提示词里被别的信息挤掉了?
我从第一次做带记忆的Agent起就养成了一个习惯:在每次请求的返回里附带一个 debug 字段,把这个请求用到了哪些记忆、每条记忆的相似度分数、短期记忆裁掉了哪几条旧消息都打出来。测试阶段直接打印在控制台,上线阶段写进日志。不要嫌难看,记忆系统是Agent里最难调试的部分,没有可观测性你只能靠猜,越猜越离谱。
说白了,给Agent装上记忆从来不只是“加个向量库”那么简单。它更像是在模拟人的认知过程:什么值得记住、什么时候想起来、什么时候该忘。按着写入、检索、遗忘这套框架去拆解,再一步步落到代码里,你的Agent才算是从“会说话”进化到了“记得住”。下一步你就可以在这个基础上去做自动化的记忆合并策略,或者按记忆置信度动态调整提示词权重,那基本就是往更成熟的记忆体系靠近了。