AI Agent 记忆系统设计：从短期上下文到长期知识持久化的工程实践

一、多轮对话的"失忆"困境：Agent 为何总是记不住？

大模型驱动的 Agent 在实际落地中，最常被用户吐槽的问题不是"回答不够聪明"，而是"记不住之前说过什么"。单轮对话中，LLM 表现出色；但跨会话、跨任务的场景下，Agent 的记忆断裂导致重复提问、上下文丢失、决策不一致。这种"失忆"在客服、项目管理和代码辅助等长周期场景中尤为致命。

生产环境的 Agent 系统需要三类记忆能力：短期记忆（当前会话的上下文窗口）、工作记忆（跨轮次的任务状态追踪）和长期记忆（用户偏好、历史决策和领域知识的持久化存储）。三者协同，才能让 Agent 在复杂任务中保持连贯性。

二、记忆分层架构：从 Token 窗口到向量持久化

Agent 记忆系统的核心挑战在于：LLM 的上下文窗口有限（即使 128K 也无法覆盖长周期任务），而外部存储的检索又存在延迟和相关性衰减。合理的分层架构是解决这一矛盾的关键。

graph TB
    subgraph 记忆分层
        A[短期记忆<br/>Context Window] --> B[工作记忆<br/>Redis/State Store]
        B --> C[长期记忆<br/>Vector DB + KV Store]
    end

    subgraph 记忆操作
        D[写入] --> A
        A -->|滑动窗口淘汰| B
        B -->|重要性评分| C
        C -->|相关性检索| A
    end

    subgraph 记忆类型
        E[对话历史] --> A
        F[任务状态] --> B
        G[用户画像] --> C
        H[领域知识] --> C
    end

短期记忆直接映射到 LLM 的上下文窗口，受 Token 限制，采用滑动窗口策略保留最近 N 轮对话。工作记忆存储当前任务的结构化状态（如待办列表、中间结果），使用 Redis 等低延迟存储。长期记忆则通过向量数据库（如 Milvus、Qdrant）存储历史对话摘要、用户偏好和领域知识，按相关性检索后注入上下文。

三、生产级记忆系统实现

3.1 短期记忆管理器

from dataclasses import dataclass, field
from typing import List, Optional
import tiktoken

@dataclass
class Message:
    role: str
    content: str
    token_count: int = 0

class ShortTermMemory:
    """短期记忆：管理上下文窗口内的对话历史"""

    def __init__(self, max_tokens: int = 8000, model: str = "gpt-4"):
        self.max_tokens = max_tokens
        self.messages: List[Message] = []
        # 使用 tiktoken 精确计算 Token 数，而非粗略估算
        self.encoder = tiktoken.encoding_for_model(model)
        self.system_prompt_tokens = 0

    def _count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def add(self, role: str, content: str) -> None:
        """添加消息并自动淘汰超出窗口的旧消息"""
        token_count = self._count_tokens(content)
        msg = Message(role=role, content=content, token_count=token_count)
        self.messages.append(msg)
        self._evict()

    def _evict(self) -> None:
        """滑动窗口淘汰：保留系统提示 + 最近的消息"""
        total = sum(m.token_count for m in self.messages) + self.system_prompt_tokens
        # 从最早的非系统消息开始淘汰
        while total > self.max_tokens and len(self.messages) > 1:
            removed = self.messages.pop(0)
            total -= removed.token_count

    def get_context(self) -> List[dict]:
        return [{"role": m.role, "content": m.content} for m in self.messages]

3.2 长期记忆与重要性评分

import numpy as np
from datetime import datetime

@dataclass
class MemoryEntry:
    content: str
    embedding: np.ndarray
    importance: float  # 0-1 重要性评分
    access_count: int = 0
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)

class LongTermMemory:
    """长期记忆：基于向量检索 + 重要性评分的持久化存储"""

    def __init__(self, embedding_dim: int = 1536):
        self.embeddings: np.ndarray = np.empty((0, embedding_dim))
        self.entries: List[MemoryEntry] = []

    def store(self, content: str, embedding: np.ndarray,
              importance: float) -> None:
        """存储记忆，importance 由 LLM 评分或规则计算"""
        entry = MemoryEntry(
            content=content,
            embedding=embedding,
            importance=importance
        )
        self.entries.append(entry)
        self.embeddings = np.vstack([self.embeddings, embedding.reshape(1, -1)])

    def retrieve(self, query_embedding: np.ndarray,
                 top_k: int = 5,
                 recency_weight: float = 0.3,
                 importance_weight: float = 0.4,
                 relevance_weight: float = 0.3) -> List[str]:
        """混合检索：相关性 + 重要性 + 时效性"""
        if len(self.entries) == 0:
            return []

        # 计算余弦相似度
        norms = np.linalg.norm(self.embeddings, axis=1)
        query_norm = np.linalg.norm(query_embedding)
        similarities = (self.embeddings @ query_embedding) / (norms * query_norm + 1e-8)

        # 时效性衰减：越近期的记忆权重越高
        now = datetime.now()
        recency_scores = np.array([
            1.0 / (1.0 + (now - e.last_accessed).total_seconds() / 86400)
            for e in self.entries
        ])

        importance_scores = np.array([e.importance for e in self.entries])

        # 加权融合
        combined = (relevance_weight * similarities +
                   importance_weight * importance_scores +
                   recency_weight * recency_scores)

        top_indices = np.argsort(combined)[-top_k:][::-1]
        results = []
        for idx in top_indices:
            self.entries[idx].access_count += 1
            self.entries[idx].last_accessed = now
            results.append(self.entries[idx].content)
        return results

3.3 重要性评分策略

记忆写入长期存储前，需要评估其"值得记住"的程度。评分维度包括：信息密度（是否包含决策或偏好）、情感强度（用户是否表达了强烈态度）、任务关联度（是否与当前目标相关）。实践中，通过 LLM 对记忆内容打分（0-1）比纯规则评分更准确，但需要控制调用成本。

四、记忆系统的 Trade-offs 与边界分析

延迟与质量的矛盾：长期记忆检索需要向量相似度计算，在高并发场景下，检索延迟可能达到 50-200ms。如果每轮对话都需要检索，端到端响应时间会显著增加。折中方案是对高频访问的记忆做缓存，仅在缓存未命中时触发向量检索。

存储成本与遗忘策略：长期记忆无限增长会导致检索质量下降和存储成本上升。需要设计遗忘机制：基于重要性评分和访问频率，定期淘汰低分记忆。但遗忘策略过于激进会导致关键信息丢失，过于保守则检索噪声增大。

一致性风险：长期记忆中的过时信息可能误导 Agent 决策。例如用户偏好已改变，但旧记忆仍被检索到。解决方案是在记忆中增加版本标记和过期时间，检索时优先返回最新版本。

适用边界：记忆系统适合长周期、多轮次的交互场景（客服、项目管理、个人助手）。对于单轮问答或无状态工具调用场景，引入记忆层反而增加复杂度，得不偿失。

五、总结

Agent 记忆系统的设计核心在于分层：短期记忆保证当前对话的连贯性，工作记忆追踪任务状态，长期记忆实现跨会话的知识持久化。工程落地上，滑动窗口管理短期上下文，向量检索 + 重要性评分驱动长期记忆的存取，三者通过统一的记忆管理器协调。

关键落地步骤：首先实现短期记忆的滑动窗口，确保基础对话连贯；其次引入工作记忆管理任务状态；最后接入向量数据库实现长期记忆。每一步都需要配合性能监控——记忆检索延迟、命中率、遗忘率——才能在质量与成本之间找到平衡点。