Agent 记忆系统设计：长期记忆与上下文管理的工程方案

Agent 记忆系统设计：长期记忆与上下文管理的工程方案

一、健忘的 Agent：重复对话背后的记忆缺位

用过 Agent 的人都有个体感：它今天教过的东西，明天就忘。同一个问题反复问，每次都从头解释。长对话进行到第二十轮，前面提到的关键约束已经丢失。这不是模型不够聪明，是它根本没有"记忆"这一层。

大模型本身是无状态的。每一次推理，都是一次从零开始的计算。所谓的"对话历史"，其实是把上文重新塞进上下文窗口。窗口一满，老内容就被截断或压缩。

信息流失不可避免。生产场景里，这个问题更刺眼。客服 Agent 记不住用户上周的诉求。编程 Agent 记不住项目约定的代码风格。

运维 Agent 记不住上个月处理过同类故障的方案。每一次都在重复踩坑，每一次都在重新发现轮子。记忆系统要解决的不只是"存下来"。还要解决"什么时候检索"、"检索什么"、"什么时候遗忘"。

存而不检，等于没存；检而不准，等于噪声。不遗忘，记忆库会越来越脏，最终拖垮检索质量。本文探讨 Agent 记忆系统的工程方案。从短期、工作、长期三层记忆切入，落到检索与遗忘策略。目标是让 Agent 真正"积累经验"，而不是每次从零开始。

二、分层记忆：短期、工作、长期的协作机制

记忆不是单一容器，而是分层结构。不同层负责不同时间尺度的信息。各司其职，才能兼顾速度与容量。

短期记忆即上下文窗口。存放当前对话的近期消息。读取零延迟，但容量受 token 限制。越靠后的内容越容易被截断。

工作记忆是 scratchpad 草稿区。Agent 在推理过程中产生的中间结论。比如"用户提到的约束"、"已尝试过的方案"。不进对话历史，但参与下一轮决策。

长期记忆是持久化的向量库。跨会话存储事实、偏好、历史经验。通过语义检索按需调取。容量几乎无限，但检索有延迟和成本。

三者协作的链路如下：

flowchart TD A[用户输入] --> B[短期记忆: 上下文窗口] A --> C[工作记忆: scratchpad] A --> D[长期记忆: 向量检索] B --> E[拼装 Prompt] C --> E D --> E E --> F[模型推理] F --> G[输出响应] F --> H[更新工作记忆] F --> I[写入长期记忆] style D fill:#fff3e0 style I fill:#e8f5e9

关键在"按需检索"。不是每次都把长期记忆全塞进 prompt。而是先用用户输入做语义检索，只取相关的几条。否则 prompt 臃肿，token 成本飙升，模型注意力也被稀释。

检索质量决定记忆系统的上限。向量召回不准，相关记忆取不到。召回过多，噪声淹没信号。需要在 top-k 数量、相似度阈值、rerank 之间找平衡。

遗忘同样关键。过时信息不清理，会污染检索结果。比如用户上周说要 A 方案，这周改主意要 B 方案。旧记忆不遗忘，Agent 还会按 A 行事。需要有 TTL、版本标记、显式覆盖等机制。

三、生产级实现：分层记忆管理器

下面用 Python 实现一个分层记忆管理器。聚焦长期记忆的写入与检索，含错误处理与遗忘策略。

