拓冰网站定制 
我做商用 Agent 快两年了被问得最多的一个问题是为什么我的 Agent 聊着聊着就像失忆了一样早上还交代得好好的偏好设置下午再问它它一脸无辜。这不是模型笨也不是 Prompt 写得不够长而是记忆架构压根没搭起来。大多数 Agent 项目把记忆当成一个缓存来用随手塞个 List 就了事结果一旦离开单次会话的舒适区各种失忆症状全冒出来。这篇文章我把商用 Agent 的记忆架构拆开揉碎从会话记忆、短期记忆、长期记忆到遗忘机制全链路过一遍每一层都给出能直接跑的 Python 代码。内容比较干建议收藏后照着敲一遍。适合正在做 Agent 应用的开发者也适合想从 Demo 走向生产的团队——你会发现记忆不是加一个数据库那么简单它是一个需要刻意设计的子系统。1. 先把记忆分层这件事说透1.1 为什么模型天生健忘大模型本身是无状态的。你把一段对话扔给它它返回一段文字仅此而已。它不记得你昨天问过什么也不记得你上个月配置过什么规则。所谓“记忆”在模型眼里只是 Prompt 里拼接进去的上下文。一切记忆问题本质上都是上下文管理问题。这个特性跟人类记忆做一个类比就很好理解模型的固定权重像是你的大脑皮层结构它决定了你如何思考但不包含你具体经历了什么而真正负责“今天见了谁、昨天吃了什么”的那部分在 Agent 系统里必须由外部存储来承担。任何宣称“模型自带记忆”的能力比如某些模型支持的 System Prompt 缓存也只是把静态配置塞得更高效并不是真正的动态记忆。商用环境里我们需要明确区分四层记忆会话记忆负责当前对话的完整性短期记忆负责多轮内的工作状态长期记忆负责跨会话的用户事实遗忘机制负责让记忆系统保持健康和合规。很多团队在做 Agent 的时候只实现了第一层甚至第一层都只是简单地把 messages 数组越拼越长直到某天爆掉 Token 上限才来补救。1.2 四层记忆的职责边界在设计记忆架构之前我们必须先把职责边界画清楚。很多记忆系统的混乱根源在于边界模糊——短期记忆跑去做长期记忆的活长期记忆存了一堆一次性对话内容遗忘机制完全缺席。四层记忆可以用一张表格先立个整体认知记忆层级生命周期典型载体核心问题会话记忆单次会话内上下文窗口、滚动消息列表不丢上下文、不爆 Token短期记忆(工作内存)单次会话内可跨数轮摘要缓冲、结构化状态对象窗口有限时怎么保留关键信息长期记忆跨会话、跨天向量库、关系数据库、键值存储怎么快速找回用户历史事实遗忘机制按策略触发TTL、LRU、重要性评分、显式删除什么时候该忘、怎么合规地忘这个分层不是学术概念而是工程现实。会话记忆解决“现在聊得顺不顺”短期记忆解决“当前任务进行中哪些信息不能丢”长期记忆解决“下次回来你还认不认识我”遗忘机制解决“记忆不是越多越好”。值得强调的是商用系统里这四层不是彼此孤立的而是数据流动的一条链会话记忆攒下来的关键信息经过提炼会升级为短期记忆短期记忆里那些跨会话仍然重要的事实会沉淀到长期记忆所有记忆都会进入遗忘策略的筛选范围。后面每一节都会沿着这条链往下走。2. 会话记忆先让上下文别断2.1 无状态接口下的有状态伪装要说会话记忆得从大模型 API 的调用方式说起。你每次调用接口时传入的是一个 messages 数组里面包含 system、user、assistant 三种角色消息。模型不会主动记住上一次调用的内容你需要把历史消息一条不少地拼进下一次请求。这就是会话记忆最朴素的实现方式——一个 List 而已。但商用环境不会这么简单。你不可能无限地把历史消息拼下去因为上下文窗口是有上限的。GPT 级别的模型动辄 128K 起步看着很大但真实对话往往塞满了工具返回结果、检索片段、思考过程一个复杂任务十分钟就能吃掉大半窗口。先看一个最基本的滚动窗口实现这是后续所有方案的基础。核心思路只保留最近 N 轮对话。from collections import deque import json class ConversationMemory: 基于滚动窗口的会话记忆保留最近 max_turns 轮对话 def __init__(self, max_turns: int 10, system_prompt: str ): self.max_turns max_turns self.system_prompt system_prompt # deque 可以高效地从左侧弹出过期消息 self.messages deque() def add_user_message(self, content: str): self.messages.append({role: user, content: content}) def add_assistant_message(self, content: str): self.messages.append({role: assistant, content: content}) def add_tool_result(self, tool_call_id: str, content: str): # 工具调用结果在 OpenAI 协议里需要和 tool_call 配对 self.messages.append({ role: tool, tool_call_id: tool_call_id, content: content }) def trim(self): # 只保留最近 max_turns 轮一轮一条 user一条 assistant while len(self.messages) self.max_turns * 2: self.messages.popleft() def get_context(self) - list: self.trim() return [{role: system, content: self.system_prompt}] list(self.messages) def to_json(self) - str: return json.dumps({messages: list(self.messages)}, ensure_asciiFalse) def load_from_json(self, data: str): self.messages deque(json.loads(data)[messages])这段代码的核心是 deque 和 trim 策略。deque 在左端弹出时是 O(1)比 list.pop(0) 的 O(n) 高效很多。trim 在每次获取上下文时触发保证消息列表有界。但滚动窗口的问题是它丢的往往是对话最前面的信息。如果用户在第五轮提到过“我叫李明”到第二十轮时这条消息早被弹出去了Agent 还是会问“请问怎么称呼您”。所以滚动窗口只适合纯闲聊场景商用 Agent 光靠它远远不够。2.2 按 Token 数裁剪窗口滚动窗口按轮数裁剪看似简单实际容易踩坑。原因是每轮对话的 Token 消耗差异极大用户问“你好”可能只要 5 个 Token而一次工具调用返回的 JSON 可能有几千 Token。按轮数裁剪无法精确控制窗口占用最好的方式是按 Token 数设置预算。一种务实的做法是以 Tokens 为计量单位动态丢弃最旧的消息直到总 Token 数低于阈值。要做到这一点你需要一个 Token 计数函数。在实际工程中用 tiktoken 做精确计数最可靠。import tiktoken class TokenBudgetMemory: 按 Token 预算裁剪的会话记忆 def __init__(self, max_tokens: int 6000, model: str gpt-4o): self.max_tokens max_tokens self.encoder tiktoken.encoding_for_model(model) self.messages [] def _count_tokens(self, messages: list) - int: # 按 OpenAI Chat 格式估算 Token实际比纯文本计数多约 10% total 0 for msg in messages: total 4 # 每条消息的元数据开销 total len(self.encoder.encode(msg.get(content, ))) # 角色标识额外算 2~3 个 Token total len(self.encoder.encode(msg.get(role, ))) if msg.get(name): total len(self.encoder.encode(msg[name])) total 2 # 回复开头压头 return total def append(self, message: dict): self.messages.append(message) def get_context(self, system_prompt: str ) - list: context [] if system_prompt: context.append({role: system, content: system_prompt}) context.extend(self.messages) # 从最旧的消息开始丢直到满足预算 while self._count_tokens(context) self.max_tokens and len(context) 2: # 保留 system 不丢丢最早的非 system 消息 for i, msg in enumerate(context): if msg[role] ! system: context.pop(i) break return context这里有两个容易被忽视的点。第一Token 预算不要拉满整个上下文窗口要给模型生成留出空间。通用做法是给模型输出预留 20% 到 30% 的 Token。比如模型上下文窗口是 128K你只把记忆预算设在 80K剩下的是给当前这一轮生成用的。第二裁剪时机要在发起请求之前而不是在响应的过程中才想起要清理否则一轮请求可能直接报超出上下文限制的错误。2.3 会话记忆的持久化会话记忆如果只存在进程内存里服务一重启就全部丢失。商用场景必须支持持久化和恢复。最轻量的做法是用 Redis 存 JSON 字符串以会话 ID 作为 Key。import redis import json class PersistentConversationMemory: 基于 Redis 的会话记忆持久化 def __init__(self, redis_url: str redis://localhost:6379/0, ttl: int 86400): self.client redis.Redis.from_url(redis_url) # 会话记忆默认保留一天超时自动清理 self.default_ttl ttl def save_message(self, session_id: str, message: dict): key fsession:{session_id} # Redis 列表尾部追加消息 self.client.rpush(key, json.dumps(message, ensure_asciiFalse)) self.client.expire(key, self.default_ttl) # 防止列表无限增长设置最大长度 self.client.ltrim(key, -200, -1) # 最多保留最近 200 条 def load_messages(self, session_id: str) - list: key fsession:{session_id} raw_list self.client.lrange(key, 0, -1) return [json.loads(raw) for raw in raw_list]Redis 列表天然适合做消息队列式的追加操作rpush 追加、ltrim 裁剪、expire 设置过期三个命令就把会话记忆的持久化、容量控制、自动过期全部解决了。但要注意Redis 中存的是序列化的消息列表取出来之后依然要过一遍 Token 预算裁剪逻辑不能直接全量塞进 Prompt。会话记忆持久化为啥重要因为商用 Agent 要支持多实例水平扩展。请求可能被负载均衡到任意一台机器如果记忆只存在单机内存里用户第二次请求落到另一台机器对话就断了。Redis 或者数据库集中存储是会话记忆走向生产的第一步。3. 短期记忆让 Agent 在长任务里抓住重点3.1 滚动摘要压缩会话记忆最头疼的问题是对话超过十几轮之后早期信息往往与当前任务高度相关但直接保留会占用大量 Token。滚动摘要Rolling Summary是解决这个问题的主流方案。思路是当消息超过一定轮数后把最旧的一批消息交给模型提炼成一段摘要用摘要代替原始消息。我实际做的实现是参考 LangChain 的 SummaryBufferMemory 思路但做了简化。核心逻辑是在消息队列里维护两个区——原始消息区和摘要区。新消息都进入原始区当原始区超过阈值时把最老的一部分和已有摘要一起交给模型生成新的摘要这段原始消息就可以丢掉了。class SummaryBufferMemory: 滚动摘要记忆原始消息只保留最近几轮更早的压成摘要 def __init__(self, llm, max_original_pairs: int 6, summary_prompt_template: str ): self.llm llm self.max_pairs max_original_pairs self.summary self.recent_messages [] self.summary_template summary_prompt_template or ( 请把下面的对话内容压缩成一段中文摘要尽可能保留其中的关键事实 用户提到的个人信息、偏好、已经做出的决定、任务状态。\n 已有的摘要{existing_summary}\n 新的对话内容{new_lines}\n 请直接输出更新后的摘要不要加前言 ) def add_message(self, role: str, content: str): self.recent_messages.append({role: role, content: content}) self._maybe_compress() def _maybe_compress(self): user_count sum(1 for m in self.recent_messages if m[role] user) if user_count self.max_pairs * 2: # 取最旧的一半消息做压缩 compress_batch self.recent_messages[:self.max_pairs] self.recent_messages self.recent_messages[self.max_pairs:] new_lines \n.join( f{m[role]}: {m[content]} for m in compress_batch ) prompt self.summary_template.format( existing_summaryself.summary, new_linesnew_lines ) self.summary self.llm(prompt) def get_context(self) - list: context [] if self.summary: context.append({ role: system, content: f以下是这场对话的早期内容摘要请作为背景信息\n{self.summary} }) context.extend(self.recent_messages) return context这里最关键的技巧是摘要是渐进更新的不是一次性把整个历史喂给模型。每次只拿最老的一批消息和已有摘要合并生成新摘要这样模型每次处理的 Token 量是可控的不至于在做摘要时自己先把窗口爆掉。踩过的坑是摘要生成时的信息折损。摘要天然会丢失细节如果你把用户说过的电话号码存在摘要里压缩两次之后就变成了一串错误数字。所以我的纪律是摘要只承载“全局背景”任何精确数据必须进结构化记忆。3.2 结构化工作内存摘要适合保存模糊背景但任务执行中更需要的是精确状态。比如用户正在让 Agent 帮忙写周报你至少得记住项目名称、截止日期、已确定的段落结构。这些信息放进摘要里既不精确也不好用更好的做法是单独维护一个结构化对象注入 Prompt 时用 JSON 呈现。class WorkingMemory: 结构化短期记忆保存当前任务状态、用户临时偏好、待办清单 def __init__(self): self.task_state {} self.user_preferences {} self.todo_list [] self.data_slots {} def update_task_state(self, key: str, value): self.task_state[key] value def set_preference(self, key: str, value): self.user_preferences[key] value def add_todo(self, item: str): self.todo_list.append(item) self.todo_list self.todo_list[-10:] # 最多保存 10 个待办 def set_data_slot(self, key: str, value): 数据槽用来保存用户提到的结构化信息比如姓名、日期 self.data_slots[key] value def get_context_block(self) - str: import json context { current_task: self.task_state, known_facts: self.data_slots, user_temporary_preferences: self.user_preferences, pending_todos: self.todo_list } return 【当前任务状态】\n json.dumps(context, ensure_asciiFalse, indent2)为什么要单独搞一套结构化工作内存而不是全部塞进摘要两个原因一是精确检索你需要能直接从记忆里拿到“截止日期是周五”而不是让模型去摘要里推断二是便于更新用户中途改口说“截止日期改到下周”你直接覆盖 data_slots[deadline] 就行摘要模式做不到这种精准纠正。这个工作内存对象的使用方式是把它序列化成一段 JSON 拼在 system prompt 里。它天然就是给 Agent 的“便利贴”。在实际运行中Agent 的每次操作步骤都应该有一个显式的“更新便利贴”动作防止信息失真。4. 长期记忆跨会话记住用户4.1 向量检索从“存进去”到“找得回”长期记忆的核心场景是今天用户说“我喜欢简洁的回复风格”一周后用户再来Agent 应该还记得这个偏好。这类跨会话记忆没法靠摘要或滚动窗口实现必须落到独立的记忆数据库里。实现长期记忆有两条路线。第一条是结构化路线把用户偏好整理成“键值对”或“标签-属性”的形式存进数据库查询时精确匹配。第二条是语义路线把记忆片段向量化用相似度检索找回相关内容。商用系统通常两条都上但语义路线的泛化能力更强能覆盖“我记得你好像提过某个项目的技术栈”这类模糊回忆需求。先看一个轻量的向量库实现。为了这篇博文能直接跑通我用的是 SQLite 存向量 余弦相似度暴力计算。生产环境请换成 pgvector、Milvus 或 Chroma 这类专用工具原理一致。import sqlite3 import numpy as np import json class VectorMemory: 基于 SQLite 的长期记忆文本片段 向量检索 def __init__(self, db_path: str agent_memory.db, embed_dim: int 768): self.conn sqlite3.connect(db_path) self.embed_dim embed_dim self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT DEFAULT fact, importance REAL DEFAULT 0.5, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_access TIMESTAMP, access_count INTEGER DEFAULT 0, embedding TEXT ) ) self.conn.commit() def store_memory(self, user_id: str, content: str, embedding: list, memory_type: str fact, importance: float 0.5): self.conn.execute( INSERT INTO memories (user_id, content, memory_type, importance, embedding) VALUES (?, ?, ?, ?, ?), (user_id, content, memory_type, importance, json.dumps(embedding)) ) self.conn.commit() def search_memories(self, user_id: str, query_embedding: list, top_k: int 5, min_score: float 0.6) - list: cursor self.conn.execute( SELECT id, content, importance, embedding FROM memories WHERE user_id ?, (user_id,) ) query_vec np.array(query_embedding, dtypenp.float32) results [] for row in cursor: mem_id, content, importance, emb_str row emb np.array(json.loads(emb_str), dtypenp.float32) # 余弦相似度 cos_sim np.dot(query_vec, emb) / ( np.linalg.norm(query_vec) * np.linalg.norm(emb) 1e-8 ) # 综合得分相似度 重要性加权 final_score cos_sim * 0.7 importance * 0.3 if cos_sim min_score: results.append({ id: mem_id, content: content, importance: importance, score: float(final_score) }) results.sort(keylambda x: x[score], reverseTrue) return results[:top_k] def update_access(self, mem_id: int): 更新访问时间与次数供遗忘策略消费 self.conn.execute( UPDATE memories SET access_count access_count 1, last_access CURRENT_TIMESTAMP WHERE id ?, (mem_id,) ) self.conn.commit()把这段代码和会话记忆对比核心差异就在 embedding 这一步。会话记忆是顺序拼接长期记忆是语义检索。query 进来以后先算一次 embedding然后去库里找相似片段。不同的 embedding 模型维度不一样代码里的 embed_dim 参数就是为这个准备的。4.2 记忆的写入时机不是所有的对话都值得记长期记忆最隐蔽的坑是写入策略。我看到很多团队直接把每轮对话全部向量化入库结果向量库越来越臃肿检索噪音越来越大。正确的做法是写之前先判断这条信息是否有跨会话价值。什么样的信息值得进入长期记忆我的判断标准有三个用户主动表达的偏好“我更喜欢你说话简洁点”、关键事实“我在某某公司负责增长”、明确说过要记住的内容“请记住这个地址”。”判断可以交给 Agent 自己来完成——每次对话结束时让 Agent 过一遍消息列表输出候选记忆列表和重要性打分再执行写入。class MemoryExtraction: 从对话中抽取值得长期记忆的内容 def __init__(self, llm): self.llm llm def extract_candidates(self, dialogue: str) - list: prompt f 这是一段用户与 AI 助手的对话。请从中抽取值得长期记住的信息。 值得记住的信息包括 1. 用户主动表达的偏好、习惯、风格要求 2. 用户提到的个人与工作背景 3. 用户明确要求记住的事项 4. 正在进行的重要项目及其关键参数 不需要记录的内容包括 1. 与当前任务无关的寒暄 2. 一次性任务的中间计算 3. 用户临时指定、但明确表示只此一次的内容 对话内容 {dialogue} 请按如下 JSON 格式返回列表 [{{content: 用户偏好简洁回复风格, importance: 0.8, memory_type: preference}}] response self.llm(prompt) return parse_json(response)这个抽取器让我想明白了记忆系统的本质记忆不是一个存储问题而是一个判断问题。存储最多影响容量判断决定质量。商用 Agent 的记忆库宁可少存也不要存垃圾——多存一条垃圾信息的代价不只是存储成本而是所有后续检索都会被噪音污染最终表现就是 Agent 回答问题变得东拉西扯。4.3 记忆注入检索出来还要用得上长期记忆检索出来之后注入也有讲究。我见过直接把几大段检索结果拼进 system prompt 的操作结果模型关注的是中间某条不相干的记忆真正用户关心的那条反而被淹没。所以注入时必须做“重排裁剪”。常见做法是给每条检索结果加一个相关性标签或来源说明让 Agent 知道哪些信息是“记忆检索得到的历史背景”哪些是“用户本次对话中的直接指令”。以我的项目为例注入格式大致如下def build_memory_prompt(retrieved_memories: list) - str: if not retrieved_memories: return memory_blocks [] for i, mem in enumerate(retrieved_memories): # 按记忆类型添加标签帮助模型区分来源 type_label { preference: 用户偏好, fact: 用户背景, task: 历史任务, }.get(mem.get(memory_type, fact), 历史信息) memory_blocks.append(f{i1}. [{type_label}] {mem[content]}) return ( 从用户的长期记忆中检索到以下与当前对话相关的历史信息。 这些信息可能是较早之前记录的请作为背景参考但不要编造不存在的细节\n \n.join(memory_blocks) )记忆注入时的顺序也很重要。放在 system prompt 较后位置的记忆比放在前面的更容易被模型注意到。另外每次注入的条数不要超过 5 条检索分数低于阈值的宁可不注入。你的目标是给模型吃“干粮”不是给它端上一锅乱炖。5. 遗忘机制会忘事的 Agent 才可靠5.1 遗忘不是缺陷是功能聊完记忆的存取最后这块往往被忽略——遗忘机制。很多开发者觉得记忆系统只要“存得进去、找得回来”就完事了但商用场景下不会遗忘的记忆系统是定时炸弹。三个理由一是成本无限增长的向量库和关系表会让检索越来越慢二是质量旧的不相关记忆会像水军一样刷屏把真正有用的信息淹没三是合规用户隐私条例和监管法规都要求你提供数据删除能力。遗忘机制做得好其实是在帮记忆系统做“注意力管理”。人类大脑就是这么运作的——你记不住去年某天午餐吃了什么这是优点不是缺陷因为大脑把资源留给了更重要的信息。Agent 也一样。5.2 显式遗忘与 TTL 过期遗忘机制第一层是显式遗忘。这条最直接用户说“把你记住的我的信息都删了吧”你必须做到。这种删除不是把向量库里的记录抹掉那么简单而是要连同相关的会话记录、摘要缓存、工作内存一起清干净。合规层面的显式遗忘要做到“可追溯”也就是你至少得知道删了什么、什么时候删的。class ForgetMechanism: 遗忘机制显式删除 TTL 过期 LRU 淘汰 def __init__(self, conn, llm_memory_storeNone): self.conn conn self.llm_memory_store llm_memory_store def explicit_forget(self, user_id: str, keywords: str ALL): 用户主动要求删除记忆 if keywords ALL: cursor self.conn.execute( SELECT id, content FROM memories WHERE user_id ?, (user_id,) ) ids [row[0] for row in cursor.fetchall()] else: # 支持按关键词删除比如用户说“把我关于项目A的记忆删掉” cursor self.conn.execute( SELECT id, content FROM memories WHERE user_id ? AND content LIKE ?, (user_id, f%{keywords}%) ) ids [row[0] for row in cursor.fetchall()] if ids: format_ids ,.join(? * len(ids)) self.conn.execute( fDELETE FROM memories WHERE id IN ({format_ids}), ids ) self.conn.commit() return {deleted_count: len(ids)} return {deleted_count: 0} def ttl_forget(self, ttl_days: int 90): 按时间过期超过 N 天的记忆自动清理例外重要记忆 cursor self.conn.execute( SELECT id, content, importance, last_access FROM memories WHERE created_at datetime(now, ?) AND importance 0.7 , (f-{ttl_days} days,)) ids [row[0] for row in cursor.fetchall()] if ids: format_ids ,.join(? * len(ids)) self.conn.execute(fDELETE FROM memories WHERE id IN ({format_ids}), ids) self.conn.commit() return {expired_count: len(ids)} def lru_forget(self, keep_top_n: int 5000): 容量控制超过容量上限时把从未访问且重要性低的记忆清掉 实际项目里往往用 LRU最近最少使用或 LFU最不经常使用策略 cursor self.conn.execute( SELECT id, content, access_count, last_access, importance FROM memories ORDER BY last_access IS NULL DESC, last_access ASC, importance ASC ) rows cursor.fetchall() if len(rows) keep_top_n: return {evicted_count: 0} evict_list rows[keep_top_n:] evict_ids [r[0] for r in evict_list] format_ids ,.join(? * len(evict_ids)) self.conn.execute(fDELETE FROM memories WHERE id IN ({format_ids}), evict_ids) self.conn.commit() return {evicted_count: len(evict_ids)}我把遗忘策略分成了三条线显式删除走的是用户指令TTL 走的是时间维度LRU 走的是容量维度。三条线并行共同维持记忆库的健康水位。实际经验里TTL 遗忘要特别注意“重要性”这个字段。直接一刀切“90 天前的都删”会误删重要记忆。比如用户一年前提过的“我是公司的合规负责人”这条信息虽然旧但很重要影响后续很多交互。所以我加了条件importance 低于 0.7 才参与过期淘汰。这个阈值不建议定得太高否则记忆库永远清理不干净。5.3 可解释性让遗忘成为一个可审计事件商用 Agent 的遗忘机制还必须考虑可审计性。不是说删完就完了你得回答这三个问题删了什么、为什么删、什么时候删的。尤其是涉及用户隐私数据时监管和合规审计会问到这些细节。最简单的做法是引入一张 forget_log 表CREATE TABLE IF NOT EXISTS forget_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, memory_ids TEXT, reason TEXT, trigger_source TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );每次删除操作都把动机和对象记录在案。做这件事的意义不是满足工程洁癖而是给 Agent 建立信任基础——用户有权知道系统记得什么、忘了什么。我甚至在部分项目里做了一个“记忆管理”面板用户可以在前端看到自己的记忆列表并手动删除任意一条。这个功能上线之后用户对 Agent 的信任度明上升了不少。6. 落地这些记忆模块的实战经验6.1 记忆层之间的数据流转前面每一层单独看了现在把整个链路串起来。一次典型的 Agent 对话流程是这样的用户发消息 → 从 Redis 读取会话记忆 → 构建短期记忆摘要结构化工作内存 → 依据当前问题从向量库检索长期记忆 → 拼装最终 Prompt 发给模型 → 模型返回响应 → 把新消息写入会话记忆 → 在合适的时机触发摘要压缩 → 抽取值得长期记忆的片段写入向量库 → 按策略执行遗忘清理。这个链路里的每一个环节都可能成为故障点。团队在场最容易出问题的不是某个记忆模块本身而是模块之间的衔接时机。比如长期记忆的抽取不该在每轮对话后都执行否则模型每轮都要额外跑一次抽取任务成本和延迟都吃不消。我通常设置的节奏是对话结束用户不再发消息超过 5 分钟或任务完成节点执行抽取而不是每轮都抽。6.2 令牌预算和持久化选型的工程建议关于记忆系统的工程选型我按项目规模给三个档位的建议原型验证单机内存 SQLite 就够。别一上来就上向量数据库先用暴力余弦相似度跑通逻辑重点是验证记忆策略是否有效。小规模商用Redis 存会话记忆PostgreSQL pgvector 存长期记忆。这个组合可以支撑几十万用户的量级运维复杂度也不高。大规模商用独立记忆服务Milvus/Weaviate 做向量检索消息队列做记忆抽取的异步管道遗忘机制做成定时任务或事件驱动任务。选型的唯一原则是不要为了架构而架构。先让你的记忆逻辑在业务上跑通再考虑扩展。6.3 每一层值得记录的“坑”层级典型问题我的解决经验会话记忆窗口裁剪把关键用户信息丢了对 user.role 为主的消息优先保留裁剪时把 assistant 的纯寒暄消息先丢短期记忆摘要越压越失真结构化字段和摘要分离精确数据走结构化模糊背景走摘要长期记忆向量检索召回大量噪声用重要性和时间衰减加权阈值宁高勿低遗忘机制误删高价值记忆遗忘策略必须引入重要性门槛重要记忆不参与自动过期这个表格里的每条都对应具体代码里的一个分支是实打实踩出来的。6.4 调试记忆系统的好用手段记忆系统是黑盒里最黑的一部分调试起来很痛苦。我的建议是必须是可见的。三个手段让记忆系统变得透明。第一记录每次请求的最终 Prompt 快照。把拼接好的完整 Prompt 存一份日志出了问题可以直接回放看到底是检索错了还是注入顺序错了。第二给检索召回的记忆打上分数显示。前端或日志里把每条记忆的相关性分数展示出来你能直接看出召回质量为什么差。第三做 A/B 测试时单独开关记忆模块。加一个全局配置项可以一键关闭长期记忆、只保留会话记忆方便对比记忆模块对业务指标的影响。调试用的日志我一般用 JSON Lines 格式每行一个事件结构化字段包含记忆层级、用户 ID、Session ID、操作类型、Token 数。配合日志检索系统可以快速定位某一类记忆异常。6.5 从代码到商用的最后一步最后补充一点上面所有代码都还只是记忆引擎的内核距离商用还差一层服务化封装。你需要把记忆模块做成独立服务对外提供 HTTP 接口或者 SDK——get_context、save_memory、search_memory、forget——这样业务层不用关心记忆是怎么存的只需要调用接口。这个抽象层的价值在团队协作时尤其明显业务开发的同事不需要知道你是用向量库还是 Redis 实现的他只需要保证调接口时塞对参数。说实话我见过很多团队在记忆系统这个环节翻车翻得最多的不是技术选型而是“不重视”。Agent 应用跑 Demo 的时候单轮短对话根本看不出记忆系统的差距一旦进入真实商用用户会持续使用好几天记忆系统的质量直接决定了 Agent 的专业度。一个会记得用户偏好和历史的 Agent和一个每次交流都像第一次见面的 Agent用户体验差着数量级。我做这个系统最深的体会是记忆架构的本质不是存储而是“取舍”。你要决定什么信息值得留、什么信息必须忘、什么信息怎么被想起来这些决策才是记忆系统的灵魂。按文章里的四层结构动手做一遍你会对 Agent 的能力边界有个全新的认识。 商用Agent记忆架构全解析：从会话记忆到遗忘机制的落地实践
我做商用 Agent 快两年了被问得最多的一个问题是为什么我的 Agent 聊着聊着就像失忆了一样早上还交代得好好的偏好设置下午再问它它一脸无辜。这不是模型笨也不是 Prompt 写得不够长而是记忆架构压根没搭起来。大多数 Agent 项目把记忆当成一个缓存来用随手塞个 List 就了事结果一旦离开单次会话的舒适区各种失忆症状全冒出来。这篇文章我把商用 Agent 的记忆架构拆开揉碎从会话记忆、短期记忆、长期记忆到遗忘机制全链路过一遍每一层都给出能直接跑的 Python 代码。内容比较干建议收藏后照着敲一遍。适合正在做 Agent 应用的开发者也适合想从 Demo 走向生产的团队——你会发现记忆不是加一个数据库那么简单它是一个需要刻意设计的子系统。1. 先把记忆分层这件事说透1.1 为什么模型天生健忘大模型本身是无状态的。你把一段对话扔给它它返回一段文字仅此而已。它不记得你昨天问过什么也不记得你上个月配置过什么规则。所谓“记忆”在模型眼里只是 Prompt 里拼接进去的上下文。一切记忆问题本质上都是上下文管理问题。这个特性跟人类记忆做一个类比就很好理解模型的固定权重像是你的大脑皮层结构它决定了你如何思考但不包含你具体经历了什么而真正负责“今天见了谁、昨天吃了什么”的那部分在 Agent 系统里必须由外部存储来承担。任何宣称“模型自带记忆”的能力比如某些模型支持的 System Prompt 缓存也只是把静态配置塞得更高效并不是真正的动态记忆。商用环境里我们需要明确区分四层记忆会话记忆负责当前对话的完整性短期记忆负责多轮内的工作状态长期记忆负责跨会话的用户事实遗忘机制负责让记忆系统保持健康和合规。很多团队在做 Agent 的时候只实现了第一层甚至第一层都只是简单地把 messages 数组越拼越长直到某天爆掉 Token 上限才来补救。1.2 四层记忆的职责边界在设计记忆架构之前我们必须先把职责边界画清楚。很多记忆系统的混乱根源在于边界模糊——短期记忆跑去做长期记忆的活长期记忆存了一堆一次性对话内容遗忘机制完全缺席。四层记忆可以用一张表格先立个整体认知记忆层级生命周期典型载体核心问题会话记忆单次会话内上下文窗口、滚动消息列表不丢上下文、不爆 Token短期记忆(工作内存)单次会话内可跨数轮摘要缓冲、结构化状态对象窗口有限时怎么保留关键信息长期记忆跨会话、跨天向量库、关系数据库、键值存储怎么快速找回用户历史事实遗忘机制按策略触发TTL、LRU、重要性评分、显式删除什么时候该忘、怎么合规地忘这个分层不是学术概念而是工程现实。会话记忆解决“现在聊得顺不顺”短期记忆解决“当前任务进行中哪些信息不能丢”长期记忆解决“下次回来你还认不认识我”遗忘机制解决“记忆不是越多越好”。值得强调的是商用系统里这四层不是彼此孤立的而是数据流动的一条链会话记忆攒下来的关键信息经过提炼会升级为短期记忆短期记忆里那些跨会话仍然重要的事实会沉淀到长期记忆所有记忆都会进入遗忘策略的筛选范围。后面每一节都会沿着这条链往下走。2. 会话记忆先让上下文别断2.1 无状态接口下的有状态伪装要说会话记忆得从大模型 API 的调用方式说起。你每次调用接口时传入的是一个 messages 数组里面包含 system、user、assistant 三种角色消息。模型不会主动记住上一次调用的内容你需要把历史消息一条不少地拼进下一次请求。这就是会话记忆最朴素的实现方式——一个 List 而已。但商用环境不会这么简单。你不可能无限地把历史消息拼下去因为上下文窗口是有上限的。GPT 级别的模型动辄 128K 起步看着很大但真实对话往往塞满了工具返回结果、检索片段、思考过程一个复杂任务十分钟就能吃掉大半窗口。先看一个最基本的滚动窗口实现这是后续所有方案的基础。核心思路只保留最近 N 轮对话。from collections import deque import json class ConversationMemory: 基于滚动窗口的会话记忆保留最近 max_turns 轮对话 def __init__(self, max_turns: int 10, system_prompt: str ): self.max_turns max_turns self.system_prompt system_prompt # deque 可以高效地从左侧弹出过期消息 self.messages deque() def add_user_message(self, content: str): self.messages.append({role: user, content: content}) def add_assistant_message(self, content: str): self.messages.append({role: assistant, content: content}) def add_tool_result(self, tool_call_id: str, content: str): # 工具调用结果在 OpenAI 协议里需要和 tool_call 配对 self.messages.append({ role: tool, tool_call_id: tool_call_id, content: content }) def trim(self): # 只保留最近 max_turns 轮一轮一条 user一条 assistant while len(self.messages) self.max_turns * 2: self.messages.popleft() def get_context(self) - list: self.trim() return [{role: system, content: self.system_prompt}] list(self.messages) def to_json(self) - str: return json.dumps({messages: list(self.messages)}, ensure_asciiFalse) def load_from_json(self, data: str): self.messages deque(json.loads(data)[messages])这段代码的核心是 deque 和 trim 策略。deque 在左端弹出时是 O(1)比 list.pop(0) 的 O(n) 高效很多。trim 在每次获取上下文时触发保证消息列表有界。但滚动窗口的问题是它丢的往往是对话最前面的信息。如果用户在第五轮提到过“我叫李明”到第二十轮时这条消息早被弹出去了Agent 还是会问“请问怎么称呼您”。所以滚动窗口只适合纯闲聊场景商用 Agent 光靠它远远不够。2.2 按 Token 数裁剪窗口滚动窗口按轮数裁剪看似简单实际容易踩坑。原因是每轮对话的 Token 消耗差异极大用户问“你好”可能只要 5 个 Token而一次工具调用返回的 JSON 可能有几千 Token。按轮数裁剪无法精确控制窗口占用最好的方式是按 Token 数设置预算。一种务实的做法是以 Tokens 为计量单位动态丢弃最旧的消息直到总 Token 数低于阈值。要做到这一点你需要一个 Token 计数函数。在实际工程中用 tiktoken 做精确计数最可靠。import tiktoken class TokenBudgetMemory: 按 Token 预算裁剪的会话记忆 def __init__(self, max_tokens: int 6000, model: str gpt-4o): self.max_tokens max_tokens self.encoder tiktoken.encoding_for_model(model) self.messages [] def _count_tokens(self, messages: list) - int: # 按 OpenAI Chat 格式估算 Token实际比纯文本计数多约 10% total 0 for msg in messages: total 4 # 每条消息的元数据开销 total len(self.encoder.encode(msg.get(content, ))) # 角色标识额外算 2~3 个 Token total len(self.encoder.encode(msg.get(role, ))) if msg.get(name): total len(self.encoder.encode(msg[name])) total 2 # 回复开头压头 return total def append(self, message: dict): self.messages.append(message) def get_context(self, system_prompt: str ) - list: context [] if system_prompt: context.append({role: system, content: system_prompt}) context.extend(self.messages) # 从最旧的消息开始丢直到满足预算 while self._count_tokens(context) self.max_tokens and len(context) 2: # 保留 system 不丢丢最早的非 system 消息 for i, msg in enumerate(context): if msg[role] ! system: context.pop(i) break return context这里有两个容易被忽视的点。第一Token 预算不要拉满整个上下文窗口要给模型生成留出空间。通用做法是给模型输出预留 20% 到 30% 的 Token。比如模型上下文窗口是 128K你只把记忆预算设在 80K剩下的是给当前这一轮生成用的。第二裁剪时机要在发起请求之前而不是在响应的过程中才想起要清理否则一轮请求可能直接报超出上下文限制的错误。2.3 会话记忆的持久化会话记忆如果只存在进程内存里服务一重启就全部丢失。商用场景必须支持持久化和恢复。最轻量的做法是用 Redis 存 JSON 字符串以会话 ID 作为 Key。import redis import json class PersistentConversationMemory: 基于 Redis 的会话记忆持久化 def __init__(self, redis_url: str redis://localhost:6379/0, ttl: int 86400): self.client redis.Redis.from_url(redis_url) # 会话记忆默认保留一天超时自动清理 self.default_ttl ttl def save_message(self, session_id: str, message: dict): key fsession:{session_id} # Redis 列表尾部追加消息 self.client.rpush(key, json.dumps(message, ensure_asciiFalse)) self.client.expire(key, self.default_ttl) # 防止列表无限增长设置最大长度 self.client.ltrim(key, -200, -1) # 最多保留最近 200 条 def load_messages(self, session_id: str) - list: key fsession:{session_id} raw_list self.client.lrange(key, 0, -1) return [json.loads(raw) for raw in raw_list]Redis 列表天然适合做消息队列式的追加操作rpush 追加、ltrim 裁剪、expire 设置过期三个命令就把会话记忆的持久化、容量控制、自动过期全部解决了。但要注意Redis 中存的是序列化的消息列表取出来之后依然要过一遍 Token 预算裁剪逻辑不能直接全量塞进 Prompt。会话记忆持久化为啥重要因为商用 Agent 要支持多实例水平扩展。请求可能被负载均衡到任意一台机器如果记忆只存在单机内存里用户第二次请求落到另一台机器对话就断了。Redis 或者数据库集中存储是会话记忆走向生产的第一步。3. 短期记忆让 Agent 在长任务里抓住重点3.1 滚动摘要压缩会话记忆最头疼的问题是对话超过十几轮之后早期信息往往与当前任务高度相关但直接保留会占用大量 Token。滚动摘要Rolling Summary是解决这个问题的主流方案。思路是当消息超过一定轮数后把最旧的一批消息交给模型提炼成一段摘要用摘要代替原始消息。我实际做的实现是参考 LangChain 的 SummaryBufferMemory 思路但做了简化。核心逻辑是在消息队列里维护两个区——原始消息区和摘要区。新消息都进入原始区当原始区超过阈值时把最老的一部分和已有摘要一起交给模型生成新的摘要这段原始消息就可以丢掉了。class SummaryBufferMemory: 滚动摘要记忆原始消息只保留最近几轮更早的压成摘要 def __init__(self, llm, max_original_pairs: int 6, summary_prompt_template: str ): self.llm llm self.max_pairs max_original_pairs self.summary self.recent_messages [] self.summary_template summary_prompt_template or ( 请把下面的对话内容压缩成一段中文摘要尽可能保留其中的关键事实 用户提到的个人信息、偏好、已经做出的决定、任务状态。\n 已有的摘要{existing_summary}\n 新的对话内容{new_lines}\n 请直接输出更新后的摘要不要加前言 ) def add_message(self, role: str, content: str): self.recent_messages.append({role: role, content: content}) self._maybe_compress() def _maybe_compress(self): user_count sum(1 for m in self.recent_messages if m[role] user) if user_count self.max_pairs * 2: # 取最旧的一半消息做压缩 compress_batch self.recent_messages[:self.max_pairs] self.recent_messages self.recent_messages[self.max_pairs:] new_lines \n.join( f{m[role]}: {m[content]} for m in compress_batch ) prompt self.summary_template.format( existing_summaryself.summary, new_linesnew_lines ) self.summary self.llm(prompt) def get_context(self) - list: context [] if self.summary: context.append({ role: system, content: f以下是这场对话的早期内容摘要请作为背景信息\n{self.summary} }) context.extend(self.recent_messages) return context这里最关键的技巧是摘要是渐进更新的不是一次性把整个历史喂给模型。每次只拿最老的一批消息和已有摘要合并生成新摘要这样模型每次处理的 Token 量是可控的不至于在做摘要时自己先把窗口爆掉。踩过的坑是摘要生成时的信息折损。摘要天然会丢失细节如果你把用户说过的电话号码存在摘要里压缩两次之后就变成了一串错误数字。所以我的纪律是摘要只承载“全局背景”任何精确数据必须进结构化记忆。3.2 结构化工作内存摘要适合保存模糊背景但任务执行中更需要的是精确状态。比如用户正在让 Agent 帮忙写周报你至少得记住项目名称、截止日期、已确定的段落结构。这些信息放进摘要里既不精确也不好用更好的做法是单独维护一个结构化对象注入 Prompt 时用 JSON 呈现。class WorkingMemory: 结构化短期记忆保存当前任务状态、用户临时偏好、待办清单 def __init__(self): self.task_state {} self.user_preferences {} self.todo_list [] self.data_slots {} def update_task_state(self, key: str, value): self.task_state[key] value def set_preference(self, key: str, value): self.user_preferences[key] value def add_todo(self, item: str): self.todo_list.append(item) self.todo_list self.todo_list[-10:] # 最多保存 10 个待办 def set_data_slot(self, key: str, value): 数据槽用来保存用户提到的结构化信息比如姓名、日期 self.data_slots[key] value def get_context_block(self) - str: import json context { current_task: self.task_state, known_facts: self.data_slots, user_temporary_preferences: self.user_preferences, pending_todos: self.todo_list } return 【当前任务状态】\n json.dumps(context, ensure_asciiFalse, indent2)为什么要单独搞一套结构化工作内存而不是全部塞进摘要两个原因一是精确检索你需要能直接从记忆里拿到“截止日期是周五”而不是让模型去摘要里推断二是便于更新用户中途改口说“截止日期改到下周”你直接覆盖 data_slots[deadline] 就行摘要模式做不到这种精准纠正。这个工作内存对象的使用方式是把它序列化成一段 JSON 拼在 system prompt 里。它天然就是给 Agent 的“便利贴”。在实际运行中Agent 的每次操作步骤都应该有一个显式的“更新便利贴”动作防止信息失真。4. 长期记忆跨会话记住用户4.1 向量检索从“存进去”到“找得回”长期记忆的核心场景是今天用户说“我喜欢简洁的回复风格”一周后用户再来Agent 应该还记得这个偏好。这类跨会话记忆没法靠摘要或滚动窗口实现必须落到独立的记忆数据库里。实现长期记忆有两条路线。第一条是结构化路线把用户偏好整理成“键值对”或“标签-属性”的形式存进数据库查询时精确匹配。第二条是语义路线把记忆片段向量化用相似度检索找回相关内容。商用系统通常两条都上但语义路线的泛化能力更强能覆盖“我记得你好像提过某个项目的技术栈”这类模糊回忆需求。先看一个轻量的向量库实现。为了这篇博文能直接跑通我用的是 SQLite 存向量 余弦相似度暴力计算。生产环境请换成 pgvector、Milvus 或 Chroma 这类专用工具原理一致。import sqlite3 import numpy as np import json class VectorMemory: 基于 SQLite 的长期记忆文本片段 向量检索 def __init__(self, db_path: str agent_memory.db, embed_dim: int 768): self.conn sqlite3.connect(db_path) self.embed_dim embed_dim self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT DEFAULT fact, importance REAL DEFAULT 0.5, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_access TIMESTAMP, access_count INTEGER DEFAULT 0, embedding TEXT ) ) self.conn.commit() def store_memory(self, user_id: str, content: str, embedding: list, memory_type: str fact, importance: float 0.5): self.conn.execute( INSERT INTO memories (user_id, content, memory_type, importance, embedding) VALUES (?, ?, ?, ?, ?), (user_id, content, memory_type, importance, json.dumps(embedding)) ) self.conn.commit() def search_memories(self, user_id: str, query_embedding: list, top_k: int 5, min_score: float 0.6) - list: cursor self.conn.execute( SELECT id, content, importance, embedding FROM memories WHERE user_id ?, (user_id,) ) query_vec np.array(query_embedding, dtypenp.float32) results [] for row in cursor: mem_id, content, importance, emb_str row emb np.array(json.loads(emb_str), dtypenp.float32) # 余弦相似度 cos_sim np.dot(query_vec, emb) / ( np.linalg.norm(query_vec) * np.linalg.norm(emb) 1e-8 ) # 综合得分相似度 重要性加权 final_score cos_sim * 0.7 importance * 0.3 if cos_sim min_score: results.append({ id: mem_id, content: content, importance: importance, score: float(final_score) }) results.sort(keylambda x: x[score], reverseTrue) return results[:top_k] def update_access(self, mem_id: int): 更新访问时间与次数供遗忘策略消费 self.conn.execute( UPDATE memories SET access_count access_count 1, last_access CURRENT_TIMESTAMP WHERE id ?, (mem_id,) ) self.conn.commit()把这段代码和会话记忆对比核心差异就在 embedding 这一步。会话记忆是顺序拼接长期记忆是语义检索。query 进来以后先算一次 embedding然后去库里找相似片段。不同的 embedding 模型维度不一样代码里的 embed_dim 参数就是为这个准备的。4.2 记忆的写入时机不是所有的对话都值得记长期记忆最隐蔽的坑是写入策略。我看到很多团队直接把每轮对话全部向量化入库结果向量库越来越臃肿检索噪音越来越大。正确的做法是写之前先判断这条信息是否有跨会话价值。什么样的信息值得进入长期记忆我的判断标准有三个用户主动表达的偏好“我更喜欢你说话简洁点”、关键事实“我在某某公司负责增长”、明确说过要记住的内容“请记住这个地址”。”判断可以交给 Agent 自己来完成——每次对话结束时让 Agent 过一遍消息列表输出候选记忆列表和重要性打分再执行写入。class MemoryExtraction: 从对话中抽取值得长期记忆的内容 def __init__(self, llm): self.llm llm def extract_candidates(self, dialogue: str) - list: prompt f 这是一段用户与 AI 助手的对话。请从中抽取值得长期记住的信息。 值得记住的信息包括 1. 用户主动表达的偏好、习惯、风格要求 2. 用户提到的个人与工作背景 3. 用户明确要求记住的事项 4. 正在进行的重要项目及其关键参数 不需要记录的内容包括 1. 与当前任务无关的寒暄 2. 一次性任务的中间计算 3. 用户临时指定、但明确表示只此一次的内容 对话内容 {dialogue} 请按如下 JSON 格式返回列表 [{{content: 用户偏好简洁回复风格, importance: 0.8, memory_type: preference}}] response self.llm(prompt) return parse_json(response)这个抽取器让我想明白了记忆系统的本质记忆不是一个存储问题而是一个判断问题。存储最多影响容量判断决定质量。商用 Agent 的记忆库宁可少存也不要存垃圾——多存一条垃圾信息的代价不只是存储成本而是所有后续检索都会被噪音污染最终表现就是 Agent 回答问题变得东拉西扯。4.3 记忆注入检索出来还要用得上长期记忆检索出来之后注入也有讲究。我见过直接把几大段检索结果拼进 system prompt 的操作结果模型关注的是中间某条不相干的记忆真正用户关心的那条反而被淹没。所以注入时必须做“重排裁剪”。常见做法是给每条检索结果加一个相关性标签或来源说明让 Agent 知道哪些信息是“记忆检索得到的历史背景”哪些是“用户本次对话中的直接指令”。以我的项目为例注入格式大致如下def build_memory_prompt(retrieved_memories: list) - str: if not retrieved_memories: return memory_blocks [] for i, mem in enumerate(retrieved_memories): # 按记忆类型添加标签帮助模型区分来源 type_label { preference: 用户偏好, fact: 用户背景, task: 历史任务, }.get(mem.get(memory_type, fact), 历史信息) memory_blocks.append(f{i1}. [{type_label}] {mem[content]}) return ( 从用户的长期记忆中检索到以下与当前对话相关的历史信息。 这些信息可能是较早之前记录的请作为背景参考但不要编造不存在的细节\n \n.join(memory_blocks) )记忆注入时的顺序也很重要。放在 system prompt 较后位置的记忆比放在前面的更容易被模型注意到。另外每次注入的条数不要超过 5 条检索分数低于阈值的宁可不注入。你的目标是给模型吃“干粮”不是给它端上一锅乱炖。5. 遗忘机制会忘事的 Agent 才可靠5.1 遗忘不是缺陷是功能聊完记忆的存取最后这块往往被忽略——遗忘机制。很多开发者觉得记忆系统只要“存得进去、找得回来”就完事了但商用场景下不会遗忘的记忆系统是定时炸弹。三个理由一是成本无限增长的向量库和关系表会让检索越来越慢二是质量旧的不相关记忆会像水军一样刷屏把真正有用的信息淹没三是合规用户隐私条例和监管法规都要求你提供数据删除能力。遗忘机制做得好其实是在帮记忆系统做“注意力管理”。人类大脑就是这么运作的——你记不住去年某天午餐吃了什么这是优点不是缺陷因为大脑把资源留给了更重要的信息。Agent 也一样。5.2 显式遗忘与 TTL 过期遗忘机制第一层是显式遗忘。这条最直接用户说“把你记住的我的信息都删了吧”你必须做到。这种删除不是把向量库里的记录抹掉那么简单而是要连同相关的会话记录、摘要缓存、工作内存一起清干净。合规层面的显式遗忘要做到“可追溯”也就是你至少得知道删了什么、什么时候删的。class ForgetMechanism: 遗忘机制显式删除 TTL 过期 LRU 淘汰 def __init__(self, conn, llm_memory_storeNone): self.conn conn self.llm_memory_store llm_memory_store def explicit_forget(self, user_id: str, keywords: str ALL): 用户主动要求删除记忆 if keywords ALL: cursor self.conn.execute( SELECT id, content FROM memories WHERE user_id ?, (user_id,) ) ids [row[0] for row in cursor.fetchall()] else: # 支持按关键词删除比如用户说“把我关于项目A的记忆删掉” cursor self.conn.execute( SELECT id, content FROM memories WHERE user_id ? AND content LIKE ?, (user_id, f%{keywords}%) ) ids [row[0] for row in cursor.fetchall()] if ids: format_ids ,.join(? * len(ids)) self.conn.execute( fDELETE FROM memories WHERE id IN ({format_ids}), ids ) self.conn.commit() return {deleted_count: len(ids)} return {deleted_count: 0} def ttl_forget(self, ttl_days: int 90): 按时间过期超过 N 天的记忆自动清理例外重要记忆 cursor self.conn.execute( SELECT id, content, importance, last_access FROM memories WHERE created_at datetime(now, ?) AND importance 0.7 , (f-{ttl_days} days,)) ids [row[0] for row in cursor.fetchall()] if ids: format_ids ,.join(? * len(ids)) self.conn.execute(fDELETE FROM memories WHERE id IN ({format_ids}), ids) self.conn.commit() return {expired_count: len(ids)} def lru_forget(self, keep_top_n: int 5000): 容量控制超过容量上限时把从未访问且重要性低的记忆清掉 实际项目里往往用 LRU最近最少使用或 LFU最不经常使用策略 cursor self.conn.execute( SELECT id, content, access_count, last_access, importance FROM memories ORDER BY last_access IS NULL DESC, last_access ASC, importance ASC ) rows cursor.fetchall() if len(rows) keep_top_n: return {evicted_count: 0} evict_list rows[keep_top_n:] evict_ids [r[0] for r in evict_list] format_ids ,.join(? * len(evict_ids)) self.conn.execute(fDELETE FROM memories WHERE id IN ({format_ids}), evict_ids) self.conn.commit() return {evicted_count: len(evict_ids)}我把遗忘策略分成了三条线显式删除走的是用户指令TTL 走的是时间维度LRU 走的是容量维度。三条线并行共同维持记忆库的健康水位。实际经验里TTL 遗忘要特别注意“重要性”这个字段。直接一刀切“90 天前的都删”会误删重要记忆。比如用户一年前提过的“我是公司的合规负责人”这条信息虽然旧但很重要影响后续很多交互。所以我加了条件importance 低于 0.7 才参与过期淘汰。这个阈值不建议定得太高否则记忆库永远清理不干净。5.3 可解释性让遗忘成为一个可审计事件商用 Agent 的遗忘机制还必须考虑可审计性。不是说删完就完了你得回答这三个问题删了什么、为什么删、什么时候删的。尤其是涉及用户隐私数据时监管和合规审计会问到这些细节。最简单的做法是引入一张 forget_log 表CREATE TABLE IF NOT EXISTS forget_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, memory_ids TEXT, reason TEXT, trigger_source TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );每次删除操作都把动机和对象记录在案。做这件事的意义不是满足工程洁癖而是给 Agent 建立信任基础——用户有权知道系统记得什么、忘了什么。我甚至在部分项目里做了一个“记忆管理”面板用户可以在前端看到自己的记忆列表并手动删除任意一条。这个功能上线之后用户对 Agent 的信任度明上升了不少。6. 落地这些记忆模块的实战经验6.1 记忆层之间的数据流转前面每一层单独看了现在把整个链路串起来。一次典型的 Agent 对话流程是这样的用户发消息 → 从 Redis 读取会话记忆 → 构建短期记忆摘要结构化工作内存 → 依据当前问题从向量库检索长期记忆 → 拼装最终 Prompt 发给模型 → 模型返回响应 → 把新消息写入会话记忆 → 在合适的时机触发摘要压缩 → 抽取值得长期记忆的片段写入向量库 → 按策略执行遗忘清理。这个链路里的每一个环节都可能成为故障点。团队在场最容易出问题的不是某个记忆模块本身而是模块之间的衔接时机。比如长期记忆的抽取不该在每轮对话后都执行否则模型每轮都要额外跑一次抽取任务成本和延迟都吃不消。我通常设置的节奏是对话结束用户不再发消息超过 5 分钟或任务完成节点执行抽取而不是每轮都抽。6.2 令牌预算和持久化选型的工程建议关于记忆系统的工程选型我按项目规模给三个档位的建议原型验证单机内存 SQLite 就够。别一上来就上向量数据库先用暴力余弦相似度跑通逻辑重点是验证记忆策略是否有效。小规模商用Redis 存会话记忆PostgreSQL pgvector 存长期记忆。这个组合可以支撑几十万用户的量级运维复杂度也不高。大规模商用独立记忆服务Milvus/Weaviate 做向量检索消息队列做记忆抽取的异步管道遗忘机制做成定时任务或事件驱动任务。选型的唯一原则是不要为了架构而架构。先让你的记忆逻辑在业务上跑通再考虑扩展。6.3 每一层值得记录的“坑”层级典型问题我的解决经验会话记忆窗口裁剪把关键用户信息丢了对 user.role 为主的消息优先保留裁剪时把 assistant 的纯寒暄消息先丢短期记忆摘要越压越失真结构化字段和摘要分离精确数据走结构化模糊背景走摘要长期记忆向量检索召回大量噪声用重要性和时间衰减加权阈值宁高勿低遗忘机制误删高价值记忆遗忘策略必须引入重要性门槛重要记忆不参与自动过期这个表格里的每条都对应具体代码里的一个分支是实打实踩出来的。6.4 调试记忆系统的好用手段记忆系统是黑盒里最黑的一部分调试起来很痛苦。我的建议是必须是可见的。三个手段让记忆系统变得透明。第一记录每次请求的最终 Prompt 快照。把拼接好的完整 Prompt 存一份日志出了问题可以直接回放看到底是检索错了还是注入顺序错了。第二给检索召回的记忆打上分数显示。前端或日志里把每条记忆的相关性分数展示出来你能直接看出召回质量为什么差。第三做 A/B 测试时单独开关记忆模块。加一个全局配置项可以一键关闭长期记忆、只保留会话记忆方便对比记忆模块对业务指标的影响。调试用的日志我一般用 JSON Lines 格式每行一个事件结构化字段包含记忆层级、用户 ID、Session ID、操作类型、Token 数。配合日志检索系统可以快速定位某一类记忆异常。6.5 从代码到商用的最后一步最后补充一点上面所有代码都还只是记忆引擎的内核距离商用还差一层服务化封装。你需要把记忆模块做成独立服务对外提供 HTTP 接口或者 SDK——get_context、save_memory、search_memory、forget——这样业务层不用关心记忆是怎么存的只需要调用接口。这个抽象层的价值在团队协作时尤其明显业务开发的同事不需要知道你是用向量库还是 Redis 实现的他只需要保证调接口时塞对参数。说实话我见过很多团队在记忆系统这个环节翻车翻得最多的不是技术选型而是“不重视”。Agent 应用跑 Demo 的时候单轮短对话根本看不出记忆系统的差距一旦进入真实商用用户会持续使用好几天记忆系统的质量直接决定了 Agent 的专业度。一个会记得用户偏好和历史的 Agent和一个每次交流都像第一次见面的 Agent用户体验差着数量级。我做这个系统最深的体会是记忆架构的本质不是存储而是“取舍”。你要决定什么信息值得留、什么信息必须忘、什么信息怎么被想起来这些决策才是记忆系统的灵魂。按文章里的四层结构动手做一遍你会对 Agent 的能力边界有个全新的认识。