TL;DRAI Agent 为什么能「记住」你之前说的话背后的记忆系统分为三层短期记忆上下文窗口、长期记忆向量数据库、工作记忆当前会话状态。本文详解三层记忆的原理、存储方式和代码实现。1. 为什么 AI Agent 需要记忆系统对话系统有上下文窗口限制。GPT-4o 最大 128K TokensClaude 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_messages10) 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( modeldeepseek-ai/DeepSeek-V3, messages[{role: user, content: summary_prompt}] ) # 压缩后的摘要 最近 5 条对话 compressed [ {role: system, content: f【会话摘要】{summary_response.choices[0].message.content}} ] messages[-5:] return compressed3. 第二层长期记忆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_keyyour-api-key, base_urlhttps://api.siliconflow.cn/v1 ) def get_embedding(text: str) - list: response embedding_client.embeddings.create( modeltext-embedding-3-small, inputtext ) return response.data[0].embedding # 3. 存储记忆 def store_memory(content: str, metadata: dict): 存储一条记忆到向量数据库 memory_id fmemory_{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_resultstop_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_okTrue) archive_file f{ARCHIVE_DIR}/{session_id}_{datetime.now().strftime(%Y%m%d)}.json with open(archive_file, w, encodingutf-8) as f: json.dump({ session_id: session_id, archived_at: datetime.now().isoformat(), messages: messages }, f, ensure_asciiFalse, indent2) 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}, encodingutf-8) as f: data json.load(f) return data[messages] return []5. 完整 Agent 记忆系统集成Python - Agent 记忆系统主类class AgentMemory: def __init__(self): # 三层记忆 self.working SlidingWindowMemory(max_messages20) 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( modeldeepseek-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 answer6. 常见问题Q1Token 爆炸怎么办三层记忆各司其职工作记忆满了 → 压缩 → 归档到长期记忆 → 定期清理。Q2向量数据库选哪个数据库适用场景推荐度Chroma轻量开发/本地⭐⭐⭐⭐⭐Qdrant生产环境/云部署⭐⭐⭐⭐Milvus超大规模数据⭐⭐⭐Pinecone不想运维⭐⭐⭐⭐Q3隐私数据怎么处理向量数据库可以本地部署如 Chroma/Qdrant数据不经过第三方服务器。如果对隐私要求极高建议自建。7. 总结层级存储容量检索方式生命周期工作记忆内存128K-200K Tokens直接访问单次会话长期记忆向量数据库无限制语义检索持久化档案记忆本地文件/S3无限制精确匹配归档三层记忆系统的核心思想热数据放工作记忆温数据放长期记忆冷数据归档。这样既保证了检索速度又不受上下文窗口限制。如果对你有帮助欢迎在评论区聊聊你实现的记忆系统。
