AI Agent最令人沮丧的问题之一是健忘每次对话都要重新介绍背景Agent无法利用历史经验优化决策跨会话的知识积累为零。这不是模型能力问题而是记忆系统工程问题。本文从架构角度深入探讨如何为Agent构建完善的记忆系统让Agent具备真正的长期记忆能力。## Agent记忆系统的四种类型认知科学将人类记忆分为多种类型Agent记忆系统也借鉴了这一分类### 1. 感知记忆Sensory Memory— 当前上下文对应LLM的context window是Agent当下能看到的信息。-容量有限通常32K-200K tokens-持久性会话结束即消失-特点访问最快但量有限### 2. 短期工作记忆Working Memory— 任务状态Agent当前任务的执行状态、中间结果、对话摘要。-容量适中-持久性可跨轮次任务结束清除-实现内存中的结构化数据### 3. 长期情节记忆Episodic Memory— 历史经历Agent经历过的具体事件、对话、操作记录。-容量大受存储限制-持久性永久或很长时间-实现向量数据库 时序索引### 4. 长期语义记忆Semantic Memory— 知识库Agent积累的结构化知识、用户偏好、领域事实。-容量大-持久性永久-实现关系数据库 向量数据库记忆体系架构┌──────────────────────────────────────┐│ 感知记忆Context Window ││ ┌────────────────────────────────┐ ││ │ 工作记忆Task State │ ││ └────────────────────────────────┘ │└──────────────────────────────────────┘ ↕ 检索/写入┌──────────┐ ┌────────────────────┐│情节记忆 │ │ 语义记忆知识 ││历史 │ │ 用户偏好/事实 │└──────────┘ └────────────────────┘## 记忆系统的核心工程挑战### 挑战1检索精度记忆库可能有百万条记录如何找到最相关的记忆- 纯语义搜索向量相似度擅长语义匹配- 关键词搜索BM25擅长精确词汇匹配- 混合检索两者结合2026年的主流方案### 挑战2记忆遗忘所有记忆都保留会导致噪声增加需要有策略地遗忘- 时间衰减旧记忆权重降低- 重要性评分高价值记忆优先保留- 去重合并相似记忆合并为更抽象的知识### 挑战3记忆冲突不同时间的记忆可能相互矛盾用户偏好变了、事实更新了需要冲突检测与解决机制。## 完整记忆系统实现### 核心数据结构pythonfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom enum import Enumfrom typing import Anyclass MemoryType(Enum): EPISODIC episodic # 情节记忆 SEMANTIC semantic # 语义记忆 PROCEDURAL procedural # 程序性记忆如何做某事 WORKING working # 工作记忆dataclassclass Memory: id: str content: str # 记忆内容自然语言 memory_type: MemoryType embedding: list[float] | None # 向量嵌入 # 元数据 created_at: datetime field(default_factorydatetime.now) last_accessed: datetime field(default_factorydatetime.now) access_count: int 0 importance_score: float 0.5 # 0-1重要性 # 关联数据 source_session_id: str # 来源会话 entities: list[str] field(default_factorylist) # 相关实体 tags: list[str] field(default_factorylist) # 时间衰减 property def effective_importance(self) - float: 考虑时间衰减的有效重要性 days_old (datetime.now() - self.created_at).days decay_factor 0.99 ** days_old # 每天衰减1% frequency_boost min(self.access_count / 10, 1.0) * 0.3 return self.importance_score * decay_factor frequency_boost### 记忆存储引擎pythonimport chromadbfrom sentence_transformers import SentenceTransformerclass MemoryStore: 基于ChromaDB的记忆存储引擎 def __init__(self, persist_dir: str ./memory_db): self.client chromadb.PersistentClient(pathpersist_dir) self.encoder SentenceTransformer(BAAI/bge-m3) # 多语言嵌入模型 # 不同记忆类型用不同集合 self.collections { episodic: self.client.get_or_create_collection(episodic_memories), semantic: self.client.get_or_create_collection(semantic_memories), procedural: self.client.get_or_create_collection(procedural_memories), } def add_memory(self, memory: Memory): 添加记忆到存储 if memory.embedding is None: memory.embedding self.encoder.encode(memory.content).tolist() collection self.collections.get(memory.memory_type.value) if not collection: return collection.add( ids[memory.id], embeddings[memory.embedding], documents[memory.content], metadatas[{ created_at: memory.created_at.isoformat(), importance_score: memory.importance_score, tags: ,.join(memory.tags), entities: ,.join(memory.entities), session_id: memory.source_session_id, }] ) def search(self, query: str, memory_types: list[MemoryType] None, top_k: int 5, min_relevance: float 0.7) - list[dict]: 混合检索记忆 query_embedding self.encoder.encode(query).tolist() if memory_types is None: memory_types list(MemoryType) all_results [] for mem_type in memory_types: if mem_type.value not in self.collections: continue results self.collections[mem_type.value].query( query_embeddings[query_embedding], n_resultsmin(top_k, 10), include[documents, metadatas, distances] ) for doc, meta, dist in zip( results[documents][0], results[metadatas][0], results[distances][0] ): relevance 1 - dist # 转换为相似度 if relevance min_relevance: all_results.append({ content: doc, metadata: meta, relevance: relevance, memory_type: mem_type.value }) # 按相关性排序 return sorted(all_results, keylambda x: x[relevance], reverseTrue)[:top_k]### 记忆管理器Memory Managerpythonclass AgentMemoryManager: Agent记忆系统的高层管理器 def __init__(self, agent_id: str, store: MemoryStore, llm_client): self.agent_id agent_id self.store store self.llm llm_client self.working_memory: dict {} # 当前任务的工作记忆 # ─── 记忆写入 ─────────────────────────────────────────────── def remember_conversation(self, user_msg: str, agent_response: str, session_id: str): 将对话提炼为记忆 # 让LLM提炼关键信息 extracted self.extract_key_info(user_msg, agent_response) if extracted[has_new_info]: memory Memory( idfep_{session_id}_{int(datetime.now().timestamp())}, contentextracted[summary], memory_typeMemoryType.EPISODIC, importance_scoreextracted[importance], tagsextracted[tags], entitiesextracted[entities], source_session_idsession_id, embeddingNone # 存储时自动生成 ) self.store.add_memory(memory) def learn_user_preference(self, preference: str, category: str): 记录用户偏好语义记忆 existing self.store.search( f用户偏好 {category}, memory_types[MemoryType.SEMANTIC], top_k1, min_relevance0.9 ) if existing: # 更新现有偏好记忆 self._update_memory(existing[0][id], preference) else: memory Memory( idfsem_pref_{category}_{int(datetime.now().timestamp())}, contentf用户偏好[{category}]{preference}, memory_typeMemoryType.SEMANTIC, importance_score0.8, # 用户偏好重要性高 tags[用户偏好, category], embeddingNone ) self.store.add_memory(memory) # ─── 记忆检索 ─────────────────────────────────────────────── def recall(self, query: str, context: str , max_tokens: int 2000) - str: 检索相关记忆格式化为上下文字符串 results self.store.search( query context, top_k10, min_relevance0.65 ) if not results: return # 格式化记忆为上下文 memory_context 【相关历史记忆】\n token_count 0 for r in results: mem_text f- [{r[memory_type]}] {r[content]}\n mem_tokens len(mem_text.split()) * 1.3 # 粗略估算 if token_count mem_tokens max_tokens: break memory_context mem_text token_count mem_tokens return memory_context # ─── 记忆整理 ─────────────────────────────────────────────── def consolidate_memories(self, min_importance: float 0.3): 定期整理记忆删除低重要性合并重复 # 实际生产中应在低峰期异步执行 print(f开始整理记忆Agent: {self.agent_id}...) def extract_key_info(self, user_msg: str, response: str) - dict: 用LLM提炼对话中的关键信息 prompt f分析以下对话提炼需要记住的关键信息用户{user_msg}助手{response}输出JSON格式{{ has_new_info: true/false, // 是否有值得记忆的新信息 summary: 一句话总结, importance: 0.0-1.0, // 重要性评分 tags: [标签1, 标签2], entities: [实体1, 实体2] // 人名/项目名/产品名等}} result self.llm.complete(prompt) return parse_json(result)### 集成到Agent对话循环pythonclass MemoryEnabledAgent: 具备长期记忆能力的Agent def __init__(self, agent_id: str, llm_client, memory_store: MemoryStore): self.agent_id agent_id self.llm llm_client self.memory AgentMemoryManager(agent_id, memory_store, llm_client) def chat(self, user_message: str, session_id: str) - str: 带记忆的对话 # 1. 检索相关历史记忆 relevant_memories self.memory.recall( queryuser_message, max_tokens1500 ) # 2. 构建带记忆的Prompt system_prompt f你是一个具有长期记忆的AI助手。{relevant_memories if relevant_memories else }请基于以上历史记忆如果有和当前对话给出有连贯性的回答。 # 3. 调用LLM response self.llm.chat( systemsystem_prompt, useruser_message ) # 4. 异步保存本次对话到记忆不阻塞响应 asyncio.create_task( self._save_to_memory_async(user_message, response, session_id) ) return response async def _save_to_memory_async(self, user_msg: str, response: str, session_id: str): 异步保存记忆 self.memory.remember_conversation(user_msg, response, session_id)## 记忆系统的高级特性### 记忆反射Memory Reflection受Generative Agents论文启发定期让Agent反思自己的记忆提炼更高层次的洞察pythonclass MemoryReflector: 定期从情节记忆中提炼语义记忆 def __init__(self, memory_manager: AgentMemoryManager, llm_client): self.memory memory_manager self.llm llm_client def reflect(self, recent_days: int 7): 反思最近的记忆提炼规律和洞察 # 获取最近的情节记忆 recent_memories self.memory.store.search( 最近发生的事情, memory_types[MemoryType.EPISODIC], top_k20 ) if len(recent_memories) 5: return # 记忆不足无需反思 # 让LLM提炼高层规律 memory_text \n.join([r[content] for r in recent_memories]) insights self.llm.complete(f基于以下最近的记忆请提炼3-5条高层次的洞察或规律{memory_text}请用简洁的句子描述每条洞察格式1. [洞察内容]2. [洞察内容]...) # 将洞察存为语义记忆 for i, insight in enumerate(parse_insights(insights)): insight_memory Memory( idfreflection_{datetime.now().strftime(%Y%m%d)}_{i}, contentinsight, memory_typeMemoryType.SEMANTIC, importance_score0.75, # 反思结果重要性较高 tags[反思, 洞察] ) self.memory.store.add_memory(insight_memory)## 生产部署注意事项### 记忆隐私与安全pythonclass PrivacyAwareMemoryStore(MemoryStore): 隐私感知的记忆存储 SENSITIVE_PATTERNS [ r\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b, # 信用卡 r\b1[3-9]\d{9}\b, # 手机号 r\b[A-Z]\d{17}|\d{18}|[0-9]{15}\b, # 身份证 ] def add_memory(self, memory: Memory): # 脱敏处理后再存储 memory.content self.anonymize(memory.content) super().add_memory(memory) def anonymize(self, text: str) - str: import re for pattern in self.SENSITIVE_PATTERNS: text re.sub(pattern, [已脱敏], text) return text## 总结Agent记忆系统是构建真正有记忆AI助手的基础工程。核心要点1.分层设计工作/情节/语义记忆各司其职不要混在一起2.混合检索语义搜索 关键词搜索提升召回准确率3.重要性管理不是所有记忆都同等重要用评分衰减控制记忆质量4.异步写入记忆保存不应阻塞主对话流程5.定期反思从情节记忆中提炼语义规律实现真正的知识积累6.隐私优先存储前脱敏记忆不等于日记要保护用户隐私构建好的记忆系统Agent才能从每次见面都是陌生人转变为真正了解你、能持续学习成长的智能伙伴。