记忆与缓存系统设计

Agent 的记忆不是一块铁板——它是分层的，像金字塔一样，每一层有不同的容量、速度和持久性。

前言

如果把 AI Agent 比作一个人，那么记忆系统就是它的"大脑"。但人的记忆不是单一的存储系统——我们有瞬时记忆（视觉暂留）、工作记忆（当前思考的内容）、短期记忆（今天发生的事）和长期记忆（童年经历）。Agent 的记忆系统也需要类似的分层设计。

本文将深入探讨 Agent 记忆与缓存系统的三层架构：短期 Prompt Cache、中期会话内缓存 和 长期跨会话持久化。我们将从 Anthropic 的 Prompt Caching 机制出发，逐步构建一个完整的多级缓存系统，并讨论缓存失效策略这一核心工程问题。

Agent 记忆的需求层次

三层记忆模型

借鉴认知心理学的记忆分层理论，Agent 的记忆系统可以分为三层：

每一层的访问模式、容量限制和失效策略都截然不同：

层级	容量	延迟	持久性	典型用途
短期 Prompt Cache	~100KB tokens	<100ms	分钟级	系统提示词、工具定义
中期会话缓存	~10MB	<10ms	会话级	工具输出、中间结果
长期持久化存储	~GB级	<100ms	永久	用户画像、历史记录

为什么需要分层？

单层缓存无法满足 Agent 的多样化需求。考虑以下场景：

1. 重复工具调用：Agent 在一次会话中多次读取同一个文件，第二次应该走缓存
2. 系统提示词复用：每次 API 调用都发送相同的系统提示词，应该利用 Prompt Cache
3. 用户偏好记忆：用户告诉 Agent “我喜欢用 TypeScript”，下次会话应该记住

这三个场景分别对应三个不同的缓存层级，使用不同的存储介质和失效策略。

短期缓存：Prompt Cache

Anthropic Prompt Caching 机制

Anthropic 在 2024 年推出了 Prompt Caching 功能，允许开发者标记消息中的静态部分进行缓存。被缓存的前缀（prefix）在后续请求中无需重新处理，可以显著降低首 token 延迟（TTFT）和成本。

Prompt Cache 的工作原理基于一个关键观察：Transformer 模型的 KV Cache 可以跨请求复用。当系统提示词和工具定义在多次请求中保持不变时，模型不需要重新计算这些 token 的注意力状态。

from anthropic import Anthropic

client = Anthropic()

