阅读时间：约 15 分钟
前置知识：理解 Agent 决策循环（P01）、上下文管理（P03）

---

P04 让 Agent 稳定了。但关掉对话再打开，Agent 什么都忘了。用户说"上次分析的那个方案"，它一脸茫然。怎么让它记住？

---

前言：记忆决定了 Agent 的上限

2026 年的一篇综述论文给出了一个结论："有记忆"和"没记忆"之间的差距，往往比不同 LLM 底座之间的差距还要大。

数据更直接：去掉反思机制，Generative Agents 在 48 小时内退化为重复响应。去掉技能库，Voyager 的学习速度降低 15.3 倍。用纯长上下文替代主动记忆，MemoryArena 任务完成率从 80%+ 降至约 45%。

我们花大量时间选模型、调 Prompt，但真正决定 Agent 产品差异化的记忆架构，经常一个下午草草设计。这就是问题所在。

📌 本章核心：Agent 记忆分三层：上下文就是工作记忆、近期摘要就是情景记忆、长期事实就是语义记忆。每一层用不同策略，别把所有东西往向量库里塞。

---

第一部分：三层记忆，三种策略

Agent 需要的记忆不是"一个数据库"，而是三种不同类型的信息，各自需要不同的存储和检索策略。

层	名称	存什么	存多久	怎么存	怎么取
L1	工作记忆	当前对话的每一条消息	单次会话	上下文窗口	模型直接看到
L2	情景记忆	近期对话摘要 + 关键事件	数周	结构化 JSON + 向量库	按时间 + 语义检索
L3	语义记忆	用户偏好、长期事实	永久	结构化键值对	精确键查询

三种记忆分开管理，不要混在一起。 90% 的人答"Agent 记忆怎么设计"时会直接说"接一个向量数据库"。这是错误的。向量检索适合模糊语义查询，但用户偏好、开关状态、最近进展这类信息更适合结构化存储。

📌 本章要点：三层记忆：L1 上下文（即时）、L2 摘要（近期）、L3 事实（永久）。不同记忆类型用不同策略，向量数据库不是万能的。

---

三层讲完了。先看最基础的一层：工作记忆。

第二部分：L1 工作记忆：上下文的边界

工作记忆就是当前对话的上下文窗口。看起来简单，但两个关键问题决定了 Agent 的稳定性：

问题一：窗口装满后怎么办

P03 讲过压缩重启，但压缩有代价：摘要会丢细节。所以需要"滑动窗口 + 摘要"混合策略。

