AI Agent记忆管理实战:用数据层解决大模型“失忆”难题

发布时间:2026/10/6 20:10:14
AI Agent记忆管理实战:用数据层解决大模型“失忆”难题 用过 GPT、Claude 这类大模型 Agent 的朋友大概率都经历过同一个崩溃瞬间昨天刚告诉 Agent“我在养一只叫阿橘的橘猫”今天随口问一句“阿橘该打疫苗了吧”它会一脸无辜地反问你“阿橘是谁”。这不是某个模型不够聪明而是 Agent 天生的结构性缺陷——它本质上是无状态的每次新对话都像失忆后重启。我最近在做的“Easy Data x AI”项目核心就是在解决这个“记不住”的问题用一层轻量、易用的数据服务把 Agent 的短期记忆和长期记忆统一管理起来让它跨会话、跨场景地真正记住你是谁、聊过什么、有什么偏好。这篇文章就把整套思路、架构设计、落地代码和踩过的坑完整复盘一遍。1. 先搞清楚Agent 为什么天生记不住你1.1 无状态的大模型就像每次失忆的天才大模型本身是个“每次失忆的天才”。它基于自回归机制工作给它一段输入它预测下一个 token然后继续补全。也就是说它所有的“聪明”都只在当前这一次请求内有效。你关掉页面再打开它对你一无所知这不仅是因为浏览器缓存没存对话更是因为模型在训练时就不带“用户档案”这个概念。更根本的约束是上下文窗口。即便今天很多模型的上下文已经支持 128k、200k tokens看起来很长但实际对话中系统提示、历史消息、工具调用结果、多轮演示样例都会挤占这个窗口。一旦信息超出窗口或者对话被压缩、被截断Agent 立刻“失忆”。所以想让 Agent 长期记住用户就不能只靠“把历史全部塞进去”必须有一个外部的、可持续读写的记忆系统。我在项目里反复跟同事强调一个类比大模型像极了一个高智商但患了顺行性遗忘的顾问你每次见它都要重新自我介绍。我们做的数据层就是它的“随身笔记本”。1.2 记忆不是一个大仓库要分层看很多人一听到“给 Agent 加记忆”第一反应是“找个数据库存聊天记录”。这么想没错但会把记忆系统做坏。真正要设计的是分层记忆工作记忆就是当前上下文窗口里的内容Agent 当下正在用的信息比如这轮对话的前面几轮。短期记忆同一个会话内、跨请求需要保留的状态比如用户这轮登录后选了哪些选项通常用 Redis 这类带过期时间的缓存实现。长期记忆跨会话、跨设备需要保留的用户画像、事实偏好、历史事件摘要这层必须落到持久化存储并配合向量检索做语义召回。很多现成的 Agent 框架默认只做了“工作记忆”最多加一个会话级缓存等于人只有大脑短期缓存没有长期笔记本。Easy Data 想补的恰恰是最后这一层同时把短期和长期之间的衔接处理好。1.3 项目的定位给 Agent 配一个“外脑”数据层Easy Data 是我这个项目的代号不是一个商业产品名字叫它“数据服务层”更准确。它的目标很简单把“数据读写”从 Agent 的对话逻辑里剥离出来让 Agent 不需要自己用变量硬记信息而是统一走记忆服务接口。这样做有非常实际的好处。第一可复用一个用户在多套 Agent 之间共享同一份画像第二可审计记忆里有什么、什么时候写入的、来自哪次对话全部有记录第三可测试哪个记忆导致 Agent 做出了什么决策可以单独回放复现。如果你正在搭自己的 Agent、做个人智能体或者在做多 Agent 协作的团队智能体这套思路基本可以直接照搬。项目本身不挑模型不管用的是国产大模型还是国外的官方 API记忆层都能作为中间件接进去。2. 整体架构Easy Data 如何变成 Agent 的外脑2.1 不是造一个大中台而是薄薄的一层服务项目刚开始时团队里有人建议直接上一套复杂的用户画像中台带实时特征计算、标签系统、AB 实验平台。我按住了这个冲动原因很简单给个人开发者或小团队用的 Agent首要是简单、能跑、可演进而不是一步到位建一个中台。Easy Data 最终做成的是一个很薄的记忆服务应用层Agent 本体负责对话生成、工具调用、任务编排。记忆服务层提供 REST API核心接口就四个——写记忆、读记忆、更新记忆、删除记忆。同时负责调用向量检索和 LLM 抽取。存储层PostgreSQL 存结构化记忆pgvector 存向量索引Redis 做短期记忆缓存和热点记忆加速。模型层Embedding 模型负责把记忆文本向量化LLM 负责从对话里抽取关键信息并压缩摘要。这层服务放在 Agent 和数据库之间看起来多了一跳但它把“记忆策略”集中收口了。以后想调整记忆评分公式、召回数量、过期策略都只改服务端不需要动 Agent 主体。2.2 核心数据模型一张表讲清楚记忆是什么记忆服务的核心表我设计得很克制。一开始只有一张memories表后来加了两张辅助表但主表结构一直没大变CREATE TABLE memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), agent_id TEXT NOT NULL, user_id TEXT NOT NULL, memory_type TEXT NOT NULL, -- fact / preference / event / summary content TEXT NOT NULL, importance INTEGER NOT NULL DEFAULT 5, embedding_id TEXT, metadata JSONB, source_chat_id TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), last_access_at TIMESTAMPTZ NOT NULL DEFAULT now(), superseded_by UUID ); CREATE INDEX idx_memories_user_agent ON memories (agent_id, user_id, memory_type); CREATE INDEX idx_memories_last_access ON memories (last_access_at DESC);字段设计都是有讲究的agent_id和user_id联合定位既支持单用户单 Agent也支持多 Agent 协作。同一个用户在不同 Agent 之间可以共享记忆也能互相隔离。memory_type区分事实、偏好、事件、摘要方便检索时区别对待。比如“用户说过自己过敏”是事实权重应高于一次随口说说的喜好。importance是人工或模型打出的重要度评分1 到 10后面检索重排会用到。这是决定记忆“值不值得被想起”的关键。superseded_by用来处理用户改口旧记忆不物理删除而是指向新记忆保留变更历史便于审计和回滚。2.3 记忆的生命周期写入、检索、更新、遗忘做记忆系统最容易被忽略的是“完整生命周期”。很多 Demo 代码只做两件事存进去、查出来结果运行几天后记忆库越来越脏效果反而变差。Easy Data 把记忆分成四个阶段管理写入对话结束后异步地从对话里抽取值得记住的信息清洗后结构化落库。检索收到新对话时把用户问题向量化从记忆库召回相关记忆再按重要度、时效性重排。更新用户说“我改主意了”原来那条记忆要被新内容替代但保留历史痕迹。遗忘长期没有被访问的低重要度记忆定期降权、归档甚至删除。遗忘不是缺点而是让记忆保持干净的必要手段。这套生命周期听起来不复杂但每一环的工程细节都值得单独展开。下面重点讲写入和检索这两个最核心的部分。3. 记忆写入与检索让 Agent“想起你”的技术细节3.1 写入流水线不是把聊天记录一股脑塞进去记忆不是聊天记录。用户说“今天好热想喝冰咖啡”这不是长期记忆而“我喝咖啡不喜欢加糖”就是值得长期保留的偏好。所以写入前必须做抽取和清洗。我的写入流水线是这样的对话会话结束后把整段对话发给抽取 LLM让它按 JSON 格式输出候选记忆。抽取提示词大概是请从以下对话中抽取值得长期记住的用户信息只输出 JSON 数组。 字段memory_type(fact/preference/event/summary)、content、importance(1-10)。 标准 - 只抽用户明确表达的事实和偏好不要抽临时性内容 - 重要度高于7的内容必须抽新手建议宁少勿多防止记忆污染 - 不确定的信息不要输出。 对话 {{conversation}}输出结果经过结构化校验后逐条写入memories表同时用 Embedding 模型计算向量存进 pgvector。这一步我建议异步执行不要阻塞对话返回。用户发完消息Agent 回复完再后台把记忆写入。实测用异步队列后响应延迟几乎没有变化。整个写入过程的代码骨架大概是这样的class MemoryWriter: def __init__(self, db, vector_db, extractor): self.db db self.vector_db vector_db self.extractor extractor async def write_from_conversation(self, agent_id, user_id, chat_id, messages): # 1. 用 LLM 抽取候选记忆 candidates await self.extractor.extract(messages) if not candidates: return [] saved [] for cand in candidates: memory_id str(uuid.uuid4()) # 2. 计算向量 vec await self.embed(memory_id, cand[content]) # 3. 入库 入向量库 self.db.save_memory(memory_id, agent_id, user_id, cand) self.vector_db.upsert(memory_id, vec, {agent_id: agent_id, user_id: user_id}) saved.append(memory_id) return saved3.2 检索重排让最该被想起的记忆先出现检索是记忆系统的灵魂。最简单的做法是把用户问题向量化然后从向量库取 top 5 条相似记忆拼进提示词里。但实测下来这种“纯语义相似”队列效果不好因为最相似的记忆不一定是当下最重要的。我最后用的重排公式是三因素加权score 0.6 * 语义相似度 0.3 * (importance / 10) 0.1 * 时间新鲜度其中语义相似度来自向量检索的余弦相似度范围 0 到 1重要性是写入时模型打的 1 到 10 分时间新鲜度按最后访问时间衰减一天内为 1之后指数衰减。举个实际计算例子用户问“今天推荐喝什么”检索到两条候选记忆记忆 A“用户喜欢冷萃咖啡不加糖”相似度 0.82重要性 8昨天刚访问过时间新鲜度约 0.9score 0.492 0.24 0.09 0.822。记忆 B“用户上周说想尝试冰滴咖啡”相似度 0.73重要性 5上周访问时间新鲜度约 0.5score 0.438 0.15 0.05 0.638。所以推荐时优先体现冷萃偏好同时把冰滴作为尝试选项。这个细节正是 Agent“懂你”和“只是匹配关键词”的分水岭。检索服务的核心实现class MemoryRetriever: def __init__(self, db, vector_db, embedder): self.db db self.vector_db vector_db self.embedder embedder async def retrieve(self, agent_id, user_id, query, top_k5): query_vec await self.embedder.embed(query) # 向量库先粗召回 top_k * 3留足重排空间 candidates self.vector_db.query(query_vec, top_k * 3, filters{agent_id: agent_id, user_id: user_id}) scored [] for cand in candidates: freshness math.exp(-(now - cand.last_access_at).days / 7.0) score 0.6 * cand.similarity 0.3 * (cand.importance / 10.0) 0.1 * freshness scored.append((score, cand)) scored.sort(reverseTrue) return [cand for _, cand in scored[:top_k]]检索完还需要把记忆拼接成合适的提示片段比如关于该用户的长期记忆 - [偏好] 喝咖啡不加糖偏爱冷萃 - [事实] 养了一只橘猫叫阿橘 - [事件] 上周买了便携咖啡机然后塞进系统提示或者上下文最前面。注意控制数量我一般限制在 5 到 10 条防止把上下文窗口撑爆。3.3 记忆更新用户改口时怎么办用户信息是会变的。上周还说“我不喝茶”这周就在研究茶具。如果旧记忆不处理Agent 会不断给出矛盾的回答。我的方案是“软更新”新记忆写入时如果检索到同一主题的旧记忆就在旧记录上打superseded_by 新记忆ID并在元数据里记录“被替代原因用户改口”。这样有两个好处短期看检索时优先取未过期的记录Agent 不会自相矛盾长期看保留了用户偏好变化轨迹可以作为个性化服务的数据基础。更新的判断不能全靠规则我同样用 LLM 辅助识别“这句话是不是在推翻之前说过的话”。识别命中后再自动做新老记忆的关联和替换整个过程也放进异步队列避免影响对话响应速度。4. 从零搭建一个带记忆的 Agent完整落地过程4.1 技术栈选型先跑通再优化项目初始我没有一上来就用很重的技术栈。技术选型遵循“最小组件完成闭环”的原则组件选型理由服务框架Python FastAPI生态好异步支持好写 AI 类工具最省心结构化存储先 SQLite后迁 PostgreSQL本地单机先用 SQLite需要并发和 JSONB 再迁 PG向量存储pgvector搭配 PG省一个向量库组件SQL 里直接查部署简单缓存Redis短期记忆和热点记忆加速TTL 天然合适Embeddingbge-m3 / m3 类中文模型中文效果稳定对中英混合支持好Agent 编排框架不重要记忆服务通过 HTTP 接入LangChain、Dify、自制循环都行这里有个很重要的体会Agent 框架解决的是“API 编排”它默认是不带跨会话记忆的。无论是用 LangChain 还是自研循环记忆层都必须自己接。Easy Data 从一开始就把自己定位成“框架无关的独立服务”。4.2 核心代码骨架MemoryManager 封装全部记忆操作实际项目里我封装了一个MemoryManager把写入、检索、更新对外暴露成几个简单方法。Agent 主循环只需要调用它不需要懂内部存储细节。class MemoryManager: def __init__(self, db_uri, redis_url, embedder, extractor): self.db MemoryTable(db_uri) self.cache MemoryCache(redis_url) self.embedder embedder self.extractor extractor self.writer MemoryWriter(self.db, self.vector_db, self.extractor) self.retriever MemoryRetriever(self.db, self.vector_db, self.embedder) async def remember(self, agent_id, user_id, chat_id, messages): await self.writer.write_from_conversation(agent_id, user_id, chat_id, messages) async def recall(self, agent_id, user_id, query, top_k5): return await self.retriever.retrieve(agent_id, user_id, query, top_k) async def update_memory(self, memory_id, new_content, new_importanceNone): # 处理用户改口、信息更新 ...Agent 主循环的接入就更简单了async def agent_loop(user_id, query, chat_history): # 1. 检索记忆 memories await memory_manager.recall(main_agent, user_id, query) # 2. 组装系统提示 system_prompt base_system_prompt \n\n format_memories(memories) # 3. 正常走大模型 answer await llm.chat(system_prompt, chat_history [{role: user, content: query}]) # 4. 响应后写入本次会话的关键记忆 await memory_manager.remember(main_agent, user_id, chat_id, chat_history [{role: user, content: query}]) return answer代码就这么薄。核心逻辑全部收敛在MemoryManager里外层 Agent 像挂了一个 U 盘插上就能用。4.3 效果演示第二次对话终于“认出”了用户本地测试时我用了一个特别直观的场景。第一次对话用户说“我喝咖啡不加糖只喝冷萃的最近还在减肥所以不要推荐带奶油的东西。”Agent 回复完后台自动写入两条记忆[偏好] 用户喝咖啡不加糖偏好冷萃[偏好] 用户正在减肥不推荐带奶油的饮品第二天模拟同一位用户再次打开应用直接问“帮我挑杯咖啡吧。”没有记忆的 Agent 会四平八稳地推荐“拿铁、摩卡、焦糖玛奇朵”。接上 Easy Data 之后系统提示里已经有了那两条偏好记忆Agent 的回答立刻变成“按你平时喝冷萃的习惯推荐淡美式或者冷萃黑咖啡少糖少负担下午来一杯提神也不影响减肥”。就是这一条差异让用户明显感觉到“它真的记得我”。这个 Demo 后来成为项目路演的主要素材。4.4 并发和用户隔离多用户 Agent 怎么扛住压力很多人在网上问“ai agent 怎么扛并发”其实在记忆系统场景下并发压力主要不在大模型调用上而在数据读写上。大模型 API 本身有一定延迟并发瓶颈通常先到的是存储和缓存。我做了几件关键事情第一用户级隔离。所有记忆读写都强制带agent_id user_id过滤向量库检索也带同样的过滤条件绝不让一个用户查到另一个用户的记忆。这个必须在服务层强制校验不能靠 Agent 自觉。第二Redis 缓存热点记忆。高频用户的记忆在第一次查询后写入 Redis设置合理的 TTL。比如活跃用户 24 小时缓存这样同一个用户在短时间内多次对话不需要每次都走向量检索直接从缓存读。实测单机 Redis 扛几十万 QPS 的读没问题对于个人产品和大多数中小应用完全够用。第三异步批量写入。记忆写入不阻塞对话主流程。会话结束后消息丢进队列后台批量处理抽取、向量化、落库。高峰时攒批写入降低数据库连接压力。第四连接池和超时控制。PostgreSQL 侧用连接池所有查询加超时比如向量检索超过 300ms 就降级为纯 SQL 关键词检索保证核心对话不被记忆服务拖死。顺便说一句网上经常争论“harness 和 agent 区别”我从项目里的体会是Agent 本体只是对话决策循环而真正塑造用户体验的“套件”即 harness包括了记忆、权限、工具、状态管理。记忆层恰恰是 harness 最关键的部分。你搭的 Agent 像不像样很多时候不看模型看记忆和工具链。5. 常见问题与排查技巧实录5.1 记忆污染什么都记等于什么都没记记忆系统上线第一周我发现一个典型问题Agent 把太多临时内容当成长期记忆存了下来。比如“我明天要去机场”“今天下雨带了伞”这类一次性事件把记忆库塞得满满的导致真正重要的偏好被淹没。排查后发现是抽取提示词写得不够严。修复方式是三条在抽取提示词里加“只抽用户明确表达的事实和偏好不抽临时性内容”增加重要度门槛低于 5 分的记忆默认不落长期库只留在会话缓存里新写入记忆执行“同用户同类目归并”同一偏好反复出现时只更新旧记录的重要度不新增。调整后记忆库的密度和命中率都有明显提升。这里也分享一个测试技巧准备几组“干扰对话”和“有效偏好”每次改完抽取逻辑都用同一批测试集回归防止修好一个 bug 引出新的问题。5.2 token 预算失控记忆把上下文撑爆了还有一次测试同事反馈“Agent 越聊越贵响应越来越慢”。查了一下原来是把该用户半年内的记忆全部塞进了上下文热门用户一个人的记忆都快超过 10 万 tokens。我的处理方案是三层控制数量限制单次召回最多 10 条记忆动态裁剪先按分数取 top 30再按剩余上下文预算从高到低选摘要降级如果用户记忆实在太多额外生成一份“长期摘要”把年度级信息浓缩成几百字每次只注入摘要不注入全部明细。token 预算问题不是记忆系统独有但 Agent 加了记忆后会更明显。建议把记忆检索的预算单独规定比如占总上下文长度的 20%不要挤压当前对话的空间。5.3 记忆与安全敏感信息怎么处理这里要重点提醒记忆系统天然会聚集用户的大量隐私信息必须把安全当核心功能设计而不是后期补丁。我做了四层基本防护传输加密所有接口走 HTTPS存储加密数据库开启加密保护权限隔离不同用户之间数据完全隔离服务端做二次校验敏感内容过滤写入前用正则和“敏感信息检测模型”双重过滤身份证、手机号一类信息一律不落长期记忆库。另外还有一个很容易踩的坑向量库里的向量虽然看起来是一堆浮点数组但如果用户记忆文本里有敏感词向量本身也可能被逆向分析出近似内容。所以“敏感信息先过滤再向量化”是必须的不要想着向量化后就不算存储敏感信息了。5.4 Embedding 模型怎么选中文场景的实测对比我在项目里试过好几款 Embedding 模型刚开始图省事用了 OpenAI 的text-embedding-3-small1536 维效果其实不差但有两个问题一是多轮请求有跨地区网络延迟二是维度高导致向量存储占用大。后来换成了开源的中文友好模型比如bge-m31024 维中英混合场景表现更稳定还能本地部署一次向量化费用都不用花。如果你服务的是纯中文用户建议优先考虑对中文支持更好的开源模型如果是中英混合场景选择对多语言友好的 1024 维模型。在本地跑过 benchmark余弦相似度在 0.8 以上的记忆命中率基本可以覆盖绝大多数场景。5.5 冷启动与记忆幻觉没有历史时怎么装懂最后一个高频问题新用户没有任何记忆Agent 又强行表现得“很懂你”结果开始编造用户偏好这就是“记忆幻觉”。典型表现是“根据以往记录您应该喜欢……”这个问题的根源是 Agent 把“没有记忆”误当成“有记忆但不确定”。我用了两个手段规避显式声明冷启动用户没有记忆时系统提示里明确写“没有该用户的长期记忆请基于当前对话自然询问偏好”禁止无中生有在系统提示里加入“不要编造用户历史记录不确定的偏好直接提问”。记忆幻觉对用户体验杀伤力很大宁可让 Agent 承认“我还不了解你”也不要硬装老熟人。这是我在多次用户访谈中得到的明确结论。写在最后一些小经验和后续扩展Easy Data 这个项目做下来我最大的体会是记忆系统不是一锤子买卖它需要持续迭代。第一版我只做了一张表加一个向量检索跑通后才发现每天都能碰到新的边界情况比如用户改口、临时话题混入、记忆老化、token 预算波动。与其一开始追求“完美架构”不如先用一个能跑的闭环验证价值再逐步把生命周期各个阶段补完。我至今还保留着那个以 SQLite 为核心的第一版代码偶尔翻出来看反而能提醒自己不把简单问题复杂化。还有一个很实用的小技巧给记忆系统做“定期体检”。每周自动统计每个用户的记忆总数、命中次数、被丢弃的旧记忆数量用一个小脚本生成报告。这个脚本帮我发现过记忆库只涨不清的问题也让重要度评分体系有了迭代依据。最后说说扩展方向。有朋友把这套记忆服务接进了 Obsidian让 Agent 读用户的知识笔记库再结合记忆系统给用户做选题推荐和知识回顾体验非常香。另一个方向是多 Agent 协作同一个用户身份在不同 Agent 间共享记忆比如订酒店 Agent 记住的常住偏好可以直接带给行程规划 Agent这就是典型的“多 AI 协作”场景。甚至有人提到把类似的记忆层接到机器人操作系统里让实体机器人也记住常去路线和操作习惯。记忆这个东西围绕它的坑还有很多但也正因为这样才值得持续折腾。