开源 AI 工具链：Agent 记忆系统的分层存储与检索架构设计

一、Agent 失忆症：长对话场景下的上下文丢失与检索困境

在构建轻量化 Agent 产品时，一个经常被忽视却致命的问题是——Agent 的"记忆"管理。当对话轮次超过数十轮，或 Agent 需要跨会话保持用户偏好时，简单地将所有历史消息拼接到 Prompt 中，不仅会迅速触及 Token 上限，还会导致模型注意力分散、响应质量下降。更棘手的是，在多用户并发场景下，记忆的存储与检索延迟直接影响 Agent 的响应速度。

核心痛点可以归纳为三点：第一，全量上下文注入导致 Token 消耗线性增长，成本不可控；第二，无差别的记忆检索引入大量噪声，降低生成质量；第三，跨会话的长期记忆缺乏有效的过期与合并机制，存储膨胀严重。这些问题在开源 Agent 框架中普遍存在，因为大多数框架只提供了简单的消息列表管理，缺少系统化的记忆架构。

二、分层记忆模型：从工作记忆到长期归档的架构设计

借鉴认知科学中的记忆分层理论，Agent 的记忆系统可以划分为三层：工作记忆（Working Memory）、情景记忆（Episodic Memory）和语义记忆（Semantic Memory）。每一层有不同的存储介质、检索策略和生命周期。

flowchart TB
    subgraph 输入层
        U[用户输入] --> P[预处理与意图识别]
    end

    subgraph 工作记忆
        P --> WM[当前会话上下文窗口]
        WM --> |滑动窗口淘汰| EB[过期缓冲区]
    end

    subgraph 情景记忆
        EB --> |向量化嵌入| VS[向量存储]
        VS --> |语义检索| RC[相关片段召回]
    end

    subgraph 语义记忆
        VS --> |周期性合并| KB[知识摘要库]
        KB --> |关键词检索| SK[结构化知识]
    end

    RC --> AGG[记忆聚合器]
    SK --> AGG
    WM --> AGG
    AGG --> LLM[大语言模型]

    style WM fill:#e1f5fe
    style VS fill:#fff3e0
    style KB fill:#e8f5e9

工作记忆维护当前会话的最近 N 轮对话，直接注入 Prompt，保证即时上下文的完整性。情景记忆将历史对话片段向量化存储，通过语义相似度检索与当前问题相关的历史片段。语义记忆则通过周期性的摘要合并，将高频出现的知识提炼为结构化条目，支持精确的关键词匹配。

这种分层设计的核心优势在于：每一层都可以独立选择最优的存储引擎和检索算法。工作记忆用内存中的环形缓冲区，情景记忆用向量数据库，语义记忆用关系型数据库或文档数据库。各层之间通过异步的晋升（Promotion）机制连接，避免记忆写入阻塞主流程。

三、分层记忆引擎的生产级实现

# memory_engine.py — Agent 分层记忆引擎核心实现
import time
import json
import hashlib
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from collections import deque
from typing import Optional

import numpy as np


@dataclass
class MemoryItem:
    """记忆条目的统一数据模型"""
    content: str
    role: str                    # user / assistant / system
    session_id: str
    timestamp: float = field(default_factory=time.time)
    metadata: dict = field(default_factory=dict)
    embedding: Optional[np.ndarray] = None
    importance_score: float = 0.0  # 重要性评分，用于晋升决策

    @property
    def item_id(self) -> str:
        raw = f"{self.session_id}:{self.timestamp}:{self.content[:64]}"
        return hashlib.md5(raw.encode()).hexdigest()[:16]


class WorkingMemory:
    """工作记忆：滑动窗口 + 重要性保留策略"""

    def __init__(self, max_tokens: int = 4096, reserve_ratio: float = 0.3):
        self.max_tokens = max_tokens
        # 预留比例：为情景记忆召回结果保留的 Token 空间
        self.reserve_ratio = reserve_ratio
        self._buffer: deque[MemoryItem] = deque()
        self._token_count = 0

    def _estimate_tokens(self, text: str) -> int:
        # 粗略估算：中文约 1.5 Token/字，英文约 1 Token/4字符
        cn_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        en_chars = len(text) - cn_chars
        return int(cn_chars * 1.5 + en_chars * 0.25)

    def add(self, item: MemoryItem) -> list[MemoryItem]:
        """添加记忆条目，返回被淘汰的条目列表"""
        item_tokens = self._estimate_tokens(item.content)
        evicted = []

        # 计算可用空间（扣除预留空间）
        available = self.max_tokens * (1 - self.reserve_ratio)

        while self._buffer and self._token_count + item_tokens > available:
            # 优先淘汰重要性最低且最早的条目
            oldest = self._buffer.popleft()
            self._token_count -= self._estimate_tokens(oldest.content)
            evicted.append(oldest)

        self._buffer.append(item)
        self._token_count += item_tokens
        return evicted

    def get_context(self) -> list[dict]:
        """获取当前工作记忆的上下文，格式化为 LLM 输入"""
        return [
            {"role": item.role, "content": item.content}
            for item in self._buffer
        ]

    @property
    def remaining_tokens(self) -> int:
        return int(self.max_tokens * self.reserve_ratio)


