AI Agent 记忆架构设计

记忆是 Agent 的核心能力。没有记忆，Agent 只能是无状态的工具。

---

为什么记忆重要

从 Anthropic 研究学到的

Agent 自主性研究的关键发现：

发现	数据
99.9 百分位任务时间	45+ 分钟
任务时间增长（3个月）	25min → 45min
Claude 主动暂停频率	人类中断的 2 倍

含义：Agent 需要在长时间跨度上保持上下文和状态。

记忆类型

类型	时间跨度	示例
工作记忆	当前会话	对话历史、当前任务状态
情景记忆	会话间	之前的对话、完成的任务
语义记忆	长期	用户偏好、领域知识
程序记忆	永久	技能、流程

---

架构设计

基础架构

┌─────────────────────────────────────────┐
│                  Agent                   │
├─────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────────┐   │
│  │ 工作记忆    │  │  长期记忆       │   │
│  │ (Context)   │  │  (Vector DB)    │   │
│  └─────────────┘  └─────────────────┘   │
│  ┌─────────────┐  ┌─────────────────┐   │
│  │ 会话存储    │  │  知识图谱       │   │
│  │ (KV Store)  │  │  (Graph DB)     │   │
│  └─────────────┘  └─────────────────┘   │
└─────────────────────────────────────────┘

Claude Code 的记忆设计

Claude Code 使用多层记忆：

1. CLAUDE.md - 项目级记忆
2. ~/.claude/memory/ - 全局记忆
3. 会话历史 - 工作记忆
4. 压缩机制 - 上下文管理

---

工作记忆设计

上下文窗口管理

class WorkingMemory:
    def __init__(self, max_tokens: int = 200000):
        self.max_tokens = max_tokens
        self.messages = []
        self.compressed_summary = None

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        self._maybe_compress()

    def _maybe_compress(self):
        if self._estimate_tokens() > self.max_tokens * 0.8:
            self._compress()

    def _compress(self):
        # 压缩旧消息为摘要
        old_messages = self.messages[:-10]
        self.compressed_summary = self._summarize(old_messages)
        self.messages = self.messages[-10:]

压缩策略

策略	说明
滑动窗口	保留最近 N 条消息
摘要压缩	将旧消息压缩为摘要
选择性保留	保留重要消息（用户指令、关键决策）
分层压缩	按重要性分层，高层保留更久

Claude Code 的压缩机制

Claude Code 使用 PreCompact 和 PostCompact hooks：

{
  "hooks": {
    "PreCompact": [
      {
        "hooks": [{ "type": "command", "command": "echo '即将压缩上下文'" }]
      }
    ],
    "PostCompact": [
      {
        "hooks": [{ "type": "command", "command": "./inject-context.sh" }]
      }
    ]
  }
}

---

长期记忆设计

向量数据库选择

数据库	特点	适用场景
Pinecone	托管、高性能	生产环境
Weaviate	开源、GraphQL	自托管
Qdrant	Rust 实现、高性能	高吞吐
Chroma	轻量、Python 原生	原型开发
Milvus	分布式、可扩展	大规模

记忆存储结构

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class Memory:
    id: str
    content: str
    embedding: list[float]
    metadata: dict
    created_at: datetime
    last_accessed: datetime
    access_count: int
    importance: float  # 0-1
    memory_type: str  # episodic, semantic, procedural

记忆检索

class MemoryRetriever:
    def __init__(self, vector_db, embedder):
        self.db = vector_db
        self.embedder = embedder

    def retrieve(
        self,
        query: str,
        k: int = 10,
        filters: dict = None,
        time_decay: float = 0.1
    ) -> list[Memory]:
        # 1. 向量检索
        query_embedding = self.embedder.embed(query)
        candidates = self.db.search(query_embedding, top_k=k*2)

        # 2. 重排序
        scored = []
        for mem in candidates:
            score = self._compute_score(mem, query, time_decay)
            scored.append((score, mem))

        # 3. 返回 top-k
        scored.sort(reverse=True, key=lambda x: x[0])
        return [mem for _, mem in scored[:k]]

    def _compute_score(self, mem: Memory, query: str, decay: float) -> float:
        # 综合考虑：相似度、重要性、时效性、访问频率
        recency = self._time_decay(mem.last_accessed, decay)
        return (
            mem.similarity * 0.4 +
            mem.importance * 0.3 +
            recency * 0.2 +
            min(mem.access_count / 100, 1) * 0.1
        )

---

情景记忆设计

会话存储

class SessionMemory:
    def __init__(self, storage):
        self.storage = storage

    def save_session(self, session_id: str, messages: list, metadata: dict):
        self.storage.set(f"session:{session_id}", {
            "messages": messages,
            "metadata": metadata,
            "created_at": datetime.now().isoformat()
        })

    def load_session(self, session_id: str) -> dict:
        return self.storage.get(f"session:{session_id}")

    def summarize_session(self, session_id: str) -> str:
        session = self.load_session(session_id)
        # 使用 LLM 生成摘要
        return self._llm_summarize(session["messages"])

会话恢复

def resume_session(session_id: str, memory: SessionMemory, working: WorkingMemory):
    # 1. 加载会话
    session = memory.load_session(session_id)

    # 2. 重建工作记忆
    for msg in session["messages"]:
        working.add_message(msg["role"], msg["content"])

    # 3. 注入摘要（如果会话很长）
    if len(session["messages"]) > 50:
        summary = memory.summarize_session(session_id)
        working.add_message("system", f"之前会话摘要：{summary}")

---

语义记忆设计

知识图谱

from dataclasses import dataclass
from typing import Optional

