1. 从一次“失忆事故”说起Agent 记忆到底缺在哪做 Agent 开发的朋友十有八九都经历过这种崩溃现场用户上一轮刚告诉你“我叫老张做跨境电商的主攻欧美市场”下一轮问“你记得我是做什么的吗”Agent 一脸茫然地开始瞎编。更离谱的是有些 Agent 在同一个会话里都能“翻脸”——前脚刚确认完订单信息后脚就问你订单编号是多少。这真的不是模型笨也不是 prompt 写得不好而是记忆架构压根没设计。我这两年帮好几家团队调过商用 Agent几乎每次排查“健忘”问题最后都落到同一个工程根源上记忆被当成一整块聊天记录来对待而没有按生命周期去拆分和管理。先说清楚“健忘”发生在哪里。最普遍的一层是模型的上下文窗口天然有限——哪怕现在 Claude、GPT 这类模型的上下文做到 200K token你也不可能把所有历史对话全部塞进 prompt。塞不下是一层问题塞进去之后互相干扰、关键信息被淹没是更隐蔽的另一层问题。我见过有团队图省事把全部历史消息拼进系统提示词结果 token 费用按月暴涨回答质量反而明显下降。原因很简单信息装进上下文不等于被模型“记住”在超长上下文中提取关键信息的精度会肉眼可见地衰减。商用场景里对记忆的真实需求通常可以归纳成四类会话连续性同一个会话内上下文连贯、用户画像沉淀跨会话记住用户偏好和身份信息、业务状态持久化订单、任务、进度这类状态数据不能丢、知识库接入让 Agent 基于企业文档做问答。如果只靠模型天然上下文这四类需求一个都不可靠。所以正确做法是把记忆拆成不同层级每层管自己擅长的事再用一套编排逻辑把它们串起来。这套东西就是下文要展开的“会话记忆—短期记忆—长期记忆—遗忘机制”全链路。2. 三层记忆架构怎么拆会话、短期、长期各管一摊2.1 为什么必须分层而不是一把梭很多人第一反应是“搞个 Redis 存聊天记录不就行了”这就是典型的把记忆等同于存储。记忆系统真正难的不是“存得下”而是“取得到”和“取对了”。分层设计背后的逻辑来自一个核心矛盾信息的时效性需求不同。用户刚说的“现在帮我查一下 A 订单”这是秒级时效必须立刻可用用户三天前透露的“我偏好晚上处理邮件”这是长期画像要在合适的场景被重新唤醒还有大量中间态信息——比如某个任务进行到一半的临时状态既不能丢又不能长期占着位置。时效性不同存储格式、检索方式、过期策略就全都不一样。放在一个桶里要么检索效率极低要么互相污染。我习惯把记忆分成三层。会话记忆Conversation Memory负责当前会话内的消息序列保底保证“接得上话”短期记忆Short-term Memory负责跨轮次但不过期的“工作记忆”比如用户正在进行的任务流程、刚确认的临时偏好通常自带 TTL长期记忆Long-term Memory负责结构化沉淀的用户画像、业务事实和历史关键时刻需要主动提取和写入。这三层在存储上天然分开在读取时按优先级合并遗忘策略也各有各的节奏。2.2 每一层该存什么、不该存什么这层我踩过最大的坑就是“什么都往长期记忆里写”。看起来万无一失实际上会把检索质量拖垮。分层不是简单按时间切而是按信息类型切。会话记忆只存原始消息和消息附带的元数据时间戳、角色、消息 ID原则上不改写内容只是做截断或摘要。短期记忆存的是“流程状态”和“临时结论”比如用户正在填的表单、当前选中的筛选条件、上一步操作产生的结果 ID。长期记忆存的是“事实”包括用户身份画像称呼、行业、偏好、业务关键记录订单历史、已确认的需求、以及经过提炼的“关键结论”比如用户明确说过的某条决策原因。一个简单判断标准这条信息如果三条之后还用得上放短期如果下次会话还用得上放长期如果只是当前对话的铺垫留在会话记忆里就行。按这个标准过滤长期记忆的体量能压缩掉 70% 以上检索精度完全不是一个量级。我还习惯给每层记忆加“写入审计”记录是哪个环节、基于哪条原始消息写入的出了问题好回溯——商用环境里这条对排查“记忆污染”极其重要。3. 全链路代码实战从消息存取到主动遗忘下面这部分我直接给出可运行的参考实现。为便于演示我用 Python 写存储层对 Redis 和向量库做了抽象模型调用统一走一个llm_complete函数实际项目里替换成 OpenAI SDK 或 Claude SDK 即可。3.1 会话记忆消息序列的组织与滑动窗口会话记忆最朴素的实现就是把消息追加进列表拼 prompt 时取最近 N 轮。但商用环境有个细节不能只按“轮”截断要按 token 数截断。因为用户一句话可能写 2000 token也可能一个字“好”。我见过按条数截断导致的上下文爆炸事故所以老实上 tiktoken 或对应模型的 tokenizer 做预算。from dataclasses import dataclass, field from typing import List, Dict, Optional import time, uuid, json dataclass class Message: role: str # user / assistant / system / tool content: str msg_id: str field(default_factorylambda: uuid.uuid4().hex) ts: float field(default_factorytime.time) class ConversationMemory: 会话记忆维护消息序列按 token 预算做滑动窗口截断 def __init__(self, max_tokens: int 8000, tokenizerNone): self.messages: List[Message] [] self.max_tokens max_tokens self._tokenize tokenizer or (lambda s: len(s) // 2) # 退化方案按字符估算 def append(self, role: str, content: str): self.messages.append(Message(rolerole, contentcontent)) self._enforce_budget() def _enforce_budget(self): 从旧到新丢弃消息直到总 token 数低于预算 total sum(self._tokenize(m.content) for m in self.messages) drop_count 0 while total self.max_tokens and len(self.messages) 1: dropped self.messages.pop(0) total - self._tokenize(dropped.content) drop_count 1 if drop_count: self._archive_to_short_term(drop_count) def _archive_to_short_term(self, drop_count: int): 被挤掉的消息不直接丢弃交给短期记忆做摘要——这里先留个钩子 # 实际实现见 3.3会把最早的一批消息送给 LLM 做摘要 pass def to_prompt_messages(self) - List[Dict[str, str]]: return [{role: m.role, content: m.content} for m in self.messages]这里最关键的设计是_archive_to_short_term。很多人直接把旧消息删掉导致用户上一句话“请基于我刚才发的需求文档回答”直接失效。正确做法是被挤出的旧消息进入压缩通道让 LLM 生成摘要后存入短期记忆这样既控制 token 预算又不丢关键信息。压缩摘要的触发点正是截断发生的瞬间而不是定时任务——因为只有这个时候我们才知道哪些信息即将“离开”上下文。3.2 短期记忆带 TTL 的工作记忆与 token 预算短期记忆我常用 Redis 来存字段结构非常简单key 是short_term:{user_id}:{session_id}value 是一组带优先级的记忆条目。每个条目有四个关键属性内容、优先级、写入时间、过期时间。优先级决定了上下文紧张时谁先被保留也决定了“写入 token 预算”时谁先被选中。class ShortTermMemory: 短期记忆带 TTL 和优先级的工作记忆用于流程状态与临时结论 def __init__(self, redis_client, ttl_seconds: int 1800, max_items: int 20): self.r redis_client self.ttl ttl_seconds self.max_items max_items def set(self, key: str, content: str, priority: int 5): 写入一条短期记忆。priority 越高越优先保留。 item { content: content, priority: priority, write_ts: time.time(), expire_ts: time.time() self.ttl, } self.r.hset(fst:{key}, item[content], json.dumps(item)) self._enforce_max_items(key) def get_active(self, key: str) - List[Dict]: 获取未过期且未失效的短期记忆 raw_items self.r.hgetall(fst:{key}) active [] for _, raw in raw_items.items(): item json.loads(raw) if item[expire_ts] time.time(): active.append(item) return sorted(active, keylambda x: -x[priority]) def _enforce_max_items(self, key: str): 条目超限时优先淘汰低优先级且较旧的记忆 all_items self.get_active(key) if len(all_items) self.max_items: return # 排序优先级升序、时间升序先淘汰优先级低且老的 to_drop sorted(all_items, keylambda x: (x[priority], x[write_ts])) for item in to_drop[: len(all_items) - self.max_items]: self.r.hdel(fst:{key}, item[content])短期记忆的 TTL 我一般设 30 分钟原因是商用 Agent 的典型会话间隔不会太长30 分钟足够覆盖“用户离开一会再回来继续操作”的场景。如果用户超过这个时间没回来这些临时状态本身就失去意义直接过期反而是好事——避免旧状态干扰新会话。优先级字段是实战里加出来的最初版本没有结果发现任务流程中的“当前步骤状态”经常被“用户随口说的偏好”挤掉加了优先级之后任务状态永远 priority9临时闲聊一律 priority3稳定多了。3.3 长期记忆向量化存储与提取式检索长期记忆是商用 Agent 记忆体系的压舱石。我这边用 ChromaDB 做向量库embedding 用现成模型比如 text-embedding-3-small生产环境也可以换 Qdrant 或 Milvus。核心思路是只在关键节点写入在响应用户前主动检索相关记忆并注入上下文。import chromadb from chromadb.utils import embedding_functions class LongTermMemory: 长期记忆向量化存储 元数据过滤 提取式检索 def __init__(self, collection_name: str agent_long_term): self.client chromadb.PersistentClient(path./agent_memory_store) self.ef embedding_functions.OpenAIEmbeddingFunction( api_keyYOUR_KEY, model_nametext-embedding-3-small ) self.collection self.client.get_or_create_collection( namecollection_name, embedding_functionself.ef ) def write_fact(self, user_id: str, fact_type: str, content: str, importance: float 0.5): 写入一条长期记忆。fact_type 便于业务维度过滤importance 控制检索权重。 mem_id f{user_id}:{uuid.uuid4().hex} self.collection.add( ids[mem_id], documents[content], metadatas[{ user_id: user_id, fact_type: fact_type, importance: importance, write_ts: time.time(), }], ) def recall(self, query: str, user_id: str, top_k: int 5, min_score: float 0.15, fact_type: Optional[str] None) - List[str]: 按相关性检索可叠加用户维度和业务类型过滤 where {user_id: user_id} if fact_type: where[fact_type] fact_type results self.collection.query( query_texts[query], n_resultstop_k, wherewhere, include[documents, metadatas, distances], ) summaries [] for doc, meta, dist in zip( results[documents][0], results[metadatas][0], results[distances][0], ): score 1 - dist if score min_score: continue # 用 importance 做加权避免低价值信息挤占上下文 weighted_score score * (0.5 meta[importance]) summaries.append((doc, weighted_score)) summaries.sort(keylambda x: -x[1]) return [s[0] for s in summaries[:top_k]]检索这步有个细节特别值得说不要只按向量相似度排。我最初调试时经常发现模型反而被“相似但无关”的记忆误导。比如用户问订单物流检索出来的记忆全是“用户抱怨过某次物流慢”这确实是语义相似但根本不是当前需要的订单状态。后来加上fact_type过滤和importance加权之后效果立竿见影。另一个细节是写入时机我强烈建议只在“对话产生明确事实”时写入而不是每轮都写。判断逻辑很简单——用户用了陈述性语气说出身份、偏好、决策或者系统产生了任务结果这两类才值得写长期记忆。3.4 遗忘机制显式清除、自动过期与检索弱化遗忘是整套架构里最容易被忽略、却又最能拉开体验差距的部分。我见过太多 Agent“记太好”——用户三个月前一句玩笑话也被当作画像反复引用结果显得又蠢又冒犯。商用环境里遗忘不仅是技术问题还是合规和体验问题。“该忘的时候忘不掉”比“该记的时候记不住”更致命。我的遗忘机制从三个方向做。class ForgetManager: 遗忘管理主动遗忘 被动过期 检索弱化三个层次配合 def __init__(self, long_term: LongTermMemory, short_term: ShortTermMemory, redis_client): self.ltm long_term self.stm short_term self.r redis_client # 1) 显式遗忘用户说“忘掉”或业务要求清除时直接删 def explicit_forget(self, user_id: str, keywords: Optional[List[str]] None): 按用户维度或关键词维度删除长期记忆。关键词为空则全删。 all_mem self.ltm.collection.get(where{user_id: user_id}) if keywords is None: ids all_mem[ids] else: # 简单实现关键词命中即删。生产环境建议先检索再删避免误伤。 ids [ mem_id for mem_id, doc in zip(all_mem[ids], all_mem[documents]) if any(kw in doc for kw in keywords) ] if ids: self.ltm.collection.delete(idsids) # 2) 自动过期长期记忆也设保鲜期重要度高的更长寿 def collect_expired(self, importance_decay: float 0.1): all_mem self.ltm.collection.get() now time.time() expired_ids [] for mem_id, meta in zip(all_mem[ids], all_mem[metadatas]): age now - meta.get(write_ts, now) lifespan 30 * 24 * 3600 * (1 meta.get(importance, 0.5)) if age lifespan: expired_ids.append(mem_id) if expired_ids: self.ltm.collection.delete(idsexpired_ids) # 3) 检索弱化不删但让旧记忆的检索分数自然衰减 def apply_decay_to_recall(self, recall_result: List[tuple], now: float time.time()): decayed [] for doc, score, meta in recall_result: age_days (now - meta.get(write_ts, now)) / 86400 decay_factor max(0.5, 1 - age_days * 0.01) decayed.append((doc, score * decay_factor)) return decayed显式遗忘要处理一个语义问题用户说“忘掉我上次说的地址”你不能直接把整个 user_id 的记忆全清了那是一种“核弹式遗忘”。我实际开发时给 ForgetManager 加了“记忆条目级”的追溯能力——每条记忆写入时记录来源消息 ID遗忘时定位到具体消息再顺藤摸瓜删除由它派生的所有摘要。自动过期这层我设的寿命基线是 30 天importance 越高活得越久这样“用户明确强调过的重要决策”能存半年“随口一提的偏好”两周就淡出。检索弱化是第三道防线不删除数据只降低旧数据的命中概率防止冷门旧记忆在特定场景下“诈尸”。4. 记忆“翻车”现场常见问题与排查实践4.1 检索召回结果质量差模型被记忆带偏这是我在商用项目里遇到最多的问题。现象很典型用户问“帮我看看上个月的报表”Agent 检索回来的记忆却是“用户上次抱怨过报表格式不对”然后开始道歉完全不干活。排查路径我一般走三步。第一步查 embedding 的效果把检索结果和 query 打印出来人工看相似度排名是否合理如果明显不合理先怀疑 embedding 模型选型换更适配领域数据的模型。第二步查过滤条件确认是否按 user_id 和 fact_type 做了隔离很多时候是测试时没区分用户导致 A 用户的记忆被 B 用户检索到。第三步查加权逻辑importance 权重是不是太高导致语义匹配度一般但权重高的记忆压过了真正相关的记忆。我最后把加权上限压到 0.3相关性排序稳定很多。另一个容易踩的坑是检索结果直接拼 prompt 的顺序。我一开始按分数从高到低排结果模型总是优先响应最靠前的记忆哪怕它只是边缘相关。后来改成“最相关的放中间、次相关的放前后”的布局配合 prompt 里“以下记忆仅供参考以用户当前问题为准”的约束准确率明显提升。这个经验未必普适但值得在调试时花十分钟试一试。4.2 记忆污染会话里的一句话污染了长期画像假设用户在某次会话里说“我今天想取消所有订阅”这句话被写进了长期记忆之后每次对话 Agent 都在劝用户退订——这就是典型的记忆污染。根本原因是写入逻辑太激进把“临时情绪表达”当成了“稳定事实”。我在长期记忆写入前加了一道“事实性校验”钩子用 LLM 判断该句话是否满足“陈述事实、面向未来、非情绪化”只有三类可以通过——身份类我是谁、我在哪、我的职业、偏好类我喜欢/我不喜欢/我需要、业务类已完成的任务、确认的决策。情绪化表达、假设性提问、临时状态一律拦截。这个校验本身会消耗额外 token但我算过账值得。一台商用 Agent 每天处理几千次对话如果没有校验污染记忆带来的错误回答成本远比那点校验 token 高。校验通过后写入前还会做一次“去重合并”检索已有记忆中是否已有等价条目有则更新原条目的 write_ts 和 importance而不是新增一条。不然三个月后同一个偏好存了 20 条检索时全是它也是另一种污染。4.3 并发问题同一个用户的多会话互相踩记忆商用 Agent 经常一个用户同时开着多个会话——网页端一个、App 一个、企微机器人又一个。如果记忆系统不做会话隔离很容易出现“App 里改的偏好把网页端的行为记录覆盖了”。我在设计里给每条记忆加了scope字段session级记忆只在本会话可见user级记忆跨会话共享。写入时显式指定 scope读取时根据当前上下文决定合并策略——用户级记忆永远加载会话级记忆只在本会话内加载。真正麻烦的是并发写同一份用户画像。比如用户在网页端说“改邮箱”同时在 App 里说“改手机号”两个会话同时更新 user 级记忆后写的把先写的覆盖了。我在业务层处理写入用户级记忆前先读取旧值用“合并而非覆盖”的策略把不同字段分别更新。这个逻辑不复杂但真出问题时极其隐蔽尤其是企微机器人这种回调并发高的场景不加锁或合并策略用户画像三天两头莫名丢字段。参考实现里 Redis 的hset逐字段更新天然规避了这个问题——这也是我推荐用哈希结构存画像的原因。4.4 上下文膨胀与 token 费用失控很多团队问我“记忆都接上了为什么 prompt 还是那么大”。查到最后几乎都是同一类问题把“检索到的记忆”和“全部短期记忆”不加筛选地拼进上下文。三层记忆的设计初衷就是控制 token但控制必须在读取侧完成。我建议的读取策略是会话记忆控制在 60% 预算短期记忆按优先级取前 3-5 条、约占 15%长期记忆只取检索命中的前 2-3 条、约占 10%剩余 15% 留作系统指令和输出预留。这个比例不是拍脑袋是我在多个项目里调出来的经验值核心原则是记忆是参考信息不是主角主角永远是当前用户输入和相关工具结果。5. 复盘我在实际项目中沉淀的几条经验这套记忆架构我在电商客服、销售线索管理、企业知识问答三个方向的 Agent 项目里跑过说几条最实在的体会。第一记忆系统的迭代节奏永远跟着“用户可感知的错误”走而不是跟着“技术完备度”走。我最早做了很复杂的记忆图谱但用户根本感知不到差别反而增加排查成本。后来砍掉 80% 的功能只保留“会话、短期、长期、遗忘”四件套稳定性反而大幅提升。商用环境里记忆系统最重要的指标不是“多智能”而是“可预期”——用户说过的关键信息不丢用户没说过的话不乱编这就已经超越市面上九成 Agent。第二遗忘和写入同样重要甚至更重要。我接手的一个客户项目之前 Agent 被吐槽“越来越陌生”排查发现是长期记忆里积累了太多过期业务状态比如三个月前的库存数字、两个月前的促销价格检索时反复命中导致回答看起来永远在说旧数据。上了自动过期和重要性衰减之后这类错误消失了大半。如果你的 Agent 也出现“越用越蠢”的迹象先检查记忆系统是不是“只进不出”。第三所有记忆操作都要留审计日志。商用环境一旦出事你需要在 10 分钟内回答“这条错误记忆是什么时候、基于什么消息、由哪个模块写入的”。没有审计排查就是大海捞针有审计十分钟定位。这条我在多个项目里反复验证属于“平时没用、出事救命”的设计。最后分享一个调试技巧我会在开发环境给每条注入 Agent 的记忆加一行不可见的注释标记比如!-- MEM:user_profile#20240901 --这样模型回答时如果引用了某条记忆你能直接在输出里看到它引用了哪条、来自什么时候。这个技巧帮我发现过不少“模型脑补记忆”的问题——它明明没检索到任何记忆却假装记得有了标记就能一眼识破。希望这套链路和代码能帮你把那个“健忘的 Agent”治好。