Agent记忆架构设计2026:让AI记住重要的事
记忆是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 proceduraldataclassclass Memory: id: str content: str memory_type: MemoryType user_id: str created_at: datetime field(default_factorydatetime.now) last_accessed: datetime field(default_factorydatetime.now) access_count: int 0 importance: float 0.5 # 0-1影响遗忘速度 tags: list[str] field(default_factorylist) metadata: dict field(default_factorydict)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, threshold0.95) if existing: # 更新已有记忆的重要性而不是创建重复记忆 await self._reinforce_memory(existing.id) return existing memory Memory( idself._generate_id(), contentcontent, memory_typememory_type, user_iduser_id, importanceimportance, tagstags or [], ) # 生成向量嵌入 embedding await self.embedder.encode(content) # 存入向量库用于语义检索 await self.vector_store.upsert( idmemory.id, vectorembedding, 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( vectorquery_embedding, top_ktop_k * 2, # 多检索一些后面重排序 filtersfilters ) # 重排序结合语义相似度和时间新近度 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(keylambda x: x[0], reverseTrue) 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(days30) deleted await self.db.delete_where( memories, user_iduser_id, importance_lt0.3, last_accessed_ltcutoff_date ) return deleted elif strategy summarize_old: # 将旧记忆归纳为摘要减少存储量但保留信息 old_memories await self.db.query( memories, user_iduser_id, created_at_ltdatetime.now() - timedelta(days90) ) if len(old_memories) 20: summary await self._summarize_memories(old_memories) await self.remember(summary, user_id, MemoryType.SEMANTIC, importance0.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( queryuser_query, user_iduser_id, top_k5 ) 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_asciiFalse, indent2)}提取规则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( contentitem[content], user_iduser_id, memory_typeMemoryType[item[type].upper()], importanceitem.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(r1[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_iduser_id) await self.vector_store.delete_by_filter({user_id: user_id}) return count—## 总结Agent记忆系统是让AI真正有用的关键基础设施。设计好的记忆系统需要1.分类管理情节、语义、程序性记忆各司其职2.智能检索结合语义相似度和时间衰减的综合排序3.自动遗忘避免记忆库无限膨胀保持检索质量4.隐私保护用户数据的控制权始终归用户5.透明注入把记忆以结构化方式注入上下文让模型知道为什么知道这些信息记忆系统的复杂度应该与应用场景匹配——简单应用用Redis的简单KV存储就够了复杂的多用户长期助手才需要向量检索关系数据库的组合方案。
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2586638.html
如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!