Agent 记忆系统设计：从短期上下文到持久化知识

大多数 Agent 的记忆停留在"把最近几轮对话塞进 context window"。这在 demo 里够用，但在生产环境中，用户会期望 Agent 记住上次讨论的方案、理解长期偏好、并基于历史经验做出更好的决策。本文将拆解 Agent 记忆的四层架构，并给出每层的实现代码。

为什么"把所有东西塞进 context"行不通

直接将历史对话全部塞入 context 有三个致命问题：

• Token 成本线性增长：10 轮对话后，每轮的推理成本翻倍
• 检索效率低下：LLM 从 50K token 的上下文中找到关键信息的可靠性远低于从 500 token 的精准检索结果中找到
• 无法跨会话持久化：context window 在会话结束后消失，用户明天回来时 Agent 什么都不记得

真正的记忆系统需要分层设计，每层有不同的存储介质、检索策略和生命周期。

四层记忆架构

┌─────────────────────────────────────────┐
│  Layer 1: Working Memory (工作记忆)       │  ← 当前对话的 context window
├─────────────────────────────────────────┤
│  Layer 2: Episodic Memory (情景记忆)     │  ← 近期对话摘要 + 关键事件
├─────────────────────────────────────────┤
│  Layer 3: Semantic Memory (语义记忆)     │  ← 向量存储的知识和事实
├─────────────────────────────────────────┤
│  Layer 4: Procedural Memory (程序记忆)   │  ← 学到的行为模式和技能
└─────────────────────────────────────────┘

Layer 1：工作记忆

工作记忆就是当前 context window。不需要额外的存储，但需要精心管理：

from dataclasses import dataclass, field

@dataclass
class WorkingMemory:
    system_prompt: str
    recent_messages: list[dict] = field(default_factory=list)
    max_tokens: int = 8000
    reserved_for_output: int = 2000

    def add_message(self, role: str, content: str):
        self.recent_messages.append({"role": role, "content": content})
        self._trim_if_needed()

    def _trim_if_needed(self):
        """当预估 token 数接近上限时，移除最早的消息"""
        estimated_tokens = sum(
            len(m["content"]) // 3 for m in self.recent_messages
        )
        budget = self.max_tokens - self.reserved_for_output - len(self.system_prompt) // 3
        while estimated_tokens > budget and len(self.recent_messages) > 2:
            removed = self.recent_messages.pop(0)
            estimated_tokens -= len(removed["content"]) // 3

    def get_context(self) -> list[dict]:
        return [
            {"role": "system", "content": self.system_prompt}
        ] + self.recent_messages

设计要点：为输出预留空间（reserved_for_output），避免 context 塞满导致输出被截断。

Layer 2：情景记忆

情景记忆存储近期事件的关键信息。当工作记忆溢出时，被移除的内容不是直接丢弃，而是压缩后存入情景记忆。

from datetime import datetime, timedelta

@dataclass
class Episode:
    timestamp: datetime
    summary: str
    key_entities: list[str]
    importance: float  # 0.0 ~ 1.0
    raw_excerpt: str | None = None  # 保留原始片段用于高重要性事件

