Agent内存架构四层设计与生产实现

你的 Agent 在演示中成功解决了问题。接着，你又问了一个后续问题——它却一脸茫然地看着你，仿佛之前一个小时的工作从未发生过。它不记得自己应用了什么修复、改动了哪些文件，也不记得自己已经做过什么决定。

这是 2026 年演示型 Agent 与生产级 Agent 之间最大的差距。推理模型已经取得了显著进步，tool-calling 变得可靠，MCP 也解决了“如何与现实世界交互”的问题。但 memory——让 Agent 能够跨调用、跨会话、跨天延续上下文的那一层——大多数团队依然接得不对。

本文将通过实践讲解真正具备扩展能力的 memory 架构：一个四层分类模型、一套现在就能运行的 Python 实现，以及那些会让朴素实现在生产环境中失败的常见问题。

为什么 memory 成了瓶颈

过去 12 个月发生的三项变化，使 Agent memory 成了关键的架构决策：

Agent 的执行轨迹现在可以持续数小时。Agent 不再只回答一个问题，而是会执行多步骤任务：对客服工单进行分类处理、更新 CRM、起草回复、安排后续跟进。每一步都是一次独立的 LLM 调用；除非你构建 memory，否则这些步骤之间唯一能够传递的只有 context window。

Context window 有限，而且成本高昂。即便模型支持 200K token，你也不可能在每次调用时都塞入完整的对话历史和所有检索到的文档。成本会线性增长，延迟会超线性增长；一旦相关上下文被埋没在噪声中，输出质量也会下降。

状态存在于对话之外。用户偏好、项目约定、数据库 schema、可用工具——这些信息都不会自动出现在 prompt 中。如果没有将它们放进 memory 层，Agent 每次运行时都得重新发现一遍。

2026 年，那些能够交付可靠 Agent 的团队不会把 memory 简单地视为“一个向量数据库”。他们将其视为一种层级结构，为每一层采用不同的存储、检索和淘汰策略。

四层 memory 模型

大多数团队只实现了第三层（向量存储），然后就认为大功告成了。这种做法会因为一个不易察觉的原因而失败：semantic memory 用来保存事实，而不是保存对话。把整段对话嵌入向量存储，然后指望余弦相似度找到“相关部分”，只会让 Agent 检索到某项决策中错误的那一半。Episodic memory 与 semantic memory 需要不同的索引和检索策略。

一套可运行的实现

下面是一套最小化、但具备生产级架构形态的 memory 层。它使用 chromadb 进行向量存储，并使用受 token 预算约束的滚动缓冲区保存对话上下文。安装命令如下：

pip install chromadb tiktoken

1. Semantic memory——通过向量检索事实

import chromadb
from chromadb.utils import embedding_functions

class SemanticMemory:
    """Facts and knowledge, retrieved by similarity. The RAG layer."""

    def __init__(self, path: str = "./memory_store"):
        self.client = chromadb.PersistentClient(path=path)
        self.embedder = embedding_functions.DefaultEmbeddingFunction()
        self.collection = self.client.get_or_create_collection(
            name="semantic", embedding_function=self.embedder
        )

    def store(self, fact: str, metadata: dict | None = None):
        """Store a single fact. Split long facts into atomic statements first."""
        self.collection.add(
            documents=[fact],
            metadatas=[metadata or {"source": "agent"}],
            ids=[f"fact_{self.collection.count()}"],
        )

    def recall(self, query: str, n: int = 5) -> list[str]:
        """Return the n most relevant stored facts."""
        results = self.collection.query(query_texts=[query], n_results=n)
        return results["documents"][0] if results["documents"] else []

真正重要的规则是：存储原子化事实，而不是段落。“用户更喜欢 Django，而不是 FastAPI”能够被准确检索出来。一条描述调试过程的 200 词日志，检索出来的只会是噪声。当 Agent 学到新信息时，先让它提取值得长期保留的事实，再写入 memory：

def extract_and_store(conversation: str, memory: SemanticMemory):
    facts = llm.extract_facts(conversation)  # returns list of atomic statements
    for f in facts:
        memory.store(f)

2. Episodic memory——具有 token 预算的对话记忆

原始对话历史的存储成本很低，但重新放回上下文的成本很高。真正具备扩展能力的模式是“滚动窗口 + 摘要”：逐字保留最近 N 个 token，因为 Agent 需要准确的近期上下文；将更早的内容压缩成摘要，并让这些摘要仍然可供检索。

import tiktoken

