商用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_ascii=False) 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 = f"session:{session_id}" # Redis 列表尾部追加消息 self.client.rpush(key, json.dumps(message, ensure_ascii=False)) self.client.expire(key, self.default_ttl) # 防止列表无限增长，设置最大长度 self.client.ltrim(key, -200, -1) # 最多保留最近 200 条 def load_messages(self, session_id: str) -> list: key = f"session:{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_summary=self.summary, new_lines=new_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_ascii=False, indent=2)

为什么要单独搞一套结构化工作内存，而不是全部塞进摘要？两个原因：一是精确检索，你需要能直接从记忆里拿到“截止日期是周五”，而不是让模型去摘要里推断；二是便于更新，用户中途改口说“截止日期改到下周”，你直接覆盖 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, dtype=np.float32) results = [] for row in cursor: mem_id, content, importance, emb_str = row emb = np.array(json.loads(emb_str), dtype=np.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(key=lambda x: x["score"], reverse=True) 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"{i+1}. [{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_store=None): 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( f"DELETE 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(f"DELETE 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(f"DELETE 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 的能力边界有个全新的认识。