人类有工作记忆、情节记忆、语义记忆和程序性记忆——这四类记忆协同工作，让我们能够在不同时间尺度上有效地学习和行动。AI Agent 的记忆系统设计，本质上是对人类认知系统的工程化模仿。本文从记忆类型分类到完整实现，带你构建一个真正「有记忆」的 AI Agent。

一、为什么记忆是 Agent 的核心能力？无记忆的 Agent 每次对话都从零开始，表现得像「金鱼」——永远只记得最近几秒。这导致：- 用户每次都要重新解释背景- 无法从过去的错误中学习- 无法积累领域知识- 多轮任务中容易丢失上下文有了记忆系统，Agent 才能从「工具」升级为「助手」。## 二、记忆类型分类### 2.1 四类记忆模型借鉴认知科学，将 Agent 记忆分为四类：┌─────────────────────────────────────────────────────────────┐│ AI Agent 记忆体系 ││ ││ 工作记忆（Working Memory） ││ - 当前上下文窗口（Context Window） ││ - 临时变量和计算结果 ││ - 持续时间：单次会话内 ││ ││ 情节记忆（Episodic Memory） ││ - 历史对话记录（摘要形式） ││ - 任务执行历史 ││ - 持续时间：跨会话，可遗忘 ││ ││ 语义记忆（Semantic Memory） ││ - 关于用户的事实（偏好、习惯、背景） ││ - 领域知识库 ││ - 持续时间：长期，主动维护 ││ ││ 程序性记忆（Procedural Memory） ││ - 已学会的工作流和最佳实践 ││ - 工具使用经验 ││ - 持续时间：永久，随经验更新 │└─────────────────────────────────────────────────────────────┘## 三、工作记忆：上下文窗口管理这是最基础的记忆层，前文已有详细论述。核心要点：pythonclass WorkingMemory: """管理当前会话的上下文窗口""" def __init__(self, max_tokens: int = 4000): self.messages = [] self.scratch_pad = {} # 临时键值存储 self.max_tokens = max_tokens self._token_count = 0 def add_message(self, role: str, content: str): msg = {"role": role, "content": content, "timestamp": time.time()} tokens = estimate_tokens(content) self.messages.append(msg) self._token_count += tokens # 超限时丢弃最旧的用户-助手对 while self._token_count > self.max_tokens and len(self.messages) > 2: removed = self.messages.pop(1) # 保留 system msg self._token_count -= estimate_tokens(removed["content"]) def set(self, key: str, value: any): """临时存储中间结果""" self.scratch_pad[key] = value def get(self, key: str, default=None) -> any: return self.scratch_pad.get(key, default)## 四、情节记忆：对话历史管理### 4.1 自动摘要与存储pythonimport jsonfrom datetime import datetimefrom openai import OpenAIclient = OpenAI()class EpisodicMemory: """跨会话的对话历史记忆""" def __init__(self, storage_path: str, agent_id: str): self.storage_path = storage_path self.agent_id = agent_id self.episodes = self._load() def _load(self) -> list: try: with open(f"{self.storage_path}/{self.agent_id}_episodes.json") as f: return json.load(f) except FileNotFoundError: return [] def _save(self): import os os.makedirs(self.storage_path, exist_ok=True) with open(f"{self.storage_path}/{self.agent_id}_episodes.json", 'w') as f: json.dump(self.episodes, f, ensure_ascii=False, indent=2) async def save_session(self, session_id: str, messages: list[dict], outcomes: list[str] = None): """将一次会话压缩保存为情节记忆""" # 生成会话摘要 summary = await self._summarize_session(messages) episode = { "session_id": session_id, "timestamp": datetime.utcnow().isoformat(), "summary": summary, "key_facts": await self._extract_facts(messages), "outcomes": outcomes or [], "message_count": len(messages), } self.episodes.append(episode) # 只保留最近 100 个情节 if len(self.episodes) > 100: self.episodes = self.episodes[-100:] self._save() async def _summarize_session(self, messages: list[dict]) -> str: """生成会话摘要""" conversation = "\n".join([ f"{m['role']}: {m['content'][:200]}" for m in messages[-20:] # 只取最后20条 ]) response = client.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"请用2-3句话总结这段对话的主要内容和结果：\n\n{conversation}" }], max_tokens=200 ) return response.choices[0].message.content async def _extract_facts(self, messages: list[dict]) -> list[str]: """从会话中抽取关键事实""" conversation = "\n".join([ f"{m['role']}: {m['content'][:300]}" for m in messages ]) response = client.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"""从以下对话中提取关于用户的重要事实（最多5条），用JSON数组格式输出，每条不超过50字：{conversation}只输出JSON数组。""" }], max_tokens=300 ) try: return json.loads(response.choices[0].message.content) except: return [] def retrieve_recent(self, n: int = 5) -> list[dict]: """获取最近 n 个情节""" return self.episodes[-n:] def search(self, query: str, top_k: int = 3) -> list[dict]: """语义搜索相关情节（简化版，实际应用建议用向量搜索）""" # 简单关键词匹配（生产环境应替换为向量相似度搜索） query_words = set(query.lower().split()) scored = [] for ep in self.episodes: summary_words = set(ep["summary"].lower().split()) overlap = len(query_words & summary_words) if overlap > 0: scored.append((overlap, ep)) scored.sort(key=lambda x: x[0], reverse=True) return [ep for _, ep in scored[:top_k]]## 五、语义记忆：用户画像与知识库### 5.1 用户画像管理pythonclass SemanticMemory: """长期语义记忆：用户偏好和事实知识""" def __init__(self, user_id: str, storage_path: str): self.user_id = user_id self.storage_path = storage_path self.profile = self._load_profile() def _load_profile(self) -> dict: try: with open(f"{self.storage_path}/{self.user_id}_profile.json") as f: return json.load(f) except FileNotFoundError: return { "preferences": {}, "facts": [], "skills": [], "context": {}, "last_updated": None, } async def update_from_conversation(self, messages: list[dict]): """从对话中更新用户画像""" conversation = "\n".join([ f"{m['role']}: {m['content']}" for m in messages ]) response = client.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"""从以下对话中提取关于用户的新信息，以JSON格式输出：{{ "new_facts": ["用户新透露的事实，如工作、偏好等"], "preferences": {{"key": "value格式的偏好信息"}}, "skills": ["用户展示的技能"]}}对话内容：{conversation}只输出JSON，没有新信息则返回空数组/对象。""" }], max_tokens=400 ) try: updates = json.loads(response.choices[0].message.content) self._merge_profile(updates) self._save_profile() except Exception as e: print(f"[ProfileUpdate] 解析失败: {e}") def _merge_profile(self, updates: dict): """合并更新到现有画像""" if updates.get("new_facts"): # 去重添加 existing = set(self.profile["facts"]) for fact in updates["new_facts"]: if fact not in existing: self.profile["facts"].append(fact) existing.add(fact) if updates.get("preferences"): self.profile["preferences"].update(updates["preferences"]) if updates.get("skills"): existing_skills = set(self.profile["skills"]) for skill in updates["skills"]: if skill not in existing_skills: self.profile["skills"].append(skill) existing_skills.add(skill) self.profile["last_updated"] = datetime.utcnow().isoformat() def get_context_for_prompt(self) -> str: """生成可注入提示词的用户上下文""" if not self.profile["facts"] and not self.profile["preferences"]: return "" parts = ["[关于用户的已知信息]"] if self.profile["facts"]: parts.append("事实：" + "；".join(self.profile["facts"][:10])) if self.profile["preferences"]: prefs = "、".join(f"{k}={v}" for k, v in list(self.profile["preferences"].items())[:5]) parts.append(f"偏好：{prefs}") if self.profile["skills"]: parts.append("技能：" + "、".join(self.profile["skills"][:5])) return "\n".join(parts)## 六、程序性记忆：经验积累pythonclass ProceduralMemory: """程序性记忆：工具使用经验和最佳实践""" def __init__(self, agent_id: str, storage_path: str): self.agent_id = agent_id self.storage_path = storage_path self.patterns = self._load_patterns() def record_success(self, task_type: str, approach: str, result: str, tools_used: list[str]): """记录成功的方案""" pattern_key = f"{task_type}_{hash(approach)%10000}" if pattern_key not in self.patterns: self.patterns[pattern_key] = { "task_type": task_type, "approach": approach, "tools": tools_used, "success_count": 0, "failure_count": 0, "last_used": None, } self.patterns[pattern_key]["success_count"] += 1 self.patterns[pattern_key]["last_used"] = datetime.utcnow().isoformat() self._save_patterns() def get_best_practices(self, task_type: str) -> list[dict]: """获取某类任务的最佳实践""" relevant = [ p for p in self.patterns.values() if p["task_type"] == task_type and p["success_count"] > p["failure_count"] ] # 按成功率排序 relevant.sort( key=lambda x: x["success_count"] / max( x["success_count"] + x["failure_count"], 1 ), reverse=True ) return relevant[:3] # 返回 top-3 最佳实践## 七、完整 Agent 记忆系统集成pythonclass MemoryEnabledAgent: """集成四类记忆的完整 Agent""" def __init__(self, agent_id: str, user_id: str, storage_dir: str = "./memory"): self.agent_id = agent_id self.user_id = user_id # 四层记忆 self.working_memory = WorkingMemory(max_tokens=4000) self.episodic_memory = EpisodicMemory(storage_dir, agent_id) self.semantic_memory = SemanticMemory(user_id, storage_dir) self.procedural_memory = ProceduralMemory(agent_id, storage_dir) def build_system_prompt(self, base_prompt: str) -> str: """构建注入记忆的系统提示词""" parts = [base_prompt] # 注入用户画像（语义记忆） user_context = self.semantic_memory.get_context_for_prompt() if user_context: parts.append(user_context) # 注入最近情节（情节记忆） recent = self.episodic_memory.retrieve_recent(n=3) if recent: summaries = "\n".join(f"- {ep['summary']}" for ep in recent) parts.append(f"[近期交互摘要]\n{summaries}") return "\n\n".join(parts) async def respond(self, user_input: str) -> str: # 1. 添加到工作记忆 self.working_memory.add_message("user", user_input) # 2. 搜索相关情节 relevant_episodes = self.episodic_memory.search(user_input) # 3. 获取当前任务类型的最佳实践 # task_type = classify_task(user_input) # best_practices = self.procedural_memory.get_best_practices(task_type) # 4. 构建带记忆的消息 system = self.build_system_prompt("你是一个有记忆能力的AI助手。") messages = [{"role": "system", "content": system}] messages.extend(self.working_memory.messages) # 5. 调用 LLM response = client.chat.completions.create( model="gpt-4o", messages=messages ) answer = response.choices[0].message.content self.working_memory.add_message("assistant", answer) return answer async def end_session(self, session_id: str): """会话结束时持久化记忆""" # 保存情节记忆 await self.episodic_memory.save_session( session_id, self.working_memory.messages ) # 更新用户画像 await self.semantic_memory.update_from_conversation( self.working_memory.messages )记忆系统是 Agent 从「能用」到「好用」的关键跨越。四层记忆体系的设计需要结合具体业务场景做取舍：对于简单的客服机器人，情节记忆 + 语义记忆足够；对于长期陪伴型 Agent，四层完整体系才能支撑真正的「了解你」体验。随着向量数据库和持久化技术的成熟，2026年构建记忆完善的 Agent 已经比以往任何时候都更容易。