ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Agent记忆架构实战:会话、短期、长期与遗忘机制四层设计

Agent记忆架构实战:会话、短期、长期与遗忘机制四层设计 1. 从“健忘”说起Agent 记忆架构到底在解决什么问题做 Agent 开发的人几乎都经历过这样一个尴尬时刻你花了一下午调好的对话助手第一轮聊得好好的第二轮它就把你刚才说的名字、偏好、任务目标全忘了。用户骂它“人工智障”你心里清楚——不是模型不行是记忆架构没搭对。我最早做 Agent 项目时也踩过这个坑。当时用最朴素的方式把整段对话历史拼进 prompt 里。头几轮还行聊到第十轮token 直接爆了成本飙升不说模型还开始“胡言乱语”把早期不相关的信息当成当前上下文。后来我才明白Agent 的记忆不是“把历史全塞进去”这么简单它是一套分层、有生命周期、需要主动管理的系统。这套系统大致分四层会话记忆Session Memory、短期记忆Short-term Memory、长期记忆Long-term Memory以及最容易被忽略但极其关键的遗忘机制Forgetting。会话记忆管的是“这一次对话的上下文”短期记忆管的是“最近几轮的任务状态和临时事实”长期记忆管的是“跨会话沉淀下来的用户画像、知识、偏好”而遗忘机制决定了“哪些该丢、哪些该留、什么时候丢”。为什么必须分层因为不同记忆的访问频率、存储成本、时效性完全不同。会话记忆要求毫秒级读取、随对话结束即销毁长期记忆要求持久化、可检索、能跨会话复用但读取慢一点没关系。如果你用一套存储硬扛所有需求要么贵得离谱要么慢得没法用。这就像你家里的东西钥匙放门口挂钩会话今天要用的文件放书桌短期房产证放保险柜长期过期的传单直接扔遗忘。全堆在桌上桌子迟早塌。这篇文章适合谁看如果你正在做 Agent 开发不管是基于 LangChain、AutoGPT 还是自研框架只要涉及到多轮对话、任务连续性、个性化服务这套记忆架构你迟早要面对。我会把四层记忆的设计思路、代码实现、参数计算、踩坑经验全部拆开讲代码用 Python 写但思路是语言无关的。看完你至少能搭出一个“不健忘”的 Agent 骨架。2. 四层记忆架构的整体设计与选型逻辑2.1 为什么是这四层而不是三层或五层先说我为什么把记忆分成这四层。市面上有些方案分三层会话、短期、长期有些分五层加了“工作记忆”“情景记忆”。我实测下来四层是最平衡的再少一层职责会混再多一层维护成本陡增。具体来说会话记忆和短期记忆经常被合并但我坚持分开。原因是它们的生命周期触发条件不同会话记忆随“连接建立/断开”而生灭短期记忆随“任务完成/超时”而生灭。一个用户可能在一个会话里发起三个任务每个任务的短期记忆应该独立但共享同一个会话上下文。如果合并任务 A 的临时状态会污染任务 B。长期记忆和短期记忆的区别更明显短期记忆是“这次任务里记住的”长期记忆是“下次还需要的”。比如用户说“帮我订明天去上海的机票”短期记忆存“目的地上海、时间明天”任务结束就可以清但如果用户说“我以后都坐靠窗座位”这就该进长期记忆。至于遗忘机制它不是一层“存储”而是一套作用于所有层的策略。很多人把它当成长期记忆的附属功能这是错的。会话记忆也需要遗忘——比如滑动窗口截断短期记忆也需要遗忘——比如任务完成后清理。遗忘是横切的关注点。2.2 存储选型为什么不全用向量数据库新手最容易犯的错是一上来所有记忆都往向量数据库里塞。我试过结果是会话记忆的读取延迟从 5ms 涨到 80ms因为向量检索再快也比不上内存字典。而且会话记忆根本不需要“语义检索”它就是按 session_id 精确取。我的选型是这样的记忆层存储介质读取方式典型延迟生命周期会话记忆内存Redis/进程内字典精确 key 查询5ms会话期间短期记忆内存 轻量持久化精确 key 时间序20ms任务期间长期记忆向量库 关系库语义检索 结构化查询50-200ms永久可归档遗忘策略定时任务 触发器批量扫描离线持续运行这个表是我踩了无数坑总结的。核心原则是越靠近当前对话的记忆读取越要快存储越要轻越远离的越要能检索存储越要稳。向量库我一般选轻量级的方案关系库用 SQLite 或 Postgres 都行。这里不展开具体产品因为选型要看你的部署环境思路比工具重要。2.3 数据流转一次对话里记忆是怎么流动的理解架构最好的方式是跟一遍数据流。假设用户说“我叫老王帮我查下北京明天的天气我一般喜欢看摄氏度。”会话记忆这句话连同 session_id 一起写入会话缓存同时把最近 N 轮历史拼进 prompt。短期记忆Agent 解析出“查天气”这个任务把{任务: 天气查询, 城市: 北京, 日期: 明天}写入短期记忆。长期记忆Agent 识别到“我叫老王”“喜欢摄氏度”是持久偏好写入长期记忆打上user_profile标签。遗忘会话结束后会话记忆按 TTL 过期任务完成后短期记忆被清理长期记忆里的偏好保留但会定期做去重和置信度衰减。这个流转里写入时机的判断是最难的。不是所有信息都值得记。我的经验是显式声明“我叫X”“我喜欢Y”优先记长期任务参数记短期闲聊内容只留会话。3. 会话记忆上下文窗口的精细化管理3.1 会话记忆的本质是“有损压缩”很多人以为会话记忆就是把历史消息存下来。错。会话记忆的本质是在有限 token 预算下最大化保留对当前对话有用的信息。这是一个有损压缩问题。为什么必须有损因为模型的上下文窗口是有限的。就算你用 128K 窗口的模型聊上几十轮也会满而且 token 越多推理越慢、越贵。我算过一笔账假设每轮对话平均 200 token128K 窗口理论上能装 640 轮但实际到 100 轮左右模型对早期信息的注意力就明显衰减了。所以不是窗口够大就不用管理。我的会话记忆结构长这样class SessionMemory: def __init__(self, session_id, max_tokens4000, max_turns20): self.session_id session_id self.messages [] # 完整消息列表 self.summary # 早期对话的摘要 self.max_tokens max_tokens self.max_turns max_turns def add_message(self, role, content): self.messages.append({role: role, content: content}) self._compress_if_needed() def _compress_if_needed(self): # 超过轮数上限把最老的几轮压缩成摘要 if len(self.messages) self.max_turns: old self.messages[:5] self.summary self._summarize(old, self.summary) self.messages self.messages[5:] def get_context(self): # 摘要 最近消息拼成最终上下文 ctx [] if self.summary: ctx.append({role: system, content: f早期对话摘要{self.summary}}) ctx.extend(self.messages) return ctx这段代码的关键在_compress_if_needed。我没有用简单的“截断最老的”而是把最老的几轮压缩成摘要。为什么因为直接截断会丢失信息比如用户第一轮说的名字截断后模型就忘了。摘要虽然也有损但保留了关键实体。3.2 摘要压缩的触发阈值怎么定max_turns20这个值不是拍脑袋定的。我的计算逻辑是先确定 token 预算再反推轮数。假设你的模型上下文窗口是 8K你要留 2K 给系统 prompt 和输出那可用上下文是 6K。每轮对话平均 150 token用户助手那理论上能装 40 轮。但为了给摘要和检索结果留空间我一般取理论值的 50%也就是 20 轮。这个“50% 安全系数”是我多次实测的经验值——留太少会频繁触发压缩留太多会浪费窗口。注意摘要本身也消耗 token。如果你的摘要写得太长压缩就失去意义了。我一般把摘要控制在 200 token 以内只保留人名、偏好、未完成任务这些关键信息。3.3 会话记忆的持久化与恢复会话记忆默认在内存里但有个问题服务重启或用户换设备会话就断了。所以生产环境我会做轻量持久化——把会话记忆定期快照到 RedisTTL 设为 30 分钟。为什么是 30 分钟因为大部分对话的间隔不会超过这个时间。超过 30 分钟用户大概率是开始新话题了旧会话恢复反而会带来干扰。这个值可以根据你的业务调整客服场景可以短一点10 分钟教育场景可以长一点2 小时。恢复逻辑很简单用户带着 session_id 回来先从 Redis 取快照取不到就新建。这里有个坑快照要存摘要和最近消息不要存全量。全量快照又大又慢恢复后还是要压缩不如直接存压缩后的状态。4. 短期记忆任务状态的临时账本4.1 短期记忆和会话记忆的边界在哪这是最容易被混淆的地方。我用一个例子说清楚用户在一个会话里说“帮我规划一个三天的北京行程第一天去故宫第二天去长城第三天待定”。会话记忆记的是用户说了这句话助手回了什么。短期记忆记的是{任务: 行程规划, 城市: 北京, 天数: 3, 已定: [故宫, 长城], 待定: [第三天]}。区别在于会话记忆是原始对话流短期记忆是结构化任务状态。会话记忆给模型看短期记忆给程序逻辑用。当用户后来说“第三天改成颐和园”程序从短期记忆里取出任务状态更新待定字段而不是去翻对话历史。为什么必须分开因为程序逻辑不该依赖自然语言解析。如果每次更新任务状态都要重新解析对话既慢又容易出错。短期记忆就是把这个解析结果缓存下来。4.2 短期记忆的数据结构设计短期记忆我一般用“任务槽Task Slot”模型class TaskSlot: def __init__(self, task_id, task_type): self.task_id task_id self.task_type task_type # 任务类型如 travel_plan self.slots {} # 槽位如 {city: 北京, days: 3} self.status in_progress # in_progress / completed / abandoned self.created_at time.time() self.updated_at time.time() def update_slot(self, key, value): self.slots[key] value self.updated_at time.time() def is_expired(self, ttl1800): return time.time() - self.updated_at ttlslots字典是核心。它的 key 是预定义的槽位名value 是当前值。预定义的好处是可控——你知道这个任务有哪些槽位不会无限膨胀。如果槽位是动态的短期记忆会变成另一个“什么都往里塞”的垃圾堆。ttl180030 分钟是任务超时时间。超过 30 分钟没更新任务大概率被用户放弃了该清理。这个值也要按业务调订票类任务可以短10 分钟写作类任务可以长2 小时。4.3 短期记忆的读写时机短期记忆的写入时机有三个任务创建时Agent 识别到新任务创建 TaskSlot。槽位更新时用户补充信息更新对应槽位。任务完成时标记 status 为 completed触发清理。读取时机主要是两个生成回复前把当前任务状态注入 prompt让模型知道“现在进行到哪了”。执行动作前程序逻辑检查槽位是否齐全不齐就追问用户。这里有个实操心得注入 prompt 的任务状态要精简。我见过有人把整个 slots 字典序列化后塞进去结果模型被一堆 JSON 搞晕了。正确做法是转成自然语言比如“当前任务规划北京三日游已确定故宫和长城第三天待定”。提示短期记忆的清理一定要主动。我早期偷懒靠 TTL 自动过期结果内存里堆了一堆僵尸任务。后来改成任务完成时立即清理 定时扫描兜底内存占用降了 60%。5. 长期记忆跨会话的知识沉淀5.1 长期记忆存什么不存什么长期记忆是最容易“贪多嚼不烂”的地方。新手恨不得把所有对话都存进去结果检索时全是噪声。我的原则是只存三类东西。第一类是用户画像姓名、偏好、习惯、禁忌。比如“老王”“喜欢摄氏度”“不吃辣”。这类信息复用率最高几乎每次对话都用得上。第二类是领域知识用户在对话中提供的、可复用的知识。比如用户说“我们公司的报销流程是 A→B→C”这个流程下次还可能用到。第三类是历史决策用户做过的选择用于保持一致性。比如“上次选了经济舱”这次推荐时可以默认经济舱。不存什么闲聊、一次性查询、临时情绪。用户问“今天天气怎么样”这个不需要长期记忆。用户说“我今天心情不好”除非他明确说“以后都这样”否则也不存。5.2 长期记忆的存储结构向量 结构化双写长期记忆我用双写策略向量库存语义关系库存结构。class LongTermMemory: def __init__(self, vector_store, relational_store): self.vector_store vector_store # 语义检索 self.relational_store relational_store # 结构化查询 def write(self, user_id, content, memory_type, metadataNone): # 双写向量库用于语义检索关系库用于精确查询 doc_id self.vector_store.add( textcontent, metadata{user_id: user_id, type: memory_type, **(metadata or {})} ) self.relational_store.insert( doc_iddoc_id, user_iduser_id, contentcontent, memory_typememory_type, metadatametadata or {}, created_attime.time() ) return doc_id def recall(self, user_id, query, top_k5): # 语义检索 semantic_results self.vector_store.search( queryquery, filter{user_id: user_id}, top_ktop_k ) # 结构化补充比如取用户画像 profile self.relational_store.query( user_iduser_id, memory_typeuser_profile ) return {semantic: semantic_results, profile: profile}为什么要双写因为语义检索和精确查询解决不同问题。用户问“我上次说的那个偏好是什么”语义检索能找到程序要取“用户的所有画像字段”结构化查询更快更准。只用向量库取画像要遍历只用关系库语义匹配做不了。双写的代价是一致性维护。两个库可能不同步比如向量库写成功、关系库失败。我的做法是以关系库为准向量库异步补偿。写入时先写关系库再异步写向量库如果向量库失败进重试队列。读取时如果向量库缺失回退到关系库做关键词匹配。5.3 长期记忆的检索策略怎么让召回更准长期记忆的检索质量直接决定 Agent 聪不聪明。我踩过的坑是检索回来的东西太多太杂模型反而被干扰。我的优化策略有三层第一层是过滤。检索前先按 user_id 和 memory_type 过滤只取相关的。比如当前是订票任务就只取user_profile和travel_preference不取work_knowledge。第二层是重排。向量检索返回 top_k 后用一个轻量模型或规则做重排。我的规则是时间越近的权重越高被引用次数越多的权重越高。一个偏好如果被用过 10 次说明它很重要。第三层是截断。最终注入 prompt 的长期记忆不超过 3 条每条不超过 100 token。为什么这么少因为长期记忆是“锦上添花”不是“雪中送炭”。当前对话的上下文才是主角长期记忆太多会喧宾夺主。检索阶段操作目的过滤按 user_id type 筛选缩小范围降噪召回向量检索 top 20保证召回率重排时间 引用次数加权提升相关性截断取 top 3每条限长控制 token 预算这个表是我调了很多次才稳定的。早期我不做重排直接取 top 5结果经常召回一些“语义相似但实际无关”的内容。加了时间衰减后准确率明显提升。6. 遗忘机制被低估的“记忆卫生”6.1 为什么遗忘比记忆更重要我见过太多 Agent 项目记忆只进不出最后变成一个“什么都记得但什么都记不清”的糊涂蛋。遗忘不是缺陷是功能。人脑每天遗忘大量信息正是这种遗忘让重要的信息更突出。Agent 的遗忘要解决三个问题过期清理不再需要的信息、去重合并重复的信息、置信度衰减可能过时的信息。过期清理最简单按 TTL 删。会话记忆 30 分钟短期记忆任务完成后立即删长期记忆里的临时信息 7 天删。去重合并稍复杂。用户可能在不同会话里说了三次“我喜欢摄氏度”长期记忆里就有三条。我的做法是写入时先做相似度检查相似度超过 0.9 就合并更新引用次数和时间戳。置信度衰减是最容易被忽略的。用户三年前说“我喜欢摄氏度”现在可能变了。我的策略是长期记忆的置信度随时间衰减每次被引用时重置。衰减公式用指数衰减confidence base_confidence * exp(-λ * days_since_last_reference)λ我取 0.01意味着大约 70 天衰减一半。这个值可以调快消品场景可以大一点医疗场景可以小一点。6.2 遗忘策略的代码实现class ForgettingEngine: def __init__(self, ltm, decay_lambda0.01, min_confidence0.3): self.ltm ltm self.decay_lambda decay_lambda self.min_confidence min_confidence def run(self): # 1. 过期清理 self._clean_expired() # 2. 去重合并 self._deduplicate() # 3. 置信度衰减 self._decay_confidence() def _clean_expired(self): now time.time() self.ltm.relational_store.delete_where( created_at ? AND memory_type temporary, (now - 7 * 86400,) ) def _deduplicate(self): # 按 user_id 分组组内做相似度合并 for user_id in self.ltm.get_all_users(): memories self.ltm.relational_store.query(user_iduser_id) merged self._merge_similar(memories) if merged: self.ltm.batch_update(merged) def _decay_confidence(self): now time.time() for mem in self.ltm.relational_store.query_all(): days (now - mem[last_referenced_at]) / 86400 new_conf mem[confidence] * math.exp(-self.decay_lambda * days) if new_conf self.min_confidence: self.ltm.relational_store.delete(mem[doc_id]) else: self.ltm.relational_store.update( mem[doc_id], confidencenew_conf )这段代码里min_confidence0.3是删除阈值。低于这个值说明这条记忆已经“不可信”了删掉。decay_lambda0.01控制衰减速度。这两个参数需要根据业务调我一般先跑一周看删除量再微调。注意遗忘引擎一定要离线跑不要放在请求链路里。我早期图省事在每次对话结束时同步跑遗忘结果响应时间从 200ms 涨到 800ms。后来改成定时任务每小时跑一次响应时间恢复正常。6.3 遗忘的“软删除”与可恢复直接删数据有风险——万一删错了呢所以我用软删除标记deleted_at不物理删除。保留 30 天后再由另一个任务物理清理。软删除的好处是可恢复。如果用户投诉“你怎么忘了我说过的话”你可以从软删除里捞回来。这个机制在客服场景特别有用我靠它救过好几次场。7. 全链路代码实战串起四层记忆7.1 整体调用流程把四层记忆串起来一次对话的完整流程是这样的class AgentMemorySystem: def __init__(self, session_store, task_store, ltm, forgetting): self.session_store session_store self.task_store task_store self.ltm ltm self.forgetting forgetting def process(self, user_id, session_id, user_input): # 1. 取会话记忆 session self.session_store.get_or_create(session_id) session.add_message(user, user_input) # 2. 取短期记忆当前任务状态 task self.task_store.get_active_task(session_id) # 3. 检索长期记忆 ltm_context self.ltm.recall(user_id, user_input, top_k3) # 4. 组装 prompt prompt self._build_prompt(session, task, ltm_context) # 5. 调用模型 response self._call_llm(prompt) # 6. 更新记忆 session.add_message(assistant, response) self._update_task(task, user_input, response) self._maybe_write_ltm(user_id, user_input, response) return response def _build_prompt(self, session, task, ltm_context): parts [] if ltm_context[profile]: parts.append(f用户画像{ltm_context[profile]}) if ltm_context[semantic]: parts.append(f相关记忆{ltm_context[semantic]}) if task: parts.append(f当前任务{task.to_natural_language()}) parts.append(对话历史) parts.extend(session.get_context()) return \n.join(str(p) for p in parts)这个流程里第 6 步的更新逻辑是重点。_update_task负责解析用户输入更新任务槽位_maybe_write_ltm负责判断是否值得写长期记忆。7.2 判断“是否写长期记忆”的规则def _maybe_write_ltm(self, user_id, user_input, response): # 规则1显式声明 if any(kw in user_input for kw in [我叫, 我是, 我喜欢, 我偏好, 记住]): self.ltm.write(user_id, user_input, user_profile) return # 规则2任务完成后的结论 if self._is_task_conclusion(response): self.ltm.write(user_id, response, task_conclusion) return # 规则3用户纠正 if any(kw in user_input for kw in [不对, 错了, 应该是]): self.ltm.write(user_id, user_input, correction)这三条规则覆盖了大部分需要长期记忆的场景。规则 1 抓显式声明规则 2 抓任务结论规则 3 抓纠正。其他情况默认不写避免噪声。7.3 参数计算token 预算怎么分配最后说一个实操中必须算清楚的账token 预算分配。假设模型窗口 8K输出预留 1K系统 prompt 占 500那可用 6.5K。我的分配是部分预算说明长期记忆3003 条每条 100任务状态200自然语言描述会话摘要200早期对话压缩最近对话5000约 25 轮缓冲800防止超限这个分配不是固定的要根据任务类型调。任务型对话订票、查询可以压缩最近对话多留任务状态闲聊型对话可以压缩任务状态多留对话。我踩过的坑是没留缓冲。有一次用户输入特别长加上检索结果直接超了窗口模型报错。后来我强制留 800 token 缓冲再没出过问题。8. 常见问题与排查技巧实录8.1 记忆串味不同用户/会话的记忆混了这是最严重的问题。我遇到过一次用户 A 的偏好被注入到用户 B 的对话里差点造成隐私事故。排查下来是过滤条件写漏了——向量检索时忘了加 user_id 过滤。排查思路先查检索代码的 filter 参数再看存储时 user_id 是否正确写入。我的经验是所有记忆操作强制带 user_id 和 session_id在数据层做校验不依赖调用方传参。8.2 记忆膨胀存储越来越大检索越来越慢长期记忆只进不出三个月后检索延迟从 50ms 涨到 500ms。解决靠遗忘引擎但更根本的是写入时就要克制。我的规则是写入前先查重相似度超 0.9 不写写入时打 TTL 标签临时信息 7 天自动清。8.3 摘要失真压缩后关键信息丢了摘要模型如果太弱会把“用户要订去上海的机票”压缩成“用户要订票”丢了目的地。我的做法是摘要用规则 模型结合。规则提取实体人名、地名、时间模型负责组织语言。这样关键实体不会丢。8.4 常见问题速查表问题可能原因排查方向解决记忆串味过滤条件缺失检查 filter 参数强制带 user_id检索慢数据膨胀看存储量启用遗忘引擎摘要失真模型太弱对比摘要和原文规则模型结合任务状态丢失TTL 太短看任务间隔调大 TTL长期记忆不召回向量库不同步查双写日志异步补偿这张表是我从多次故障里总结的基本覆盖了 80% 的记忆相关问题。遇到问题先查表能省不少时间。8.5 独家避坑技巧最后分享几个文档里不会写的技巧。第一记忆写入要幂等。同一个信息重复写入要能去重否则重试机制会导致重复。第二检索结果要打时间戳。模型看到“用户三个月前说喜欢摄氏度”会自己判断是否过时。第三遗忘引擎要有 dry-run 模式。上线前先跑一遍看会删什么确认无误再真删。我靠 dry-run 避免过一次误删用户画像的事故。这套架构我用了大半年从最初的“聊三轮就失忆”到现在能稳定支撑多轮复杂任务核心就是把记忆当成一个有生命周期的系统来设计而不是一个简单的存储桶。会话、短期、长期、遗忘四层各司其职缺一层都会出问题。代码不用照抄但分层的思路和那些参数背后的计算逻辑应该能帮你少走不少弯路。
返回列表