class EpisodicMemory:
    def __init__(self, max_tokens: int = 4000, summarize_at: int = 3000):
        self.max_tokens = max_tokens
        self.summarize_at = summarize_at
        self.history: list[dict] = []          # verbatim recent turns
        self.summaries: list[str] = []          # compressed older turns
        self.enc = tiktoken.get_encoding("cl100k_base")

    def add_turn(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        if self._token_count() > self.summarize_at:
            self._roll_up()

    def _token_count(self) -> int:
        return sum(len(self.enc.encode(t["content"])) for t in self.history)

    def _roll_up(self):
        """Summarize the oldest half of history and drop it from verbatim storage."""
        old = self.history[: len(self.history) // 2]
        text = "\n".join(f"{t['role']}: {t['content']}" for t in old)
        summary = llm.summarize(text)  # one call, ~50 lines of conversation -> 3 lines
        self.summaries.append(summary)
        self.history = self.history[len(self.history) // 2 :]

    def context(self) -> list[dict]:
        """What actually goes into the next LLM call."""
        return (
            [{"role": "system", "content": "## Prior context\n" + "\n".join(self.summaries)}]
            + self.history
        )

注意这里的不对称性：摘要可以以很低的成本永久保存——每天一个会话也不过几 KB；逐字保存的历史则成本高昂。roll-up 可以让上下文始终处于 token 预算以内，同时又不会丢失长期任务的主线。

3. 统一的 MemoryManager

Agent 不应该知道，也不需要关心自己正在与哪一层交互：

class MemoryManager:
    def __init__(self):
        self.semantic = SemanticMemory()
        self.episodic = EpisodicMemory()

    def remember(self, observation: str, kind: str = "episodic"):
        if kind == "episodic":
            self.episodic.add_turn("user", observation)
        elif kind == "semantic":
            self.semantic.store(observation)

    def build_prompt(self, task: str) -> list[dict]:
        facts = self.semantic.recall(task)
        context = self.episodic.context()
        system = (
            "You are an agent with memory.\n"
            f"## Relevant facts\n{chr(10).join(facts) if facts else '(none yet)'}"
        )
        return [{"role": "system", "content": system}] + context

所有严肃的 Agent framework 最终都会收敛到这种架构形态。LangGraph 将其称为 checkpointing + store，LlamaIndex 将其称为 StorageContext，但底层契约完全相同：相关事实 + 压缩后的历史 + 最近逐字保留的对话轮次，在每次调用时组装起来。

MCP：让 memory 成为工具，而不是 framework 的内置功能

到了 2026 年，暴露 memory 最简洁的方式是将其实现为 MCP server。你的 Agent 会获得 memory_store、memory_recall 和 memory_forget 工具，而任何兼容 MCP 的 Agent——无论是 Claude、你自己的 Agent，还是团队成员的 Agent——都可以共享同一个 memory store。这样一来，memory 层就成了基础设施，而不是嵌入某个 framework 的功能：

from fastmcp import FastMCP

mcp = FastMCP("Agent Memory")

@mcp.tool()
def memory_store(fact: str) -> str:
    """Store a durable fact about the user or project."""
    manager.semantic.store(fact)
    return "stored"

@mcp.tool()
def memory_recall(query: str) -> list[str]:
    """Retrieve relevant stored facts."""
    return manager.semantic.recall(query)

@mcp.tool()
def memory_forget(substring: str) -> str:
    """Delete facts containing substring (privacy/consent requests)."""
    manager.semantic.delete(substring)
    return "deleted"

这样做有两项收益：第一，Agent 可以自行选择何时持久化信息——它会学着判断哪些内容值得记忆，而不是把一切都记录下来；第二，你运行的所有 Agent 都可以使用同一份共享 memory。

会让朴素 memory 实现失败的常见问题

存储一切。把原始对话日志放进向量数据库并不叫 memory，这叫囤积。随着存储中充满噪声，检索质量会迅速崩溃。提取事实，其余内容则直接丢弃。

没有淘汰或整合机制。事实会过时——例如用户已经从 Postgres 迁走，但上个月的 memory 仍然说用户在用 Postgres。为事实添加时效性 metadata，并通过定期整合任务合并或清除旧条目。

只在构建 prompt 时检索一次。长时间运行的 Agent 应该在多次 tool call 之间重新查询 memory，而不是只在开始时查一次。它在第 7 步需要的事实，往往在第 1 步时并不相关。

遗忘 working memory 层。如果每一步都重新读取完整历史，那么每次调用都要承担全部上下文成本。当前任务应当保持最小化且明确；已经完成的工作则移入 episodic 或 semantic 层。

忽视隐私与用户同意。用户有权知道 Agent 记住了哪些关于他们的信息——memory_forget 不是可选功能，而是最基本的要求；在欧盟，它也日益成为一项法律要求。

生产环境检查清单

[ ] 写入 semantic storage 前提取原子化事实

[ ] 采用具有 token 预算和摘要 roll-up 机制的 episodic buffer

[ ] 将 memory 作为工具暴露（MCP server），而不是隐藏在内部实现中

[ ] 长任务运行时，在各步骤之间重新查询

[ ] 时效性 metadata + 定期整合任务

[ ] 面向用户数据权利的遗忘与导出 endpoint

Memory 决定了一个 Agent 究竟只是能完成任务，还是能够持续学习。四层模型——working、episodic、semantic、procedural——为每一层分别采用不同的存储和检索方式，正是区分演示玩具与能够持续积累能力的系统的关键。从上面的代码开始，把它接到 MCP server 后面，你的 Agent 就不会再重复询问同样的问题。

欢迎关注更多关于 AI Agent 工程实践的内容——memory、MCP 与生产级模式。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。