class EpisodicMemory:
    def __init__(self, max_episodes: int = 200, ttl_days: int = 30):
        self.episodes: list[Episode] = []
        self.max_episodes = max_episodes
        self.ttl = timedelta(days=ttl_days)

    def add(self, summary: str, key_entities: list[str],
            importance: float = 0.5, raw: str | None = None):
        episode = Episode(
            timestamp=datetime.now(),
            summary=summary,
            key_entities=key_entities,
            importance=importance,
            raw_excerpt=raw,
        )
        self.episodes.append(episode)
        self._evict_if_needed()

    def retrieve_recent(self, hours: int = 24, limit: int = 10) -> list[Episode]:
        cutoff = datetime.now() - timedelta(hours=hours)
        candidates = [
            e for e in self.episodes if e.timestamp > cutoff
        ]
        # 按重要性排序，时间近的加权更高
        candidates.sort(
            key=lambda e: e.importance * self._recency_score(e),
            reverse=True,
        )
        return candidates[:limit]

    def _recency_score(self, episode: Episode) -> float:
        age_hours = (datetime.now() - episode.timestamp).total_seconds() / 3600
        return max(0.1, 1.0 - age_hours / 720)  # 30 天内线性衰减

    def _evict_if_needed(self):
        # 先淘汰过期的
        cutoff = datetime.now() - self.ttl
        self.episodes = [e for e in self.episodes if e.timestamp > cutoff]
        # 再淘汰低重要性的
        if len(self.episodes) > self.max_episodes:
            self.episodes.sort(key=lambda e: e.importance * self._recency_score(e))
            self.episodes = self.episodes[-self.max_episodes:]

关键设计：不是所有记忆都同等重要。通过 importance 字段和 recency_score 实现优先级淘汰，确保高价值记忆被保留。

Layer 3：语义记忆

语义记忆是长期知识存储层，使用向量数据库实现语义检索。

import numpy as np

class SemanticMemory:
    def __init__(self, embedding_dim: int = 1536):
        self.vectors: np.ndarray = np.empty((0, embedding_dim))
        self.documents: list[dict] = []

    def add(self, text: str, embedding: list[float], metadata: dict | None = None):
        vec = np.array(embedding).reshape(1, -1)
        self.vectors = (
            np.vstack([self.vectors, vec])
            if len(self.vectors) > 0 else vec
        )
        self.documents.append({
            "text": text,
            "metadata": metadata or {},
            "access_count": 0,
        })

    def search(self, query_embedding: list[float], top_k: int = 5,
               threshold: float = 0.7) -> list[dict]:
        if len(self.vectors) == 0:
            return []

        query = np.array(query_embedding).reshape(1, -1)
        # 余弦相似度
        norms = np.linalg.norm(self.vectors, axis=1) * np.linalg.norm(query)
        similarities = (self.vectors @ query.T).flatten() / np.clip(norms, 1e-8, None)

        # 取 top_k 并过滤低相似度结果
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        results = []
        for idx in top_indices:
            if similarities[idx] >= threshold:
                doc = self.documents[idx]
                doc["access_count"] += 1
                doc["score"] = float(similarities[idx])
                results.append(doc)
        return results

    def consolidate(self, min_access: int = 3, max_entries: int = 5000):
        """记忆整合：移除从未被访问的低价值记忆"""
        if len(self.documents) <= max_entries:
            return
        keep_indices = [
            i for i, doc in enumerate(self.documents)
            if doc["access_count"] >= min_access
        ]
        if not keep_indices:
            return
        self.vectors = self.vectors[keep_indices]
        self.documents = [self.documents[i] for i in keep_indices]

检索策略：设置相似度阈值（threshold）过滤噪声。access_count 追踪使用频率，为记忆整合提供依据。

Layer 4：程序记忆

程序记忆存储 Agent 从经验中学到的行为模式——不是"发生了什么"，而是"应该怎么做"。

@dataclass
class LearnedPattern:
    trigger: str        # 触发条件描述
    action: str         # 推荐的行为
    success_rate: float # 历史成功率
    sample_size: int    # 统计样本量

class ProceduralMemory:
    def __init__(self):
        self.patterns: list[LearnedPattern] = []

    def record_outcome(self, trigger: str, action: str, success: bool):
        existing = next(
            (p for p in self.patterns
             if p.trigger == trigger and p.action == action),
            None,
        )
        if existing:
            # 更新贝叶斯平均
            total = existing.sample_size + 1
            existing.success_rate = (
                existing.success_rate * existing.sample_size + int(success)
            ) / total
            existing.sample_size = total
        else:
            self.patterns.append(LearnedPattern(
                trigger=trigger,
                action=action,
                success_rate=float(success),
                sample_size=1,
            ))

    def get_best_action(self, trigger: str, min_samples: int = 5) -> str | None:
        candidates = [
            p for p in self.patterns
            if p.trigger == trigger and p.sample_size >= min_samples
        ]
        if not candidates:
            return None
        best = max(candidates, key=lambda p: p.success_rate)
        return best.action if best.success_rate > 0.6 else None

