1. 为什么“塞上下文”这条路注定走不到生产环境
1.1 从 Demo 到灰度:上下文窗口和账单先给你上一课
做 Agent 的第一天,没人觉得记忆系统是个问题。我一开始也是这么想的:用户多轮对话嘛,把历史消息全部塞进 prompt,模型自己会看懂上下文。直到某个项目从 Demo 走到灰度,用户会话拉到 60 轮以上,问题像约好了一样集中爆发——账单肉眼可见地涨、响应延迟往上爬、模型开始忘记 20 轮前用户亲口说过的偏好。那个时候我才意识到,上下文窗口不等于记忆,而工业级 Agent 的记忆机制,是需要像数据库一样单独设计和维护的一层基础设施。
先算一笔账。假设一个客服类 Agent 每轮请求的用户输入加上模型输出平均是 800 tokens,如果每一轮都把完整历史拼进上下文,那么第 n 次请求携带的上下文长度就是 800 × n。从第 1 轮到第 N 轮,累计消耗的 token 大概是:
[ \sum_{i=1}^{N} 800 \times i = 800 \times \frac{N(N+1)}{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_factory=time.time) updated_at: float = field(default_factory=time.time) access_count: int = 0 archived_at: float | None = None class MemoryService: def __init__(self, db, vector_store, embed_fn, write_threshold=0.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, threshold=0.92) if existing is not None: existing.access_count += 1 existing.updated_at = time.time() self.db.save(existing) return existing item = MemoryItem( user_id=user_id, content=content, memory_type=memory_type, importance=importance, ) 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, limit=500) if not candidates: return [] # 2. 多路召回 vec_hits = self.vector_store.search(self.embed_fn(query), top_k=top_k * 2) recent_hits = self.db.recent_hits(user_id, top_k=top_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(key=lambda x: x[0], reverse=True) 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 退化为无记忆模式,保证业务不中断。
最后再分享一个我自己的习惯:每次想给记忆系统加新规则之前,先到线上日志里翻出三五个真实失败案例,对着案例改,而不是凭空设计规则。拆解大厂方案的时候也一样,所有机制看懂了还不算数,落到自己的数据和失败样本里,才算是真的会用了。记忆系统这东西,从“会画架构图”到“线上跑得稳”之间隔的不是想象,而是一轮轮的线上打磨。