# Prompt Cache 的核心用法：标记静态前缀
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一个代码分析助手。你的任务是分析用户提供的代码..." * 100,
            # cache_control 标记此段为可缓存
            # Anthropic 会自动缓存此前缀，后续请求复用
            "cache_control": {"type": "ephemeral"}
        }
    ],
    tools=[
        {
            "name": "read_file",
            "description": "读取文件内容",
            "input_schema": {
                "type": "object",
                "properties": {"path": {"type": "string"}},
                "required": ["path"]
            },
            # 工具定义也可以缓存，通常占token比例不小
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[{"role": "user", "content": "分析 main.py"}]
)

# 从响应中可以看到缓存命中情况
print(f"Cache creation: {response.usage.cache_creation_input_tokens}")
print(f"Cache read: {response.usage.cache_read_input_tokens}")

Prompt Cache 的命中策略

Prompt Cache 的命中有一个重要约束：缓存前缀必须完全匹配。这意味着：

• 系统提示词必须逐字节相同
• 工具定义的顺序和内容不能改变
• 只有前缀部分可以缓存，中间的变更会导致后续缓存失效

因此，最佳实践是将不变的内容放在消息序列的最前面，将变化的内容（如用户消息）放在最后面。

缓存成本分析

Prompt Cache 并非免费。Anthropic 的定价模型是：

• 缓存写入（首次创建）：比普通输入 token 贵 25%
• 缓存读取（命中）：比普通输入 token 便宜 90%
• 缓存 TTL：默认 5 分钟，可配置

这意味着只有当同一前缀被多次请求时，缓存才有收益。对于典型的 Agent 会话（10+ 轮对话），Prompt Cache 几乎总是划算的。

中期缓存：会话内缓存

工具结果缓存

在一次 Agent 会话中，同一个工具可能被多次调用。例如，Agent 可能在分析代码时多次读取同一个文件。如果每次都重新执行工具调用，不仅浪费时间，还可能因为文件在两次读取之间被修改而导致不一致。

工具结果缓存解决了这个问题：同一个工具、相同的输入参数，在一次会话内只执行一次。

import hashlib
import json
import time
from typing import Any, Dict, Optional
from dataclasses import dataclass, field

@dataclass
class CacheEntry:
    """缓存条目 - 存储单次工具调用的结果"""
    key: str                    # 缓存键（工具名 + 参数哈希）
    value: Any                  # 缓存值（工具输出）
    created_at: float           # 创建时间戳
    ttl: int                    # 生存时间（秒）
    hit_count: int = 0          # 命中次数，用于 LRU/LFU 淘汰
    size_bytes: int = 0         # 条目大小，用于容量管理

    def is_expired(self) -> bool:
        """检查缓存是否过期"""
        return time.time() - self.created_at > self.ttl

class ToolResultCache:
    """工具结果缓存 - 会话级别的工具输出缓存"""

    def __init__(self, max_size_mb: int = 50, default_ttl: int = 600):
        self.cache: Dict[str, CacheEntry] = {}
        self.max_size_bytes = max_size_mb * 1024 * 1024
        self.default_ttl = default_ttl
        self.current_size = 0
        self.stats = {'hits': 0, 'misses': 0, 'evictions': 0}

    def _make_key(self, tool_name: str, params: Dict[str, Any]) -> str:
        """生成缓存键 - 工具名 + 参数的确定性哈希"""
        # 对参数排序以保证相同参数产生相同的键
        param_str = json.dumps(params, sort_keys=True, ensure_ascii=False)
        hash_val = hashlib.sha256(f"{tool_name}:{param_str}".encode()).hexdigest()[:16]
        return f"{tool_name}:{hash_val}"

    def get(self, tool_name: str, params: Dict[str, Any]) -> Optional[Any]:
        """查询缓存 - 返回 None 表示未命中"""
        key = self._make_key(tool_name, params)
        entry = self.cache.get(key)

        if entry is None:
            self.stats['misses'] += 1
            return None

        if entry.is_expired():
            # 过期条目，删除并返回未命中
            self._remove_entry(key)
            self.stats['misses'] += 1
            return None

        # 命中！更新命中计数
        entry.hit_count += 1
        self.stats['hits'] += 1
        return entry.value

    def put(self, tool_name: str, params: Dict[str, Any],
            result: Any, ttl: Optional[int] = None) -> None:
        """写入缓存 - 带容量管理和自动淘汰"""
        key = self._make_key(tool_name, params)
        value_bytes = len(json.dumps(result, ensure_ascii=False).encode('utf-8'))

        # 如果单条结果超过缓存总容量的 20%，不缓存
        if value_bytes > self.max_size_bytes * 0.2:
            return

        # 容量不足时执行淘汰
        while self.current_size + value_bytes > self.max_size_bytes:
            self._evict_lfu()  # 使用 LFU（最不经常使用）淘汰策略

        entry = CacheEntry(
            key=key,
            value=result,
            created_at=time.time(),
            ttl=ttl or self.default_ttl,
            size_bytes=value_bytes
        )
        self.cache[key] = entry
        self.current_size += value_bytes

    def _evict_lfu(self) -> None:
        """LFU 淘汰 - 移除命中次数最少的条目"""
        if not self.cache:
            return
        # 找到命中次数最少的条目
        lfu_key = min(self.cache, key=lambda k: self.cache[k].hit_count)
        self._remove_entry(lfu_key)
        self.stats['evictions'] += 1

    def _remove_entry(self, key: str) -> None:
        """移除缓存条目并更新大小统计"""
        entry = self.cache.pop(key, None)
        if entry:
            self.current_size -= entry.size_bytes

    def invalidate(self, tool_name: str, params: Dict[str, Any]) -> bool:
        """手动失效 - 当已知数据源变化时调用"""
        key = self._make_key(tool_name, params)
        if key in self.cache:
            self._remove_entry(key)
            return True
        return False

    def invalidate_by_prefix(self, prefix: str) -> int:
        """按前缀批量失效 - 例如失效所有 read_file 的缓存"""
        keys_to_remove = [k for k in self.cache if k.startswith(prefix)]
        for key in keys_to_remove:
            self._remove_entry(key)
        return len(keys_to_remove)

这段代码实现了一个完整的工具结果缓存，包含以下关键设计决策：

1. 缓存键生成：工具名 + 参数的确定性哈希，保证相同调用产生相同键
2. LFU 淘汰策略：在容量不足时移除命中次数最少的条目，而非最旧的条目
3. 单条目大小限制：超过总容量 20% 的结果不缓存，防止大文件挤占缓存空间
4. 手动失效机制：当已知数据源变化时可以主动使缓存失效

上下文窗口缓存

除了工具结果，Agent 还需要缓存"上下文窗口"——即当前发送给模型的完整消息序列。这个缓存的目的不是避免重复计算，而是快速恢复对话上下文。

当用户在会话中回退到某个历史节点时，Agent 需要重建该节点的上下文窗口。如果没有缓存，就需要从头遍历所有事件来重建，这在长会话中可能很慢。

长期记忆：跨会话持久化

记忆的组织方式

长期记忆需要解决的核心问题是检索效率。当用户说"上次我们讨论的那个性能优化方案"时，Agent 需要从海量历史记录中快速找到相关内容。

常见的组织方式包括：

1. 时间索引：按日期组织，适合"上周的对话"这类查询
2. 主题索引：按讨论主题分类，适合"性能优化"这类查询
3. 向量索引：将记忆编码为向量，支持语义相似度搜索
4. 图索引：记忆之间建立关联关系，支持关联查询

import json
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field

@dataclass
class Memory:
    """单条长期记忆"""
    memory_id: str
    content: str                # 记忆内容（自然语言描述）
    source_session_id: str      # 来源会话ID
    created_at: str             # 创建时间
    last_accessed: str          # 最后访问时间
    access_count: int = 0       # 访问次数
    importance: float = 0.5     # 重要性评分（0-1）
    tags: List[str] = field(default_factory=list)  # 主题标签
    embedding: Optional[List[float]] = None  # 向量表示（用于语义搜索）

class LongTermMemoryStore:
    """长期记忆存储 - 支持多维度检索"""

    def __init__(self, storage_path: str = "memories.jsonl"):
        self.storage_path = storage_path
        self.memories: Dict[str, Memory] = {}
        self._load_from_disk()

    def _load_from_disk(self) -> None:
        """从磁盘加载记忆索引"""
        try:
            with open(self.storage_path, 'r', encoding='utf-8') as f:
                for line in f:
                    if line.strip():
                        data = json.loads(line)
                        mem = Memory(**data)
                        self.memories[mem.memory_id] = mem
        except FileNotFoundError:
            pass  # 首次运行，无历史记忆

    def store(self, content: str, source_session_id: str,
              importance: float = 0.5, tags: List[str] = None) -> Memory:
        """存储新记忆"""
        now = datetime.utcnow().isoformat() + 'Z'
        memory_id = f"mem_{int(datetime.utcnow().timestamp())}_{id(content) % 10000}"

        memory = Memory(
            memory_id=memory_id,
            content=content,
            source_session_id=source_session_id,
            created_at=now,
            last_accessed=now,
            importance=importance,
            tags=tags or []
        )
        self.memories[memory_id] = memory
        self._append_to_disk(memory)
        return memory

    def retrieve_by_recency(self, limit: int = 10) -> List[Memory]:
        """按时间检索 - 最近的记忆优先"""
        sorted_memories = sorted(
            self.memories.values(),
            key=lambda m: m.created_at,
            reverse=True
        )
        return sorted_memories[:limit]

    def retrieve_by_importance(self, limit: int = 10) -> List[Memory]:
        """按重要性检索 - 评分最高的记忆优先"""
        sorted_memories = sorted(
            self.memories.values(),
            key=lambda m: m.importance,
            reverse=True
        )
        return sorted_memories[:limit]

    def retrieve_by_tags(self, tags: List[str]) -> List[Memory]:
        """按标签检索 - 返回包含任一标签的记忆"""
        return [
            m for m in self.memories.values()
            if set(tags) & set(m.tags)
        ]

    def decay_importance(self, decay_factor: float = 0.95) -> None:
        """重要性衰减 - 模拟遗忘曲线，长期未访问的记忆重要性降低"""
        now = datetime.utcnow()
        for memory in self.memories.values():
            last_access = datetime.fromisoformat(
                memory.last_accessed.rstrip('Z')
            )
            days_since_access = (now - last_access).days
            # 每天衰减一次，访问次数越多衰减越慢
            effective_decay = decay_factor ** (days_since_access / max(memory.access_count, 1))
            memory.importance *= effective_decay

    def _append_to_disk(self, memory: Memory) -> None:
        """追加写入磁盘 - Append-only 模式"""
        with open(self.storage_path, 'a', encoding='utf-8') as f:
            data = {
                'memory_id': memory.memory_id,
                'content': memory.content,
                'source_session_id': memory.source_session_id,
                'created_at': memory.created_at,
                'last_accessed': memory.last_accessed,
                'importance': memory.importance,
                'tags': memory.tags
            }
            f.write(json.dumps(data, ensure_ascii=False) + '\n')

这段长期记忆存储的实现包含了几个关键设计：

1. 多维检索：支持按时间、重要性和标签三种方式检索记忆
2. 遗忘曲线：通过 decay_importance 方法模拟人类的遗忘机制——长期未访问的记忆会逐渐降低重要性
3. Append-only 持久化：与 AG-16 中讨论的存储模式一致，记忆只追加不修改

记忆的生命周期管理

长期记忆不能无限增长。需要一个生命周期管理策略：

1. 创建：从会话中提取有价值的信息存入长期记忆
2. 激活：被检索到时更新访问计数和最后访问时间
3. 衰减：定期执行重要性衰减，模拟遗忘曲线
4. 归档：重要性低于阈值的记忆移入归档存储
5. 删除：归档超过一定时间后彻底删除

缓存失效策略

缓存失效的三大难题

Phil Karlton 说过："计算机科学中只有两件难事：缓存失效和命名。"在 Agent 系统中，缓存失效尤其困难，因为：

1. 数据源异构：文件系统、API、数据库的变化频率各不相同
2. 依赖关系复杂：一个缓存条目可能依赖多个数据源
3. 一致性要求模糊：Agent 的"最终一致性"标准比传统应用更宽松

失效策略对比

策略	原理	优点	缺点	适用场景
TTL（时间过期）	固定时间后自动失效	实现简单，可预测	可能过早或过晚失效	Prompt Cache
LRU（最近最少使用）	淘汰最久未访问的条目	适配访问模式	可能淘汰偶尔访问的大对象	工具结果缓存
LFU（最不经常使用）	淘汰访问频率最低的条目	保留热点数据	新条目容易被淘汰	会话内缓存
事件驱动失效	数据源变化时主动通知	实时性最好	需要数据源配合	文件监控缓存
版本号失效	每次写入递增版本号	精确控制	需要维护版本元数据	配置缓存

混合失效策略

实践中，Agent 系统通常采用混合策略。以文件读取缓存为例：

class HybridCacheInvalidation:
    """混合缓存失效策略 - 结合 TTL 和事件驱动"""

    def __init__(self, cache: ToolResultCache):
        self.cache = cache
        self.file_watchers: Dict[str, float] = {}  # 文件路径 -> 最后修改时间

    def should_invalidate(self, tool_name: str, params: Dict[str, Any]) -> bool:
        """判断缓存条目是否应该失效"""
        if tool_name == "read_file":
            file_path = params.get("path", "")
            # 策略1：检查文件修改时间
            try:
                import os
                current_mtime = os.path.getmtime(file_path)
                cached_mtime = self.file_watchers.get(file_path)
                if cached_mtime and current_mtime > cached_mtime:
                    return True  # 文件被修改，缓存失效
                self.file_watchers[file_path] = current_mtime
            except FileNotFoundError:
                return True  # 文件被删除，缓存失效

        elif tool_name in ("web_search", "api_call"):
            # 策略2：外部 API 结果使用短 TTL（已在缓存层处理）
            return False  # 依赖 TTL 自动过期

        elif tool_name == "list_directory":
            # 策略3：目录列表使用中等 TTL + 文件系统事件
            import os
            dir_path = params.get("path", ".")
            try:
                current_mtime = os.path.getmtime(dir_path)
                cached_mtime = self.file_watchers.get(dir_path)
                if cached_mtime and current_mtime > cached_mtime:
                    return True
                self.file_watchers[dir_path] = current_mtime
            except FileNotFoundError:
                return True

        return False

总结

Agent 的记忆与缓存系统是一个分层的、多策略的复杂工程问题。核心设计原则是：

1. 分层设计：短期（Prompt Cache）、中期（会话缓存）、长期（持久化记忆）各司其职
2. 按需选择淘汰策略：不同层级使用不同的淘汰算法
3. 遗忘曲线：长期记忆需要模拟人类的遗忘机制，避免信息过载
4. 混合失效：结合 TTL、事件驱动和版本号等多种失效策略

在实现自己的 Agent 记忆系统时，建议从最简单的 TTL 缓存开始，根据实际的命中率和使用模式逐步引入更复杂的策略。记住：缓存的目的是用空间换时间，但如果缓存管理本身的开销超过了收益，那就本末倒置了。

参考资料

1. Anthropic, “Prompt Caching” (2024) - Anthropic 官方 Prompt Caching 文档，详细说明了缓存机制和定价模型
2. Claude Code 源码 - 会话内缓存和上下文管理的实现参考
3. Laird et al., “The Soar Cognitive Architecture” (2012) - 认知架构中的记忆分层理论
4. Facebook/Meta, “Scaling Memcache at Facebook” (2013) - 大规模缓存系统的经典论文，讨论了缓存失效的实际工程挑战
5. Anthropic, “Claude Code: Best practices for agentic coding” (2025) - 官方文档中关于缓存和记忆管理的工程实践

---

本系列覆盖 AI 大模型基础、Agent 开发、MCP 协议、Skill 开发、RAG、模型微调、部署推理 七大方向，从入门到实战的全栈内容持续更新中。

所有文章的 Markdown 源文件、可运行代码、高清配图已整理成完整资料包。

👍 点赞 + ⭐ 关注，评论区扣「1」，挨个发你领取方式 👇