设计理念：程序记忆不存储原始对话，而是存储抽象后的"触发-行动-成功率"三元组。需要最少 min_samples 次观测才输出建议，避免小样本偏差。

四层协作：完整的记忆管理器

class AgentMemoryManager:
    def __init__(self, system_prompt: str):
        self.working = WorkingMemory(system_prompt=system_prompt)
        self.episodic = EpisodicMemory()
        self.semantic = SemanticMemory()
        self.procedural = ProceduralMemory()

    def build_context(self, current_query: str, query_embedding: list[float] | None = None) -> list[dict]:
        """为当前查询组装最优 context"""
        # 1. 工作记忆（始终包含）
        context = self.working.get_context()

        # 2. 语义记忆（向量检索相关知识）
        if query_embedding:
            knowledge = self.semantic.search(query_embedding, top_k=3, threshold=0.7)
            if knowledge:
                knowledge_text = "\n".join(f"- {k['text'][:200]}" for k in knowledge)
                context.append({
                    "role": "system",
                    "content": f"[相关知识]\n{knowledge_text}",
                })

        # 3. 程序记忆（检查是否有可用的行为建议）
        best_action = self.procedural.get_best_action(current_query[:100])
        if best_action:
            context.append({
                "role": "system",
                "content": f"[行为建议] 基于历史经验，建议：{best_action}",
            })

        # 3. 情景记忆（注入近期重要事件）
        recent_episodes = self.episodic.retrieve_recent(hours=48, limit=5)
        if recent_episodes:
            episode_text = "\n".join(
                f"- {e.summary}" for e in recent_episodes
            )
            context.append({
                "role": "system",
                "content": f"[近期记忆]\n{episode_text}",
            })

        return context

记忆检索的决策框架

不同场景应该使用不同的记忆层：

场景	主要记忆层	检索策略	原因
多轮对话中的指代消解	工作记忆	最近 N 条消息	信息就在上下文中
"上次我们讨论的方案是什么"	情景记忆	时间范围 + 重要性	需要时间线索
"之前有没有类似的技术方案"	语义记忆	向量相似度检索	需要语义匹配
"这种情况一般怎么处理"	程序记忆	触发条件匹配	需要经验模式
新用户首次对话	程序记忆	默认行为模式	无个人记忆可用时回退到通用经验

常见误区

误区一："记忆越多越好" 记忆质量 > 记忆数量。无差别的记忆存储会导致检索时噪声淹没信号。每层都需要淘汰机制（TTL、重要性评分、访问频率）。

误区二："向量检索万能" 向量检索擅长语义匹配，但不擅长精确匹配和时间排序。"昨天讨论的方案"用向量检索不如用情景记忆的时间索引。为查询选择正确的记忆层比调优向量模型更有效。

误区三："不需要记忆压缩，直接存原文" 原文存储的成本和检索噪声都很高。压缩不是信息损失——好的摘要保留了决策相关的信息，去除了社交客套和冗余表述。

总结

• 四层记忆各司其职：工作记忆管当前对话，情景记忆管近期事件，语义记忆管长期知识，程序记忆管行为模式
• 每层都需要独立的淘汰机制：TTL、重要性评分、访问频率，三者至少选其二
• 检索策略比存储方案更重要：为查询选择正确的记忆层
• 记忆压缩是必需品不是奢侈品：好的压缩保留信号、去除噪声
• 程序记忆是最容易被忽视的层，但在长期运行中最有价值

---

本文由 AgentList 团队整理，更多 Agent 记忆系统相关项目请浏览本站项目列表。