大模型记忆系统实战：从向量化存储到混合索引召回

1. 从“健忘”到“有记忆”为什么大模型需要记忆系统如果你用过早期的ChatGPT或者一些开箱即用的基础大模型肯定遇到过这样的场景你跟它聊了十几轮详细讨论了一个项目的技术方案然后你问它“我们刚才提到的那个数据库选型它的主要优势是什么”它可能会给你一个非常笼统、甚至完全错误的回答。它“忘记”了你们刚才长达半小时的对话细节。这就是典型的“上下文窗口依赖症”——模型只对当前输入的有限窗口内的信息敏感一旦信息被推出窗口就如同石沉大海。在货拉拉这样复杂的业务场景里这种“健忘”是致命的。想象一下一个司机师傅在月初咨询过某个区域的派单规则月底他再次询问时如果客服大模型给不出基于他历史行为的个性化回答体验就会大打折扣。又或者一个用户反复抱怨过某个功能不好用下次交互时模型却毫无察觉依然推荐那个功能这无疑会加剧用户的不满。因此给大模型装上“记忆系统”让它能记住关键的对话历史、用户偏好和业务事实并从海量记忆中精准地提取出当前对话所需的信息就成了AI工程化落地必须啃下的硬骨头。这不仅仅是技术炫技而是实实在在的业务刚需。记忆系统本质上是在扩展模型的“工作内存”和“长期记忆”使其行为更连贯、更个性化、更符合业务逻辑。今天我就结合我们在货拉拉的真实工程实践拆解记忆系统的核心流程——从“提取”到“召回”。这不是一篇纯理论的论文而是充满了踩坑、权衡和实战代码的工程笔记。我们会聚焦于记忆从哪里来提取记忆存到哪里去存储与索引以及当需要时如何又快又准地找到它召回2. 记忆的源头活水多模态信息的结构化提取记忆不是凭空产生的它的原料来自于用户与系统交互产生的所有数据。在货拉拉的场景下这些数据纷繁复杂有纯文本的客服对话、有结构化的订单信息地址、货物类型、费用、有司机的轨迹日志、甚至还有语音通话的转写文本。第一步我们要从这些多模态、非结构化的数据流中提取出能够被“记忆”的、有价值的结构化信息。这个过程我们称之为“记忆提取”Memory Extraction。它远不止是简单的关键词匹配或正则表达式抽取而是一个融合了规则引擎、轻量级模型和启发式策略的流水线。2.1 定义“记忆单元”什么值得被记住不是所有信息都值得进入长期记忆。把聊天记录全部存下来是最简单的但也是最低效的会导致记忆库迅速膨胀召回时噪声极大。我们的核心思路是只记忆那些具有持久价值、可能被未来对话反复引用的信息。我们定义了以下几类核心记忆单元用户属性与偏好例如用户是“经常搬运家具的个体户”、司机是“偏好接长途订单的老司机”。这些信息不会频繁变动但对个性化服务至关重要。关键事实与承诺例如用户曾说过“我每天下午5点后才有空”客服承诺过“针对您反馈的计费问题我们会在3个工作日内回复”。这些是对话中的核心事实点。业务实体与关系例如订单号、运单号、涉及的货物类型“大型机械设备”、“易碎品”、发货/收货地址。这些是连接对话与业务系统的桥梁。情感与问题脉络例如用户本次会话中表现出的“不满”情绪或反复咨询的“如何开发票”问题。这有助于模型理解对话的上下文和用户的潜在需求。2.2 构建提取流水线规则与模型的双轨制基于上述定义我们设计了一个分层提取流水线。第一层基于规则与模式的快速提取对于高度结构化的信息规则引擎Regex 业务字典是效率最高、确定性最强的选择。例如从文本中提取手机号、订单号符合特定编码规则、金额、日期时间等。我们维护了一个庞大的业务实体词典包含货物类型、城市区域、车型等通过精确匹配或模糊匹配进行抽取。# 示例一个简单的规则提取器用于抽取订单号和货物类型 import re class RuleBasedExtractor: def __init__(self): self.order_pattern re.compile(r订单[:]\s*([A-Z0-9]{10,15})) self.cargo_keywords [家具, 电器, 建材, 宠物, 易碎, 大型设备] def extract(self, text): memories [] # 提取订单号 order_match self.order_pattern.search(text) if order_match: memories.append({ type: business_entity, key: order_id, value: order_match.group(1), source: text }) # 提取货物类型 for keyword in self.cargo_keywords: if keyword in text: memories.append({ type: user_preference, key: cargo_type, value: keyword, source: text }) return memories第二层基于轻量级模型的语义提取对于更复杂的语义信息如用户意图、情感倾向、总结性陈述我们使用微调过的轻量级文本分类或序列标注模型。例如我们用一个简单的BERT分类模型来判断一句话是否包含了“用户偏好”或“问题投诉”。注意这里我们没有直接用千亿级大模型做提取主要是出于成本和延迟的考虑。在对话流中实时进行提取要求毫秒级响应。轻量级模型百兆级别在精度和速度上取得了更好的平衡。大模型更适合作为“校验器”或“后处理器”在异步流程中对提取结果进行润色和纠错。第三层会话级摘要与关键信息凝结单句话的提取是基础但记忆往往存在于多轮对话的上下文中。因此我们引入了会话级处理。当一段对话例如超过10轮或主题明显转换结束时我们会触发一个摘要模型为这段对话生成一个简短的文本摘要并从中提取最核心的1-3个记忆点。这相当于为一段复杂的对话生成了一个“记忆标签”。整个提取流水线的输出是一个结构化的“记忆单元”列表。每个单元都包含类型、键、值、置信度、时间戳和来源片段。这些单元就是我们要存入记忆库的原材料。3. 记忆的安身之所向量化存储与混合索引构建提取出来的记忆单元是结构化的但如何存储才能支持高效的、基于语义的召回呢传统的数据库如MySQL适合精确查询“用户A的订单号123”但无法处理模糊的语义查询“用户上次提到的关于运费的问题”。这就是向量数据库Vector Database大显身手的地方。3.1 向量化将语义转化为数学召回的核心是相似度计算。我们需要把文本用户当前的问题和记忆库里的文本历史记忆都转换成计算机能理解的数值形式——即向量Embedding然后计算它们之间的余弦相似度或欧氏距离。我们选用了开源的text2vec模型来生成向量。这个选择基于几点考虑首先它在中文语义相似度任务上表现稳健其次模型大小适中约300MB推理速度快最后其生成的向量在我们的业务语义空间上分布良好。# 示例使用 text2vec 生成记忆单元的向量 from text2vec import SentenceModel import numpy as np class MemoryEmbedder: def __init__(self, model_nameshibing624/text2vec-base-chinese): self.model SentenceModel(model_name) def embed_memory(self, memory_unit): # 根据记忆单元类型构造用于向量化的文本 if memory_unit[type] user_preference: text_to_embed f用户偏好{memory_unit[key]} 是 {memory_unit[value]} elif memory_unit[type] business_entity: text_to_embed f业务实体{memory_unit[key]} {memory_unit[value]} else: text_to_embed memory_unit.get(source, ) # 生成向量 vector self.model.encode(text_to_embed) return vector.astype(np.float32) # 统一为float32节省空间关键细节直接使用原始的source字段用户原话做向量化有时并不好因为原话可能冗长且包含无关信息。我们采用“模板填充”的方式将结构化的typekeyvalue组合成一句通顺的陈述句这样生成的向量更能捕捉记忆的语义核心而非表面字词。3.2 存储与索引向量数据库的选型与实战有了向量我们需要一个专门的数据来存储和检索它们。我们对比了Milvus、Qdrant和PGVectorPostgreSQL插件。Milvus功能强大生态成熟但运维相对复杂对于中小规模的记忆库千万级以下有点“杀鸡用牛刀”。Qdrant用Rust编写性能出色API友好云服务版也很成熟。是我们重点考虑的对象。PGVector最大的优势是与现有业务数据库PostgreSQL无缝集成无需引入新的技术栈利用现有的备份、监控体系。考虑到团队技术栈和运维成本我们最终选择了PGVector。它虽然绝对性能可能不及专门的向量数据库但对于我们初期的数据量百万级记忆单元和延迟要求P99 50ms完全够用并且大大降低了系统的复杂性和维护成本。-- 示例在 PostgreSQL 中创建存储记忆的表 CREATE TABLE memory_units ( id BIGSERIAL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, session_id VARCHAR(128), memory_type VARCHAR(50), memory_key VARCHAR(255), memory_value TEXT, source_snippet TEXT, confidence FLOAT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 使用 pgvector 扩展的 vector 类型存储 768 维向量 embedding vector(768) ); -- 创建向量索引以加速相似性搜索 CREATE INDEX ON memory_units USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);踩坑实录直接使用ivfflat索引在数据量快速增长时召回精度下降很明显。原因是ivfflat的lists参数需要根据数据分布调整。我们后来采用了HNSWHierarchical Navigable Small World索引虽然构建慢一点占用空间大一点但查询速度和精度都非常稳定更适合在线服务。CREATE INDEX ON memory_units USING hnsw (embedding vector_cosine_ops);3.3 混合索引当向量搜索不够用时纯向量搜索解决的是“语义相似”的问题。但业务中大量需求是“精确匹配”或“多条件过滤”。例如“查找用户A在昨天产生的所有关于‘运费’的记忆”。这需要结合向量搜索和传统数据库的属性过滤。PGVector的优势在这里体现得淋漓尽致。我们可以写一个SQL同时进行向量相似度搜索和属性过滤# 示例使用 pgvector 进行带过滤的混合搜索 import psycopg2 from psycopg2.extras import Json def search_memories(user_id, query_text, memory_typeNone, limit5): # 1. 将查询文本向量化 query_vector embedder.embed_query(query_text) # 2. 构建SQL sql SELECT id, memory_key, memory_value, source_snippet, 1 - (embedding %s) as similarity -- 是pgvector的余弦距离运算符 FROM memory_units WHERE user_id %s params [query_vector.tolist(), user_id] if memory_type: sql AND memory_type %s params.append(memory_type) sql ORDER BY embedding %s LIMIT %s; # 按距离排序 params.extend([query_vector.tolist(), limit]) # 3. 执行查询 with conn.cursor() as cur: cur.execute(sql, params) results cur.fetchall() return results这种“向量索引 属性过滤”的混合索引策略是我们记忆召回系统的基石它兼顾了语义的灵活性和业务查询的精确性。4. 召回策略的核心从粗排到精排的漏斗模型当用户发起一次新的对话我们需要从记忆库中召回最相关的记忆。这不是一次简单的向量搜索而是一个多阶段的、精心设计的召回漏斗。4.1 召回阶段一多路并发粗筛我们的目标是低延迟几十毫秒内返回结果因此第一阶段的召回必须快。我们同时发起多路召回以覆盖不同的相关性维度向量相似度召回如上节所述用当前用户query的向量去记忆库搜索最相似的N条例如100条记忆。这是语义召回的主力。关键词召回对当前query进行分词提取核心名词、实体词在记忆的memory_key和memory_value字段进行倒排索引检索使用PostgreSQL的GIN索引。这能抓住那些字面匹配强但语义可能稍弱的记忆。时间衰减召回用户最近的记忆通常更重要。我们会额外召回该用户最近一段时间如7天内创建的记忆并按时间加权。会话上下文召回如果当前对话有session_id我们会优先召回同一session下的历史记忆保证对话的连贯性。每一路召回都会返回一个候选记忆列表及其分数向量相似度分、关键词匹配分、时间衰减分等。4.2 召回阶段二多分数融合与重排序多路召回的结果需要合并、去重并产生一个统一的排序。我们采用了一个线性加权打分模型来进行重排序Re-ranking最终分数 w1 * 向量相似度分 w2 * 关键词匹配分 w3 * 时间衰减分 w4 * 业务规则分其中权重w1, w2, w3, w4是通过离线评估看Top K记忆的准确率和线上A/B测试调出来的。业务规则分可以加入一些先验知识比如“memory_type为user_preference的记忆权重更高”。实操心得重排序模型一开始我们想上复杂的机器学习模型如LambdaMART但后来发现对于初期阶段一个精心调校的线性模型已经能带来80%的收益且简单、稳定、可解释性强。工程上先解决“有没有”再优化“好不好”。4.3 召回阶段三上下文长度与记忆注入经过重排序我们得到了一个Top K比如5-10条的高质量记忆列表。接下来问题来了如何把这些记忆“喂”给大模型最直接的方式是把记忆文本拼接在用户query之前作为上下文。但大模型的上下文长度是有限的如4K、8K、32K Token。我们必须精打细算。我们的策略是动态长度预估对排序后的记忆按优先级估算其Token长度。优先级截断从最高优先级的记忆开始注入直到总Token数接近模型上下文窗口的某个安全阈值例如预留30%的窗口给模型本身的思考和当前对话历史。格式化提示词不是简单拼接文本。我们使用清晰的模板格式化记忆让模型更容易理解。例如以下是用户的历史信息记忆供你在回答时参考 - [偏好] 用户表示更偏好厢式货车。 - [事实] 用户上次下单的收货地址是“XX科技园B座”。 - [问题] 用户曾在2023-10-26反馈过运费计算不准的问题。 当前用户问题{当前query}这种格式化的、结构清晰的记忆注入方式比扔过去一堆原始文本能显著提升大模型对记忆的利用效率。5. 工程落地中的挑战与优化理论很美好但工程落地处处是坑。分享几个我们遇到的关键挑战和解决方案。5.1 记忆的更新、冲突与失效记忆不是只写不删的。用户的偏好会变地址会更新问题解决了就该归档。更新对于同一user_idmemory_key的记忆我们采用“软更新”策略。新记忆到来时会查找是否有同key旧记忆。如果有则不直接覆盖而是将旧记忆标记为is_obsoletetrue并插入新记忆。同时在向量库中我们会计算新旧记忆向量的差异如果差异很小可能只是补充信息我们会尝试合并。冲突检测有时用户会说自相矛盾的话。我们设计了一个简单的冲突检测规则如果新记忆与近期如24小时内的同类型记忆在关键值上直接矛盾例如偏好从“厢式货车”变成“平板车”则触发一个低置信度警报该条记忆会进入人工审核队列而不是直接入库。失效策略我们为记忆设置了“保鲜期”。例如一次性的投诉记忆在标记为“已解决”30天后自动归档长期未激活的用户其部分动态偏好记忆也会逐渐降权。这避免了记忆库无限膨胀和存储陈旧信息。5.2 冷启动与噪声记忆处理新用户没有历史记忆怎么办我们引入了“群体记忆”或“默认记忆”的概念。例如对于新司机我们可以注入“大多数司机常问的规则”作为初始记忆。但这需要非常谨慎避免造成刻板印象。另一个大问题是噪声记忆。提取环节不可能100%准确可能把玩笑话、无关信息也当成记忆存了下来。我们在召回链路的最后加了一道“轻量级过滤器”用一个极小的文本分类模型判断召回的某条记忆是否真的与当前query强相关。如果相关性分数低于阈值则将其从最终注入列表剔除。这道过滤虽然增加了少量计算但显著提升了注入记忆的整体质量。5.3 性能、监控与评估体系性能记忆系统的延迟直接影响对话响应。我们对所有环节提取、向量化、检索、重排都做了埋点和监控。PGVector的HNSW索引将核心检索的P99延迟控制在了20ms以内。对于向量化模型我们使用了TensorRT进行推理优化并部署了缓存层对相同的文本避免重复计算。监控除了常规的QPS、延迟、错误率我们还监控一些业务指标记忆提取率多少对话产生了记忆、记忆召回率有历史记忆的query中成功召回的比例、记忆利用率被召回的记忆中有多少被大模型在回复中实际引用或体现。评估评估记忆系统的效果是难点。我们采用了人工评估和自动评估结合的方式。人工评估定期抽样对话日志评估记忆的提取是否准确、召回是否相关、以及最终是否提升了回答质量。自动评估我们构建了一个基于规则的“模拟用户”测试集包含一系列多轮对话检查系统能否正确记住并在后续轮次中使用关键信息。同时我们也用召回的“命中率”和“准确率”作为辅助指标。5.4 与现有系统的整合记忆系统不是孤立的。它需要与对话管理、用户画像、知识库等系统紧密协作。与对话管理记忆的提取和召回时机由对话状态机触发。例如在对话开场或检测到用户提及历史话题时主动触发记忆召回。与用户画像记忆系统中沉淀的结构化用户偏好和事实会定期同步到用户画像系统丰富画像维度。反之画像系统提供的静态标签如用户等级也可以作为记忆召回时的过滤条件。与知识库记忆个性化、动态和知识通用、静态是互补的。我们的召回系统实际是一个“统一检索器”可以同时从记忆库和向量化的知识库中检索信息然后一起喂给大模型。从提取到召回这只是记忆系统工程实践的上半场。它解决了“记什么、存哪里、怎么找”的问题。下半场则更侧重于“怎么用”——如何将召回的记忆与LLM的推理过程更深度地结合如何评估记忆系统的整体效果以及如何设计更复杂的记忆结构如事件记忆、情节记忆。这些内容我们留到下一篇文章再详细探讨。工程之路就是不断在理想架构与现实约束之间寻找平衡点的过程。希望我们这些踩过的坑和总结的经验能为你构建自己的大模型记忆系统提供一些切实的参考。