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_name='shibing624/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字段（用户原话）做向量化有时并不好，因为原话可能冗长且包含无关信息。我们采用“模板填充”的方式，将结构化的type，key，value组合成一句通顺的陈述句，这样生成的向量更能捕捉记忆的语义核心，而非表面字词。
3.2 存储与索引：向量数据库的选型与实战
有了向量，我们需要一个专门的数据来存储和检索它们。我们对比了Milvus、Qdrant和PGVector（PostgreSQL插件）。
- 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参数需要根据数据分布调整。我们后来采用了HNSW（Hierarchical 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_type=None, limit=5): # 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_id+memory_key的记忆，我们采用“软更新”策略。新记忆到来时，会查找是否有同key旧记忆。如果有，则不直接覆盖，而是将旧记忆标记为is_obsolete=true，并插入新记忆。同时，在向量库中，我们会计算新旧记忆向量的差异，如果差异很小，可能只是补充信息，我们会尝试合并。 - 冲突检测：有时用户会说自相矛盾的话。我们设计了一个简单的冲突检测规则：如果新记忆与近期（如24小时内）的同类型记忆在关键值上直接矛盾（例如，偏好从“厢式货车”变成“平板车”），则触发一个低置信度警报，该条记忆会进入人工审核队列，而不是直接入库。
- 失效策略：我们为记忆设置了“保鲜期”。例如，一次性的投诉记忆，在标记为“已解决”30天后自动归档；长期未激活的用户，其部分动态偏好记忆也会逐渐降权。这避免了记忆库无限膨胀和存储陈旧信息。
5.2 冷启动与噪声记忆处理
新用户没有历史记忆，怎么办？我们引入了“群体记忆”或“默认记忆”的概念。例如，对于新司机，我们可以注入“大多数司机常问的规则”作为初始记忆。但这需要非常谨慎，避免造成刻板印象。
另一个大问题是噪声记忆。提取环节不可能100%准确，可能把玩笑话、无关信息也当成记忆存了下来。我们在召回链路的最后加了一道“轻量级过滤器”：用一个极小的文本分类模型，判断召回的某条记忆是否真的与当前query强相关。如果相关性分数低于阈值，则将其从最终注入列表剔除。这道过滤虽然增加了少量计算，但显著提升了注入记忆的整体质量。
5.3 性能、监控与评估体系
- 性能：记忆系统的延迟直接影响对话响应。我们对所有环节（提取、向量化、检索、重排）都做了埋点和监控。PGVector的HNSW索引将核心检索的P99延迟控制在了20ms以内。对于向量化模型，我们使用了TensorRT进行推理优化，并部署了缓存层，对相同的文本避免重复计算。
- 监控：除了常规的QPS、延迟、错误率，我们还监控一些业务指标：记忆提取率（多少对话产生了记忆）、记忆召回率（有历史记忆的query中，成功召回的比例）、记忆利用率（被召回的记忆中，有多少被大模型在回复中实际引用或体现）。
- 评估：评估记忆系统的效果是难点。我们采用了人工评估和自动评估结合的方式。
- 人工评估：定期抽样对话日志，评估记忆的提取是否准确、召回是否相关、以及最终是否提升了回答质量。
- 自动评估：我们构建了一个基于规则的“模拟用户”测试集，包含一系列多轮对话，检查系统能否正确记住并在后续轮次中使用关键信息。同时，我们也用召回的“命中率”和“准确率”作为辅助指标。
5.4 与现有系统的整合
记忆系统不是孤立的。它需要与对话管理、用户画像、知识库等系统紧密协作。
- 与对话管理：记忆的提取和召回时机由对话状态机触发。例如，在对话开场或检测到用户提及历史话题时，主动触发记忆召回。
- 与用户画像：记忆系统中沉淀的结构化用户偏好和事实，会定期同步到用户画像系统，丰富画像维度。反之，画像系统提供的静态标签（如用户等级）也可以作为记忆召回时的过滤条件。
- 与知识库：记忆（个性化、动态）和知识（通用、静态）是互补的。我们的召回系统实际是一个“统一检索器”，可以同时从记忆库和向量化的知识库中检索信息，然后一起喂给大模型。
从提取到召回，这只是记忆系统工程实践的上半场。它解决了“记什么、存哪里、怎么找”的问题。下半场则更侧重于“怎么用”——如何将召回的记忆与LLM的推理过程更深度地结合，如何评估记忆系统的整体效果，以及如何设计更复杂的记忆结构（如事件记忆、情节记忆）。这些内容，我们留到下一篇文章再详细探讨。工程之路，就是不断在理想架构与现实约束之间寻找平衡点的过程。希望我们这些踩过的坑和总结的经验，能为你构建自己的大模型记忆系统提供一些切实的参考。
