AI Agent 记忆系统的四层架构各层的职责与存储介质是什么？

聊到记忆系统的分层，其实工程界并没有一个“官方钦定”的四层结构，但结合认知科学和实际落地，我习惯把它抽象成下面这四层。每层职责明确，存储介质选型也完全不一样，面试时如果能把这四层讲清楚，基本就超出 90% 的候选人。

---

👁️ L1：感知缓冲层（Sensory Buffer）

职责：

承接摄像头、麦克风、传感器等原始流数据，只做极短时间的“寄存”，不做任何理解。它的核心任务是削峰填谷——下游模块来不及处理时，数据不丢；处理完一帧就扔掉，绝不长期占用资源。

存储介质：

内存环形队列、Redis Stream、Kafka Topic。

特点是高吞吐、自动覆盖、零持久化。

import collections
import time

class SensoryBuffer:
    """最近 3 秒的音频帧缓冲区"""
    def __init__(self, max_seconds=3):
        self.buffer = collections.deque()
        self.max_seconds = max_seconds

    def push(self, audio_chunk):
        self.buffer.append((time.time(), audio_chunk))
        self._evict()

    def get_all(self):
        self._evict()
        return [chunk for _, chunk in self.buffer]

    def _evict(self):
        now = time.time()
        while self.buffer and now - self.buffer[0][0] > self.max_seconds:
            self.buffer.popleft()

实际踩坑：这里千万别上磁盘，否则延迟一上来，整个 Agent 的实时性就崩了。

---

🧠 L2：工作上下文层（Working Context）

职责：

维持当前任务的全部“思维素材”——最近的对话轮次、中间推理步骤（CoT）、子目标、工具返回结果等。它决定了 Agent “此刻能考虑多少信息”，也是 token 预算管控的关键。

存储介质：

LLM 上下文窗口本身（消息列表）、内存变量（scratchpad）、必要时靠 Redis 做跨会话的短期暂存。

import tiktoken

class WorkingContext:
    def __init__(self, system_prompt, max_tokens=4000):
        self.messages = [{"role": "system", "content": system_prompt}]
        self.scratchpad = ""
        self.max_tokens = max_tokens
        self.enc = tiktoken.get_encoding("cl100k_base")

    def add(self, role, content):
        self.messages.append({"role": role, "content": content})
        self._trim()

    def _trim(self):
        total = sum(len(self.enc.encode(m["content"])) for m in self.messages)
        while total > self.max_tokens and len(self.messages) > 1:
            removed = self.messages.pop(1)  # 保留 system prompt
            total -= len(self.enc.encode(removed["content"]))

这一层的难点不是存，而是 怎么在信息超载时聪明地忘掉不重要的部分。

---

📝 L3：情景历史层（Episodic History）

职责：

把每次交互的事件摘要、用户反馈、决策结果打包成“经验片段”，支持按时间或语义检索。下次遇到相似场景时，Agent 能翻出“上次我们怎么解决的”。

存储介质：

向量数据库（Chroma / Milvus）存 embedding + 元数据，关系数据库（PostgreSQL）存完整日志，方便按时间范围查询。

import chromadb
from datetime import datetime
from sentence_transformers import SentenceTransformer

class EpisodicMemory:
    def __init__(self):
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("episodes")
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

    def store(self, user_input, agent_action, outcome):
        text = f"input: {user_input}\naction: {agent_action}\noutcome: {outcome}"
        emb = self.encoder.encode(text).tolist()
        self.collection.add(
            embeddings=[emb],
            documents=[text],
            metadatas=[{"timestamp": datetime.now().isoformat(), "outcome": outcome}],
            ids=[str(abs(hash(text)))]
        )

    def recall(self, query, k=3):
        emb = self.encoder.encode(query).tolist()
        return self.collection.query(query_embeddings=[emb], n_results=k)["documents"][0]

面试加分点：提到会设计 记忆衰减权重，让 Agent 更倾向于采纳近期高质量的经验。

---

📚 L4：语义知识层（Semantic Knowledge）

职责：

存领域知识、产品手册、政策文档等去情境化的稳定知识。和情景历史不同，它不关心“什么时候学的”，只关心“这个事实本身对不对”。

存储介质：

文档块 + 向量索引（FAISS / Pinecone），或结构化知识图谱（Neo4j）。通常结合 RAG 架构，检索后注入工作上下文。

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticKnowledgeBase:
    def __init__(self, documents):
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.chunks = []
        self.embeddings = []
        for doc in documents:
            emb = self.encoder.encode(doc).tolist()
            self.chunks.append(doc)
            self.embeddings.append(emb)
        self.embeddings = np.array(self.embeddings)

    def query(self, question, top_k=2):
        q_emb = np.array(self.encoder.encode(question))
        scores = np.dot(self.embeddings, q_emb) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(q_emb)
        )
        top_idx = np.argsort(scores)[-top_k:][::-1]
        return [self.chunks[i] for i in top_idx]

⚡ 四层架构总览

层级	职责	典型存储介质	生命周期
👁️ 感知缓冲	暂存原始流，等待处理	环形队列、Redis Stream	毫秒~秒
🧠 工作上下文	维持当前思考窗口	内存消息列表、scratchpad	单次任务
📝 情景历史	积累具体交互经验	向量库+元数据库	跨会话长期
📚 语义知识	存储稳定领域知识	向量索引/知识图谱	静态或定期更新

---

🔄 一个延伸思考

我们在 L3 和 L4 上都用了向量检索，但当情景记忆（“上次客户这么说是要退款”）和语义知识（“退款政策是 7 天内”）给出的指引矛盾时，Agent 该听谁的？

这背后其实是记忆检索的 优先级仲裁机制——它比选哪个向量库更能决定系统的可靠程度。

---