TL;DR：AI Agent 为什么能「记住」你之前说的话？背后的记忆系统分为三层：短期记忆（上下文窗口）、长期记忆（向量数据库）、工作记忆（当前会话状态）。本文详解三层记忆的原理、存储方式和代码实现。

1. 为什么 AI Agent 需要记忆系统

对话系统有上下文窗口限制。GPT-4o 最大 128K Tokens，Claude 3 200K Tokens——听起来很大，但一个中型项目的代码库往往超过 100 万 Tokens。

如果每次对话都把所有历史塞进上下文：

• 每次请求的 Token 成本飙升
• 超出上下文上限后无法处理
• 模型注意力分散，效果下降

解决方案：分层记忆系统。

三层记忆架构

用户输入
    ↓
┌─────────────────────────────────────┐
│  Layer 1: 工作记忆（Working Memory） │
│  当前会话的核心信息，LLM 直接访问     │
└─────────────────────────────────────┘
    ↓（压缩 → 存储）
┌─────────────────────────────────────┐
│  Layer 2: 长期记忆（Long-term Memory）│
│  向量数据库，支持语义检索             │
└─────────────────────────────────────┘
    ↓（历史归档）
┌─────────────────────────────────────┐
│  Layer 3: 档案记忆（Archive Memory） │
│  历史会话归档，冷存储                 │
└─────────────────────────────────────┘

2. 第一层：工作记忆（Working Memory）

📌 位置：LLM 上下文窗口

容量：当前模型的最大上下文窗口（如 128K / 200K Tokens）

生命周期：单次对话

工作记忆是 LLM 直接能「看到」的信息。你的 Prompt、历史对话、检索到的内容——都在这里。

工作记忆的管理策略：

2.1 滑动窗口策略

保留最近 N 条对话，旧的自动丢弃：

Python - 滑动窗口实现

class SlidingWindowMemory:
    def __init__(self, max_messages: int = 20):
        self.max_messages = max_messages
        self.messages = []

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # 超出窗口大小时，从旧到新丢弃
        if len(self.messages) > self.max_messages:
            self.messages.pop(0)

    def get_context(self) -> list:
        return self.messages

    def count_tokens(self) -> int:
        from tiktoken import encoding_for_model
        enc = encoding_for_model("gpt-4o")
        return len(enc.encode(" ".join(m["content"] for m in self.messages)))

# 使用示例
memory = SlidingWindowMemory(max_messages=10)
memory.add("user", "我想开发一个博客系统")
memory.add("assistant", "好的，你想要什么技术栈？")
memory.add("user", "用 FastAPI + Vue")
# 只保留最近 10 条

2.2 摘要压缩策略

当对话太长时，用 LLM 压缩历史：

Python - 摘要压缩

async def summarize_and_compress(messages: list, client) -> list:
    """当 token 超限时，压缩历史对话"""

    old_messages = messages[:-5]  # 保留最近 5 条，其余压缩
    summary_prompt = f"""请将以下对话压缩成一个摘要：

{messages_to_text(old_messages)}

摘要要求：
1. 保留关键信息（需求、决策、技术选型）
2. 忽略无关细节
3. 输出格式：{key_info: ..., decisions: ..., pending: ...}"""

    summary_response = await client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V3",
        messages=[{"role": "user", "content": summary_prompt}]
    )

    # 压缩后的摘要 + 最近 5 条对话
    compressed = [
        {"role": "system", "content": f"【会话摘要】{summary_response.choices[0].message.content}"}
    ] + messages[-5:]

    return compressed

3. 第二层：长期记忆（Long-term Memory）

📌 位置：向量数据库（Qdrant / Chroma / Milvus）

容量：无限制

生命周期：持久化

长期记忆是 AI Agent 的「外部大脑」。当工作记忆不够用时，从长期记忆里检索相关信息，拼进上下文。

3.1 为什么要用向量数据库

普通数据库只能做「精确匹配」（WHERE name = '张三'）。

但人类的记忆是「语义」的——你记不住「张三」这个名字，但能记住「那个帮我修过 Bug 的同事」。

向量数据库做的就是这件事：语义检索。你问「帮我回忆一下上周讨论的技术方案」，向量数据库能找到语义相近的历史记录。

3.2 完整记忆系统实现

Python - 完整 Agent 记忆系统

import chromadb
import openai
from datetime import datetime

# ========== 1. 初始化向量数据库 ==========
client_db = chromadb.Client()
collection = client_db.create_collection("agent_memory")

# ========== 2. 嵌入模型 ==========
embedding_client = openai.OpenAI(
    api_key="your-api-key",
    base_url="https://api.siliconflow.cn/v1"
)

def get_embedding(text: str) -> list:
    response = embedding_client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# ========== 3. 存储记忆 ==========
def store_memory(content: str, metadata: dict):
    """存储一条记忆到向量数据库"""
    memory_id = f"memory_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    embedding = get_embedding(content)

    collection.add(
        ids=[memory_id],
        embeddings=[embedding],
        documents=[content],
        metadatas=[{**metadata, "created_at": datetime.now().isoformat()}]
    )
    print(ff"✅ 已存储记忆：{memory_id}")

