1. 项目概述1.1 AI记忆的核心需求解析做AI应用这一年多我最大的感触是单纯接个大模型API做问答根本算不上智能。你想想看用户上午问了你我喜欢喝美式还是拿铁下午再问帮我推荐一杯咖啡如果AI完全忘了上午的对话这体验是不是跟失忆症患者聊天差不多这就是ai-memory这个方向要解决的事——给AI装上持久记忆让它在不同会话之间记住用户、记住上下文、记住偏好。我最初接触AI记忆是因为做一个用户运营助手。当时遇到的情况很典型用户每开一个新对话系统就要重新解释一遍自己的需求。后来我翻了各种memory方案从LangChain内置的ConversationBufferMemory到Mem0、Zep这类专门做记忆的中间件再到自己动手设计记忆里的数据库结构前前后后折腾了不少时间。这篇博文就是把这些踩坑经验整理出来给正在做AI Agent、客服机器人、个性化推荐系统的朋友做个参考。1.2 这个项目能解决什么先说清楚AI记忆到底解决什么问题不然很容易做成简单的日志存储。我梳理下来大概三类场景跨会话记忆用户上次聊了什么下次接着聊时能回忆起来而不用重新介绍自己。长期偏好记忆用户的喜好、习惯、说话风格可以跨项目、跨场景长期沉淀。动态上下文管理当前对话中的临时信息比如刚才提到的文件路径和长期记忆分开管理互不干扰。很多人一上来就想着把对话历史全存下来这是个误区。存下来和想起来是两码事——你需要的是按需召回而不是把整段历史都塞给模型。AI记忆的核心是存储检索遗忘的组合拳不是简单的数据库读写。所以在动手做之前我先花时间把需求拆成了三层记忆获取、记忆存储、记忆召回。后面所有的工作都围绕这三层展开。提示如果你只是为了给聊天机器人加个记住用户名字的功能用sessioncookie就够了不用上向量数据库。AI记忆项目的复杂度取决于你要记住什么、存多久、怎么用。1.3 适合谁来参考如果你属于下面这几类人这篇文章的实操部分可以直接抄做AI客服/助手的开发者需要让机器人记住用户的历史诉求。做Agent(智能体)的工程师需要让多步骤任务共享状态和上下文。做个性化推荐或内容生成的产品经理需要理解记忆在用户画像之外能补充什么。纯粹对LLM应用感兴趣的学习者想了解RAG和AI Memory之间的关系。这篇博文的代码部分用的是Python向量存储用FAISS和ChromaDB举例记忆框架会对比Mem0和自研方案的取舍。虽然不是零基础教程但我会把每个关键技术点的原理讲清楚边实操边解释为什么这么做。即使你还没碰过向量数据库跟着走一遍也能跑通。2. 记忆系统的架构设计与方案选型2.1 为什么不能直接存对话历史第一个要破除的误区是记住对话不等于存储对话。我见过不少团队直接拿Redis或者MySQL把原始对话记录存起来等下次对话时再把历史记录全部翻出来拼进prompt。这在对话轮次少的时候能用但一旦聊了几十轮token就爆炸了还容易把无关紧要的内容带进上下文。更严重的是早期对话里的关键信息会被海量闲聊稀释模型根本抓不住重点。正确的思路是像人脑一样对记忆做抽象和压缩。我们不需要记住每句话只需要记住发生了什么事用户表达了什么偏好当前任务的进度如何。2.2 记忆的三层模型我最终把记忆系统分成了三层每层的存储和检索逻辑完全不一样层级作用存储方式典型场景工作记忆当前会话的临时上下文内存/短期存储多轮对话中的中间状态情景记忆具体事件的完整记录向量数据库用户上次投诉的具体内容语义记忆提炼后的用户画像/偏好结构化数据库用户偏好、习惯、需求这种三层的设计参考了认知科学里对人类记忆的分类方式。工作记忆放最活跃的数据比如当前正在生成的文档标题。情景记忆放向量因为要语义检索用户之前提过的某某问题。语义记忆放结构化存储因为要支持精确查询比如用户所在的城市是什么。2.3 自研方案与Mem0/Zep的选型对比市面上确实存在现成的AI记忆解决方案。我实际调研和试用过Mem0、Zep以及LangChain生态里的记忆模块。最终我的选择是自研为主参考Mem0的设计思路。这么做不是因为开源方案不好而是因为它们不够可定制。拿Mem0来说它的设计很优雅核心流程是对话记录 → 提取关键记忆 → 向量化 → 存储 → 检索返回相关片段。但在实际项目中我需要把记忆和业务数据库打通。比如用户在某次对话里提到我在上海工作我希望这条记忆能同步到用户表里的disable字段而不是停留在记忆库里。Mem0提供了接口但扩展成本高不如自己在核心逻辑上复刻一套轻量实现。Zep的优势是有时间衰减和自动遗忘机制但它偏向Agent场景对普通客服机器人来说有点重。LangChain内置的Memory类说实话比较适合demo生产环境基本撑不住复杂需求。所以我的方案是底层存储自己设计核心记忆逻辑自己写但借鉴了Mem0对记忆提取-存储-召回三阶段的划分方式。注意选型的时候先列一下你的记忆要对接哪些业务系统。如果只是给LLM对话用Mem0这类方案非常省事。如果还要跟CRM、工单系统联动自研一个精简版更灵活。2.4 记忆的原子化设计这是我这套方案里最关键的设计决定。我不用对话记录作为记忆的最小单位而是用事实。每次用户对话后我会调用一个提取器从对话中抽取结构化的记忆原子。比如{ entity: user:12345, relation: prefers_language, object: Python, source_conversation: conv_6789, timestamp: 1699999999, confidence: 0.95 }这种三元组格式参考了知识图谱的实体-关系-实体结构。好处是检索的时候可以直接精确匹配也可以向量语义匹配。比如用户明确说过我喜欢Python这直接进结构化匹配如果用户说我平时主要写脚本处理数据语义向量才能归纳出和Python相关。原子化设计的另一个好处是支持记忆更新。用户上次说喜欢Python这次说其实最近都在写Go我只需要更新这一条三元组而不是把整个对话记忆重写一遍。3. 记忆存储与检索的基础设施3.1 为什么选择向量数据库既然记忆按语义被召回那向量数据库是绕不开的存储底座。以召回与当前问题相似的记忆为例。用户当前问了一句推荐几个适合团队协作的工具我要从历史记忆里找到和这句话语义相近的内容纯靠关键词匹配会丢很多信息比如用户以前说过我们在用Notion管文档最近想换一换。关键词协作工具都匹配不到但语义是相关的。向量化的做法是把这句话变成一个几百维的向量同时在存储记忆时也把记忆内容向量化。查询时计算两个向量的距离比如余弦相似度距离越近越相关。这就实现了语义级检索。我用的是ChromaDB做原型测试后续切换到FAISS做生产环境的索引管理。原因后面细说。3.2 Embedding模型的选择向量化这一步的选型决定了记忆召回的质量。我踩过的坑包括早期用OpenAI的text-embedding-ada-002效果确实好但要付费而且数据要发到第三方API。后来改用开源的BAAI/bge-small-zh-v1.5在中文场景下效果也很稳本地部署延迟低batch推理方便。我当时的对比测试数据大致如下测试集300条中文记忆片段Embedding模型召回Top-5准确率单条向量化延迟部署方式OpenAI text-embedding-ada-00289.2%35msAPIBGE-small-zh-v1.585.6%8ms本地text2vec-large-chinese86.1%12ms本地可以看到本地模型和OpenAI的差距并不大但延迟和成本优势明显。如果你对中文语义要求极高可以再试试BGE-large就是推理成本会涨一些。实操建议embedding模型选定后尽量少更换。因为更换模型意味着所有历史记忆的向量都要重新计算一遍。如果记忆量比较大这个重算成本很高。我第一次切换的时候10万条记忆向量化跑了好几个小时。3.3 向量存储和混合检索设计纯向量储存在某些精细场景下不够用。举个例子用户问我之前反馈过登录报错的问题解决了吗这种查询语义相似度能召回登录报错相关的记忆但怎么确定是哪一次反馈就需要结合时间过滤、会话ID过滤等结构化条件。所以我的存储表结构是这样设计的CREATE TABLE ai_memory ( id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, memory_type ENUM(fact, episode, preference) NOT NULL, content TEXT NOT NULL, embedding VECTOR(768), -- 向量索引配合pgvector使用 metadata JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, access_count INT DEFAULT 0, last_access_at TIMESTAMP, INDEX idx_user_id (user_id), INDEX idx_memory_type (memory_type), INDEX idx_created_at (created_at) );构建索引的时候我对embedding字段做了HNSW索引。写SQL逻辑时我把向量相似度召回和结构化条件过滤组合成一个查询SELECT id, content, metadata, 1 - (embedding :query_embedding) AS similarity FROM ai_memory WHERE user_id :user_id AND created_at :start_time ORDER BY similarity DESC LIMIT :top_k;先用结构化条件锁粗范围再按向量相似度精排效率和准确性都更好。如果不这样设计一个全局向量召回会经常捞出不相关的东西毕竟跨用户、跨时间长度的记忆太多了。这里重点说一下实际的混合检索我用的是先过滤后向量的流程而不是先向量后过滤。如果先说向量再过滤时间范围可能会把当前用户最相关的记忆排在前面但时间条件一旦应用就会丢失结果。先过滤就没这个问题还能减少向量检索的规模。3.4 记忆衰减与遗忘策略存储基础设施搭好了之后你会发现一个新问题记忆不会自己瘦身。用户聊了半年记忆库可能积累了几万条许多都是昨天中午吃了牛肉面这类价值极低的内容。所以必须设计记忆的衰减和遗忘机制。我参考了Zep的时间衰减思路但做了简化每个记忆有一个初始权重比如1.0每次被成功召回时权重增加长期不被召回则权重随时间衰减。def decay_score(score, last_access_at, half_life_days30): elapsed_days (datetime.now() - last_access_at).days decay_factor 0.5 ** (elapsed_days / half_life_days) return max(0.1, score * decay_factor)每30天权重减半。如果权重低于0.2这条记忆进入待清理状态由后台Job批量删除或归档到冷存储。这个策略的核心逻辑是越常被想起的记忆越重要越少被触碰的记忆越可能没用。听上去是常识但实际执行下来记忆库的体量能控制住召回质量也明显提升。4. 实操从零搭一个带AI记忆的对话系统4.1 技术栈与目录结构我用的技术栈是Python FastAPI ChromaDB SQLite OpenAI兼容API可用本地模型替代。目录结构如下ai-memory-demo/ ├── main.py # FastAPI 服务入口 ├── memory/ │ ├── __init__.py │ ├── extractor.py # 记忆提取器 │ ├── store.py # 记忆存储封装 │ ├── retriever.py # 记忆召回器 │ └── models.py # Pydantic 模型 ├── config.py # 配置文件 └── requirements.txt4.2 记忆提取器实现记忆提取器的作用是把用户和AI的一次对话提炼成结构化的记忆。不能用原始对话存进去否则长期记忆就变成了日志堆积。我的做法是给LLM设定一个提取提示词EXTRACT_PROMPT 你是一个记忆提取器。请从以下对话中提取需要长期记住的信息。 只提取以下类型的事实用户偏好、用户个人信息、用户目标、项目进展、明确表态。 忽略寒暄、临时状态、无关闲聊。 输出JSON数组每个元素包含 { entity: 主体标识, relation: 关系类型, object: 对象内容, memory_type: fact|episode|preference, importance: high|medium|low } 对话内容 {conversation} 比如用户说我一般下午才比较有空提取到的记忆是{ entity: user:12345, relation: preferred_work_time, object: 下午, memory_type: preference, importance: medium }对于importance字段系统决策时会更关注high的记忆低重要度记忆直接跳过来减少token占用。4.3 记忆存储封装存储层我封装了一套统一的接口方便底层在ChromaDB和FAISS之间切换。class MemoryStore: def __init__(self, collection_nameai_memory, persist_dir./db): self.client chromadb.PersistentClient(pathpersist_dir) self.collection self.client.get_or_create_collection( namecollection_name, metadata{hnsw:space: cosine} ) def add(self, memory: dict, embedding: list[float]): self.collection.add( ids[memory[id]], embeddings[embedding], documents[memory[content]], metadatas[{ user_id: memory[user_id], memory_type: memory[memory_type], timestamp: memory[timestamp], importance: memory[importance] }] ) def delete(self, memory_id: str): self.collection.delete(ids[memory_id])ChromaDB的好处是开箱即用支持持久化到本地磁盘对单机项目来说非常够用。到生产环境我切换到了FAISS PostgreSQL的一体化方案向量索引用FAISS元数据像用户ID、时间戳、访问次数放到PostgreSQL。因为FAISS的过滤语法相对原始我是先在PostgreSQL里把满足条件的所有候选ID捞出来再在FAISS里做向量检索最后按时间排序。4.4 记忆召回与上下文注入召回逻辑是整个系统性能和效果的关键很多项目跑偏都是在这块。先看一下代码def retrieve_relevant_memories(user_id: str, query: str, top_k: int 5): # 1. 生成查询向量 query_embedding embed(query) # 2. 向量检索 results collection.query( query_embeddings[query_embedding], n_resultstop_k, where{user_id: user_id} ) # 3. 过滤低相似度记忆 filtered [ (doc, meta, dist) for doc, meta, dist in zip(results[documents][0], results[metadatas][0], results[distances][0]) if dist 0.55 # 这是关键参数后面细说 ] # 4. 按时间新鲜度和重要性重排 filtered.sort(keylambda x: (x[1][importance], x[1][timestamp]), reverseTrue) return filtered[:top_k]重排的目的事前过滤掉语义相关但价值不高的记忆。比如发现用户当前在问Python历史记忆里确实有一段和Python相关但如果那是一年前问的Python安装教程重要性低、时效性差就不该排在前面。重排让答案更贴近当下的需求。召回的记忆拼接入prompt时我用了一个固定格式【相关历史记忆】 - 用户偏好下午工作语言偏好Python - 3天前反馈过登录页报错500状态未解决这段记忆信息插到system prompt的末尾作为context的一部分。模型看到相关历史后才能做出个性化的回应。4.5 相似度阈值的调优心法我上面代码里的dist 0.55这个阈值是我调了很久才确定的。这可能是整个项目里最值得记录的经验。ChromaDB默认返回的是L2距离因为我设置的是cosine空间其实返回的是余弦距离。余弦距离的范围是0到2越小越好。0.55这个阈值在BGE-small模型下能保证召回的大多数记忆都在语义上有明确相关性。怎么调呢我准备了一个小测试集里面放了20个正确应该召回的查询和20个不应该召回的干扰查询。然后枚举不同的threshold计算准召率阈值精确率召回率0.3595%41%0.4588%67%0.5574%83%0.6553%92%最终我选0.55求一个平衡。如果你想调这个参数别拍脑袋要像我一样逐档测试看你的场景更看重别带无关信息还是别漏掉关键信息。注意不同embedding模型的向量空间分布不同换模型后阈值必须重新调。之前从ada-002切到BGE后0.55这个值直接让召回率掉了20%因为BGE的向量余弦距离整体偏大。如果你用不同模型必须先跑一轮校准。4.6 完整对话流程串起来每收到一条用户消息完整的记忆链路是这样的从请求里读出user_id和message。先调retrieve_relevant_memories召回相关记忆拼入system prompt。调LLM生成回复把回复和用户消息一起交给extractor。extractor提取新事实写入MemoryStore。返回回复给用户同时更新相关记忆的access_count和last_access_at。这里有一个顺序问题第一步和第四步容易搞混。必须先召回再生成生成后再存新记忆。存新记忆的时候要特别注意去重如果用户这轮说我才喜欢Python跟昨天的我喜欢Python特征相似度很高就该做合并更新而不是再插一条重复数据。我在store层加了一个去重逻辑def add_memory_deduplicated(memory, threshold0.92): existing collection.query( query_embeddings[embed(memory[content])], n_results1, where{user_id: memory[user_id]} ) if existing[distances][0][0] threshold: # 合并更新旧的记忆而不是插入新的 update_memory(existing[ids][0], memory) else: add(memory)这样做的原因是避免记忆碎片化。如果同一条偏好被存入多条相似记忆召回时会同时带出5条占用的token是5倍的信息量却没有增加。5. 常见问题与排查技巧实录5.1 记忆检索不准换阈值、换检索方式、换模型这是被问得最多的问题我召回来的记忆明显不相关怎么回事我的排查顺序是第一看相似度得分到底是多少。如果召回结果的距离集中在0.6-0.7说明阈值太松了类似上面的调优。如果得分集中在0.3-0.4但内容依然不相关那就不是阈值问题而是embedding模型的问题。第二看查询语句本身。用户问的很短很口语化比如上次那个事这种query做向量化效果极差。我目前的解决办法是加一个查询改写环节用LLM把上次那个事改写为用户上次提到的未解决问题然后再embedding和召回。改造后召回准确率提升了近15%。第三如果前两步都没问题就要考虑做rerank。可以引入一个粗排精排的二级检索架构向量检索做粗排取Top-50再交给一个rerank模型。我自己的业务对延迟没那么敏感就在召回后接了一个cross-encoder开源模型。质量提升很显著代价是每查询增加了几十毫秒。5.2 记忆冲突新旧记忆打架怎么办这种情况经常出现。用户上个月说我喜欢喝美式这周又说最近想试试拿铁。两条记忆语义上都是用户对咖啡的偏好但内容冲突了。如果简单地把两条都塞给LLM模型会迷惑到底哪个是对的。我的解决办法是在记忆提取阶段就引入冲突检测新提取出来的对象先和已有的同entity、同relation的记忆做对照。如果发现矛盾判断依据是recency最新优先和importance高重要隐私优先。如果recency相同就保留importance更高的。实际执行的时候我会把旧记忆标记为superseded_by_new_id而不是直接删除。这样可以追溯到用户的偏好变迁历史对于做推荐系统的人来说这本身就是有价值的数据。5.3 记忆膨胀存储和token成本失控前面说过通过遗忘策略控制规模。实际运营下来还有两个很有效的技巧第一降采样。对话日志里大量重复的日常性记忆如果只保留7天之后只保留摘要。我给记忆系统加了一个汇总Job每天对前一天的记忆做聚类然后生成一个日摘要。日摘要之间存在相似性再聚合成周摘要。这样长期来看高层摘要负责宏观偏好短期记忆负责细节上下文不会无限膨胀。第二按用户分级。我一直以为所有用户都应该存下全部记忆后来运营数据打脸90%的普通用户记忆保留30天就够了只有深度用户或付费用户开启长期记忆权限。分级存储不仅成本降低而且普通用户的召回准确率还提升了因为少了大量低价值旧记忆的干扰。5.4 隐私与数据安全容易被忽略的关键环节做记忆系统的都明白记忆本身就是隐私数据你存了用户的偏好、对话、甚至是个人信息。这是逃不掉的责任。我在合规上有几个做法供你参考敏感信息脱敏在提取记忆存入存储之前用一个脱敏模型检测手机号、身份证号等PII处理后再存储。用户删除权提供了清除记忆API按user_id一键删除全部数据。记忆导出的权限隔离严格按user_id隔离记忆表。生产环境运维时禁止执行全库向量查询这类操作。日志审计所有记忆的写入、读取、更新都记录审计日志保留操作者防止内部滥用。这些设计不是锦上添花而是上线前必须完成的。我见过有同行把AI记忆做成产品因为默认开启全局记忆、没有用户控制面板被用户投诉隐私问题最后只能下线整改。5.5 性能排查从召回到生成的完整耗时分析如果用户反馈机器人变慢了不要猜先看数据。我的一个排障案例可以说明问题某天客服机器人的响应延迟从800ms飙到2.5s直觉是LLM太慢一查慢在了记忆召回环节。再排查是某个用户的历史记忆达到3万条向量检索一次要扫全量慢得很。后来优化了两点第一按用户维度分collection。每个用户一个collection查询时天然缩小范围。但collection数量太多会有管理开销我改成了按用户分桶比如按user_id hash到16个collection。第二缓存高频召回结果。用户反复询问同一个主题时短时间内召回结果高度重合加了LRU缓存命中时零延迟。实测下缓存命中率在30%左右平均响应时间降了12%。6. 踩坑总结与经验分享6.1 我应该早点想清楚的三个问题第一不要把RAG和Memory混为一谈。RAG是从外部知识库检索文档Memory是从历史交互中找回上下文。两者虽然都依赖向量检索但数据来源、更新频率、存储结构完全不同。很多团队把记忆做成了一般的RAG结果就是对话里带上一堆百科内容而不是用户偏好的信息。第二记忆的质量比数量重要。宁可每天只存下三条高置信度的偏好也不要盲目存储一百条废话。质量低的信息会对召回造成干扰挤占上下文窗口。我在迭代中一直在提高记忆提取器的过滤标准账户里存下的记忆条数少了2/3但召回的用户满意度反而更高了。第三记忆系统要重度依赖遗忘。人脑的遗忘不是缺陷是保护机制。AI记忆如果没有遗忘旧信息会污染新判断存储成本会拖垮性能。我曾经因为忘了设计记忆清理半年后上线的大模型开始胡言乱语——用户已经改了偏好模型还在按三个月前的旧偏好回复。6.2 这个项目的后续扩展方向做完基础版的ai-memory之后我觉得这三个方向是最有价值的扩展多模态记忆存储现在文本记忆只占一个维度图片和音视频里的信息也应该能被提取出来这对智能助手场景很实用。记忆跨设备同步用户手机上聊的口味到PC端打开App也要能想起来这就需要记忆库的云端同步和统一身份系统。Agent记忆协作如果任务是靠多Agent协作完成的记忆就不能只挂在用户维度还要挂在任务维度、Agent本身维度。任务记忆的设计比用户记忆复杂得多因为中间状态很多状态转移的路径也不固定。6.3 最后一句掏心话我最后想说的是AI记忆工程真正难的地方不是向量数据库怎么连、embedding用哪个模型而是你怎么定义什么才值得记住。这需要在产品逻辑、用户体验和工程成本之间反复权衡。我第一次做出ai-memory的demo时感觉特别简单两三百行代码就实现了基本对话记忆。可真正把它推到线上每天处理几万次对话、召回几千条记忆之后才发现记忆系统是需要持续运营和调优的。它会伴随你的业务一起涨大也会在某个临界点逼着你去重构。现在再做类似项目的话我会一上来就把记忆的提取、存储、召回、遗忘四个阶段都想明白不急着写代码。把数据模型和管理策略理顺了再动手。这套方法论比任何一份开源代码都值钱。 AI记忆系统实战：架构设计、向量检索与记忆召回
1. 项目概述1.1 AI记忆的核心需求解析做AI应用这一年多我最大的感触是单纯接个大模型API做问答根本算不上智能。你想想看用户上午问了你我喜欢喝美式还是拿铁下午再问帮我推荐一杯咖啡如果AI完全忘了上午的对话这体验是不是跟失忆症患者聊天差不多这就是ai-memory这个方向要解决的事——给AI装上持久记忆让它在不同会话之间记住用户、记住上下文、记住偏好。我最初接触AI记忆是因为做一个用户运营助手。当时遇到的情况很典型用户每开一个新对话系统就要重新解释一遍自己的需求。后来我翻了各种memory方案从LangChain内置的ConversationBufferMemory到Mem0、Zep这类专门做记忆的中间件再到自己动手设计记忆里的数据库结构前前后后折腾了不少时间。这篇博文就是把这些踩坑经验整理出来给正在做AI Agent、客服机器人、个性化推荐系统的朋友做个参考。1.2 这个项目能解决什么先说清楚AI记忆到底解决什么问题不然很容易做成简单的日志存储。我梳理下来大概三类场景跨会话记忆用户上次聊了什么下次接着聊时能回忆起来而不用重新介绍自己。长期偏好记忆用户的喜好、习惯、说话风格可以跨项目、跨场景长期沉淀。动态上下文管理当前对话中的临时信息比如刚才提到的文件路径和长期记忆分开管理互不干扰。很多人一上来就想着把对话历史全存下来这是个误区。存下来和想起来是两码事——你需要的是按需召回而不是把整段历史都塞给模型。AI记忆的核心是存储检索遗忘的组合拳不是简单的数据库读写。所以在动手做之前我先花时间把需求拆成了三层记忆获取、记忆存储、记忆召回。后面所有的工作都围绕这三层展开。提示如果你只是为了给聊天机器人加个记住用户名字的功能用sessioncookie就够了不用上向量数据库。AI记忆项目的复杂度取决于你要记住什么、存多久、怎么用。1.3 适合谁来参考如果你属于下面这几类人这篇文章的实操部分可以直接抄做AI客服/助手的开发者需要让机器人记住用户的历史诉求。做Agent(智能体)的工程师需要让多步骤任务共享状态和上下文。做个性化推荐或内容生成的产品经理需要理解记忆在用户画像之外能补充什么。纯粹对LLM应用感兴趣的学习者想了解RAG和AI Memory之间的关系。这篇博文的代码部分用的是Python向量存储用FAISS和ChromaDB举例记忆框架会对比Mem0和自研方案的取舍。虽然不是零基础教程但我会把每个关键技术点的原理讲清楚边实操边解释为什么这么做。即使你还没碰过向量数据库跟着走一遍也能跑通。2. 记忆系统的架构设计与方案选型2.1 为什么不能直接存对话历史第一个要破除的误区是记住对话不等于存储对话。我见过不少团队直接拿Redis或者MySQL把原始对话记录存起来等下次对话时再把历史记录全部翻出来拼进prompt。这在对话轮次少的时候能用但一旦聊了几十轮token就爆炸了还容易把无关紧要的内容带进上下文。更严重的是早期对话里的关键信息会被海量闲聊稀释模型根本抓不住重点。正确的思路是像人脑一样对记忆做抽象和压缩。我们不需要记住每句话只需要记住发生了什么事用户表达了什么偏好当前任务的进度如何。2.2 记忆的三层模型我最终把记忆系统分成了三层每层的存储和检索逻辑完全不一样层级作用存储方式典型场景工作记忆当前会话的临时上下文内存/短期存储多轮对话中的中间状态情景记忆具体事件的完整记录向量数据库用户上次投诉的具体内容语义记忆提炼后的用户画像/偏好结构化数据库用户偏好、习惯、需求这种三层的设计参考了认知科学里对人类记忆的分类方式。工作记忆放最活跃的数据比如当前正在生成的文档标题。情景记忆放向量因为要语义检索用户之前提过的某某问题。语义记忆放结构化存储因为要支持精确查询比如用户所在的城市是什么。2.3 自研方案与Mem0/Zep的选型对比市面上确实存在现成的AI记忆解决方案。我实际调研和试用过Mem0、Zep以及LangChain生态里的记忆模块。最终我的选择是自研为主参考Mem0的设计思路。这么做不是因为开源方案不好而是因为它们不够可定制。拿Mem0来说它的设计很优雅核心流程是对话记录 → 提取关键记忆 → 向量化 → 存储 → 检索返回相关片段。但在实际项目中我需要把记忆和业务数据库打通。比如用户在某次对话里提到我在上海工作我希望这条记忆能同步到用户表里的disable字段而不是停留在记忆库里。Mem0提供了接口但扩展成本高不如自己在核心逻辑上复刻一套轻量实现。Zep的优势是有时间衰减和自动遗忘机制但它偏向Agent场景对普通客服机器人来说有点重。LangChain内置的Memory类说实话比较适合demo生产环境基本撑不住复杂需求。所以我的方案是底层存储自己设计核心记忆逻辑自己写但借鉴了Mem0对记忆提取-存储-召回三阶段的划分方式。注意选型的时候先列一下你的记忆要对接哪些业务系统。如果只是给LLM对话用Mem0这类方案非常省事。如果还要跟CRM、工单系统联动自研一个精简版更灵活。2.4 记忆的原子化设计这是我这套方案里最关键的设计决定。我不用对话记录作为记忆的最小单位而是用事实。每次用户对话后我会调用一个提取器从对话中抽取结构化的记忆原子。比如{ entity: user:12345, relation: prefers_language, object: Python, source_conversation: conv_6789, timestamp: 1699999999, confidence: 0.95 }这种三元组格式参考了知识图谱的实体-关系-实体结构。好处是检索的时候可以直接精确匹配也可以向量语义匹配。比如用户明确说过我喜欢Python这直接进结构化匹配如果用户说我平时主要写脚本处理数据语义向量才能归纳出和Python相关。原子化设计的另一个好处是支持记忆更新。用户上次说喜欢Python这次说其实最近都在写Go我只需要更新这一条三元组而不是把整个对话记忆重写一遍。3. 记忆存储与检索的基础设施3.1 为什么选择向量数据库既然记忆按语义被召回那向量数据库是绕不开的存储底座。以召回与当前问题相似的记忆为例。用户当前问了一句推荐几个适合团队协作的工具我要从历史记忆里找到和这句话语义相近的内容纯靠关键词匹配会丢很多信息比如用户以前说过我们在用Notion管文档最近想换一换。关键词协作工具都匹配不到但语义是相关的。向量化的做法是把这句话变成一个几百维的向量同时在存储记忆时也把记忆内容向量化。查询时计算两个向量的距离比如余弦相似度距离越近越相关。这就实现了语义级检索。我用的是ChromaDB做原型测试后续切换到FAISS做生产环境的索引管理。原因后面细说。3.2 Embedding模型的选择向量化这一步的选型决定了记忆召回的质量。我踩过的坑包括早期用OpenAI的text-embedding-ada-002效果确实好但要付费而且数据要发到第三方API。后来改用开源的BAAI/bge-small-zh-v1.5在中文场景下效果也很稳本地部署延迟低batch推理方便。我当时的对比测试数据大致如下测试集300条中文记忆片段Embedding模型召回Top-5准确率单条向量化延迟部署方式OpenAI text-embedding-ada-00289.2%35msAPIBGE-small-zh-v1.585.6%8ms本地text2vec-large-chinese86.1%12ms本地可以看到本地模型和OpenAI的差距并不大但延迟和成本优势明显。如果你对中文语义要求极高可以再试试BGE-large就是推理成本会涨一些。实操建议embedding模型选定后尽量少更换。因为更换模型意味着所有历史记忆的向量都要重新计算一遍。如果记忆量比较大这个重算成本很高。我第一次切换的时候10万条记忆向量化跑了好几个小时。3.3 向量存储和混合检索设计纯向量储存在某些精细场景下不够用。举个例子用户问我之前反馈过登录报错的问题解决了吗这种查询语义相似度能召回登录报错相关的记忆但怎么确定是哪一次反馈就需要结合时间过滤、会话ID过滤等结构化条件。所以我的存储表结构是这样设计的CREATE TABLE ai_memory ( id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, memory_type ENUM(fact, episode, preference) NOT NULL, content TEXT NOT NULL, embedding VECTOR(768), -- 向量索引配合pgvector使用 metadata JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, access_count INT DEFAULT 0, last_access_at TIMESTAMP, INDEX idx_user_id (user_id), INDEX idx_memory_type (memory_type), INDEX idx_created_at (created_at) );构建索引的时候我对embedding字段做了HNSW索引。写SQL逻辑时我把向量相似度召回和结构化条件过滤组合成一个查询SELECT id, content, metadata, 1 - (embedding :query_embedding) AS similarity FROM ai_memory WHERE user_id :user_id AND created_at :start_time ORDER BY similarity DESC LIMIT :top_k;先用结构化条件锁粗范围再按向量相似度精排效率和准确性都更好。如果不这样设计一个全局向量召回会经常捞出不相关的东西毕竟跨用户、跨时间长度的记忆太多了。这里重点说一下实际的混合检索我用的是先过滤后向量的流程而不是先向量后过滤。如果先说向量再过滤时间范围可能会把当前用户最相关的记忆排在前面但时间条件一旦应用就会丢失结果。先过滤就没这个问题还能减少向量检索的规模。3.4 记忆衰减与遗忘策略存储基础设施搭好了之后你会发现一个新问题记忆不会自己瘦身。用户聊了半年记忆库可能积累了几万条许多都是昨天中午吃了牛肉面这类价值极低的内容。所以必须设计记忆的衰减和遗忘机制。我参考了Zep的时间衰减思路但做了简化每个记忆有一个初始权重比如1.0每次被成功召回时权重增加长期不被召回则权重随时间衰减。def decay_score(score, last_access_at, half_life_days30): elapsed_days (datetime.now() - last_access_at).days decay_factor 0.5 ** (elapsed_days / half_life_days) return max(0.1, score * decay_factor)每30天权重减半。如果权重低于0.2这条记忆进入待清理状态由后台Job批量删除或归档到冷存储。这个策略的核心逻辑是越常被想起的记忆越重要越少被触碰的记忆越可能没用。听上去是常识但实际执行下来记忆库的体量能控制住召回质量也明显提升。4. 实操从零搭一个带AI记忆的对话系统4.1 技术栈与目录结构我用的技术栈是Python FastAPI ChromaDB SQLite OpenAI兼容API可用本地模型替代。目录结构如下ai-memory-demo/ ├── main.py # FastAPI 服务入口 ├── memory/ │ ├── __init__.py │ ├── extractor.py # 记忆提取器 │ ├── store.py # 记忆存储封装 │ ├── retriever.py # 记忆召回器 │ └── models.py # Pydantic 模型 ├── config.py # 配置文件 └── requirements.txt4.2 记忆提取器实现记忆提取器的作用是把用户和AI的一次对话提炼成结构化的记忆。不能用原始对话存进去否则长期记忆就变成了日志堆积。我的做法是给LLM设定一个提取提示词EXTRACT_PROMPT 你是一个记忆提取器。请从以下对话中提取需要长期记住的信息。 只提取以下类型的事实用户偏好、用户个人信息、用户目标、项目进展、明确表态。 忽略寒暄、临时状态、无关闲聊。 输出JSON数组每个元素包含 { entity: 主体标识, relation: 关系类型, object: 对象内容, memory_type: fact|episode|preference, importance: high|medium|low } 对话内容 {conversation} 比如用户说我一般下午才比较有空提取到的记忆是{ entity: user:12345, relation: preferred_work_time, object: 下午, memory_type: preference, importance: medium }对于importance字段系统决策时会更关注high的记忆低重要度记忆直接跳过来减少token占用。4.3 记忆存储封装存储层我封装了一套统一的接口方便底层在ChromaDB和FAISS之间切换。class MemoryStore: def __init__(self, collection_nameai_memory, persist_dir./db): self.client chromadb.PersistentClient(pathpersist_dir) self.collection self.client.get_or_create_collection( namecollection_name, metadata{hnsw:space: cosine} ) def add(self, memory: dict, embedding: list[float]): self.collection.add( ids[memory[id]], embeddings[embedding], documents[memory[content]], metadatas[{ user_id: memory[user_id], memory_type: memory[memory_type], timestamp: memory[timestamp], importance: memory[importance] }] ) def delete(self, memory_id: str): self.collection.delete(ids[memory_id])ChromaDB的好处是开箱即用支持持久化到本地磁盘对单机项目来说非常够用。到生产环境我切换到了FAISS PostgreSQL的一体化方案向量索引用FAISS元数据像用户ID、时间戳、访问次数放到PostgreSQL。因为FAISS的过滤语法相对原始我是先在PostgreSQL里把满足条件的所有候选ID捞出来再在FAISS里做向量检索最后按时间排序。4.4 记忆召回与上下文注入召回逻辑是整个系统性能和效果的关键很多项目跑偏都是在这块。先看一下代码def retrieve_relevant_memories(user_id: str, query: str, top_k: int 5): # 1. 生成查询向量 query_embedding embed(query) # 2. 向量检索 results collection.query( query_embeddings[query_embedding], n_resultstop_k, where{user_id: user_id} ) # 3. 过滤低相似度记忆 filtered [ (doc, meta, dist) for doc, meta, dist in zip(results[documents][0], results[metadatas][0], results[distances][0]) if dist 0.55 # 这是关键参数后面细说 ] # 4. 按时间新鲜度和重要性重排 filtered.sort(keylambda x: (x[1][importance], x[1][timestamp]), reverseTrue) return filtered[:top_k]重排的目的事前过滤掉语义相关但价值不高的记忆。比如发现用户当前在问Python历史记忆里确实有一段和Python相关但如果那是一年前问的Python安装教程重要性低、时效性差就不该排在前面。重排让答案更贴近当下的需求。召回的记忆拼接入prompt时我用了一个固定格式【相关历史记忆】 - 用户偏好下午工作语言偏好Python - 3天前反馈过登录页报错500状态未解决这段记忆信息插到system prompt的末尾作为context的一部分。模型看到相关历史后才能做出个性化的回应。4.5 相似度阈值的调优心法我上面代码里的dist 0.55这个阈值是我调了很久才确定的。这可能是整个项目里最值得记录的经验。ChromaDB默认返回的是L2距离因为我设置的是cosine空间其实返回的是余弦距离。余弦距离的范围是0到2越小越好。0.55这个阈值在BGE-small模型下能保证召回的大多数记忆都在语义上有明确相关性。怎么调呢我准备了一个小测试集里面放了20个正确应该召回的查询和20个不应该召回的干扰查询。然后枚举不同的threshold计算准召率阈值精确率召回率0.3595%41%0.4588%67%0.5574%83%0.6553%92%最终我选0.55求一个平衡。如果你想调这个参数别拍脑袋要像我一样逐档测试看你的场景更看重别带无关信息还是别漏掉关键信息。注意不同embedding模型的向量空间分布不同换模型后阈值必须重新调。之前从ada-002切到BGE后0.55这个值直接让召回率掉了20%因为BGE的向量余弦距离整体偏大。如果你用不同模型必须先跑一轮校准。4.6 完整对话流程串起来每收到一条用户消息完整的记忆链路是这样的从请求里读出user_id和message。先调retrieve_relevant_memories召回相关记忆拼入system prompt。调LLM生成回复把回复和用户消息一起交给extractor。extractor提取新事实写入MemoryStore。返回回复给用户同时更新相关记忆的access_count和last_access_at。这里有一个顺序问题第一步和第四步容易搞混。必须先召回再生成生成后再存新记忆。存新记忆的时候要特别注意去重如果用户这轮说我才喜欢Python跟昨天的我喜欢Python特征相似度很高就该做合并更新而不是再插一条重复数据。我在store层加了一个去重逻辑def add_memory_deduplicated(memory, threshold0.92): existing collection.query( query_embeddings[embed(memory[content])], n_results1, where{user_id: memory[user_id]} ) if existing[distances][0][0] threshold: # 合并更新旧的记忆而不是插入新的 update_memory(existing[ids][0], memory) else: add(memory)这样做的原因是避免记忆碎片化。如果同一条偏好被存入多条相似记忆召回时会同时带出5条占用的token是5倍的信息量却没有增加。5. 常见问题与排查技巧实录5.1 记忆检索不准换阈值、换检索方式、换模型这是被问得最多的问题我召回来的记忆明显不相关怎么回事我的排查顺序是第一看相似度得分到底是多少。如果召回结果的距离集中在0.6-0.7说明阈值太松了类似上面的调优。如果得分集中在0.3-0.4但内容依然不相关那就不是阈值问题而是embedding模型的问题。第二看查询语句本身。用户问的很短很口语化比如上次那个事这种query做向量化效果极差。我目前的解决办法是加一个查询改写环节用LLM把上次那个事改写为用户上次提到的未解决问题然后再embedding和召回。改造后召回准确率提升了近15%。第三如果前两步都没问题就要考虑做rerank。可以引入一个粗排精排的二级检索架构向量检索做粗排取Top-50再交给一个rerank模型。我自己的业务对延迟没那么敏感就在召回后接了一个cross-encoder开源模型。质量提升很显著代价是每查询增加了几十毫秒。5.2 记忆冲突新旧记忆打架怎么办这种情况经常出现。用户上个月说我喜欢喝美式这周又说最近想试试拿铁。两条记忆语义上都是用户对咖啡的偏好但内容冲突了。如果简单地把两条都塞给LLM模型会迷惑到底哪个是对的。我的解决办法是在记忆提取阶段就引入冲突检测新提取出来的对象先和已有的同entity、同relation的记忆做对照。如果发现矛盾判断依据是recency最新优先和importance高重要隐私优先。如果recency相同就保留importance更高的。实际执行的时候我会把旧记忆标记为superseded_by_new_id而不是直接删除。这样可以追溯到用户的偏好变迁历史对于做推荐系统的人来说这本身就是有价值的数据。5.3 记忆膨胀存储和token成本失控前面说过通过遗忘策略控制规模。实际运营下来还有两个很有效的技巧第一降采样。对话日志里大量重复的日常性记忆如果只保留7天之后只保留摘要。我给记忆系统加了一个汇总Job每天对前一天的记忆做聚类然后生成一个日摘要。日摘要之间存在相似性再聚合成周摘要。这样长期来看高层摘要负责宏观偏好短期记忆负责细节上下文不会无限膨胀。第二按用户分级。我一直以为所有用户都应该存下全部记忆后来运营数据打脸90%的普通用户记忆保留30天就够了只有深度用户或付费用户开启长期记忆权限。分级存储不仅成本降低而且普通用户的召回准确率还提升了因为少了大量低价值旧记忆的干扰。5.4 隐私与数据安全容易被忽略的关键环节做记忆系统的都明白记忆本身就是隐私数据你存了用户的偏好、对话、甚至是个人信息。这是逃不掉的责任。我在合规上有几个做法供你参考敏感信息脱敏在提取记忆存入存储之前用一个脱敏模型检测手机号、身份证号等PII处理后再存储。用户删除权提供了清除记忆API按user_id一键删除全部数据。记忆导出的权限隔离严格按user_id隔离记忆表。生产环境运维时禁止执行全库向量查询这类操作。日志审计所有记忆的写入、读取、更新都记录审计日志保留操作者防止内部滥用。这些设计不是锦上添花而是上线前必须完成的。我见过有同行把AI记忆做成产品因为默认开启全局记忆、没有用户控制面板被用户投诉隐私问题最后只能下线整改。5.5 性能排查从召回到生成的完整耗时分析如果用户反馈机器人变慢了不要猜先看数据。我的一个排障案例可以说明问题某天客服机器人的响应延迟从800ms飙到2.5s直觉是LLM太慢一查慢在了记忆召回环节。再排查是某个用户的历史记忆达到3万条向量检索一次要扫全量慢得很。后来优化了两点第一按用户维度分collection。每个用户一个collection查询时天然缩小范围。但collection数量太多会有管理开销我改成了按用户分桶比如按user_id hash到16个collection。第二缓存高频召回结果。用户反复询问同一个主题时短时间内召回结果高度重合加了LRU缓存命中时零延迟。实测下缓存命中率在30%左右平均响应时间降了12%。6. 踩坑总结与经验分享6.1 我应该早点想清楚的三个问题第一不要把RAG和Memory混为一谈。RAG是从外部知识库检索文档Memory是从历史交互中找回上下文。两者虽然都依赖向量检索但数据来源、更新频率、存储结构完全不同。很多团队把记忆做成了一般的RAG结果就是对话里带上一堆百科内容而不是用户偏好的信息。第二记忆的质量比数量重要。宁可每天只存下三条高置信度的偏好也不要盲目存储一百条废话。质量低的信息会对召回造成干扰挤占上下文窗口。我在迭代中一直在提高记忆提取器的过滤标准账户里存下的记忆条数少了2/3但召回的用户满意度反而更高了。第三记忆系统要重度依赖遗忘。人脑的遗忘不是缺陷是保护机制。AI记忆如果没有遗忘旧信息会污染新判断存储成本会拖垮性能。我曾经因为忘了设计记忆清理半年后上线的大模型开始胡言乱语——用户已经改了偏好模型还在按三个月前的旧偏好回复。6.2 这个项目的后续扩展方向做完基础版的ai-memory之后我觉得这三个方向是最有价值的扩展多模态记忆存储现在文本记忆只占一个维度图片和音视频里的信息也应该能被提取出来这对智能助手场景很实用。记忆跨设备同步用户手机上聊的口味到PC端打开App也要能想起来这就需要记忆库的云端同步和统一身份系统。Agent记忆协作如果任务是靠多Agent协作完成的记忆就不能只挂在用户维度还要挂在任务维度、Agent本身维度。任务记忆的设计比用户记忆复杂得多因为中间状态很多状态转移的路径也不固定。6.3 最后一句掏心话我最后想说的是AI记忆工程真正难的地方不是向量数据库怎么连、embedding用哪个模型而是你怎么定义什么才值得记住。这需要在产品逻辑、用户体验和工程成本之间反复权衡。我第一次做出ai-memory的demo时感觉特别简单两三百行代码就实现了基本对话记忆。可真正把它推到线上每天处理几万次对话、召回几千条记忆之后才发现记忆系统是需要持续运营和调优的。它会伴随你的业务一起涨大也会在某个临界点逼着你去重构。现在再做类似项目的话我会一上来就把记忆的提取、存储、召回、遗忘四个阶段都想明白不急着写代码。把数据模型和管理策略理顺了再动手。这套方法论比任何一份开源代码都值钱。