一文搞懂 AI Agent 的记忆系统：从短期到长期，从本地到向量数据库

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 自动存储记忆的时机

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

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：向量数据库选哪个？

Q3：隐私数据怎么处理？

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

7. 总结

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

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