# ========== 4. 检索记忆 ==========
def retrieve_memories(query: str, top_k: int = 5) -> list:
    """语义检索相关记忆"""
    query_embedding = get_embedding(query)

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )

    memories = []
    for i, doc in enumerate(results["documents"][0]):
        memories.append({
            "content": doc,
            "metadata": results["metadatas"][0][i],
            "score": results["distances"][0][i]
        })

    return memories

# ========== 5. 记忆衰减策略 ==========
def decay_memories(threshold: float = 0.8):
    """删除不重要/太久远的记忆，释放空间"""
    import time
    all_memories = collection.get()

    for memory in all_memories["ids"]:
        metadata = collection.get(where={"id": memory})["metadatas"][0]

        # 1. 时间衰减：超过 30 天的重要性 -0.5
        created = datetime.fromisoformat(metadata["created_at"])
        days_old = (datetime.now() - created).days
        importance = metadata.get("importance", 0.5) - (days_old / 60)

        if importance < threshold:
            collection.delete(ids=[memory])
            print(ff"🗑️ 删除记忆：{memory}")

3.3 自动存储记忆的时机

不是所有对话都要存储。以下几个时机触发存储：

触发时机	存储内容	重要性
用户明确的需求	需求描述、决策结论	⭐⭐⭐⭐⭐
Agent 执行的关键步骤	执行结果、遇到的问题	⭐⭐⭐⭐
代码变更	改了什么文件、为什么改	⭐⭐⭐⭐⭐
错误和修复	Bug 原因、修复方案	⭐⭐⭐⭐⭐
日常闲聊	不存储	—

4. 第三层：档案记忆（Archive Memory）

📌 位置：本地文件 / 对象存储（S3）

容量：无限制

生命周期：永久归档

档案记忆是「记忆的记忆」——长期记忆经过压缩后，归档到冷存储。

Python - 档案归档

import json
import os
from datetime import datetime

ARCHIVE_DIR = "./memory_archive"

def archive_session(session_id: str, messages: list):
    """归档一个完整的会话到本地文件"""
    os.makedirs(ARCHIVE_DIR, exist_ok=True)

    archive_file = f"{ARCHIVE_DIR}/{session_id}_{datetime.now().strftime('%Y%m%d')}.json"

    with open(archive_file, "w", encoding="utf-8") as f:
        json.dump({
            "session_id": session_id,
            "archived_at": datetime.now().isoformat(),
            "messages": messages
        }, f, ensure_ascii=False, indent=2)

    print(ff"📦 会话 {session_id} 已归档到 {archive_file}")

def load_archive(session_id: str) -> list:
    """读取归档的会话"""
    for filename in os.listdir(ARCHIVE_DIR):
        if filename.startswith(session_id):
            with open(f"{ARCHIVE_DIR}/{filename}", encoding="utf-8") as f:
                data = json.load(f)
                return data["messages"]
    return []

5. 完整 Agent 记忆系统集成

Python - Agent 记忆系统主类

class AgentMemory:

    def __init__(self):
        # 三层记忆
        self.working = SlidingWindowMemory(max_messages=20)
        self.longterm = LongTermMemory()   # 基于 Chroma
        self.archive = ArchiveMemory()

    async def process_message(self, user_input: str, client) -> str:
        # Step 1: 从长期记忆检索相关内容
        relevant_memories = self.longterm.retrieve(user_input)

        # Step 2: 构建 Prompt（工作记忆 + 相关长期记忆）
        system_prompt = f"""你是 Agent，你的记忆如下：
【相关长期记忆】
{relevant_memories}
"""

        # Step 3: 调用 LLM
        response = await client.chat.completions.create(
            model="deepseek-ai/DeepSeek-V3",
            messages=[
                {"role": "system", "content": system_prompt}
            ] + self.working.get_context() + [{"role": "user", "content": user_input}]
        )

        answer = response.choices[0].message.content

        # Step 4: 自动存储重要内容到长期记忆
        self.longterm.store_if_important(user_input, answer)

        # Step 5: 更新工作记忆
        self.working.add("user", user_input)
        self.working.add("assistant", answer)

        return answer

6. 常见问题

Q1：Token 爆炸怎么办？

三层记忆各司其职：工作记忆满了 → 压缩 → 归档到长期记忆 → 定期清理。

Q2：向量数据库选哪个？

数据库	适用场景	推荐度
Chroma	轻量开发/本地	⭐⭐⭐⭐⭐
Qdrant	生产环境/云部署	⭐⭐⭐⭐
Milvus	超大规模数据	⭐⭐⭐
Pinecone	不想运维	⭐⭐⭐⭐

Q3：隐私数据怎么处理？

向量数据库可以本地部署（如 Chroma/Qdrant），数据不经过第三方服务器。如果对隐私要求极高，建议自建。

7. 总结

层级	存储	容量	检索方式	生命周期
工作记忆	内存	128K-200K Tokens	直接访问	单次会话
长期记忆	向量数据库	无限制	语义检索	持久化
档案记忆	本地文件/S3	无限制	精确匹配	归档

三层记忆系统的核心思想：热数据放工作记忆，温数据放长期记忆，冷数据归档。这样既保证了检索速度，又不受上下文窗口限制。

如果对你有帮助，欢迎在评论区聊聊你实现的记忆系统。