记忆是Agent从"工具"到"助手"的关键跨越
没有记忆的AI Agent，每次对话都是第一次见面。它不知道你上周提过哪些需求，不记得你们达成过哪些共识，更不知道上次任务做到了哪一步。这样的Agent能处理孤立的任务，但无法支撑需要长期协作的场景。记忆系统的引入，让Agent从"每次重新出发"变成了"持续成长的伙伴"。但记忆系统的设计并不简单——什么应该记、以什么形式记、多久之后遗忘、如何在检索时快速找到相关记忆——每个问题都需要工程化的解答。本文从架构设计角度，系统梳理2026年Agent记忆系统的完整方案。—## 记忆的四种类型认知科学把人类记忆分为多种类型，这套分类对AI Agent同样适用：### 类型一：情节记忆（Episodic Memory）具体事件的记录，带有时间戳和上下文：- “用户在2026-05-01说他的团队有5人”- “上次任务执行失败，原因是API超时”- “用户偏好使用Python，不喜欢JavaScript”### 类型二：语义记忆（Semantic Memory）通用知识和事实，不绑定具体时间：- “该公司的产品线包括A、B、C三个系列”- “用户的技术栈：FastAPI + PostgreSQL + Redis”- “项目命名规范：snake_case for Python, camelCase for JS”### 类型三：程序性记忆（Procedural Memory）如何完成特定任务的知识：- “处理该用户的代码审查请求时，先检查安全性，再看性能”- “每次会议总结需要包含：决策、行动项、参与者”### 类型四：工作记忆（Working Memory）当前任务的中间状态，会在任务完成后清除（参见Context Engineering文章中的详细讨论）。—## 记忆系统架构┌─────────────────────────────────────────────┐│ Agent记忆系统 ││ ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ││ │ 短期记忆 │ │ 长期记忆 │ │ 工作记忆 │ ││ │ (会话内) │ │ (跨会话) │ │ (任务内) │ ││ └─────┬────┘ └─────┬────┘ └─────┬────┘ ││ │ │ │ ││ └──────────────┼──────────────┘ ││ ▼ ││ ┌─────────────────┐ ││ │ 记忆管理器 │ ││ │ (存储/检索/遗忘) │ ││ └────────┬────────┘ ││ ▼ ││ ┌─────────────────────────┐ ││ │ 存储后端 │ ││ │ 向量DB + 关系DB + 缓存 │ ││ └─────────────────────────┘ │└─────────────────────────────────────────────┘—## 核心实现：记忆管理器pythonfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom enum import Enumfrom typing import Anyimport jsonclass MemoryType(Enum): EPISODIC = "episodic" SEMANTIC = "semantic" PROCEDURAL = "procedural"@dataclassclass Memory: id: str content: str memory_type: MemoryType user_id: str created_at: datetime = field(default_factory=datetime.now) last_accessed: datetime = field(default_factory=datetime.now) access_count: int = 0 importance: float = 0.5 # 0-1，影响遗忘速度 tags: list[str] = field(default_factory=list) metadata: dict = field(default_factory=dict)class AgentMemoryManager: """Agent长期记忆管理器""" def __init__( self, vector_store, # 用于语义搜索 relational_db, # 用于精确查询 embedding_model, ): self.vector_store = vector_store self.db = relational_db self.embedder = embedding_model async def remember( self, content: str, user_id: str, memory_type: MemoryType = MemoryType.EPISODIC, importance: float = 0.5, tags: list[str] = None ) -> Memory: """存储一条记忆""" # 去重：检查是否已有高度相似的记忆 existing = await self._find_similar(content, user_id, threshold=0.95) if existing: # 更新已有记忆的重要性，而不是创建重复记忆 await self._reinforce_memory(existing.id) return existing memory = Memory( id=self._generate_id(), content=content, memory_type=memory_type, user_id=user_id, importance=importance, tags=tags or [], ) # 生成向量嵌入 embedding = await self.embedder.encode(content) # 存入向量库（用于语义检索） await self.vector_store.upsert( id=memory.id, vector=embedding, metadata={ "user_id": user_id, "memory_type": memory_type.value, "created_at": memory.created_at.isoformat(), "importance": importance, "tags": json.dumps(tags or []) } ) # 存入关系库（用于精确查询和管理） await self.db.insert("memories", memory.__dict__) return memory async def recall( self, query: str, user_id: str, top_k: int = 5, memory_types: list[MemoryType] = None, recency_weight: float = 0.3 ) -> list[Memory]: """检索相关记忆""" query_embedding = await self.embedder.encode(query) # 向量语义搜索 filters = {"user_id": user_id} if memory_types: filters["memory_type"] = [t.value for t in memory_types] results = await self.vector_store.search( vector=query_embedding, top_k=top_k * 2, # 多检索一些，后面重排序 filters=filters ) # 重排序：结合语义相似度和时间新近度 reranked = [] for result in results: memory = await self.db.get("memories", result.id) if memory: # 综合评分：语义相似度 + 时间衰减 + 重要性 age_days = (datetime.now() - memory.created_at).days recency_score = 1.0 / (1 + age_days * 0.1) # 时间衰减 final_score = ( (1 - recency_weight) * result.score + recency_weight * recency_score ) * memory.importance reranked.append((final_score, memory)) # 更新访问时间（加强记忆） await self._update_access(memory.id) # 按评分降序排列，返回top_k reranked.sort(key=lambda x: x[0], reverse=True) return [m for _, m in reranked[:top_k]] async def forget( self, user_id: str, strategy: str = "importance_decay" ) -> int: """根据遗忘策略清理记忆""" if strategy == "importance_decay": # 删除重要性低且长时间未访问的记忆 cutoff_date = datetime.now() - timedelta(days=30) deleted = await self.db.delete_where( "memories", user_id=user_id, importance_lt=0.3, last_accessed_lt=cutoff_date ) return deleted elif strategy == "summarize_old": # 将旧记忆归纳为摘要，减少存储量但保留信息 old_memories = await self.db.query( "memories", user_id=user_id, created_at_lt=datetime.now() - timedelta(days=90) ) if len(old_memories) > 20: summary = await self._summarize_memories(old_memories) await self.remember(summary, user_id, MemoryType.SEMANTIC, importance=0.8) await self.db.delete_many("memories", [m.id for m in old_memories]) return len(old_memories)—## 记忆注入：如何把记忆用起来存储了记忆，还需要在合适的时机注入上下文：pythonclass MemoryInjector: """在Agent对话中动态注入相关记忆""" def __init__(self, memory_manager: AgentMemoryManager): self.memory_manager = memory_manager async def build_memory_context( self, user_query: str, user_id: str, max_tokens: int = 500 ) -> str: """为当前查询构建记忆上下文""" # 检索最相关的记忆 memories = await self.memory_manager.recall( query=user_query, user_id=user_id, top_k=5 ) if not memories: return "" # 按类型分组，结构化呈现 episodic = [m for m in memories if m.memory_type == MemoryType.EPISODIC] semantic = [m for m in memories if m.memory_type == MemoryType.SEMANTIC] context_parts = [] if semantic: context_parts.append("**关于该用户的已知信息：**") for m in semantic[:3]: context_parts.append(f"- {m.content}") if episodic: context_parts.append("\n**相关历史交互：**") for m in episodic[:2]: date_str = m.created_at.strftime("%Y-%m-%d") context_parts.append(f"- [{date_str}] {m.content}") return "\n".join(context_parts) async def extract_and_store_memories( self, conversation: list[dict], user_id: str ): """从对话中提取值得记忆的信息""" # 使用LLM提取值得记忆的事实 extraction_prompt = f"""从以下对话中提取值得长期记忆的信息：{json.dumps(conversation, ensure_ascii=False, indent=2)}提取规则：1. 用户明确表达的偏好或需求2. 用户的个人背景信息（技术栈、团队规模等）3. 达成的共识或决策4. 需要跟进的行动项输出JSON数组，每条格式：{{"content": "...", "type": "semantic|episodic", "importance": 0.0-1.0}}如果没有值得记忆的信息，返回空数组 []。""" # 调用LLM提取 response = await self._call_llm(extraction_prompt) extracted = json.loads(response) # 存储提取的记忆 for item in extracted: await self.memory_manager.remember( content=item["content"], user_id=user_id, memory_type=MemoryType[item["type"].upper()], importance=item.get("importance", 0.5) )—## 隐私与安全：记忆系统的边界记忆系统存储用户的个人信息，必须认真对待隐私问题：1.用户控制权：提供接口让用户查看、删除自己的记忆2.数据最小化：只记忆任务相关的信息，不记录无关的个人细节3.访问控制：严格隔离不同用户的记忆数据4.加密存储：敏感记忆内容应该加密存储pythonclass PrivacyAwareMemoryManager(AgentMemoryManager): """具备隐私保护的记忆管理器""" SENSITIVE_PATTERNS = ["密码", "身份证", "银行卡", "手机号"] async def remember(self, content: str, user_id: str, **kwargs) -> Memory: # 脱敏处理 sanitized = self._sanitize(content) return await super().remember(sanitized, user_id, **kwargs) def _sanitize(self, content: str) -> str: import re # 脱敏手机号 content = re.sub(r'1[3-9]\d{9}', '1**********', content) # 脱敏邮箱 content = re.sub(r'\w+@\w+\.\w+', '***@***.***', content) return content async def delete_user_memories(self, user_id: str) -> int: """删除用户的所有记忆（GDPR合规）""" count = await self.db.delete_where("memories", user_id=user_id) await self.vector_store.delete_by_filter({"user_id": user_id}) return count—## 总结Agent记忆系统是让AI真正有用的关键基础设施。设计好的记忆系统需要：1.分类管理：情节、语义、程序性记忆各司其职2.智能检索：结合语义相似度和时间衰减的综合排序3.自动遗忘：避免记忆库无限膨胀，保持检索质量4.隐私保护：用户数据的控制权始终归用户5.透明注入：把记忆以结构化方式注入上下文，让模型知道为什么知道这些信息记忆系统的复杂度应该与应用场景匹配——简单应用用Redis的简单KV存储就够了，复杂的多用户长期助手才需要向量检索+关系数据库的组合方案。
