Agent记忆系统设计：三层架构与工程落地全攻略

做Agent这件事绕不开一座大山记忆。不管是自己从零撸Agent框架还是用LangGraph、Spring AI这类现成的编排层跑几个demo之后你会发现最让人头疼的不是模型调用不是工具定义而是“这Agent怎么聊着聊着就忘了前面说过的话”。这个系列写到现在把Agent的主循环、规划、工具调用、技能编排都聊过了。这篇终于轮到最容易被低估也最决定体验上限的模块记忆系统。如果你正在自研Agent或者准备给现有的Agent加上“记忆力”这篇内容就是从架构设计到落地排查的完整拆解。1. 为什么要给Agent做记忆系统没有记忆的Agent只是聊天机器人1.1 无状态对话的假象为什么上下文窗口不能直接当记忆用很多新手刚开始做Agent的时候觉得记忆很简单——“不就把历史消息拼到prompt里吗”这个想法在聊天机器人阶段没错但一旦进入真正的Agent场景问题立刻暴露。第一个问题是成本失控。模型上下文窗口是有限的随着对话变长你很快会发现历史消息占用的token越来越多。第二个问题是噪音淹没重点如果历史里有一百条消息真正对当前任务有用的可能只有三条其余全是寒暄和中间过程。把全部原文塞给模型它容易被无关信息干扰注意力被稀释回答质量反而下降。第三个问题也是最核心的问题上下文窗口本质上是“临时便签”它不是记忆。用户昨天跟你说过“我喜欢简洁风格”今天新开一个会话模型什么也不记得。跨会话的持久化、跨任务的积累、长期偏好的学习——这些都不是一个窗口能解决的。所以做记忆系统的本质是给Agent设计一套“分层存储和按需调取”的机制该放在手边的放在手边该归档的归档该遗忘的遗忘。这和人脑的工作方式是一样的——你不会把大学学过的全部知识都放在脑子里惦记着但需要用的时候能想起来。1.2 记忆系统的三层模型缓冲层、工作层、持久层我目前见过最合理的Agent记忆架构是分三层。第一层叫短期缓冲Short-term Buffer存的是最近几轮对话的原始消息。这一层的目的很单纯保证上下文连贯。它不需要结构化直接拼接到prompt里就行。关键在于“裁剪”策略——保留最近N轮或者按token预算保留最近的量超出部分进入下一层处理。第二层叫工作记忆Working Memory存的是当前正在执行的任务的中间状态。比如订单Agent处理一笔退款“订单号是A10023”、“用户已同意方案B”、“退款接口已调用但还没返回结果”——这些就是工作记忆。它们在任务进行期间有极高的时效性任务结束后大部分失去价值少数值得沉淀到长期记忆。工作记忆通常放在内存里用字典或者专门的上下文对象维护伴随Agent运行周期生灭。我自己做的时候直接用了一个带Schema校验的上下文对象每个字段代表一项任务状态比扔在prompt里让模型自己维护要稳得多。第三层叫长期记忆Long-term Memory是真正意义上的“档案库”。存的是跨会话依然有效的信息用户偏好、领域知识、项目事实、过往经验。比如“用户不吃辣”、“项目X部署在Y环境”、“上次排查出这个问题是因为配置错误”。这层需要用结构化存储加向量索引查询时按相关性和时效性召回。三层之间的关系是流动的短期缓冲里的信息经过抽取有价值的部分进长期记忆任务状态里的关键结论在任务结束后沉降到长期记忆。反过来长期记忆里召回的信息注入到工作记忆或短期缓冲里参与当前的推理。2. 记忆的核心生命周期写入、检索、冲突处理与遗忘2.1 写入阶段抽什么、怎么抽、存到哪里记忆系统的第一个关键环节不是查询是写入。写什么进长期记忆直接决定这个系统有没有用。我的经验是遵循一个原则存“事实”而不是存“原文”。怎么说假设用户说“我不是很喜欢吃辣的东西而且最近在健身晚饭尽量少碳水。”这句话如果原封不动存进去以后每次召回都能匹配到“晚饭”“健身”这些字但真正有用的记忆项是两条事实口味偏好不吃辣、健康目标健身中晚餐低碳水。把对话压缩成结构化的事实条目召回效率和后续触达质量会好很多。在实操里我一般把记忆项分成三种类型每类的抽取逻辑不同实体类记忆包括用户ID、项目名、订单号、报错消息中的关键标识。这类记忆用规则抽取就够了不需要大模型参与正则加简单词表就能处理速度快成本低。偏好类记忆来自用户对话中的主观表达比如“我觉得/我喜欢/我不想”。这类信息上下文依赖强需要让大模型做一次轻量抽取输出固定格式的JSON包含属性名和属性值。注意限定抽取范围不然模型会把寒暄也当成偏好存下来。事件类记忆指发生过的事比如“2024-06-12部署了v2版本”、“上周已经处理过这个客户的投诉”。这类信息通常来自工具调用结果或者用户陈述结构化程度较高适合在Agent主循环里埋点记录也可以在对话结束后让模型做一次总结。在存储之前还有一个很重要的步骤去重与合并。同样的实体可能在不同时间被写入多次比如“用户偏好-不吃辣”出现过两次如果不管后面查询会召回两条重复记忆浪费上下文空间。我用的办法是在写入时做一次embedding相似度比对如果新记忆和已有记忆的相似度超过阈值通常是0.92以上就不新增而是更新原有条目的updated_at时间戳或者合并补充信息。2.2 检索阶段什么时候查、怎么查、召回顺序有了记忆库接下来是检索。这里第一个原则是不是每一轮都要查记忆。如果每轮用户输入都做一次向量检索成本高且容易注入噪音。我设计了一个简单的“查询触发机制”根据用户输入是否包含明确的实体引用、疑问代词、上下文依赖词比如“刚才说的那个”“上次提到”来决定是否触发深层检索。一个经验参数是大约只有三成到四成的用户输入会真正触发长期记忆检索其余时候靠短期缓冲就够了。当触发检索之后不要只做单一的向量相似度召回。我踩过不少纯向量检索的坑后面第4章会详细说这里先讲架构多路召回加合并重排。多路召回一般是三路并行向量相似度召回按语义找相关记忆、关键词召回用BM25或者简单倒排索引处理专有名词、最近写入召回时间维度很多刚沉淀的短期事实往往最有用。三路召回各自拿出top 10到top 20的候选然后合并去重进入重排阶段。重排的评分公式我用的是一套加权打分核心思想是融合三个维度final_score 0.5 * relevance 0.3 * recency 0.2 * importancerelevance是向量相似度分数recency是时间衰减因子importance是记忆条目自带的重要程度标签1到5分在写入时通过规则或模型标注。recency的具体计算我用了指数衰减recency 0.9 ^ (hours_since / half_life)half_life设置为七天。也就是说一条记忆在七天后权重衰减到九成一个月后大约衰减到六成。这套参数实际跑下来效果不错既记住近期事实又不会让旧的重大信息完全被淹没。重排后的结果按最终分数取前3到5条格式化为“记忆片段”注入到prompt的system消息里标注为“以下是该用户的历史记忆供参考”。这里还有一个小细节注入的位置放在system消息末尾、接近对话内容的位置比放在最开头效果要好——模型对靠近“当前问题”的上下文更敏感别把记忆埋在一长串system指令里。2.3 更新与冲突处理记忆不是只增不改很多人的记忆系统做成了“只写不更新”这是隐患。用户今天说喜欢吃辣明天体检发现胃炎他说“以后不吃辣了”——旧记忆和新记忆冲突系统怎么处理如果新记忆覆盖旧记忆那旧记忆在某些场景下可能还有价值如果不覆盖检索时两条记忆同时命中模型会糊涂。实践里我采用的策略是带置信度的版本化存储。新写入时如果检测到与已有记忆存在冲突不直接覆盖而是对比置信度高置信度覆盖低置信度比如直接陈述的置信度高于推测的置信度和时间戳相同置信度下新条目替代旧条目。被替代的旧记忆不物理删除而是打上superseded标记在常规检索中排除但当用户明确询问“我之前说过什么”这类元问题时可以追溯到历史版本。这个设计让我少踩了很多坑。之前用最朴素的“后写覆盖先写”方案出现过系统刚刚更新的偏好又被旧记忆翻出来混淆了判断的情况。版本化听起来复杂实现起来就是一个status字段加一个valid_from时间戳的事收益却很大。遗忘机制同样重要。长期记忆不能无限膨胀否则未来的每一次检索都是大海捞针。我定期做两类清理一类是自动过期比如临时的会议安排、一次性的事件记录在事件时间过去之后标记过期另一类是重要性衰减长时间未被命中的低重要性记忆定期降级或者归档到冷存储。一句话经验舍得删记忆系统的价值不在存得多在查得准。3. 技术选型与实现细节从存储方案到主循环接入3.1 存储方案对比别一开始就上重型向量库记忆系统选存储最大的误区是一上来就上专业的向量数据库。我见过不少项目记忆还没积累到几千条就开始折腾部署Qdrant或者Milvus运维成本直接拉满。其实不同阶段的选型应该完全不一样。原型验证阶段我用的是SQLite加JSON列事实条目存在一张表里embedding向量存成二进制或者JSON数组。召回时直接SQL分页全量扫描再在内存里算余弦相似度。数据量在几万条以内这个方案的性能和准确性完全够用而且部署成本为零重启不丢数据日志好排查。数据量上来之后需要引入真正的向量索引。这时的选择也未必是独立的向量数据库可以优先考虑给现有的数据库加向量能力——比如PostgreSQL的pgvector扩展或者SQLite的sqlite-vec。我把同一套SQLite方案迁移到pgvector只改了查询语句所有业务代码不用动。只要业务的数据量到不了亿级pgvector非常稳事务也顺手。单条记忆的表结构我给一个参考设计CREATE TABLE memories ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, session_id TEXT, content TEXT NOT NULL, memory_type TEXT NOT NULL, -- entity / preference / event importance INTEGER DEFAULT 3, version INTEGER DEFAULT 1, status TEXT DEFAULT active, -- active / superseded / expired metadata TEXT DEFAULT {}, embedding VECTOR(1024), created_at TIMESTAMP DEFAULT now(), updated_at TIMESTAMP DEFAULT now() ); CREATE INDEX idx_memories_user_status ON memories (user_id, status); CREATE INDEX idx_memories_updated ON memories (updated_at);这里比较关键的是user_id和status的联合索引几乎所有查询都带着用户维度这个索引能让常规的记忆召回快好几个数量级。另外我把content和metadata分开存content是给模型看的自然语言metadata是给程序看的标签和关联信息——这俩混在一起会让解析和过滤都很痛苦。3.2 Embedding模型选择与文本切分策略embedding模型的选择对检索质量影响很大这块的坑比很多人想象的多。我试过几类通用英文模型OpenAI的text-embedding-3系列、中文专项模型比如BGE系列、以及多语言大模型。结论是如果你的Agent主要服务中文用户直接用中文优化过的开源模型比如bge-m3性价比很高。英文项目再考虑OpenAI或者Cohere。通用模型处理中文专有名词时召回质量有明显差距这个差异在日常对话里还能忍一旦涉及人名、订单号、产品名会频繁翻车。embedding还有一个容易被忽视的细节文本切分粒度。切太粗一块记忆里包含多个事实向量语义被拉平召回精度下降切太细每个记忆太碎片化缺乏上下文召回后模型也读不出完整信息。我常用的粒度是以语义段落为单位切分每个记忆条目控制在200到500字之间。比如网页转markdown的知识片段按markdown的二级标题分块块内不足200字的和下一条合并超过500字的按句号二次切分相邻块保留50字重叠避免切在语义边界上。切分之后写入前的最后一步是加“记忆前缀”。直接存原文的召回效果不如给向量加一个语义角色前缀比如存用户偏好时内容改写成“用户偏好不喜欢吃辣健身中晚餐低碳水”存项目知识时改写成“项目知识服务X部署在生产环境Y”。这个小技巧让向量能区分“这是关于谁的什么信息”召回命中率提升明显而且基本不增加成本。3.3 记忆模块与Agent主循环的接入方式到这里记忆系统的基本组件都有了但真正让它跑起来的是和Agent主循环的接入方式。我见过不少项目把记忆功能做成一个工具让Agent自己决定要不要调用memory_save和memory_lookup结果很不稳定——模型经常忘记调用或者在不需要的时候乱调用。我自己的方案是“主循环内置挂钩”不让模型自由决定而是让框架自动处理。完整的接入逻辑分三步用户输入到达时先做一次记忆召回把命中的长期记忆注入到prompt的上下文区。这一步发生在模型推理之前用户无感知。模型响应和工具执行结束后把这一轮的对话与工具结果交给“记忆抽取器”由它提取有价值的信息并写入记忆库。这一步通常是异步的避免阻塞用户侧的主链路。任务状态发生关键变化时比如订单状态变更、用户明确提出新偏好立刻同步更新工作记忆同时判断是否需要沉淀到长期记忆。简化版的代码骨架大致是这个样子class MemorySystem: def __init__(self, vector_store, embedder, extractor): self.store vector_store self.embedder embedder self.extractor extractor def recall(self, user_id, query, top_k5): # 1. 判断是否需要深层检索 if not self._need_recall(query): return [] # 2. 多路召回 candidates self.store.multi_recall(user_id, query, self.embedder) # 3. 重排打分 ranked self._rerank(candidates, query) return ranked[:top_k] def absorb(self, user_id, session_id, dialogue): # 1. 从对话中抽取结构化记忆 items self.extractor.extract(dialogue) for item in items: # 2. 去重合并 冲突检测 if self._is_dup(item): self.store.merge(item) continue if self._has_conflict(item): self._resolve_conflict(item) continue # 3. 向量化写入 vec self.embedder.embed(item.content) self.store.insert(user_id, session_id, item, vec)主循环那边挂两个钩子就够了入口钩子调用recall只对特定user_id和session生效出口钩子调用absorb异步执行用队列解耦。这套方案跑下来记忆系统的参与是稳定且可预期的不会出现“靠模型自觉”的不可控状态。4. 落地过程中踩过的坑和排查实录4.1 上下文窗口打爆token预算必须提前规划这是所有Agent记忆系统落地时第一个撞见的坑。早期的我短期缓冲和长期记忆一股脑全塞进上下文用户多聊几轮就爆窗口报错刷屏。后来总结出一套“token预算分配法”。以常见的200k上下文窗口为例输出侧留20%也就是40k因为输出长度不可控必须预留输入侧可用160k。这160k里系统提示占30k短期缓冲最多占40k长期记忆注入占20k工作记忆占10k剩下60k是动态余量用来处理单轮长输入或者特殊情况。预算不是摆设要在代码里硬编码成阈值每次构造prompt前先算一遍当前急需占用的token总量超出预算就触发裁剪。短期缓冲的裁剪逻辑我采用的是“部分摘要加丢弃”一旦缓冲超过预算的七成就把最早的几轮对话送入摘要程序生成一段浓缩的事件摘要替换掉原始消息。注意摘要只保留“对后续任务可能有用的信息”寒暄直接丢掉。实测中一个三小时的长会话最终在上下文里留下的可能只有二十轮的原文和三段摘要效果和全量粘贴相差不大成本却少了六成以上。4.2 检索噪声与记忆串扰隔离和阈值缺一不可纯向量检索的噪声问题我在第3章说过这是记忆系统里最隐蔽的敌人。举个例子用户聊到“最近项目部署有点慢”向量检索出来的top1是“用户上个月抱怨过部署流程复杂”看似相关实际对当前问题没有直接帮助反而把模型的注意力带偏。这里的核心解法是双管齐下。一个是检索阈值校准跑一批真实对话样例记录每条召回的向量相似度分数画出分数分布然后选择分布上“相关记忆”和“无关噪音”的边界作为threshold低于阈值直接不召回。我做中文场景的经验bge系列模型下阈值通常在0.30到0.42之间。但一定不要盲抄网上参数embedding模型不同分布差异很大。另一个是记忆串扰的隔离。如果你的Agent服务的不是一个用户而是多用户甚至多Agent最痛苦的问题是A用户的记忆串到了B用户的召回结果里。这个坑我踩得很惨做了全局向量索引之后测试阶段一切正常一上线就被用户投诉“这个Agent怎么知道我的私人信息”排查很久才发现是忘了加user_id过滤器。后来学乖了所有查询都必须携带user_id维度SQL里加上WHERE user_id ?向量查询必须用带过滤条件的接口从架构上杜绝串扰。多Agent拿同一个记忆库时也一样每个Agent要有agent_id每个用户会话要有session_id三层维度限定死记忆不会串味。4.3 记忆冲突与错误记忆宁可少存不可错存还有一个经常被忽视的问题记忆系统写入错误信息之后错误的记忆会一遍遍被召回到prompt里导致Agent在错误认知上越走越远。这比没有记忆还可怕。最常见的错误来源是“模型在抽取时过度推断”。比如用户问“这种菜如果少放点辣椒会不会好吃”抽取器可能会错误地抽成“用户偏好喜欢少辣”。用户明明只是在询问不是陈述偏好。这个问题我在前面提过解决办法是约束抽取器的保守程度只抽取有明确主谓结构的陈述句疑问句、假设句、否定句式需要额外的标记字段才允许入库存。第二类错误来源是用户改口。处理方式在2.3节已经讲过版本化加superseded标记。我这里补充一个前端体验层面的经验当Agent检测到用户信息与已有记忆冲突时最稳的处理不是默默更新而是在回复里说一句“好的我更新一下对您的了解”给用户一个确认和纠正的机会。这个细节极大地改善了用户的掌控感也让错误记忆在暴露后能被及时发现。4.4 记忆系统的安全与持久化别让记忆成为新的泄露面记忆系统存储了大量对话事实和用户偏好天然是敏感信息集中的地方。所以“Agent安全”在记忆模块这里有一个特殊含义记忆的读取权限必须严格遵守最小化原则。在实现上我给记忆库的访问加了两层限制。第一层是功能级限制召回接口只接受带user_id的查询任何不带租户维度的请求直接拒绝第二层是内容级限制有些记忆条目带有“高敏感”标签比如支付账号、身份证号在写入时会做一次敏感信息检测命中标签的内容只允许存脱敏版本。另外不要用大模型直接处理包含高敏感信息的对话原文应该在主循环的最外层先做脱敏再把脱敏结果送进抽取器。这个点的安全边界值得特别留意记忆系统在能力和便利性上为Agent加了很大的分但同时它也是攻击面最大的模块——如果能拿到记忆库的读取权限等于拿到了用户的行为画像。我自己的习惯是定期审核记忆库里到底存了什么把那些长期未命中且高敏感的项目手动清理掉。宁可少一些记忆的“大数据”也不要留下一堆随时可能引爆的“隐私地雷”。5. 一份直接可抄的最小实现骨架前面讲了那么多设计如果你只是想快速跑通一个带记忆的Agent不需要完整复刻所有机制一个人工可控的最小实现就够了。我把这套最小骨架整理在下面它融合了上面聊到的核心策略约200行Python即可覆盖主链路。import json import time import sqlite3 import numpy as np from dataclasses import dataclass dataclass class Memory: user_id: str content: str memory_type: str importance: int 3 created_at: float time.time() class TinyMemory: def __init__(self, db_pathmemory.db, embed_fnNone): self.conn sqlite3.connect(db_path) self.embed_fn embed_fn or (lambda text: np.zeros(8)) self._init_db() def _init_db(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT, content TEXT, type TEXT, importance INTEGER, embedding TEXT, created_at FLOAT, updated_at FLOAT ) ) def save(self, memory: Memory): vec self.embed_fn(memory.content) self.conn.execute( INSERT INTO memories (user_id, content, type, importance, embedding, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?), (memory.user_id, memory.content, memory.memory_type, memory.importance, json.dumps(vec.tolist()), memory.created_at, memory.created_at) ) self.conn.commit() def recall(self, user_id: str, query: str, top_k: int 5) - list[Memory]: query_vec np.array(self.embed_fn(query), dtypenp.float32) rows self.conn.execute( SELECT content, type, importance, created_at, embedding FROM memories WHERE user_id ? AND status active, (user_id,) ).fetchall() scored [] for row in rows: stored_vec np.array(json.loads(row[4]), dtypenp.float32) sim float(np.dot(stored_vec, query_vec) / (np.linalg.norm(stored_vec) * np.linalg.norm(query_vec) 1e-9)) recency 0.9 ** ((time.time() - row[3]) / (7 * 86400)) final 0.5 * sim 0.2 * recency 0.3 * (row[2] / 5.0) if final 0.35: scored.append((final, row[0], row[1], row[2])) scored.sort(reverseTrue) return [Memory(user_id, r[1], r[2], r[3]) for r in scored[:top_k]]这样一个TinyMemory可以直接挂到你的Agent主循环用户输入进来后调用recall拿历史记忆拼入system message。这一轮对话结束后从原始对话中抽取可能是事实的句子保守抽法只抽肯定陈述句调用save存入。等数据量涨到几千条再把内部的SQLite扫描换成pgvector接口保持不变上层逻辑零改动。这个最小骨架最核心的价值在于它把一个“记忆系统”降维成了一块可插拔芯片任何一个用LangChain、Spring AI或者手搓循环写的Agent都能在半小时内接上跑几天真实负载你就知道记忆系统里什么参数需要根据自己场景调哪里需要加复杂逻辑了。个人经验与后续扩展从最早被用户说“你像金鱼只有七秒记忆”到后来把记忆系统从一张字段表扩展成三层架构我最大的体会有三点先让短期缓冲可靠再碰长期记忆顺序反了你会被各种召回噪音淹死宁可少存不可错存错误记忆对Agent的破坏性远大于缺失记忆记忆必须可视化、可删除你要给用户足够的解释权和修正权而不是一个神秘的“黑盒档案柜”。这套系统后续我还在迭代的方向是让长期记忆能够沉淀成可复用的“技能经验”——比如Agent在一次任务中发现某个工具的调用方式很有效把这段经验存成技能描述下次类似任务直接复用。这个方向如果做扎实了Agent才真正开始具备“越用越顺手”的属性而不只是一个拥有完美记忆的复读机。