class EpisodicMemory:
    """情景记忆：基于向量相似度的片段检索"""

    def __init__(self, embed_dim: int = 768, top_k: int = 5,
                 similarity_threshold: float = 0.65):
        self.embed_dim = embed_dim
        self.top_k = top_k
        self.similarity_threshold = similarity_threshold
        # 生产环境应替换为 Milvus / Qdrant / Pinecone
        self._vectors: list[np.ndarray] = []
        self._items: list[MemoryItem] = []

    def store(self, item: MemoryItem) -> None:
        """存储记忆条目及其向量表示"""
        if item.embedding is None:
            return  # 无嵌入向量的条目不存入情景记忆
        self._vectors.append(item.embedding)
        self._items.append(item)

    def recall(self, query_embedding: np.ndarray,
               max_tokens: int = 1024) -> list[MemoryItem]:
        """根据查询向量召回最相关的记忆片段"""
        if not self._vectors:
            return []

        # 批量计算余弦相似度
        query_norm = query_embedding / (np.linalg.norm(query_embedding) + 1e-8)
        matrix = np.array(self._vectors)
        matrix_norm = matrix / (np.linalg.norm(matrix, axis=1, keepdims=True) + 1e-8)
        similarities = matrix_norm @ query_norm

        # 按相似度排序，过滤低分结果
        ranked_indices = np.argsort(similarities)[::-1]
        results = []
        token_budget = max_tokens

        for idx in ranked_indices:
            if similarities[idx] < self.similarity_threshold:
                break
            item = self._items[idx]
            item_tokens = len(item.content)  # 简化估算
            if token_budget - item_tokens < 0:
                break
            results.append(item)
            token_budget -= item_tokens
            if len(results) >= self.top_k:
                break

        return results


class SemanticMemory:
    """语义记忆：结构化知识的摘要与合并"""

    def __init__(self):
        # 生产环境应替换为 PostgreSQL / MongoDB
        self._knowledge: dict[str, dict] = {}

    def merge(self, items: list[MemoryItem], summary_fn) -> None:
        """将一组记忆条目合并为结构化知识摘要"""
        if not items:
            return

        # 按主题聚类后分别生成摘要
        cluster_key = items[0].metadata.get("topic", "general")
        contents = [item.content for item in items]
        combined = "\n".join(contents)

        # 调用外部摘要函数（可以是 LLM 调用）
        summary = summary_fn(combined)

        self._knowledge[cluster_key] = {
            "summary": summary,
            "source_count": len(items),
            "updated_at": time.time(),
        }

    def query(self, keywords: list[str]) -> list[dict]:
        """关键词检索结构化知识"""
        results = []
        for key, value in self._knowledge.items():
            # 简单的关键词匹配，生产环境可用全文检索
            if any(kw in key or kw in value["summary"] for kw in keywords):
                results.append({"topic": key, **value})
        return results


