1. 为什么“塞上下文”这条路注定走不到生产环境1.1 从 Demo 到灰度上下文窗口和账单先给你上一课做 Agent 的第一天没人觉得记忆系统是个问题。我一开始也是这么想的用户多轮对话嘛把历史消息全部塞进 prompt模型自己会看懂上下文。直到某个项目从 Demo 走到灰度用户会话拉到 60 轮以上问题像约好了一样集中爆发——账单肉眼可见地涨、响应延迟往上爬、模型开始忘记 20 轮前用户亲口说过的偏好。那个时候我才意识到上下文窗口不等于记忆而工业级 Agent 的记忆机制是需要像数据库一样单独设计和维护的一层基础设施。先算一笔账。假设一个客服类 Agent 每轮请求的用户输入加上模型输出平均是 800 tokens如果每一轮都把完整历史拼进上下文那么第 n 次请求携带的上下文长度就是 800 × n。从第 1 轮到第 N 轮累计消耗的 token 大概是[ \sum_{i1}^{N} 800 \times i 800 \times \frac{N(N1)}{2} ]也就是说第 20 轮累计消耗约 16.8 万 tokens第 60 轮累计消耗约 146 万 tokens。同样一个模型不管单位价格是多少这个量级摆在那里成本一定是爆炸的。这不是“优化一下 prompt 就能解决”的问题而是算法层面的复杂度问题。更麻烦的是延迟。Transformer 推理时间随输入长度近似线性增长当上下文从 8k 涨到 48k首 token 延迟会明显抬升。我们当时灰度环境的 p95 响应时间从 1.2 秒涨到 3 秒以上产品同学直接跑到我工位上来问“是不是模型服务挂了”。我说模型服务没挂是每轮都在把整部小说喂给它读。这类问题在 Demo 阶段完全不存在因为你只有三个用户在测一上灰度成本、延迟、稳定性三个硬指标同时向你扑过来。1.2 长上下文不等于有记忆注意力稀释与状态丢失如果说成本和延迟是“肉痛”那效果劣化就是“骨折”。把全部历史塞进上下文模型确实看得到所有内容但“看得到”和“用得上”是两回事。业界对长上下文有个公认的观察当输入文本很长时模型对中间部分内容的注意力会明显衰减信息更容易被开头和结尾的内容盖过去。也就是说假设用户在第 5 轮说过“结算账号用尾号 8890 那个”到第 30 轮 Agent 处理付款动作时这条信息大概率躺在上下文的中间某个位置模型很可能就漏了它。于是 Agent 按默认账号发起结算用户回来投诉你查日志发现这条信息明明出现过模型就是没用上。这还只是“漏读”。更隐蔽的问题是 Agent 的状态一致性。在长会话里Session 内的关键约束会散落在多轮消息里例如“发票抬头”“配送地址”“审批人”“预算上限”。当上下文越来越长你会忍不住做截断来压成本——无论是滑动窗口还是摘要压缩只要截断了某些状态就永久丢失。Agent 后续的每一步都建立在残缺状态上动作链越长偏差越大。这就是我常说的长会话里最怕的不是模型笨而是“上下文里有但模型没有命中”。你没有一套显式的记忆检索机制就是在赌模型每次都刚好注意到要用的那一段。对于线上 SLA 来说这种赌法迟早会输。1.3 记忆的本质是存取策略不是存储空间把事情想清楚之后我给自己定了个框架上下文是工作台记忆是文件柜。工作台的桌面再大东西堆多了照样找不到文件柜的价值不是“能放更多东西”而是你有一套索引和检索规则知道什么该放、什么该丢、什么该归档。一个工业级 Agent 记忆系统本质上就是在回答三个问题什么信息值得写进记忆库用户当前需求下应该从记忆库里取哪些内容出来记忆如何更新、发生冲突时怎么处理、何时遗忘这三个问题拆开看每一项都不复杂。难的是把它们组合成一个在真实流量下稳定工作的系统。接下来我把我们从公开分享和团队实践里提炼出来的记忆骨架拆给你看然后再讲每个部分的落地细节。2. 拆解工业级 Agent 的记忆骨架四类存储各司其职2.1 工作记忆会话内必须随叫随到的状态如果只看那些技术分享的标题很多人会以为大厂记忆系统就是一个巨型向量数据库。实际拆开看成熟方案普遍把记忆分成了多层工作记忆就是最基础的一层。工作记忆对应的是当前 Session 内部的临时状态。例如用户选了哪个商品、填写到哪一步、当前页面的筛选条件这类数据生命周期只在这个会话内有效。它的特点是“高频写入、高频读取、短生命周期”最适合的载体是内存键值库直接以 session_id 为 key 存一份可序列化的状态对象设置 TTL 让它随会话超时自动过期。这一层最关键的取舍在于到底是全量保留原始对话还是保留提炼后的状态全量保留的缺点我们在上一节已经说了上下文膨胀。提炼状态则需要额外的抽槽和更新逻辑。我们用的方案是混合结构上下文里始终放两块内容——一块是“会话摘要”另一块是“最近 N 轮原始对话”。摘要每隔固定轮数由模型生成一次把历史信息压缩成两三句话最近几轮原文则保证当下动作的连贯性。这样做的好处是既不丢关键状态又把上下文控制在稳定长度附近。像“用户已经确认了收货地址是 xxx”“当前审批状态是待财务确认”这类信息放进工作记忆后后续每一步 Agent 的动作都可以直接读取不用再去上下文中翻。2.2 语义记忆把“可复用的事实”变成可检索的向量工作记忆解决的是“当下”语义记忆解决的是“长期复用”。它存储的是跨会话仍然有效、且需要通过语义相似度来召回的信息比如用户的长期偏好、项目的背景约束、某个领域的关键结论。这一层的标准实现是 embedding 加向量检索。但我想强调一点向量数据库不是语义记忆的核心写入口径才是。很多人一上来就往向量库里灌原始对话记录结果检索时召回一堆“今天天气不错”“嗯嗯好的”这类屁话对 Agent 一点帮助没有。语义记忆库只应该接收“值得长期记住的信息”。判断标准可以很简单这条信息如果下次会话还用得上并且属于可复用的事实才写进去。举个例子“用户偏好简洁回答”值得写“用户今天问了退货政策”不值得写后者去查知识库就行。每条语义记忆还应该带一些辅助权重重要性、访问次数、创建时间。这些字段在召回排序阶段非常有用后面我会给出完整的打分公式。2.3 结构化事实记忆精确键值比向量检索合适语义记忆适合做模糊回忆但有一类信息用向量检索反而是浪费——精确键值对。比如用户当前的会员等级、默认结算账号、所属组织架构、功能开关状态这些数据用一条查询语句直接命中就可以了做 embedding 有什么意义呢你不可能期望用向量检索去回答“这个用户是不是 VIP”这种二值问题。结构化事实记忆通常以用户维度为主键存成关系表或键值结构。它的特点是强一致、可覆盖、可版本化。用户修改了某条资料直接 UPDATE 记录旧值要么被覆盖要么归档保留做审计。这类数据如果塞进语义记忆库会发生一件很尴尬的事用户明确说不喜欢 A 了但老记忆里还躺着“用户喜欢 A”向量检索时新老记录同时被召回模型就懵了。所以我们的经验是能精确匹配的事实永远不要丢给向量检索。语义检索是兜底方案结构化检索是确定性方案两者必须并存。2.4 它们如何配合一趟完整的记忆生命周期说了这么多层它们不是各自为战的。我把一次完整用户会话里记忆系统的参与过程梳理一下用户进入会话系统加载该用户的结构化事实记忆比如身份、偏好设置注入初始上下文。对话进行中每轮消息到达后先做一轮判断属于会话内临时状态的更新工作记忆属于用户画像或关键偏好的写入结构化事实表属于可复用的经验、项目背景、决策记录判断重要性后写入语义记忆库。每一次模型调用之前先走一次记忆检索结构化查询加向量召回加最近活跃记忆把命中的记忆片段拼成一个“记忆上下文块”和当前对话一起交给模型。会话结束或超时工作记忆压缩生成会话摘要有长期价值的沉淀到语义记忆库其余直接丢弃。这一套流程跑通之后Agent 每一次决策看到的都是“精选后的记忆”而不是堆成山的原文。这也是大厂方案里最值得抄的一点记忆系统不是一股脑地存而是分层的、带取舍的、有生命周期的。3. 落地第一步记忆数据模型与存储选型深度对比3.1 存储选型别为了向量检索引入重型组件拆解完机制开始谈落地。很多团队走到这一步容易犯一个错误一听说要上语义记忆立刻去部署一套专用向量数据库。我的建议是先看看你手上已经有什么。我整理了一张选型对比表覆盖了最常见的四类存储方案方案擅长场景延迟运维成本注意点内存键值库会话状态、热记忆缓存、结构化事实微秒到毫秒级低数据容易丢需要持久化配合关系库加向量插件全量长期记忆、需要多条件过滤毫秒级低到中索引类型和写入频率需要调优专用向量检索服务超大规模向量召回毫秒级高不适合复杂结构化过滤和关联查询倒排索引检索引擎文本检索加复杂过滤、标签组合毫秒级中到高存储开销比较大Lucene 类方案要定期维护结论很明确如果你的记忆量级在千万条以下关系库加向量插件就是最平衡的选择高并发场景下可以把热记忆放到内存键值库做缓存关系库存全量只有当你的记忆规模到了亿级别、而且检索条件非常复杂时再去考虑专用向量服务或倒排引擎。我见过一个反面案例某个项目为了做记忆功能引入了全套搜索集群结果线上数据量就只有几万条整个集群大半时间在空转光运维工时就吃掉了一个人。记住存储选型是匹配规模不是越重越好。3.2 记忆记录的数据模型一张表的核心字段设计下面给出一份可以直接落到关系库里的核心表结构。我基于实际项目打磨过很多版这份是目前比较稳的CREATE TABLE memory_records ( id BIGSERIAL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, session_id VARCHAR(64), memory_type VARCHAR(16) NOT NULL, -- semantic / fact / summary content TEXT NOT NULL, -- 给模型读的自然语言文本 importance FLOAT DEFAULT 0.5, access_count INT DEFAULT 0, embedding VECTOR(768), metadata JSONB, -- 给代码读的过滤条件 created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now(), archived_at TIMESTAMPTZ ); CREATE INDEX idx_memory_user_type ON memory_records(user_id, memory_type); CREATE INDEX idx_memory_active ON memory_records(user_id, archived_at);逐字段解释一下核心设计的理由memory_type区分记录的类型事实走覆盖逻辑语义走相似度合并逻辑摘要走归档逻辑。content存的是给模型看的自然语言文本比如“用户结算账号尾号为 8890”。不要存结构化对象模型读起来吃力。metadata是给代码读的比如{topic: payment, source: session_20240115}用于精确过滤。importance和access_count是重排阶段的原料重要性决定基础权重访问次数反映活跃度。archived_at做软删除。冲突消解和遗忘归档都改这个字段而不是物理删行方便排查问题。一份典型记忆记录长这样{ user_id: u_10086, memory_type: fact, content: 用户结算账号尾号为 8890, importance: 0.9, metadata: {topic: payment, source: session_20240115}, embedding_version: embed-v1 }这里有个细节很容易被忽略embedding_version字段。你的 embedding 模型不可能永远不变一旦升级新旧向量分布就不同混在一起检索会出现召回的记录看起来相关、实际不在同一个语义空间里的情况。更稳妥的做法是写入时记录版本查询时按版本过滤必要时触发历史记忆的重算任务。3.3 为什么要把向量召回和 SQL 过滤分开刚开始搭的时候我一度想把所有逻辑都交给向量库用户问什么直接对整个记忆库做向量搜索让相似度来排序。结果就是乱。比如用户问“我上次的支付信息”向量检索确实能召回到几条关于支付的记忆但里面可能混着“支付渠道偏好”“支付失败处理流程”“某次订单的支付时间”等完全不相关的记录因为它们在语义空间里都离“支付”很近。正确姿势是先用 SQL 把候选集缩到一个可控范围再在这个范围内跑向量排序。顺序是 filter → recall → rerank。比如“找出用户最近 7 天内关于支付的记忆”先执行SELECT * FROM memory_records WHERE user_id u_10086 AND archived_at IS NULL AND created_at now() - interval 7 days LIMIT 500;然后在返回的 500 条候选里做向量相似度排序取前 5 条。这样做的好处有两个一是把全库搜索变成小范围排序性能稳定二是过滤条件是确定性的不会像“纯向量召回”那样把语义相近但不该出现的内容捞进来。4. 写入与召回记忆系统的两个核心管线的设计4.1 写入管线什么内容值得进记忆库记忆系统的质量一半取决于写入策略。如果垃圾信息都进去了后续召回和 Agent 输出质量都是灾难。我设计的写入管线分成两步规则过滤加重要性打分。规则过滤是为了挡掉明确不该入库的内容包括明显的隐私数据身份证号、银行卡号、密码、一次性指令“帮我查下今天的天气”、纯寒暄“好的”“谢谢”。这一步不需要模型用关键词规则加正则就能挡掉大部分。重要性打分可以用一条简单的规则函数实现我这里是结合经验和线上效果调出来的def score_importance(content: str, memory_type: str) - float: score 0.5 # 默认基线 if memory_type fact: score 0.3 # 结构化事实天然高于语义记忆 if 偏好 in content or 喜欢 in content or 不要 in content: score 0.3 # 用户明确表达的偏好 if 结论 in content or 决定 in content or 确认 in content: score 0.2 # 决策性信息 if len(content) 4: score - 0.3 # 太短的碎片信息大概率无价值 return min(score, 1.0) WRITE_THRESHOLD 0.6超过阈值的才进入写入流程。这个阈值的设定直接影响写入规模和记忆质量阈值设太高好东西也丢了设太低垃圾满天飞。我们一开始设为 0.5结果库涨得飞快命中率反而下降后来调到 0.6 才平衡。写入时还有一个容易被忽略的点合并写而不是每次都插新行。每次写入之前先跑一次相似度碰撞如果新内容与已有记忆的向量相似度超过 0.92 且类型一致执行“触碰更新”——把已有记录的 access_count 加一、更新 updated_at不发新记录。这个机制能避免同一个事实被重复写几十遍。4.2 召回调优多路召回加权重重排只靠向量相似度召回效果是很单薄的。真实场景里用户问“我上次那个账号的尾号是多少”时仅仅做一次 query embedding 再搜 TopK结果可能不如预期因为用户表述太模糊了缺少实体词。我采用的召回方案是多路召回向量召回对当前 query 做 embedding从候选池里取 TopN。关键词命中如果 query 里出现了记忆 metadata 里的标签词、内容里的实体词直接命中一组记录。最近活跃用户最近几次会话里被访问过的记忆记录直接取一组出来。然后把三路结果合并去重进入重排阶段。重排公式我调试过很多版目前用的是[ score 0.5 \times sim 0.3 \times recency_norm 0.2 \times importance ]其中sim是向量相似度recency_norm是时间衰减系数importance是写入时的权重。时间衰减用指数形式import math def recency_norm(updated_at_ts: float, now_ts: float, tau: float 7 * 86400): delta max(0, now_ts - updated_at_ts) # 秒 return math.exp(-delta / tau)tau取 7 天意味着 7 天前没被触碰的记忆时间权重衰减到约 0.37一个月前的基本就不太可能靠时间因素进入 TopK。这个设计模拟人的记忆曲线高活跃记忆会被反复顶上冷记忆慢慢沉底。另外一个非常关键的细节是 query 改写。用户真实输入通常很口语化比如“上次那个”直接拿去做 embedding 效果很差。可以在检索前加一层改写器让模型把短 query 扩展成一句完整描述或者生成两三个变体分别召回再合并。我们实测下来简单加一个规则改写词表命中率能提升七八个百分点。4.3 更新与遗忘记忆没有“永久”这回事长期跑下来你会发现记忆库最麻烦的不是写入而是“改”和“删”。用户昨天说喜欢 A今天说 A 不好这两条记忆如果同时存在Agent 一定输出矛盾内容。我们的做法是在写入管线里加一个冲突消解逻辑当新记忆的语义包含否定或纠正信号时先在旧记忆库里检索相关记录命中后把旧记录置为 archived而不是删除。这样做可以保留历史痕迹同时让检索时不再命中旧版本。def resolve_conflict(existing: MemoryItem, new_content: str): negative_markers [不要, 不喜欢, 取消, 改为, 更正] if any(m in new_content for m in negative_markers): existing.archived_at time.time() # 同时记录 superseded_by 指向新记忆 id遗忘策略是另一个容易被忽视的设计。记忆不是越多越好长期不命中的低价值记忆留着只会增加检索噪音。我建议定期跑一个归档任务连续 90 天没有命中记录、且 importance 低于 0.7 的记忆统一置为 archived。如果日后用户旧事重提仍可以通过关键词检索把 archive 状态改为常驻。5. 线上跑起来的实战问答命中率、准确率与可观测性5.1 先定指标别用“感觉”评估记忆系统记忆系统上线后怎么评估很多人第一反应是看用户反馈太慢了。要把记忆系统当作一个独立组件来做指标监控我重点盯四个指标指标计算口径说明记忆命中率检索返回的 TopK 中实际被模型引用进回答的比例衡量召回结果是否真的有用引用准确率模型引用记忆内容与实际事实一致的比例需要人工抽检或事后核验写入有效率写入记忆库后未来 7 天内被再次命中的比例衡量写入阈值是否合理检索延迟记忆检索服务的 p95 响应时间必须压到 50ms 内可接受这四个指标一上监控记忆系统的健康度就变成数字了。写入有效率尤其能说明问题如果大量写入的记忆在 7 天内都没被再次命中说明写入规则太宽松了需要提高阈值或收紧类型过滤。5.2 五个坑我挨个踩过第一个坑是 embedding 模型版本漂移。有一次我把 embedding 服务升级了一个版本然后线上立刻出现召回质量下降明明完全相同的 query命中结果完全变了。排查之后发现是向量空间变了旧记忆的向量和新查询向量不在同一个坐标系里。从那以后所有记忆记录一律带 embedding_version检索时过滤版本升级后跑离线重算任务。第二个坑是短 query 召回差。用户问“那个呢”“上次说的”直接 embedding 基本什么都召不回来。后来加了 query 改写器模型先行把短 query 扩写成完整描述习惯表述“上次说的”之前先查一下用户当前会话的主题改写效果才上来。这个组件很值得做代价小收益大。第三个坑是写入风暴。上线初期每一轮用户消息都触发一次记忆写入判断遇到活跃用户一晚上能写几千条记录关系库写入压力直接拉满。后来我们把写入改成异步队列批量合并对同一用户同时只允许一个写任务加上相似度去重写入量降了一个数量级。第四个坑是召回矛盾内容。用户在某次会话里明确说“以后不要默认发邮件改发短信”但因为老记忆和他之前的偏好记录都还在检索时两条同时命中Agent 回答就自相矛盾。这个问题必须靠冲突消解机制解决检索时对 archived 记录做排除同时在写入时主动触发消解动作。第五个坑是 embedding 服务抖动导致主链路崩溃。曾经有一次 embedding 服务超时整个主流程抛异常用户消息直接处理失败。现在所有记忆检索都包了一层降级逻辑检索失败返回空列表让模型只基于当前对话回答同时记录一条告警日志。记忆系统是加分项永远不能因为加分项把主流程拉没了。5.3 链路追踪与降级设计记忆系统的链路追踪太重要了不然出了问题根本不知道是哪一步。我们每个请求都会带 trace_id检索结果写入结构化日志{ event: memory_hit, trace_id: trace_8f3a1b, user_id: u_10086, retrieved_count: 5, sources: [vector, recent, keyword], latency_ms: 45 }有了这份日志用户说 Agent 回答不对的时候我们能快速回溯到底有没有命中记忆命中了几条来源是向量还是关键词延迟是正常还是异常如果没有可观测性排查记忆问题就像闭着眼找针。降级策略再强调一遍检索失败走空结果路径写入失败走日志队列绝不因为记忆组件的问题把用户请求放在主线程上干等。6. 小而完整的记忆服务骨架代码6.1 代码骨架写入、召回、合并一把梭理论讲完直接上一个可以改来用的 Python 骨架。为了不引入额外依赖这里只用了标准库和接口抽象外部组件的具体实现你可以按自己的环境替换。import time import uuid from dataclasses import dataclass, field dataclass class MemoryItem: user_id: str content: str memory_type: str semantic # semantic / fact / summary importance: float 0.5 created_at: float field(default_factorytime.time) updated_at: float field(default_factorytime.time) access_count: int 0 archived_at: float | None None class MemoryService: def __init__(self, db, vector_store, embed_fn, write_threshold0.6): self.db db self.vector_store vector_store self.embed_fn embed_fn self.write_threshold write_threshold def _score_importance(self, content: str, memory_type: str) - float: score 0.5 if memory_type fact: score 0.3 for marker in [偏好, 喜欢, 不要, 结论, 确认]: if marker in content: score 0.15 break if len(content) 4: score - 0.3 return min(score, 1.0) def add(self, user_id: str, content: str, memory_type: str semantic): importance self._score_importance(content, memory_type) if importance self.write_threshold: return None existing self._find_similar(user_id, content, threshold0.92) if existing is not None: existing.access_count 1 existing.updated_at time.time() self.db.save(existing) return existing item MemoryItem( user_iduser_id, contentcontent, memory_typememory_type, importanceimportance, ) embedding self.embed_fn(content) self.db.insert(item) self.vector_store.upsert(item.id, embedding, {user_id: user_id}) return item def retrieve(self, user_id: str, query: str, top_k: int 5): # 1. SQL 过滤候选集 candidates self.db.filter_by_user(user_id, limit500) if not candidates: return [] # 2. 多路召回 vec_hits self.vector_store.search(self.embed_fn(query), top_ktop_k * 2) recent_hits self.db.recent_hits(user_id, top_ktop_k) # 3. 合并去重 merged self._merge(vec_hits, recent_hits) # 4. 重排 ranked self._rerank(query, merged) return ranked[:top_k] def _rerank(self, query: str, items: list): query_vec self.embed_fn(query) scored [] for item in items: emb self.vector_store.get_embedding(item.id) sim self._cosine(query_vec, emb) recency recency_norm(item.updated_at, time.time()) score 0.5 * sim 0.3 * recency 0.2 * item.importance scored.append((score, item)) scored.sort(keylambda x: x[0], reverseTrue) return [item for _, item in scored]代码里缺的函数像_find_similar、_merge、_cosine都是直白的功能接入具体组件时补上就行。这个骨架的作用是给你一个完整的思维框架写入要过滤打分检索要分阶段召回重排更新是触碰而不是乱插。6.2 接入 Agent 主流程的接入点记忆服务不是独立跑着玩的必须接进 Agent 的调用链里。接入位置非常固定模型调用前取当前用户的问题调用retrieve()拿到记忆列表拼成一个“记忆上下文块”注入 system prompt。模型调用后根据模型的最终动作和用户反馈异步调用add()写入新记忆同时触发冲突消解。注入 prompt 的格式我试过很多种目前最稳定的是以下是关于用户的记忆片段请参考但不编造 - [mem-1] 用户结算账号尾号为 8890 - [mem-2] 用户偏好简洁回答避免冗长解释每条记忆附带mem-id这样模型在回答中可能引用对应 id方便你做归因分析。别小看这个设计它让“记忆是否真的被使用了”这个问题可追踪了。6.3 运行前要补的生产化清单骨架代码只是起点上生产之前至少还要补齐这些东西写入异步化add()不能同步阻塞主链路丢到消息队列里批量消费。用户隔离多租户场景下user_id 之上还要有 tenant_id所有查询强过滤防止数据串味。隐私合规记忆内容里可能包含敏感字段入库前先做脱敏查询时按权限控制出参。监控告警命中率低于阈值要告警embedding 服务延迟升高要告警写入队列堆积要告警。一键降级做一个总开关记忆服务异常时可以直接关闭让 Agent 退化为无记忆模式保证业务不中断。最后再分享一个我自己的习惯每次想给记忆系统加新规则之前先到线上日志里翻出三五个真实失败案例对着案例改而不是凭空设计规则。拆解大厂方案的时候也一样所有机制看懂了还不算数落到自己的数据和失败样本里才算是真的会用了。记忆系统这东西从“会画架构图”到“线上跑得稳”之间隔的不是想象而是一轮轮的线上打磨。