@dataclass
class Entity:
    id: str
    name: str
    type: str
    properties: dict

@dataclass
class Relation:
    source: str  # entity id
    target: str  # entity id
    type: str
    properties: dict

class KnowledgeGraph:
    def __init__(self, graph_db):
        self.db = graph_db

    def add_entity(self, entity: Entity):
        self.db.create_node(entity.id, entity.type, entity.properties)

    def add_relation(self, relation: Relation):
        self.db.create_edge(
            relation.source,
            relation.target,
            relation.type,
            relation.properties
        )

    def query_related(self, entity_id: str, depth: int = 2) -> list[Entity]:
        return self.db.traverse(entity_id, depth)

用户偏好存储

class UserPreferences:
    def __init__(self, storage):
        self.storage = storage

    def get(self, user_id: str, key: str, default=None):
        return self.storage.get(f"user:{user_id}:pref:{key}", default)

    def set(self, user_id: str, key: str, value):
        self.storage.set(f"user:{user_id}:pref:{key}", value)

    def learn_from_feedback(self, user_id: str, action: str, feedback: str):
        # 从用户反馈中学习偏好
        preference = self._extract_preference(action, feedback)
        if preference:
            self.set(user_id, preference["key"], preference["value"])

---

记忆更新策略

遗忘机制

class ForgettingMechanism:
    def __init__(self, half_life_days: float = 30):
        self.half_life = half_life_days

    def compute_retention(self, memory: Memory) -> float:
        """计算记忆保留概率"""
        days_old = (datetime.now() - memory.last_accessed).days
        decay = 0.5 ** (days_old / self.half_life)
        return decay * memory.importance

    def should_forget(self, memory: Memory) -> bool:
        return self.compute_retention(memory) < 0.1

    def cleanup(self, memories: list[Memory]) -> list[Memory]:
        return [m for m in memories if not self.should_forget(m)]

记忆强化

class MemoryReinforcement:
    def reinforce(self, memory: Memory):
        """访问时强化记忆"""
        memory.access_count += 1
        memory.last_accessed = datetime.now()
        # 重要性随访问增加
        memory.importance = min(1.0, memory.importance + 0.1)

    def consolidate(self, memories: list[Memory]):
        """将短期记忆转化为长期记忆"""
        for mem in memories:
            if mem.access_count > 5:
                mem.memory_type = "semantic"
                mem.importance = min(1.0, mem.importance + 0.2)

---

实战：完整的记忆系统

class AgentMemorySystem:
    def __init__(
        self,
        vector_db,
        graph_db,
        kv_store,
        embedder,
        llm
    ):
        self.working = WorkingMemory(max_tokens=200000)
        self.long_term = MemoryRetriever(vector_db, embedder)
        self.knowledge = KnowledgeGraph(graph_db)
        self.sessions = SessionMemory(kv_store)
        self.preferences = UserPreferences(kv_store)
        self.forgetting = ForgettingMechanism()
        self.llm = llm

    def remember(self, content: str, metadata: dict = None):
        """存储新记忆"""
        embedding = self.embedder.embed(content)
        memory = Memory(
            id=str(uuid.uuid4()),
            content=content,
            embedding=embedding,
            metadata=metadata or {},
            created_at=datetime.now(),
            last_accessed=datetime.now(),
            access_count=0,
            importance=0.5,
            memory_type="episodic"
        )
        self.vector_db.insert(memory)

    def recall(self, query: str, k: int = 10) -> list[Memory]:
        """检索相关记忆"""
        return self.long_term.retrieve(query, k)

    def get_context(self, query: str) -> str:
        """构建上下文"""
        # 1. 检索相关记忆
        memories = self.recall(query, k=5)

        # 2. 获取相关实体
        entities = self.knowledge.query_related(query)

        # 3. 获取用户偏好
        prefs = self.preferences.get_all()

        # 4. 组装上下文
        context = self._assemble_context(memories, entities, prefs)
        return context

    def learn(self, interaction: dict):
        """从交互中学习"""
        # 1. 提取实体和关系
        entities, relations = self._extract_knowledge(interaction)

        # 2. 更新知识图谱
        for e in entities:
            self.knowledge.add_entity(e)
        for r in relations:
            self.knowledge.add_relation(r)

        # 3. 存储情景记忆
        self.remember(
            interaction["content"],
            {"type": "interaction", "entities": [e.id for e in entities]}
        )

        # 4. 学习用户偏好
        if "feedback" in interaction:
            self.preferences.learn_from_feedback(
                interaction["user_id"],
                interaction["action"],
                interaction["feedback"]
            )

---

最佳实践

1. 分层存储

• 热数据 → 内存
• 温数据 → Redis
• 冷数据 → 向量数据库

2. 主动遗忘

• 定期清理低价值记忆
• 避免存储膨胀
• 保持检索效率

3. 记忆压缩

• 定期总结旧记忆
• 提取关键信息
• 丢弃冗余细节

4. 隐私保护

• 敏感信息加密
• 用户可控删除
• 合规性检查

---

参考资源

• Anthropic Agent 自主性研究
• Claude Code 文档
• MemGPT 架构
• Letta Agent 记忆

---

总结

Agent 记忆设计需要考虑：

1. 工作记忆 - 上下文窗口管理、压缩策略
2. 长期记忆 - 向量数据库、检索算法
3. 情景记忆 - 会话存储、恢复机制
4. 语义记忆 - 知识图谱、用户偏好
5. 遗忘机制 - 价值衰减、主动清理

好的记忆系统让 Agent 能够：

• 保持长时间对话的一致性
• 从过去经验中学习
• 个性化服务用户
• 处理复杂、多步骤的任务