class MemoryEngine:
    """记忆引擎：协调三层记忆的读写与晋升"""

    def __init__(self, working: WorkingMemory, episodic: EpisodicMemory,
                 semantic: SemanticMemory, embed_fn=None, summarize_fn=None):
        self.working = working
        self.episodic = episodic
        self.semantic = semantic
        self._embed_fn = embed_fn      # 文本向量化函数
        self._summarize_fn = summarize_fn  # 摘要生成函数

    def add_memory(self, content: str, role: str, session_id: str,
                   metadata: dict = None) -> None:
        """写入新记忆，自动处理晋升流程"""
        item = MemoryItem(
            content=content,
            role=role,
            session_id=session_id,
            metadata=metadata or {},
        )

        # 重要性评分：基于内容长度和元数据标记
        item.importance_score = self._calc_importance(item)

        # 写入工作记忆，获取被淘汰的条目
        evicted = self.working.add(item)

        # 被淘汰的条目晋升到情景记忆
        for evicted_item in evicted:
            if self._embed_fn:
                evicted_item.embedding = self._embed_fn(evicted_item.content)
            self.episodic.store(evicted_item)

    def recall(self, query: str, session_id: str = None) -> list[dict]:
        """多路召回：工作记忆 + 情景记忆 + 语义记忆"""
        context = self.working.get_context()
        remaining = self.working.remaining_tokens

        # 情景记忆召回
        if self._embed_fn and remaining > 0:
            query_emb = self._embed_fn(query)
            episodic_results = self.episodic.recall(query_emb, remaining)
            for item in episodic_results:
                context.append({
                    "role": "system",
                    "content": f"[历史相关] {item.content}"
                })

        # 语义记忆召回
        keywords = query.split()[:5]  # 简化关键词提取
        semantic_results = self.semantic.query(keywords)
        for entry in semantic_results:
            context.append({
                "role": "system",
                "content": f"[知识摘要] {entry['summary']}"
            })

        return context

    def _calc_importance(self, item: MemoryItem) -> float:
        """计算记忆条目的重要性评分"""
        score = 0.0
        # 包含决策或结论的内容更重要
        decision_keywords = ["决定", "选择", "确认", "取消", "important", "must"]
        if any(kw in item.content.lower() for kw in decision_keywords):
            score += 0.3
        # 用户偏好类内容更重要
        if item.metadata.get("type") == "preference":
            score += 0.4
        # 较长的内容可能包含更多信息
        score += min(len(item.content) / 500, 0.3)
        return min(score, 1.0)

    def consolidate(self, topic: str = None) -> int:
        """执行记忆合并：将情景记忆中的条目摘要后存入语义记忆"""
        # 简化实现：按主题分组后合并
        items_to_merge = [
            item for item in self.episodic._items
            if topic is None or item.metadata.get("topic") == topic
        ]
        if items_to_merge and self._summarize_fn:
            self.semantic.merge(items_to_merge, self._summarize_fn)
        return len(items_to_merge)

上述实现中，MemoryEngine 作为门面类协调三层记忆的读写。工作记忆使用滑动窗口策略，被淘汰的条目自动晋升到情景记忆。情景记忆基于向量相似度检索，语义记忆通过周期性的合并操作将碎片化的记忆提炼为结构化知识。

四、分层存储的权衡：延迟、一致性与存储成本的三角博弈

分层记忆架构并非银弹，它在解决上下文丢失问题的同时，引入了新的工程权衡。

延迟问题：情景记忆的向量检索在数据量增大时延迟会显著上升。当向量库达到百万级时，即使使用 HNSW 索引，单次检索延迟也可能从毫秒级上升到百毫秒级。缓解方案是在情景记忆层引入缓存——对高频查询的 Top-K 结果进行 LRU 缓存，将热路径的检索延迟压回个位数毫秒。

一致性问题：三层记忆之间的晋升是异步的，存在短暂的不一致窗口。例如，一条记忆刚从工作记忆淘汰但尚未完成向量化存入情景记忆，此时如果发生检索，就会丢失该条记忆。对于大多数 Agent 场景，这种最终一致性是可以接受的；但对于金融交易等强一致性场景，需要引入同步写入模式，在写入情景记忆完成后再返回响应。

存储成本：向量存储的膨胀速度远超预期。一条 100 Token 的对话，其 768 维向量需要约 3KB 存储空间，加上元数据索引，实际占用约 5KB。日活 10 万用户的 Agent，每天新增向量数据可达数 GB。必须设计合理的过期与归档策略——超过 30 天未被召回的情景记忆应降级为冷存储，超过 90 天的应执行摘要合并后删除原始向量。

适用边界：分层记忆架构适用于对话轮次多、需要跨会话保持上下文的 Agent 产品。对于简单的单轮问答工具，引入三层记忆反而增加了不必要的复杂度，直接使用固定窗口的上下文管理即可。

五、总结

Agent 记忆系统的分层设计，本质上是将认知科学中的记忆模型映射到工程架构中。工作记忆保证即时响应，情景记忆提供语义关联，语义记忆沉淀结构化知识。三层之间通过晋升与合并机制实现数据的自动流转。在落地时，需要根据业务场景的延迟要求、一致性等级和存储预算，灵活调整每层的存储引擎和检索策略。对于轻量化 Agent 产品，建议从工作记忆 + 情景记忆两层起步，待数据量增长后再引入语义记忆层。