真正有用的AI Agent必须能记住。记住用户偏好、历史决策、工具使用经验——这就是Agent记忆系统的核心价值。本文从工程角度系统介绍如何构建一个生产级的Agent记忆架构。一、为什么Agent记忆如此关键想象两个场景场景A无记忆用户告诉Agent我喜欢用Python代码风格参考PEP8。下次对话Agent完全不记得又开始用Java写代码。场景B有记忆Agent记住了用户偏好每次都主动按PythonPEP8风格输出甚至记得上次任务卡在了哪里继续推进。场景B就是我们追求的目标。要实现它需要理解Agent记忆的四种类型| 记忆类型 | 类比 | 内容 | 持久性 ||---------|------|------|--------|| 工作记忆 | 人的注意力 | 当前任务上下文 | 会话级 || 情景记忆 | 个人经历 | 过去的对话和事件 | 长期 || 语义记忆 | 知识 | 学到的事实和规则 | 长期 || 程序记忆 | 技能 | 成功的操作序列 | 长期 |## 二、工作记忆当前任务的上下文管理工作记忆是Agent在单次任务中的RAMpythonfrom dataclasses import dataclass, fieldfrom typing import Any, Optionalfrom datetime import datetimeimport uuiddataclassclass WorkingMemory: Agent工作记忆存储当前任务的所有相关上下文 task_id: str field(default_factorylambda: str(uuid.uuid4())) task_description: str # 当前步骤和进度 current_step: int 0 total_steps: int 0 # 工具调用历史当前任务 tool_calls: list[dict] field(default_factorylist) # 中间结果 intermediate_results: dict[str, Any] field(default_factorydict) # 当前状态变量 variables: dict[str, Any] field(default_factorydict) # 错误历史 errors: list[dict] field(default_factorylist) created_at: datetime field(default_factorydatetime.now) def add_tool_call(self, tool: str, params: dict, result: Any, success: bool): 记录工具调用 self.tool_calls.append({ tool: tool, params: params, result: result, success: success, timestamp: datetime.now().isoformat(), step: self.current_step }) def set_variable(self, key: str, value: Any): 存储中间变量 self.variables[key] value def get_context_summary(self) - str: 生成工作记忆摘要用于注入LLM上下文 lines [f任务{self.task_description}] lines.append(f进度第{self.current_step}/{self.total_steps}步) if self.variables: lines.append(当前变量) for k, v in list(self.variables.items())[-5:]: # 最近5个变量 lines.append(f - {k}: {str(v)[:100]}) if self.tool_calls: lines.append(f已执行 {len(self.tool_calls)} 次工具调用) # 只展示最近3次 for call in self.tool_calls[-3:]: status ✓ if call[success] else ✗ lines.append(f {status} {call[tool]}({list(call[params].keys())})) if self.errors: lines.append(f⚠️ 遇到 {len(self.errors)} 个错误最近错误{self.errors[-1].get(message, )[:50]}) return \n.join(lines)## 三、情景记忆用向量数据库存储过往经历情景记忆存储历史对话和任务执行记录通过语义搜索检索相关经历pythonfrom typing import Optionalimport jsonimport hashlibclass EpisodicMemory: 情景记忆存储和检索历史对话与任务 def __init__(self, vector_store, embedder, max_episodes: int 10000): self.vector_store vector_store # 可以是ChromaDB、Weaviate等 self.embedder embedder self.max_episodes max_episodes def store_episode(self, episode_type: str, # conversation, task, tool_use content: str, metadata: dict None) - str: 存储一条情景记忆 episode_id hashlib.md5( f{episode_type}{content}{datetime.now().isoformat()}.encode() ).hexdigest() # 生成embedding embedding self.embedder.encode(content) episode_metadata { episode_id: episode_id, type: episode_type, content: content, timestamp: datetime.now().isoformat(), importance: self._calculate_importance(content, metadata), **(metadata or {}) } # 存入向量数据库 self.vector_store.add( ids[episode_id], embeddings[embedding.tolist()], documents[content], metadatas[episode_metadata] ) return episode_id def retrieve_relevant(self, query: str, episode_type: Optional[str] None, top_k: int 5, min_similarity: float 0.7) - list[dict]: 检索与当前查询最相关的历史情景 query_embedding self.embedder.encode(query) where_filter {} if episode_type: where_filter[type] episode_type results self.vector_store.query( query_embeddings[query_embedding.tolist()], n_resultstop_k * 2, # 多查一些再过滤 wherewhere_filter if where_filter else None ) episodes [] for i, (doc, meta, dist) in enumerate(zip( results[documents][0], results[metadatas][0], results[distances][0] )): similarity 1 - dist # 将距离转为相似度 if similarity min_similarity: episodes.append({ content: doc, similarity: similarity, type: meta.get(type), timestamp: meta.get(timestamp), importance: meta.get(importance, 0.5) }) # 按 相似度 * 重要性 综合排序 episodes.sort( keylambda x: x[similarity] * x[importance], reverseTrue ) return episodes[:top_k] def _calculate_importance(self, content: str, metadata: dict None) - float: 计算记忆重要性0-1 importance 0.5 # 基础重要性 if metadata: # 任务成功的记忆更重要 if metadata.get(task_success): importance 0.2 # 用户明确确认的记忆更重要 if metadata.get(user_confirmed): importance 0.2 # 错误/失败的记忆也很重要避免重蹈覆辙 if metadata.get(is_error): importance 0.15 # 较长的内容通常包含更多信息 if len(content) 500: importance 0.1 return min(1.0, importance) def forget_old_episodes(self, max_age_days: int 90): 遗忘旧的低重要性记忆 cutoff_date (datetime.now() - timedelta(daysmax_age_days)).isoformat() # 查找旧的低重要性记忆 old_episodes self.vector_store.query( query_embeddings[[0] * self.embedder.get_sentence_embedding_dimension()], n_results100, where{ $and: [ {timestamp: {$lt: cutoff_date}}, {importance: {$lt: 0.4}} ] } ) if old_episodes[ids][0]: self.vector_store.delete(idsold_episodes[ids][0]) print(f遗忘了 {len(old_episodes[ids][0])} 条旧记忆)## 四、语义记忆存储用户偏好和领域知识语义记忆存储的是事实而非经历pythonclass SemanticMemory: 语义记忆存储用户偏好、规则和领域知识 def __init__(self, storage_path: str): self.storage_path storage_path self.memory: dict self._load() def _load(self) - dict: try: with open(self.storage_path, r, encodingutf-8) as f: return json.load(f) except FileNotFoundError: return { user_preferences: {}, domain_facts: {}, rules: [], entities: {} } def _save(self): with open(self.storage_path, w, encodingutf-8) as f: json.dump(self.memory, f, ensure_asciiFalse, indent2) def remember_preference(self, category: str, key: str, value: Any): 记住用户偏好 if category not in self.memory[user_preferences]: self.memory[user_preferences][category] {} self.memory[user_preferences][category][key] { value: value, updated_at: datetime.now().isoformat(), confidence: 1.0 } self._save() def remember_fact(self, domain: str, fact: str, source: str None): 记住领域事实 fact_id hashlib.md5(fact.encode()).hexdigest()[:8] if domain not in self.memory[domain_facts]: self.memory[domain_facts][domain] {} self.memory[domain_facts][domain][fact_id] { fact: fact, source: source, added_at: datetime.now().isoformat() } self._save() def add_rule(self, rule: str, priority: int 5): 添加行为规则如永远不要删除用户文件 self.memory[rules].append({ rule: rule, priority: priority, # 1-10越高越重要 added_at: datetime.now().isoformat() }) # 按优先级排序 self.memory[rules].sort(keylambda x: x[priority], reverseTrue) self._save() def get_user_preferences(self) - str: 生成用户偏好摘要 prefs self.memory[user_preferences] if not prefs: return lines [## 用户偏好] for category, items in prefs.items(): lines.append(f\n### {category}) for key, info in items.items(): lines.append(f- {key}: {info[value]}) return \n.join(lines) def get_rules_for_context(self) - str: 获取规则列表 if not self.memory[rules]: return high_priority [r for r in self.memory[rules] if r[priority] 7] return \n.join(f- {r[rule]} for r in high_priority)## 五、程序记忆记住成功的操作序列程序记忆存储怎么做——成功的工具调用序列和策略pythonclass ProceduralMemory: 程序记忆存储成功的操作序列技能 def __init__(self, storage_path: str): self.storage_path storage_path self.skills: dict self._load() def learn_skill(self, task_description: str, tool_sequence: list[dict], success_outcome: str, context: dict None): 从成功的任务执行中学习新技能 skill_id hashlib.md5(task_description.encode()).hexdigest()[:12] # 提取可复用的模式 pattern self._extract_pattern(task_description, tool_sequence) if skill_id in self.skills: # 已有技能更新成功次数 self.skills[skill_id][success_count] 1 self.skills[skill_id][last_used] datetime.now().isoformat() else: # 新技能 self.skills[skill_id] { skill_id: skill_id, description: task_description, pattern: pattern, tool_sequence: tool_sequence, success_outcome: success_outcome, context: context or {}, success_count: 1, failure_count: 0, created_at: datetime.now().isoformat(), last_used: datetime.now().isoformat() } self._save() return skill_id def find_applicable_skills(self, task_description: str, min_success_rate: float 0.7) - list[dict]: 找到适用于当前任务的历史技能 applicable [] for skill in self.skills.values(): # 计算成功率 total skill[success_count] skill[failure_count] success_rate skill[success_count] / total if total 0 else 0 if success_rate min_success_rate: continue # 简单的关键词匹配实际可以用embedding相似度 task_words set(task_description.lower().split()) skill_words set(skill[description].lower().split()) overlap len(task_words skill_words) / len(task_words | skill_words) if overlap 0.3: applicable.append({ **skill, relevance: overlap, success_rate: success_rate }) applicable.sort(keylambda x: x[relevance] * x[success_rate], reverseTrue) return applicable[:3] def _extract_pattern(self, description: str, tools: list[dict]) - str: 提取任务模式描述 tool_names [t.get(tool, unknown) for t in tools] return f任务{description[:50]}... → 工具序列{ → .join(tool_names)} def _save(self): with open(self.storage_path, w, encodingutf-8) as f: json.dump(self.skills, f, ensure_asciiFalse, indent2) def _load(self) - dict: try: with open(self.storage_path, r, encodingutf-8) as f: return json.load(f) except FileNotFoundError: return {}## 六、统一记忆管理器将四种记忆整合到一个统一接口pythonclass AgentMemorySystem: 统一的Agent记忆系统 def __init__(self, agent_id: str, config: dict): self.agent_id agent_id base_path config.get(storage_path, f./memory/{agent_id}) # 初始化四种记忆 self.working WorkingMemory() self.episodic EpisodicMemory( vector_storeself._init_vector_store(base_path), embedderself._init_embedder(config) ) self.semantic SemanticMemory(f{base_path}/semantic.json) self.procedural ProceduralMemory(f{base_path}/procedural.json) def build_memory_context(self, current_query: str) - str: 构建注入LLM的记忆上下文 parts [] # 1. 语义记忆用户偏好和规则总是包含 prefs self.semantic.get_user_preferences() if prefs: parts.append(prefs) rules self.semantic.get_rules_for_context() if rules: parts.append(f## 重要规则\n{rules}) # 2. 情景记忆相关历史经历 relevant_episodes self.episodic.retrieve_relevant( current_query, top_k3 ) if relevant_episodes: ep_text \n.join( f- [{e[type]}] {e[content][:200]} for e in relevant_episodes ) parts.append(f## 相关历史\n{ep_text}) # 3. 程序记忆适用的历史技能 skills self.procedural.find_applicable_skills(current_query) if skills: skill_text \n.join( f- {s[pattern]}成功率{s[success_rate]:.0%} for s in skills ) parts.append(f## 可参考的历史方法\n{skill_text}) # 4. 工作记忆当前任务状态 if self.working.task_description: parts.append(f## 当前任务状态\n{self.working.get_context_summary()}) return \n\n.join(parts) def after_task_complete(self, task: str, tools_used: list[dict], outcome: str, success: bool): 任务完成后更新记忆 # 存储情景记忆 self.episodic.store_episode( episode_typetask, contentf任务{task}\n结果{outcome}, metadata{ task_success: success, tools_count: len(tools_used) } ) # 如果任务成功学习技能 if success and len(tools_used) 0: self.procedural.learn_skill( task_descriptiontask, tool_sequencetools_used, success_outcomeoutcome ) def learn_from_user_feedback(self, feedback: str): 从用户反馈中提取偏好 # 简单的规则提取实际可用LLM解析 if 喜欢 in feedback or 偏好 in feedback: self.episodic.store_episode( episode_typepreference, contentfeedback, metadata{user_confirmed: True} )## 七、记忆系统的实际集成示例pythonclass AgentWithMemory: 集成记忆系统的Agent示例 def __init__(self, agent_id: str, llm_client): self.memory AgentMemorySystem(agent_id, { storage_path: f./agent_memory/{agent_id} }) self.llm llm_client async def process_request(self, user_message: str) - str: # 1. 从记忆系统获取相关上下文 memory_context self.memory.build_memory_context(user_message) # 2. 构建包含记忆的提示 system_prompt f你是一个有记忆的AI助手。{memory_context}---基于以上背景知识和历史经验回应用户的请求。 # 3. 调用LLM response await self.llm.chat([ {role: system, content: system_prompt}, {role: user, content: user_message} ]) # 4. 将本次对话存入情景记忆 self.memory.episodic.store_episode( episode_typeconversation, contentf用户{user_message}\n助手{response[:200]}, metadata{session: datetime.now().date().isoformat()} ) # 5. 尝试提取偏好 if any(kw in user_message for kw in [喜欢, 不喜欢, 希望, 偏好]): self.memory.learn_from_user_feedback(user_message) return response## 八、记忆系统的最佳实践原则一记忆分层按需检索不要每次都把所有记忆塞进上下文。按当前任务的相关性动态检索控制记忆注入的token量。原则二重要性权重不是每条信息都值得记住。用户明确纠正、任务成功的操作序列、高频偏好——这些值得更高的重要性权重。原则三遗忘机制允许Agent遗忘低重要性的旧记忆防止记忆库膨胀影响检索质量。定期清理monthly是好习惯。原则四记忆隐私如果Agent为多用户服务记忆必须严格按用户隔离绝不能跨用户泄露。原则五可解释性当Agent的决策受到历史记忆影响时应该能向用户解释我是因为记得上次你说X才这样做的。## 九、总结2026年记忆系统已经是生产级AI Agent不可缺少的组件。一个完整的记忆架构需要四个维度协同工作工作记忆保证当前任务的连贯性情景记忆让Agent能从历史经历中学习语义记忆存储用户偏好使Agent越用越懂你程序记忆让Agent掌握可复用的技能。构建好记忆系统是AI Agent从工具进化为助手的关键一步。
AI Agent记忆系统设计2026：短期、长期与情景记忆的工程实现
真正有用的AI Agent必须能记住。记住用户偏好、历史决策、工具使用经验——这就是Agent记忆系统的核心价值。本文从工程角度系统介绍如何构建一个生产级的Agent记忆架构。一、为什么Agent记忆如此关键想象两个场景场景A无记忆用户告诉Agent我喜欢用Python代码风格参考PEP8。下次对话Agent完全不记得又开始用Java写代码。场景B有记忆Agent记住了用户偏好每次都主动按PythonPEP8风格输出甚至记得上次任务卡在了哪里继续推进。场景B就是我们追求的目标。要实现它需要理解Agent记忆的四种类型| 记忆类型 | 类比 | 内容 | 持久性 ||---------|------|------|--------|| 工作记忆 | 人的注意力 | 当前任务上下文 | 会话级 || 情景记忆 | 个人经历 | 过去的对话和事件 | 长期 || 语义记忆 | 知识 | 学到的事实和规则 | 长期 || 程序记忆 | 技能 | 成功的操作序列 | 长期 |## 二、工作记忆当前任务的上下文管理工作记忆是Agent在单次任务中的RAMpythonfrom dataclasses import dataclass, fieldfrom typing import Any, Optionalfrom datetime import datetimeimport uuiddataclassclass WorkingMemory: Agent工作记忆存储当前任务的所有相关上下文 task_id: str field(default_factorylambda: str(uuid.uuid4())) task_description: str # 当前步骤和进度 current_step: int 0 total_steps: int 0 # 工具调用历史当前任务 tool_calls: list[dict] field(default_factorylist) # 中间结果 intermediate_results: dict[str, Any] field(default_factorydict) # 当前状态变量 variables: dict[str, Any] field(default_factorydict) # 错误历史 errors: list[dict] field(default_factorylist) created_at: datetime field(default_factorydatetime.now) def add_tool_call(self, tool: str, params: dict, result: Any, success: bool): 记录工具调用 self.tool_calls.append({ tool: tool, params: params, result: result, success: success, timestamp: datetime.now().isoformat(), step: self.current_step }) def set_variable(self, key: str, value: Any): 存储中间变量 self.variables[key] value def get_context_summary(self) - str: 生成工作记忆摘要用于注入LLM上下文 lines [f任务{self.task_description}] lines.append(f进度第{self.current_step}/{self.total_steps}步) if self.variables: lines.append(当前变量) for k, v in list(self.variables.items())[-5:]: # 最近5个变量 lines.append(f - {k}: {str(v)[:100]}) if self.tool_calls: lines.append(f已执行 {len(self.tool_calls)} 次工具调用) # 只展示最近3次 for call in self.tool_calls[-3:]: status ✓ if call[success] else ✗ lines.append(f {status} {call[tool]}({list(call[params].keys())})) if self.errors: lines.append(f⚠️ 遇到 {len(self.errors)} 个错误最近错误{self.errors[-1].get(message, )[:50]}) return \n.join(lines)## 三、情景记忆用向量数据库存储过往经历情景记忆存储历史对话和任务执行记录通过语义搜索检索相关经历pythonfrom typing import Optionalimport jsonimport hashlibclass EpisodicMemory: 情景记忆存储和检索历史对话与任务 def __init__(self, vector_store, embedder, max_episodes: int 10000): self.vector_store vector_store # 可以是ChromaDB、Weaviate等 self.embedder embedder self.max_episodes max_episodes def store_episode(self, episode_type: str, # conversation, task, tool_use content: str, metadata: dict None) - str: 存储一条情景记忆 episode_id hashlib.md5( f{episode_type}{content}{datetime.now().isoformat()}.encode() ).hexdigest() # 生成embedding embedding self.embedder.encode(content) episode_metadata { episode_id: episode_id, type: episode_type, content: content, timestamp: datetime.now().isoformat(), importance: self._calculate_importance(content, metadata), **(metadata or {}) } # 存入向量数据库 self.vector_store.add( ids[episode_id], embeddings[embedding.tolist()], documents[content], metadatas[episode_metadata] ) return episode_id def retrieve_relevant(self, query: str, episode_type: Optional[str] None, top_k: int 5, min_similarity: float 0.7) - list[dict]: 检索与当前查询最相关的历史情景 query_embedding self.embedder.encode(query) where_filter {} if episode_type: where_filter[type] episode_type results self.vector_store.query( query_embeddings[query_embedding.tolist()], n_resultstop_k * 2, # 多查一些再过滤 wherewhere_filter if where_filter else None ) episodes [] for i, (doc, meta, dist) in enumerate(zip( results[documents][0], results[metadatas][0], results[distances][0] )): similarity 1 - dist # 将距离转为相似度 if similarity min_similarity: episodes.append({ content: doc, similarity: similarity, type: meta.get(type), timestamp: meta.get(timestamp), importance: meta.get(importance, 0.5) }) # 按 相似度 * 重要性 综合排序 episodes.sort( keylambda x: x[similarity] * x[importance], reverseTrue ) return episodes[:top_k] def _calculate_importance(self, content: str, metadata: dict None) - float: 计算记忆重要性0-1 importance 0.5 # 基础重要性 if metadata: # 任务成功的记忆更重要 if metadata.get(task_success): importance 0.2 # 用户明确确认的记忆更重要 if metadata.get(user_confirmed): importance 0.2 # 错误/失败的记忆也很重要避免重蹈覆辙 if metadata.get(is_error): importance 0.15 # 较长的内容通常包含更多信息 if len(content) 500: importance 0.1 return min(1.0, importance) def forget_old_episodes(self, max_age_days: int 90): 遗忘旧的低重要性记忆 cutoff_date (datetime.now() - timedelta(daysmax_age_days)).isoformat() # 查找旧的低重要性记忆 old_episodes self.vector_store.query( query_embeddings[[0] * self.embedder.get_sentence_embedding_dimension()], n_results100, where{ $and: [ {timestamp: {$lt: cutoff_date}}, {importance: {$lt: 0.4}} ] } ) if old_episodes[ids][0]: self.vector_store.delete(idsold_episodes[ids][0]) print(f遗忘了 {len(old_episodes[ids][0])} 条旧记忆)## 四、语义记忆存储用户偏好和领域知识语义记忆存储的是事实而非经历pythonclass SemanticMemory: 语义记忆存储用户偏好、规则和领域知识 def __init__(self, storage_path: str): self.storage_path storage_path self.memory: dict self._load() def _load(self) - dict: try: with open(self.storage_path, r, encodingutf-8) as f: return json.load(f) except FileNotFoundError: return { user_preferences: {}, domain_facts: {}, rules: [], entities: {} } def _save(self): with open(self.storage_path, w, encodingutf-8) as f: json.dump(self.memory, f, ensure_asciiFalse, indent2) def remember_preference(self, category: str, key: str, value: Any): 记住用户偏好 if category not in self.memory[user_preferences]: self.memory[user_preferences][category] {} self.memory[user_preferences][category][key] { value: value, updated_at: datetime.now().isoformat(), confidence: 1.0 } self._save() def remember_fact(self, domain: str, fact: str, source: str None): 记住领域事实 fact_id hashlib.md5(fact.encode()).hexdigest()[:8] if domain not in self.memory[domain_facts]: self.memory[domain_facts][domain] {} self.memory[domain_facts][domain][fact_id] { fact: fact, source: source, added_at: datetime.now().isoformat() } self._save() def add_rule(self, rule: str, priority: int 5): 添加行为规则如永远不要删除用户文件 self.memory[rules].append({ rule: rule, priority: priority, # 1-10越高越重要 added_at: datetime.now().isoformat() }) # 按优先级排序 self.memory[rules].sort(keylambda x: x[priority], reverseTrue) self._save() def get_user_preferences(self) - str: 生成用户偏好摘要 prefs self.memory[user_preferences] if not prefs: return lines [## 用户偏好] for category, items in prefs.items(): lines.append(f\n### {category}) for key, info in items.items(): lines.append(f- {key}: {info[value]}) return \n.join(lines) def get_rules_for_context(self) - str: 获取规则列表 if not self.memory[rules]: return high_priority [r for r in self.memory[rules] if r[priority] 7] return \n.join(f- {r[rule]} for r in high_priority)## 五、程序记忆记住成功的操作序列程序记忆存储怎么做——成功的工具调用序列和策略pythonclass ProceduralMemory: 程序记忆存储成功的操作序列技能 def __init__(self, storage_path: str): self.storage_path storage_path self.skills: dict self._load() def learn_skill(self, task_description: str, tool_sequence: list[dict], success_outcome: str, context: dict None): 从成功的任务执行中学习新技能 skill_id hashlib.md5(task_description.encode()).hexdigest()[:12] # 提取可复用的模式 pattern self._extract_pattern(task_description, tool_sequence) if skill_id in self.skills: # 已有技能更新成功次数 self.skills[skill_id][success_count] 1 self.skills[skill_id][last_used] datetime.now().isoformat() else: # 新技能 self.skills[skill_id] { skill_id: skill_id, description: task_description, pattern: pattern, tool_sequence: tool_sequence, success_outcome: success_outcome, context: context or {}, success_count: 1, failure_count: 0, created_at: datetime.now().isoformat(), last_used: datetime.now().isoformat() } self._save() return skill_id def find_applicable_skills(self, task_description: str, min_success_rate: float 0.7) - list[dict]: 找到适用于当前任务的历史技能 applicable [] for skill in self.skills.values(): # 计算成功率 total skill[success_count] skill[failure_count] success_rate skill[success_count] / total if total 0 else 0 if success_rate min_success_rate: continue # 简单的关键词匹配实际可以用embedding相似度 task_words set(task_description.lower().split()) skill_words set(skill[description].lower().split()) overlap len(task_words skill_words) / len(task_words | skill_words) if overlap 0.3: applicable.append({ **skill, relevance: overlap, success_rate: success_rate }) applicable.sort(keylambda x: x[relevance] * x[success_rate], reverseTrue) return applicable[:3] def _extract_pattern(self, description: str, tools: list[dict]) - str: 提取任务模式描述 tool_names [t.get(tool, unknown) for t in tools] return f任务{description[:50]}... → 工具序列{ → .join(tool_names)} def _save(self): with open(self.storage_path, w, encodingutf-8) as f: json.dump(self.skills, f, ensure_asciiFalse, indent2) def _load(self) - dict: try: with open(self.storage_path, r, encodingutf-8) as f: return json.load(f) except FileNotFoundError: return {}## 六、统一记忆管理器将四种记忆整合到一个统一接口pythonclass AgentMemorySystem: 统一的Agent记忆系统 def __init__(self, agent_id: str, config: dict): self.agent_id agent_id base_path config.get(storage_path, f./memory/{agent_id}) # 初始化四种记忆 self.working WorkingMemory() self.episodic EpisodicMemory( vector_storeself._init_vector_store(base_path), embedderself._init_embedder(config) ) self.semantic SemanticMemory(f{base_path}/semantic.json) self.procedural ProceduralMemory(f{base_path}/procedural.json) def build_memory_context(self, current_query: str) - str: 构建注入LLM的记忆上下文 parts [] # 1. 语义记忆用户偏好和规则总是包含 prefs self.semantic.get_user_preferences() if prefs: parts.append(prefs) rules self.semantic.get_rules_for_context() if rules: parts.append(f## 重要规则\n{rules}) # 2. 情景记忆相关历史经历 relevant_episodes self.episodic.retrieve_relevant( current_query, top_k3 ) if relevant_episodes: ep_text \n.join( f- [{e[type]}] {e[content][:200]} for e in relevant_episodes ) parts.append(f## 相关历史\n{ep_text}) # 3. 程序记忆适用的历史技能 skills self.procedural.find_applicable_skills(current_query) if skills: skill_text \n.join( f- {s[pattern]}成功率{s[success_rate]:.0%} for s in skills ) parts.append(f## 可参考的历史方法\n{skill_text}) # 4. 工作记忆当前任务状态 if self.working.task_description: parts.append(f## 当前任务状态\n{self.working.get_context_summary()}) return \n\n.join(parts) def after_task_complete(self, task: str, tools_used: list[dict], outcome: str, success: bool): 任务完成后更新记忆 # 存储情景记忆 self.episodic.store_episode( episode_typetask, contentf任务{task}\n结果{outcome}, metadata{ task_success: success, tools_count: len(tools_used) } ) # 如果任务成功学习技能 if success and len(tools_used) 0: self.procedural.learn_skill( task_descriptiontask, tool_sequencetools_used, success_outcomeoutcome ) def learn_from_user_feedback(self, feedback: str): 从用户反馈中提取偏好 # 简单的规则提取实际可用LLM解析 if 喜欢 in feedback or 偏好 in feedback: self.episodic.store_episode( episode_typepreference, contentfeedback, metadata{user_confirmed: True} )## 七、记忆系统的实际集成示例pythonclass AgentWithMemory: 集成记忆系统的Agent示例 def __init__(self, agent_id: str, llm_client): self.memory AgentMemorySystem(agent_id, { storage_path: f./agent_memory/{agent_id} }) self.llm llm_client async def process_request(self, user_message: str) - str: # 1. 从记忆系统获取相关上下文 memory_context self.memory.build_memory_context(user_message) # 2. 构建包含记忆的提示 system_prompt f你是一个有记忆的AI助手。{memory_context}---基于以上背景知识和历史经验回应用户的请求。 # 3. 调用LLM response await self.llm.chat([ {role: system, content: system_prompt}, {role: user, content: user_message} ]) # 4. 将本次对话存入情景记忆 self.memory.episodic.store_episode( episode_typeconversation, contentf用户{user_message}\n助手{response[:200]}, metadata{session: datetime.now().date().isoformat()} ) # 5. 尝试提取偏好 if any(kw in user_message for kw in [喜欢, 不喜欢, 希望, 偏好]): self.memory.learn_from_user_feedback(user_message) return response## 八、记忆系统的最佳实践原则一记忆分层按需检索不要每次都把所有记忆塞进上下文。按当前任务的相关性动态检索控制记忆注入的token量。原则二重要性权重不是每条信息都值得记住。用户明确纠正、任务成功的操作序列、高频偏好——这些值得更高的重要性权重。原则三遗忘机制允许Agent遗忘低重要性的旧记忆防止记忆库膨胀影响检索质量。定期清理monthly是好习惯。原则四记忆隐私如果Agent为多用户服务记忆必须严格按用户隔离绝不能跨用户泄露。原则五可解释性当Agent的决策受到历史记忆影响时应该能向用户解释我是因为记得上次你说X才这样做的。## 九、总结2026年记忆系统已经是生产级AI Agent不可缺少的组件。一个完整的记忆架构需要四个维度协同工作工作记忆保证当前任务的连贯性情景记忆让Agent能从历史经历中学习语义记忆存储用户偏好使Agent越用越懂你程序记忆让Agent掌握可复用的技能。构建好记忆系统是AI Agent从工具进化为助手的关键一步。