import time import hashlib from dataclasses import dataclass, field from typing import Optional @dataclass class MemoryItem: """长期记忆的单条记录：内容、元数据、过期时间""" content: str scope: str # 所属会话或用户标识 tags: list[str] = field(default_factory=list) created_at: float = field(default_factory=time.time) ttl: Optional[float] = None # 秒；None 表示永久 content_hash: str = "" def __post_init__(self) -> None: # 用内容哈希做去重键，避免相同事实反复写入 self.content_hash = hashlib.md5(self.content.encode()).hexdigest() class LongTermMemory: """长期记忆库：向量检索 + TTL 遗忘 + 去重""" def __init__(self, embed_fn, top_k: int = 5, min_score: float = 0.6) -> None: # embed_fn 由调用方注入，便于切换模型与 mock 测试 self._embed = embed_fn self._store: dict[str, MemoryItem] = {} self._vectors: dict[str, list[float]] = {} self.top_k = top_k self.min_score = min_score def add(self, item: MemoryItem) -> bool: # 去重：同 scope 下相同内容不重复写入 key = f"{item.scope}:{item.content_hash}" if key in self._store: return False try: vec = self._embed(item.content) except Exception as e: # 嵌入失败不阻断主流程，仅记录并跳过 # 真实系统接日志与告警，这里简化 print(f"[memory] embed failed: {e}") return False self._store[key] = item self._vectors[key] = vec return True def search(self, query: str, scope: str) -> list[MemoryItem]: try: q_vec = self._embed(query) except Exception as e: print(f"[memory] query embed failed: {e}") return [] scored = [] for key, item in list(self._store.items()): if not item.scope == scope: continue # 过期清理：检索时顺手剔除 TTL 到期的记忆 if item.ttl and time.time() - item.created_at > item.ttl: self._store.pop(key, None) self._vectors.pop(key, None) continue score = self._cosine(q_vec, self._vectors[key]) if score >= self.min_score: scored.append((score, item)) # 按相似度降序取 top_k scored.sort(key=lambda x: x[0], reverse=True) return [item for _, item in scored[: self.top_k]] @staticmethod def _cosine(a: list[float], b: list[float]) -> float: # 余弦相似度：向量检索的基础度量 # 真实系统用 faiss 或 milvus，这里内联便于阅读 dot = sum(x * y for x, y in zip(a, b)) na = sum(x * x for x in a) ** 0.5 nb = sum(y * y for y in b) ** 0.5 if na == 0 or nb == 0: return 0.0 return dot / (na * nb) if __name__ == "__main__": # mock 一个嵌入函数，真实环境替换为线上模型 def fake_embed(text: str) -> list[float]: return [float(len(text)), float(text.count(" "))] mem = LongTermMemory(embed_fn=fake_embed, top_k=3, min_score=0.5) mem.add(MemoryItem(content="用户偏好 Python", scope="u1", ttl=3600)) mem.add(MemoryItem(content="用户偏好 Python", scope="u1")) # 去重，不会写入 print(f"命中 {len(mem.search('用户喜欢什么语言', 'u1'))} 条")

真实系统会在几个方向上扩展。向量库换成 faiss、milvus 或 pgvector，支撑百万级记忆。检索前加 rerank 模型，提升 top-k 的精度。写入前做事实抽取，只存结构化结论，不存原始对话。遗忘策略叠加"显式覆盖"：用户改主意时，主动作废旧记忆。

四、Agent 记忆系统设计的代价与边界

记忆系统能力强，但坑也深。

召回质量不稳。向量相似度高不等于语义相关。"用户喜欢 Python"和"用户讨厌 Python"向量很近。需要配合 rerank 或基于元数据的过滤。

否则错误记忆会误导决策。

记忆污染。Agent 把幻觉或错误结论写进长期记忆。下一轮又检索出来当事实用，错误被放大。写入前必须做事实校验或人工审核。

高风险记忆要标记"待确认"状态。

隐私与合规。长期记忆里可能沉淀用户敏感信息。跨用户共享、跨境传输、留存周期都成问题。需要按 scope 严格隔离，敏感字段脱敏。

并支持按用户请求彻底删除。

成本与延迟。每次检索都要做向量计算和 rerank。高频对话下，记忆系统的开销可能超过推理本身。需要对热查询做缓存，对冷记忆做分层存储。

记忆系统的"治理"比"建库"更难。把记忆写进去只是第一步，真正的工程量在长期运营：定期审计记忆库里的脏数据、监控召回质量指标、处理用户的"遗忘请求"。另一个被忽视的点是"记忆的可解释性"：Agent 引用了一条记忆做决策，必须能追溯到来源时间与上下文，否则出错时无法定位是哪条记忆在作祟。最后，记忆系统要预留"关停开关"，在召回质量下滑或合规风险出现时，能快速降级为无记忆模式，保住主链路可用。

五、总结

Agent 的记忆系统，本质是让无状态模型获得跨会话的经验积累。机制上用分层结构兼顾速度与容量，用向量检索按需调取。工程上靠去重、TTL、rerank 守住召回质量与成本。落地路线：先上短期记忆管理上下文窗口；再接工作记忆做中间结论暂存；长期记忆用向量库加 rerank；最后补审计、隔离与遗忘治理。记忆不是越多越好，而是越准越好。