class WorkingMemory:
    """L1 工作记忆：管理当前对话上下文"""
    
    def __init__(self, max_tokens=16000):
        self.messages = []
        self.max_tokens = max_tokens
    
    def add(self, role, content):
        """添加消息，超出上限自动压缩"""
        self.messages.append({"role": role, "content": str(content)})
        if self._estimate_tokens() > self.max_tokens:
            self._compress()
    
    def _estimate_tokens(self):
        """粗略估算 token 数（实际应用用 tiktoken）"""
        return sum(len(str(m.get("content", ""))) // 2 for m in self.messages)
    
    def _compress(self):
        """保留最近 4 条消息，其余压缩为摘要"""
        recent = self.messages[-4:]  # 最近两轮对话
        older = self.messages[:-4]   # 更早的历史
        if not older:
            return
        
        # 生成摘要
        summary_text = "\n".join(
            f"[{m['role']}]: {str(m.get('content', ''))[:100]}"
            for m in older
        )
        summary = f"[历史摘要] {summary_text}"
        
        # 用摘要替代旧历史
        self.messages = [
            {"role": "system", "content": summary},
            *recent
        ]
    
    def get_context(self):
        """返回当前上下文供 LLM 使用"""
        return self.messages

问题二：哪些信息应该留在上下文里

窗口里不是越多越好。三类信息应该留在上下文：

1. 当前任务的目标和约束：Agent 必须时刻知道自己在做什么
2. 最近两轮的工具调用结果：推理链路的直接依据
3. 用户的最近一条消息：当前要解决的问题

其他东西（历史对话细节、之前的工具结果、用户偏好）应该外移到 L2 或 L3。P03 的外化记忆和即时加载就是为这个服务的。

📌 本章要点：L1 工作记忆用滑动窗口 + 摘要混合策略。只保留当前任务必需的信息，其余外移到更深层记忆。

---

上下文满了可以压缩。但关了对话，上下文直接消失。跨会话的记忆怎么办？

第三部分：L2 情景记忆：让 Agent 记得"上周发生了什么"

情景记忆存的是"某时某地发生了什么"：上周三用户反馈了什么问题、三天前 Agent 做了一个什么决策、昨天讨论过什么方案。

写入：每轮对话结束后异步提取

不要在对话进行中做记忆提取（会增加延迟），而是在对话结束后异步执行。

import json
import time
import hashlib

class EpisodicMemory:
    """L2 情景记忆：近期对话摘要与关键事件"""
    
    def __init__(self, storage_path="episodic_memory.json"):
        self.path = storage_path
        self.entries = self._load()
    
    def _load(self):
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                return json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            return []
    
    def write(self, conversation_summary, key_events):
        """
        写入一条情景记忆
        conversation_summary: 本轮对话的一句话摘要
        key_events: 本轮对话中的关键事件列表
        """
        entry = {
            "id": hashlib.md5(
                f"{time.time()}{conversation_summary}".encode()
            ).hexdigest()[:12],
            "timestamp": time.time(),
            "summary": conversation_summary,
            "events": key_events,   # [{"type": "decision", "detail": "..."}, ...]
            "importance": self._score_importance(key_events)
        }
        self.entries.append(entry)
        
        # 只保留最近 200 条
        if len(self.entries) > 200:
            self.entries = sorted(
                self.entries, key=lambda e: e["importance"], reverse=True
            )[:200]
        
        self._save()
    
    def _score_importance(self, events):
        """根据事件类型评分：决策 > 错误 > 偏好 > 普通"""
        scores = {"decision": 5, "error": 4, "preference": 3, "info": 1}
        return max((scores.get(e.get("type", "info"), 1) for e in events), default=0)
    
    def recall(self, query, top_k=5):
        """
        检索相关情景记忆
        简单实现用关键词匹配，生产环境用向量检索
        """
        results = []
        for entry in self.entries:
            score = 0
            # 关键词匹配
            if query.lower() in entry["summary"].lower():
                score += 3
            for event in entry.get("events", []):
                if query.lower() in event.get("detail", "").lower():
                    score += 1
            if score > 0:
                results.append((score, entry))
        
        results.sort(key=lambda x: x[0], reverse=True)
        return [r[1] for r in results[:top_k]]
    
    def get_recent(self, days=7):
        """获取最近 N 天的情景记忆"""
        cutoff = time.time() - days * 86400
        return [e for e in self.entries if e["timestamp"] > cutoff]
    
    def _save(self):
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(self.entries, f, ensure_ascii=False, indent=2)

检索：新对话开始时注入相关记忆

每次新对话启动时，检索相关的情景记忆注入 System Prompt：

def build_system_prompt(episodic, user_query):
    """构建包含情景记忆的系统提示"""
    recent = episodic.get_recent(days=7)
    relevant = episodic.recall(user_query, top_k=3)
    
    context_blocks = []
    
    if recent:
        context_blocks.append("## 近期记录")
        for e in recent[-5:]:  # 最近 5 条
            context_blocks.append(f"- {e['summary']}")
    
    if relevant:
        context_blocks.append("## 相关历史")
        for e in relevant:
            context_blocks.append(f"- {e['summary']}")
    
    return "\n".join(context_blocks) if context_blocks else ""

清理：过期的记忆要遗忘

记忆系统最容易被忽略的环节是 Manage：去重、合并、遗忘。情景记忆不能只增不减，否则会退化。

def cleanup_episodic_memory(episodic):
    """定期清理过期和重复的情景记忆"""
    now = time.time()
    keep = []
    seen_summaries = set()
    
    for entry in sorted(episodic.entries, key=lambda e: e["timestamp"], reverse=True):
        # 规则 1：超过 90 天的低重要性记忆丢弃
        if now - entry["timestamp"] > 90 * 86400 and entry["importance"] < 3:
            continue
        
        # 规则 2：相似摘要去重（保留最新的）
        key = entry["summary"][:50]
        if key in seen_summaries:
            continue
        seen_summaries.add(key)
        
        keep.append(entry)
    
    episodic.entries = keep
    episodic._save()

📌 本章要点：L2 情景记忆记录"什么时候发生了什么"。写入异步执行（不拖慢对话），检索在对话开始时注入。关键的是管理环节：去重、遗忘、重要性评分。

---

情景记忆让 Agent 记住了近期的事。但用户的长期偏好：“我喜欢简洁的回复”“我是后端工程师”：应该存在最底层。

第四部分：L3 语义记忆：用户永远不会重复说第二遍

语义记忆存的是长期、稳定的事实：用户名、技术栈偏好、汇报对象、常用城市。这些信息一旦提取就很少改变，除非用户明确更新。

为什么不用向量数据库

L3 适合结构化存储，不是向量数据库。原因有三：

1. 需要精确匹配，不是模糊检索。 "用户的城市是北京"和"用户的城市是上海"在向量空间里很近，但含义完全相反。向量检索可能把旧值当成新值返回。
2. 需要覆盖更新，不是追加。 “用户城市：北京"→ 用户搬家了 → 应该直接覆盖为"上海”，而不是追加一条新记录让两条冲突。
3. 查询模式固定。 L3 的查询永远是"用户 X 的偏好 Y 是什么"，不需要语义搜索。

class SemanticMemory:
    """L3 语义记忆：长期事实，结构化键值存储"""
    
    def __init__(self, storage_path="semantic_memory.json"):
        self.path = storage_path
        self.facts = self._load()  # {"user_id": {"key": "value", ...}}
    
    def _load(self):
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                return json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            return {}
    
    def set(self, user_id, key, value):
        """设置一个事实，直接覆盖旧值"""
        if user_id not in self.facts:
            self.facts[user_id] = {}
        
        old_value = self.facts[user_id].get(key)
        self.facts[user_id][key] = value
        self._save()
        
        if old_value and old_value != value:
            return f"已更新: {key} 从 '{old_value}' 改为 '{value}'"
        return f"已记录: {key} = '{value}'"
    
    def get(self, user_id, key):
        """精确查询一个事实"""
        return self.facts.get(user_id, {}).get(key)
    
    def get_all(self, user_id):
        """获取用户的所有已知事实"""
        return self.facts.get(user_id, {})
    
    def extract_from_conversation(self, user_id, conversation_summary):
        """
        从对话摘要中提取长期事实
        生产环境中这里调用 LLM 做抽取
        """
        # 简化示例：基于关键词提取
        extracted = {}
        summary_lower = conversation_summary.lower()
        
        if "北京" in summary_lower or "上海" in summary_lower:
            for city in ["北京", "上海", "深圳", "杭州"]:
                if city in conversation_summary:
                    extracted["preferred_city"] = city
                    break
        
        if "python" in summary_lower:
            extracted["tech_stack"] = "Python"
        if "后端" in summary_lower or "backend" in summary_lower:
            extracted["role"] = "后端工程师"
        
        for key, value in extracted.items():
            self.set(user_id, key, value)
        
        return extracted
    
    def forget(self, user_id, key):
        """删除一个事实"""
        if user_id in self.facts and key in self.facts[user_id]:
            del self.facts[user_id][key]
            self._save()
    
    def _save(self):
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(self.facts, f, ensure_ascii=False, indent=2)

何时写入、何时读取

L3 的节奏和 L2 不同。L2 每次对话都写（情景记忆累积），L3 只在检测到新事实时写（语义记忆覆盖）。

# L3 写入时机：对话结束后 LLM 扫描是否有新增长期事实
def after_conversation(l2_memory, l3_memory, user_id):
    """对话结束后的异步记忆整理"""
    recent = l2_memory.get_recent(days=1)
    
    # 把最近一天的情景记忆喂给 LLM，提取可能的新事实
    summary_text = "\n".join(e["summary"] for e in recent)
    
    # 生产环境这里调用 LLM：
    # "从以下对话中提取用户的长期偏好和事实，以 JSON 格式返回 {key: value}"
    l3_memory.extract_from_conversation(user_id, summary_text)

# L3 读取时机：新对话开始时注入 System Prompt
def inject_semantic_memory(system_prompt, l3_memory, user_id):
    """将用户的长期事实注入 System Prompt"""
    facts = l3_memory.get_all(user_id)
    if not facts:
        return system_prompt
    
    fact_lines = "\n".join(f"- {k}: {v}" for k, v in facts.items())
    return f"""{system_prompt}

## 用户背景
{fact_lines}

在回复时参考以上用户背景信息。"""

📌 本章要点：L3 语义记忆用结构化键值存储，不用向量库。原因：需要精确覆盖而非模糊检索、需要更新而非追加、查询模式固定。写入节奏比 L2 慢，只在检测到新事实时才写。

---

三层拆完了。但最容易出问题的不是任何一层，而是层与层之间的流转。

第五部分：Write-Manage-Read 闭环

Agent 记忆的工程难度不在"存"和"取"，而在"管"。2026 年综述论文指出：Write 和 Read 大部分系统做得不错，但 Manage（维护、剪枝、压缩、矛盾消解）几乎被普遍忽视。

阶段	含义	难度	容易出错的地方
Write	对话结束后提取关键信息写入	⭐⭐	提取不完整、噪音混入
Manage	去重、压缩、更新、遗忘、冲突解决	⭐⭐⭐⭐⭐	最常被忽略，系统退化的根源
Read	新对话时检索相关记忆注入	⭐⭐⭐	召回不准、压入错误记忆

Manage 环节最容易被忽视的问题：

1. 过时记忆不清理：用户半年前说喜欢 Python，现在转 Rust 了。旧偏好还在记忆里不断被召回。
2. 冲突不解决："用户城市：北京"和"用户城市：上海"同时存在。读出来哪个？
3. 重要性不区分：200 条记忆里有 180 条是废话。每次检索都在噪声里捞。

一个好的记忆系统，写和读各占 20% 的精力，Manage 应该占 60%。

📌 本章要点：Write 和 Read 简单，Manage 最难也最重要。不做 Manage 的记忆系统，半年后就会退化为噪声仓库。

---

第六部分：实战：完整的三层记忆系统

把三层串起来，写一个完整可用的记忆管理器：

# ═══════════════════════════════════════════
# MemoryManager：三层记忆统一管理
# ═══════════════════════════════════════════

class MemoryManager:
    """整合 L1 工作记忆 + L2 情景记忆 + L3 语义记忆"""
    
    def __init__(self, user_id="default"):
        self.user_id = user_id
        self.working = WorkingMemory()                  # L1
        self.episodic = EpisodicMemory()                # L2
        self.semantic = SemanticMemory()                # L3
    
    # ── 对话前：注入 L2 + L3 记忆 ──
    def start_session(self, user_message):
        """新对话开始时，检索并注入记忆"""
        # L3: 用户长期事实
        facts = self.semantic.get_all(self.user_id)
        if facts:
            fact_text = "\n".join(
                f"- {k}: {v}" for k, v in facts.items()
            )
            self.working.add("system",
                f"[用户背景]\n{fact_text}"
            )
        
        # L2: 近期情景记忆
        recent = self.episodic.get_recent(days=7)
        relevant = self.episodic.recall(user_message, top_k=3)
        merged = {e["id"]: e for e in recent + relevant}
        if merged:
            memory_text = "\n".join(
                f"- {e['summary']}" for e in list(merged.values())[:5]
            )
            self.working.add("system",
                f"[近期记录]\n{memory_text}"
            )
        
        self.working.add("user", user_message)
        return self.working.get_context()
    
    # ── 对话中：记录工作记忆 ──
    def record(self, role, content):
        self.working.add(role, content)
    
    # ── 对话后：整理 L2 + L3 ── 
    def end_session(self, conversation_summary, key_events):
        """对话结束后异步整理记忆"""
        # L2: 写入情景记忆
        self.episodic.write(conversation_summary, key_events)
        
        # L3: 提取长期事实
        self.semantic.extract_from_conversation(
            self.user_id, conversation_summary
        )
        
        # Manage: 清理过期记忆
        cleanup_episodic_memory(self.episodic)
    
    # ── 用户更新偏好 ──
    def update_preference(self, key, value):
        self.semantic.set(self.user_id, key, value)
    
    # ── 用户删除记忆 ──
    def forget(self, key=None):
        if key:
            self.semantic.forget(self.user_id, key)
        else:
            self.semantic.facts.pop(self.user_id, None)
            self.semantic._save()


# 使用示例
memory = MemoryManager(user_id="user_123")

# 对话前：自动注入用户背景和近期记录
context = memory.start_session("帮我分析一下上周讨论的架构方案")
# context 里已经包含了用户的长期偏好和最近 7 天的对话摘要

# 对话中：正常记录
memory.record("assistant", "你上周讨论的是微服务拆分方案，核心争议在数据库拆分策略...")
memory.record("user", "对，帮我对比一下分库和分表的优劣")

# 对话后：异步整理
memory.end_session(
    conversation_summary="讨论了微服务拆分中的数据库策略，对比了分库和分表方案",
    key_events=[
        {"type": "decision", "detail": "决定采用分表方案，暂不分库"},
        {"type": "preference", "detail": "用户倾向于渐进式架构演进"}
    ]
)

📌 本章要点：完整记忆系统 = 对话前注入 L2+L3 + 对话中记录 L1 + 对话后整理 L2+L3。关键：L2 每次写，L3 检测到新事实才写，Manage 定期做。

---

总结：读完这篇，你应该带走这几件事

1. 记忆分层不是可选项。三层记忆（工作/情景/语义）各用不同策略，不要把一切都往向量库里塞。
2. 向量数据库不是默认识别。它适合模糊语义检索（L2），不适合精确覆盖和快更新（L3）。ChatGPT 的记忆是四层纯结构化设计，不是接向量库。
3. Manage 比 Write 和 Read 更重要。不去重、不遗忘、不解决冲突的记忆系统，半年后退化为噪声仓库。
4. 写入要异步，不要拖慢对话。L2 和 L3 的写入都在对话结束后批量执行，不影响对话延迟。
5. 三层记忆的生命周期不同：L1 随会话消失，L2 保留数周到数月，L3 永久保留直到用户明确更新或删除。

🤔 思考一下：你的 Agent 现在有记忆吗？如果有，它是在每轮对话结束后做整理，还是只写不清理？

---

思维导图

• Agent Memory 系统
  • 三层记忆
    • L1 工作记忆：上下文窗口，滑动窗口+压缩
    • L2 情景记忆：近期摘要，异步写入，向量检索
    • L3 语义记忆：长期事实，结构化键值，精确覆盖
  • Write-Manage-Read 闭环
    • Write：对话结束后提取（难度 ⭐⭐）
    • Manage：去重/合并/遗忘/冲突解决（难度 ⭐⭐⭐⭐⭐）
    • Read：对话开始时注入（难度 ⭐⭐⭐）
  • 关键原则
    • 不同记忆类型用不同存储策略
    • 向量数据库不是万能的
    • 写入异步，不拖慢对话
    • Manage 应该占 60% 精力
  • 核心 Takeaways
    • 记忆分层决定 Agent 上限
    • 三层各自独立管理
    • 不做 Manage 的记忆系统会退化

---

下一篇预告

P06. Multi-Agent：什么时候该用、什么时候不该用

单 Agent 的记忆搞定了。但有人说：上 Multi-Agent 吧，架构更灵活。另一个人说：别上，复杂度爆炸。多人协作什么时候是乘法、什么时候是除法？

🤔 思考一下：你的场景里，多个 Agent 真的能并行工作，还是只是在互相等结果？