尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

为AI助手装上三层记忆:跨会话记忆的工程实践

为AI助手装上三层记忆:跨会话记忆的工程实践 如果你的 AI 助手还是“一聊就忘”每次新会话都要用户重新交代背景那问题不在模型本身而在于没有跨会话记忆。跨会话记忆解决的是很具体的痛点同一个用户在不同会话里说过的关键信息能不能在下次对话时被准确想起。我落地这套方案时给 AI 助手装了“三层记忆”工作记忆、情景记忆、语义记忆。这篇文章就是完整拆解适合已经能跑通基础对话、下一步想把 Agent 做得更像长期助理的开发者。先给结论三层记忆里工作记忆负责当前这一轮能不能接上话情景记忆负责“上次聊到哪”语义记忆负责“用户长期偏好是什么”。三层不能混在一个表里也不能全塞进上下文。真正好用的记忆系统不是把更多资料检索出来堆给模型而是让 Agent 学会在合适时机回忆、在合适时机遗忘。这句话听起来简单实际做起来要解决写入、召回、压缩、更新、冲突一整套问题。下面按落地顺序拆开讲。1. 先想清楚跨会话记忆到底在解决什么问题1.1 一聊就忘的本质状态没有持久化大语言模型本身是无状态的。用户每次发消息模型拿到的是“系统提示 当前输入”它对之前聊过的内容一无所知。即使你用的是上下文窗口很大的模型也只是把历史消息塞进本次输入一旦请求结束这些状态就没了。很多人第一反应是“那我多传几轮历史不就行了”。短会话可以但连续用下来问题非常明显Token 成本上升、响应变慢、无关历史干扰判断。更关键的是用户开一个新会话后你根本不知道“该带哪段历史”。这时候就需要把记忆单独抽出来管理。所谓跨会话记忆本质是把对话中值得保留的信息做持久化和结构化处理在下一次对话开始前按需把相关部分放回模型上下文。它不是“把所有聊天记录原样搬回来”而是经过筛选、压缩、重组之后的一小段有效信息。这个理解很重要。很多从零开发的工程会卡在“都存了为什么还是想不起来”根源就是没有区分存储和读取策略。1.2 三层记忆不是三个数据库是三种读取策略我接触过不少 Agent 记忆方案最后沉淀下来的三层记忆对应的是认知记忆里常用的分类。第一层工作记忆Working Memory。保存当前会话最近的消息保证模型能接得住上下文。第二层情景记忆Episodic Memory。保存一段对话的摘要、关键决策、遗留任务。它描述的是“发生过什么”。第三层语义记忆Semantic Memory。保存用户偏好、长期事实、稳定的画像信息。它描述的是“用户是谁、习惯是什么”。下面这个表格可以直接作为设计参考记忆分层保存内容更新时机读取时机工作记忆最近 N 条对话、本轮暂存信息每轮对话后追加每次请求情景记忆历史会话摘要、决策、待办会话结束 / 触发摘要新会话开始、话题相关时语义记忆用户偏好、画像、禁止项用户明确或行为确认每次请求按需检索在很多实现里工作记忆可以直接放在内存或 Redis情景记忆适合放文档型存储加摘要字段语义记忆适合用向量库或键值存储。但技术选型不是重点重点是读取策略要有区分度。如果三层全都无脑灌进 Prompt那和三年前把所有聊天记录拼接起来没有本质区别。1.3 记忆系统不是把更多东西检索出来而是让 Agent 学会“回忆”“记忆系统不是把更多东西检索出来而是让 Agent 学会回忆”我设计时反复用这句话来校准方案。它说的不是文学修辞而是工程取舍。如果你每次请求都检索二十条记忆然后全塞给模型模型大概率分不清哪些重要。真实情况里一条“用户上周说项目要赶在月底上线”比十条“用户看过某篇文章”更有价值。所以记忆系统要做两件事精确召回只挑和当前问题相关的记忆。压缩表达把多条事实合并成一句完整自然的话。我常用的验证例子是这样。用户在会话 A 说“我是做量化策略的目前主要用 Python。”会话 A 结束时系统生成情景摘要。用户会话 B 第一句话是“帮我看看我现在适合学什么框架。”如果记忆只检索出一条“用户是量化策略开发者使用 Python”模型就能给出合理建议。如果检索出来的是十条零散对话记录模型可能在无关内容里打转。这也是为什么我建议把“记忆召回结果”单独打到日志里。你要能观察到模型最终看到的是什么记忆否则没法判断是模型不会用还是记忆本身就没对上。2. 三层记忆的分工和触发条件2.1 工作记忆当前会话能不能接上话工作记忆是最基础的一层也是很多人容易做过头的一层。它的职责只有一个让模型在当前会话里理解刚才聊了什么。通常做法是把对话历史按时间顺序放入上下文但要注意控制轮数和 Token。我一般设置一个上限最多保留最近 30 条消息如果总长度超过阈值就丢弃更早的普通消息优先保留系统消息、函数调用结果和最后几条关键对话。为什么优先保留这些因为系统消息通常包含用户目标或约束函数调用结果经常是任务状态最后几条对话是当前上下文主线。普通寒暄被挤掉影响不大。给个例子。用户先问“杭州有什么适合办活动的园区”又问“从城西出发方便吗”。如果工作记忆只剩后一句模型就不知道“城西出发”是要去哪个园区。但如果保留最后三轮这个问题就能接上。工作记忆一般不需要写数据库一个以 conversation_id 为 key 的队列内存结构就够了。多实例部署时才需要放到 Redis 这类共享存储。2.2 情景记忆记住“上次我们聊到哪”情景记忆解决的是跨会话的“情节连续性”。它不需要保存完整聊天记录而是要保存一段摘要。摘要里至少包含用户这次会话的核心目标已经确认的决策尚未解决的待办出现过的重要数据比如金额、日期、版本号用户当时的状态或情绪背景但只保留客观描述我采用的触发方式是“会话结束”和“Token 阈值”双触发。如果用户明确说了结束语就立刻生成摘要。如果聊天太长还没结束就先根据最近窗口生成一份临时摘要后续再更新。这里要特别注意摘要本身也是 LLM 生成的会丢细节。所以我不只存摘要还会把“关键事件”或“结构化结果”单独存一份。比如用户说“地址是杭州市西湖区”业务是订餐助手这个地址就必须作为字段存进用户档案而不是期待摘要在压缩时保留它。如果一个会话触发了多次摘要不要每次都从头生成。更稳的做法是在上一份摘要基础上做增量追加把新发生的关键事件并进去。这样既能控制摘要长度也能避免把早期信息覆盖掉。2.3 语义记忆用户偏好和长期事实语义记忆是三层里最接近“人设”的一层适合存这些信息用户职业、公司、技术栈语言偏好、称呼偏好常用约束比如“不要周末发消息”对某个工具、模型、方案的偏好明确说过的禁止事项写入时机比想象中保守。不是每一句话都值得写入语义记忆。我倾向用三种判断用户明确说“记住以后都这样”用户纠正过助手而且纠正内容具有持续性同一个需求或同一类表述重复出现两次以上如果没有明确的写入逻辑先不要用模型自由抽取否则会写入大量噪声。比如用户随口说“今天天气不错”模型可能会写成“用户喜欢晴天”这明显是过度推导。每条语义记忆建议控制在 20 字以内能带时间就带时间。例如“2024 年 12 月决定从 Python 迁移到 Go”比“用户用 Go”更准确。长期记忆不是越多越好而是越准确越好。3. 代码落地从零搭一个可运行的三层记忆模块3.1 先定义统一接口不急着选数据库很多项目一上来就选定 MongoDB、pgvector、Chroma结果原型开发时一直卡在环境配置上。我的建议相反先定义一个抽象的 MemoryStore 接口实现一个能用 JSON 文件落地的版本跑通闭环后再替换成正式数据库。一个最小接口只需要四个核心能力from abc import ABC, abstractmethod from typing import Any class MemoryStore(ABC): abstractmethod def load_working_context(self, conversation_id: str) - list[dict]: 读取当前会话的最近消息 abstractmethod def save_working_context(self, conversation_id: str, messages: list[dict]) - None: 保存当前会话消息 abstractmethod def load_episodic_summary(self, user_id: str) - list[str]: 读取用户的历史会话摘要 abstractmethod def save_episodic_summary(self, user_id: str, summary: dict) - None: 保存一份会话摘要 abstractmethod def load_semantic_memories(self, user_id: str, query: str, top_k: int 3) - list[str]: 按当前问题召回用户长期记忆 abstractmethod def save_semantic_memory(self, user_id: str, content: str, metadata: dict[str, Any]) - None: 写入一条用户长期记忆先定义接口有个额外好处后续不管是换 SQLite、Redis 还是 Chroma业务调度代码完全不用动。原型阶段不要把接口拆得太细方法过多反而难跑通。3.2 工作记忆先用显式列表过度设计反而卡住自己工作记忆最简单的实现就是为每个会话保存一个消息列表。我直接用一个 JSON 文件存单用户数据调试时打开文件就能看到所有状态。import json import os from datetime import datetime class JsonMemoryStore(MemoryStore): def __init__(self, base_dir./memory_store): self.base_dir base_dir os.makedirs(base_dir, exist_okTrue) def _user_path(self, user_id: str) - str: return os.path.join(self.base_dir, fuser_{user_id}.json) def _ensure_user_file(self, user_id: str) - dict: path self._user_path(user_id) if not os.path.exists(path): return {conversations: {}, summaries: [], semantic_memories: []} with open(path, r, encodingutf-8) as f: return json.load(f) def _write_user_file(self, user_id: str, data: dict) - None: path self._user_path(user_id) with open(path, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) def load_working_context(self, conversation_id: str) - list[dict]: user_id conversation_id.split(_)[0] data self._ensure_user_file(user_id) return data[conversations].get(conversation_id, []) def save_working_context(self, conversation_id: str, messages: list[dict]) - None: user_id conversation_id.split(_)[0] data self._ensure_user_file(user_id) data[conversations][conversation_id] messages self._write_user_file(user_id, data)注意这里我把用户 ID 编进会话 ID 里原型阶段够用。实际项目中更建议把 user_id 作为单独字段传给所有方法。这个版本只适合开发和测试真正上线到多用户并发时JSON 文件会存在写入竞争。届时要切换成 SQLite 或 Redis。但在跑通闭环之前不要花太多时间在存储选型上。3.3 摘要记忆历史变长时的自动压缩情景摘要不能等到用户关会话才生成因为很多用户不会说“结束了”他们会直接开新会话。更稳的方式是每次保存消息后估算当前会话的 Token 数量超过阈值就触发一次压缩。Token 数在所有模型上没有一个通用精确值所以这里我用估算函数代替真实项目里建议用模型自带的 tokenizer。def estimate_tokens(text: str) - int: # 中文场景的简化估算 # 真实项目建议用模型 tokenizer 精确统计 return max(1, len(text) // 1) MAX_WORKING_TOKENS 4000 MAX_WORKING_MESSAGES 40 def maybe_summarize(store: JsonMemoryStore, user_id: str, conversation_id: str, llm_summarize_fn) - None: messages store.load_working_context(conversation_id) total sum(estimate_tokens(m.get(content, )) for m in messages) if total MAX_WORKING_TOKENS or len(messages) MAX_WORKING_MESSAGES: summary llm_summarize_fn(messages) store.save_episodic_summary(user_id, { summary: summary, conversation_id: conversation_id, created_at: datetime.now().isoformat() }) # 保留最近 10 条消息其余压缩进摘要防止会话窗口继续膨胀 messages messages[-10:] store.save_working_context(conversation_id, messages)这里的 llm_summarize_fn 就是现有的模型调用封装。摘要生成是一次额外模型调用会带来成本和延迟所以要放在消息保存后的后台任务里不要让用户在对话中等待。3.4 语义记忆向量检索加用户级存储语义记忆的召回需要“相关度判断”。最简单可靠的做法是向量检索把记忆条目预先生成向量用户提问时生成问题向量查相似度取前 top_k 条。在原型里甚至可以先用关键词相似度模拟。把用户问题里的词和记忆条目过滤一遍得分高的才返回。等数据量大了再迁移到向量库。迁移时只需要改动 save_semantic_memory 和 load_semantic_memories 两个方法的内部实现对外接口不用动。class SimpleSemanticMemory: def __init__(self, store: JsonMemoryStore, embed_fnNone): self.store store self.embed_fn embed_fn def save(self, user_id: str, content: str, metadata: dict) - None: data self.store._ensure_user_file(user_id) data[semantic_memories].append({ content: content, metadata: metadata, created_at: datetime.now().isoformat() }) self.store._write_user_file(user_id, data) def recall(self, user_id: str, query: str, top_k: int 3) - list[str]: data self.store._ensure_user_file(user_id) candidates data[semantic_memories] if not candidates: return [] # 先做简单关键词打分等数据量上来再替换成 embed_fn 向量召回 scored [] query_tokens [w for w in query if w.strip()] for mem in candidates: score sum(1 for w in query_tokens if w in mem[content]) scored.append((score, mem[content])) scored.sort(keylambda x: x[0], reverseTrue) return [content for score, content in scored[:top_k] if score 0]我把 embed_fn 参数预留出来但注释里说明当前先用关键词打分。这样读者本地没有 embedding 服务也能跑通等条件允许再升级。关键词打分对“Python”“量化策略”这类词很有效对语义相近但表达不同的查询就开始失效了那时就必须上向量。4. 把记忆注入 AI 助手的请求流程4.1 请求到达时按优先级装配记忆一个请求进来我会按这个顺序组装系统提示和上下文系统提示助手人设语义记忆用户画像摘要情景记忆与该话题相关的历史摘要最近对话工作记忆当前用户输入这里用一段伪代码说明def build_prompt(user_id: str, conversation_id: str, user_input: str, store: MemoryStore): working store.load_working_context(conversation_id) semantic store.load_semantic_memories(user_id, user_input, top_k3) episodic select_relevant_summaries(store.load_episodic_summary(user_id), user_input) memory_block 用户长期记忆\n \n.join(semantic) if semantic else episode_block 历史会话摘要\n \n.join(episodic) if episodic else system_prompt SYSTEM_PROMPT \n\n memory_block \n\n episode_block messages [ {role: system, content: system_prompt.strip()}, *working, {role: user, content: user_input} ] return messages这个优先级值得解释。语义记忆放最前面是为了先在模型头脑里建立“这个用户是谁”的基线。情景摘要放中间是为了给出任务背景。工作记忆保持原顺序避免把对话时间线打乱。最后才是当前输入。如果你用的是支持自动拼接历史消息的 SDK仍然建议把记忆内容放进 system prompt而不是混在用户消息里。记忆是事实背景不是当前发言放进 system prompt 更能让模型按“长期事实”对待而不是当成对话内容去回引。4.2 模型回复后再决定写入哪些记忆组装只是读路径的一半写路径同样重要。不要在每次回复后都把整段对话塞进记忆而是做一次“记忆写入判断”。写入规则可以分两类。一类是明确规则比如用户说“记住 / 以后都 / 老是忘”命中后就写入。另一类是模型判断把最近几轮对话交给模型让它提取“可长期记忆的用户偏好、事实、待办”输出 JSON。MEMORY_EXTRACT_PROMPT 请从最近对话中提取需要长期记住的信息。 输出 JSON 数组每项包含 type 和 content。 type 可以是user_preference / personal_fact / pending_task / prohibition。 如果没有值得长期记忆的信息输出 []。 def after_reply(user_id: str, conversation_id: str, store: MemoryStore, extract_fn) - None: recent_messages store.load_working_context(conversation_id)[-6:] extracted extract_fn(MEMORY_EXTRACT_PROMPT, recent_messages) for item in extracted: if should_deduplicate(item[content], store.load_semantic_memories(user_id, item[content], top_k5)): continue store.save_semantic_memory( user_id, item[content], {type: item[type], conversation_id: conversation_id} )去重逻辑要单独抽一个函数。最简单的是字符串相似度比较比如用户说“我喜欢 Python”十次系统不应该存十条一样的记录。流程上先查已有记忆如果相似度超过阈值就跳过写入或更新原条目的“最后确认时间”。4.3 用日志和两轮会话验证效果开发阶段建议在三个位置加日志写入日志记录某用户新增了哪条语义记忆召回日志记录用户输入后召回了哪些记忆组装日志记录最终送进模型的 prompt 里包含多少记忆内容验证流程很简单就测两轮会话。第一轮用户 A 新建会话发消息“我是做量化策略的主要用 Python最近在选向量数据库。”系统回复后日志里应新增语义记忆或情景摘要。第二轮用户 A 再次新建会话发消息“帮我继续看向量数据库上次说要从哪里开始”此时看召回日志应当能召回“用户做量化策略、使用 Python、最近在选向量数据库”相关记忆。如果第二轮模型回答得像失忆顺着日志一步步查基本能在五分钟内定位问题。我第一次跑通这个闭环时问题就出在会话 ID 不一致。第一轮生成的会话 ID 和第二轮差一个分隔符写库和读库根本不在同一个 key 上日志一打就露馅了。5. 参数、成本与稳定性控制5.1 Token 阈值、检索数量和相似度阈值怎么配直接给一组可用的初始值但落地时以你的模型和环境为准。参数初始值影响MAX_WORKING_MESSAGES30太小容易忘前文太大成本高MAX_WORKING_TOKENS4000控制上下文长度中文场景常用范围SUMMARY_TRIGGER_MESSAGES40触发摘要的会话消息条数SEMANTIC_TOP_K3每条请求带入的长期记忆条数SIMILARITY_THRESHOLD0.6低于该分数的记忆不召回MEMORY_EXPIRE_DAYS30超过未触达的语义记忆可归档经验是top_k 不要超过 5。记忆不是越多越好模型反而抓不到重点。Embedding 相似度阈值需要先用测试集跑一遍不同模型的分数分布差异很大不要照搬别人的值。成本上也要算清楚。摘要生成和记忆提取每次都是额外模型调用。如果一个助手每天处理十万次请求每次多调用一次摘要成本会明显上升。所以写入判断要尽量用规则先挡掉大部分没必要写的内容只有真正值得记的才走模型提取。5.2 记忆过期、冲突与用户主动更新三层记忆里最容易出问题的不是写入而是更新和冲突。举例用户上个月说“我主要用 Python”这个月说“我全面转到 Go 了”。如果系统只追加不更新下次召回会同时出现“Python”和“Go”模型就会迷惑。处理方式维护一个“记忆最后确认时间”。当新记忆和旧记忆指向同一主题时新记忆写入时把旧记忆标记为过期或直接删除。冲突判定可以先按主题和类型做关键词匹配再用模型辅助判断两段内容是否矛盾。记忆也不是永久有效的。30 天没被召回的语义记忆应该进入归档或由用户确认后删除。长期排不上用场的记忆留在检索库里只会拉低召回精度。遗忘机制不是缺陷而是记忆系统的一部分。5.3 隐私、删除权与边界这里要多写一点因为很多从零到一的项目会忽略。记忆系统存储的是用户真实信息一定要在设计早期就留好“删除”能力。至少要有用户可以查看自己有哪些长期记忆用户可以删除某条记忆用户可以一键清空某个用户目录日志里不能打印完整敏感字段不要把 API Key、密码、身份证号、支付信息写入记忆。即使模型在对话里提到了提取层也应设置关键词黑名单或字段过滤。跨会话记忆持续越久隐私风险越高。这不是上线后补的事而是开发阶段就要考虑的设计约束。如果你的产品面向企业或公开用户最好在记忆写入前给用户明确提示并提供“关闭记忆”的开关。个人开发者的助手可以默认全开但商业产品必须把选择权交给用户。6. 排查指南记忆存了却想不起来从哪开始查6.1 先看写入日志有没有真正落库最典型的“失忆”场景是用户明明在上一轮说了偏好新会话里助手完全不记得。第一步先看写入日志。常见问题包括写入方法被异步任务跳过异常时没有重试embedding 调用失败导致整条记忆写入失败JSON 文件权限不足保存时静默失败写入的 user_id 和读取时不一致排查时不要猜。先查看用户记忆文件或数据库记录确认里面是否有预期内容。没有记录就回到写入链路一层层看看是否走到了 save 方法看调用参数是否正确。6.2 再看召回条件候选项有没有被选出如果记录里有但新会话还是想不起来问题大概率在召回。排查顺序是当前用户 ID 是否正确。多用户系统经常把匿名会话 ID 当成用户 ID导致新旧会话不在同一个命名空间。相似度阈值是否过高。关键词召回时没有得分向量召回时距离太远。是否加了时间过滤。比如只召回 7 天内的记忆但关键记忆是 10 天前的。是否只有“当前会话”记忆缺少“用户级”记忆。工作记忆无法跨会话这是设计边界不是 Bug。如果用的是关键词召回先检查用户问题里是否出现了和记忆条目重合的词。比如记忆是“用户研究量化策略”问题是“帮我看看框架”两者没有直接重合词关键词召回就会漏。这也是关键词阶段最大的硬伤。6.3 看提示词组装模型是否真的看到了记忆召回成功和模型看到是两个环节。有几次我排查到内存记录里召回出了三条记忆但组装 prompt 时因为上下文超长被截断了又或者记忆被放在 system prompt 里但某次请求用了旧版本逻辑没有把 memory_block 拼进去。建议把最终发送给模型的 messages 打到日志里至少打印每条消息的前 200 字。一眼就能看出记忆有没有进去。这一步看起来笨但排查效率最高。如果你发现记忆进入了 prompt模型还是没用上再考虑是 prompt 表达不够明确。记忆内容不要直接丢给模型可以加引导语“以下是关于用户的长期记忆请结合这些信息回答。”明确提示会让模型更愿意使用。6.4 最后查缓存、并发和会话 ID 规则还有一类隐蔽问题写入后立刻读取正常但换个进程或实例就读不到。原因通常是这样使用了进程内内存存储多实例部署时没有共享Redis key 没有设置合理过期策略导致数据倾斜会话 ID 生成规则变化比如之前用时间戳后来改成 UUID缓存层命中旧数据没及时失效这类问题不常见但在“明明代码一样换个环境就失忆”的场景里优先怀疑。我建议在记忆系统的关键方法入口都打上 user_id 和 conversation_id 的日志。排查时先确认这两个 ID 在写入和读取时是否完全一致。很多定位不到的问题最后都是 ID 规则不一致导致的。三层记忆真正落地时最该盯住的不是功能列表而是写入、召回、组装这条闭环是否跑通。先把单用户、单会话跑稳再考虑向量库、并发和批量生产。记忆系统不是一上来就搞复杂而是先让“写了能读回、读了能生效”这个最小闭环成立。踩过几次之后我发现很多“失忆”问题不是工具能力不够而是前置环境、输入字段和调用顺序没有处理干净。希望这套拆解能帮你少走一段弯路。
返回列表