请介绍一下 AI Agent 的记忆机制？实际开发中怎么设计记忆模块？

#RAG#Agent#面试八股

请介绍一下 AI Agent 的记忆机制？实际开发中怎么设计记忆模块？

大模型本身没有记忆，它每次调用都是一张白纸。所谓"Agent 的记忆"，全部是外部系统在每一轮把相关内容挑出来、塞进上下文的结果。 所以记忆机制的本质是两件事：写什么、存在哪，以及 什么时候取、取多少放回去。前者决定能力上限，后者决定成本与准确率。

1. 结论：三层记忆，各管一段时间尺度

层管多久存什么存放位置取用方式
工作记忆（Working Memory）单次任务内当前目标、已执行步骤、工具原始结果进程内存 / 上下文窗口全量拼接，按需截断
短期记忆（Short-term Memory）一次会话内最近 N 轮对话、阶段性结论会话表（Redis / MySQL）滑动窗口 + 增量摘要
长期记忆（Long-term Memory）跨会话、长期用户偏好、事实、历史决策与教训向量库 + 关系库 + KV语义召回 + 结构化过滤

三层不是替代关系，是逐层沉淀：工作记忆里的重要结论会写进短期记忆，短期记忆里稳定不变的部分会被抽取成长期记忆。

2. 机制：一次对话的读写路径

用户输入
   │
   ├─▶ ① 召回：长期记忆按 query 向量检索 top-k ─┐
   ├─▶ ② 拼接：系统提示 + 长期记忆 + 短期窗口  ─┤─▶ 送给模型
   │                                            │
   └─▶ ③ 反思/抽取：本轮是否产生值得记的事实？─┘
              │ 是
              ▼
        ④ 去重与合并 ──▶ ⑤ 写入（关系库存事实，向量库存语义）
              │
              ▼
        ⑥ 滚动：窗口超阈值则摘要最旧的一段

写入环节比读取更容易被忽略。绝大多数"记忆不好用"的案例，不是召回算法差，而是写进去的是流水账：把整段对话原文塞进向量库，检索出来的片段又长又散，模型还得自己从中提炼。

3. 粒度与存储选型

记忆内容建议粒度存储写入时机备注
用户偏好（"回答要简短"）一条断言关系库 KV首次出现即写需支持更新与失效
用户事实（"在杭州，做后端"）结构化字段关系库会话结束抽取冲突时以新值为准
历史事件（"上周排查过 Redis 雪崩"）事件摘要 + 时间向量库事件结束带时间戳便于时序召回
任务教训（"这个接口要传 X-Env"）一句话经验向量库反思通过后高价值，优先召回
原始对话单轮对象存储 / 日志实时仅用于回溯，不进向量库

关键设计：长期记忆只存"结论"，不存"过程"。过程留在日志里，靠 trace 查；进记忆的必须是可复用、可验证的短句。

4. 代码：一个够用的记忆模块

接口只要四个方法：remember、recall、forget、summarize。先写接口，再换实现，是这类模块最容易演进的方式。

class Memory:
    def __init__(self, store, embedder, window=20, budget_tokens=6000):
        self.store, self.embedder = store, embedder
        self.window, self.budget = window, budget_tokens

    def remember(self, user_id, text, kind="fact", ttl_days=180):
        """写入前先去重：同义或高度相似的条目做合并，而不是追加"""
        vec = self.embedder(text)
        hit = self.store.search(user_id, vec, top_k=1)
        if hit and hit[0].score > 0.92:
            self.store.update(hit[0].id, text=text)      # 覆盖而非新增
        else:
            self.store.insert(user_id, text, vec, kind=kind, ttl_days=ttl_days)

    def recall(self, user_id, query, top_k=5, min_score=0.35):
        vec = self.embedder(query)
        items = self.store.search(user_id, vec, top_k=top_k * 2)
        items = [i for i in items if i.score >= min_score and i.kind != "raw"]
        return self.fit(items, top_k)                    # 超预算就截断

    def build_context(self, user_id, query, history):
        facts = self.recall(user_id, query)              # 长期：按需
        turns = history[-self.window:]                   # 短期：滚动窗口
        return {"system_extra": render(facts), "messages": turns}

配套的存储表，两个索引就能覆盖绝大多数场景：

CREATE TABLE agent_memory (
  id          BIGINT PRIMARY KEY,
  user_id     VARCHAR(64)  NOT NULL,
  kind        VARCHAR(16)  NOT NULL,   -- fact / event / lesson / preference
  content     TEXT         NOT NULL,
  embedding   VECTOR(1024) NOT NULL,   -- 语义召回
  created_at  DATETIME     NOT NULL,
  expire_at   DATETIME     NULL,       -- TTL，过期自动清理
  KEY idx_user_kind (user_id, kind),
  VECTOR INDEX idx_vec (embedding)
);

kind 字段很关键：召回时可以按类型配比，比如偏好最多 2 条、事件最多 3 条，避免某一类记忆霸占上下文。

5. 常见追问

问：短期记忆为什么不能只靠上下文窗口？ 窗口再大也有上限，而且长上下文的"中间遗忘"（lost in the middle）很明显；更现实的是成本——每轮都把全部历史发一遍，费用随轮数平方增长。

问：什么时候做摘要？ 按 token 阈值触发（如窗口使用率超 70%），而不是每轮都做。摘要保留"结论、决定、未完成事项"，丢弃寒暄与重复确认。

问：记忆冲突怎么办？ 写一条覆盖一条，并在条目上带 updated_at 与置信度，召回时同主题只取最新或置信度最高的。不要指望模型自己从两条矛盾事实里挑对。

问：怎么防止记忆被污染（用户诱导写入错误事实）？ 写入前过一道判定：只允许从"用户明确陈述"或"工具返回结果"中抽取，且与既有记忆冲突时不覆盖高置信条目，转入待确认队列。生产环境还应支持用户查看和删除自己的记忆。

问：多 Agent 共享记忆吗？ 共享"事实层"，不共享"工作记忆"。工作记忆按 Agent 隔离，事实层用同一份带命名空间的存储，避免互相踩脚。

小结

Agent 记忆 = 分层存储 + 精准写入 + 按预算召回。工程上先把"写什么"定清楚（结论而非流水账），再把"取多少"用 token 预算卡住，最后补上去重、冲突处理和 TTL。比起换更强的向量库，这三件事对效果的提升要直接得多。

笔记加载中…