Agent 的记忆之术从金鱼脑到长期记忆AI 智能体记忆机制的设计哲学一、金鱼脑的 Agent为什么无状态的智能体永远不够聪明你跟 ChatGPT 聊了半小时关掉窗口再打开它已经不记得你是谁、你们聊过什么。这就是无状态 Agent 的困境——每次交互都是全新的开始没有上下文积累没有经验复用没有个性记忆。这就像一条金鱼每 7 秒重新认识一次世界。AI Agent 的记忆机制决定了它的智商上限。没有记忆的 Agent 只能做单次推理有短期记忆的 Agent 能完成多轮对话有长期记忆的 Agent 能跨会话积累知识有反思记忆的 Agent 能从错误中学习。记忆的层次越深Agent 的能力越强。我养了一只英短猫叫 Tensor它的记忆系统很有意思——它记得零食柜的位置长期记忆记得我昨天没喂它短期记忆还记得上次偷吃被抓的教训反思记忆。Agent 的记忆设计也可以借鉴这种分层结构。二、Agent 记忆机制架构从工作记忆到反思记忆的四层模型Agent 记忆的核心思路是工作记忆当前上下文→ 短期记忆会话级→ 长期记忆跨会话→ 反思记忆经验总结逐层抽象、逐层持久化。flowchart TD A[Agent 记忆四层模型] -- B[工作记忆: Working Memory] A -- C[短期记忆: Short-Term Memory] A -- D[长期记忆: Long-Term Memory] A -- E[反思记忆: Reflective Memory] B -- B1[当前对话上下文] B -- B2[系统 Prompt] B -- B3[工具调用结果] B -- B4[容量: 4-8K tokens] C -- C1[会话历史摘要] C -- C2[当前任务状态] C -- C3[用户偏好缓存] C -- C4[存储: Redis/内存] D -- D1[用户画像: 偏好/习惯] D -- D2[知识库: 事实/规则] D -- D3[交互历史: 关键事件] D -- D4[存储: 向量数据库] E -- E1[成功经验: 什么有效] E -- E2[失败教训: 什么无效] E -- E3[策略模板: 何时用何法] E -- E4[存储: 结构化知识图谱] B --|摘要压缩| C C --|重要信息持久化| D D --|经验提炼| E E --|策略指导| B D --|知识检索| B C --|上下文补充| B style B fill:#e1f5fe style C fill:#fff3e0 style D fill:#e8f5e9 style E fill:#fce4ec2.1 记忆系统实现# agent_memory.py — Agent 记忆系统 # 设计意图实现四层记忆架构支持记忆的存储、检索、压缩和反思 # 让 Agent 从金鱼脑进化为拥有长期记忆的智能体 from dataclasses import dataclass, field from typing import List, Dict, Optional, Any from datetime import datetime from enum import Enum import hashlib import json import logging logger logging.getLogger(__name__) class MemoryType(Enum): WORKING working # 工作记忆 SHORT_TERM short_term # 短期记忆 LONG_TERM long_term # 长期记忆 REFLECTIVE reflective # 反思记忆 dataclass class MemoryItem: 记忆条目 id: str content: str memory_type: MemoryType timestamp: datetime importance: float 0.5 # 重要性评分 [0, 1] access_count: int 0 # 访问次数 last_accessed: datetime field(default_factorydatetime.now) metadata: Dict[str, Any] field(default_factorydict) embedding: Optional[List[float]] None # 向量嵌入 class WorkingMemory: 工作记忆当前对话的上下文窗口 容量有限4-8K tokens遵循 LRU 淘汰策略 当容量不足时将旧内容摘要压缩后转移到短期记忆 def __init__(self, max_tokens: int 4096): self.max_tokens max_tokens self.items: List[MemoryItem] [] self._current_tokens 0 def add(self, content: str, role: str user) - MemoryItem: 添加记忆条目 item MemoryItem( idhashlib.md5(f{content}_{datetime.now().isoformat()}.encode()).hexdigest()[:12], contentcontent, memory_typeMemoryType.WORKING, timestampdatetime.now(), metadata{role: role}, ) self.items.append(item) self._current_tokens len(content) // 4 # 粗略估算 token 数 # 超出容量时淘汰旧记忆 while self._current_tokens self.max_tokens and len(self.items) 1: evicted self.items.pop(0) self._current_tokens - len(evicted.content) // 4 logger.debug(f工作记忆淘汰: {evicted.id}) return item def get_context(self) - str: 获取当前工作记忆的完整上下文 return \n.join( f[{item.metadata.get(role, unknown)}]: {item.content} for item in self.items ) def get_overflow_items(self) - List[MemoryItem]: 获取超出容量的记忆条目用于转移到短期记忆 return self.items[:-4] if len(self.items) 4 else [] class ShortTermMemory: 短期记忆会话级的历史摘要 存储当前会话的关键信息摘要会话结束后可选择持久化到长期记忆 def __init__(self, max_items: int 50): self.max_items max_items self.items: List[MemoryItem] [] def add(self, content: str, importance: float 0.5) - MemoryItem: 添加短期记忆 item MemoryItem( idhashlib.md5(f{content}_{datetime.now().isoformat()}.encode()).hexdigest()[:12], contentcontent, memory_typeMemoryType.SHORT_TERM, timestampdatetime.now(), importanceimportance, ) self.items.append(item) # 超出容量时按重要性淘汰 if len(self.items) self.max_items: self.items.sort(keylambda x: x.importance) self.items.pop(0) return item def add_from_working_memory(self, items: List[MemoryItem]): 从工作记忆转移过来的条目 for item in items: # 压缩提取关键信息而非保留原文 summary self._summarize(item.content) self.add(summary, importanceitem.importance) staticmethod def _summarize(content: str) - str: 摘要压缩 实际项目中应调用 LLM 生成摘要 这里用简单的截断作为示例 if len(content) 200: return content[:200] ... return content def get_recent(self, n: int 10) - List[MemoryItem]: 获取最近的 N 条记忆 return self.items[-n:] def get_important(self, threshold: float 0.7) - List[MemoryItem]: 获取重要性超过阈值的记忆 return [item for item in self.items if item.importance threshold] class LongTermMemory: 长期记忆跨会话的持久化知识 使用向量数据库存储支持语义检索 存储用户画像、知识库、关键交互历史 def __init__(self, vector_storeNone): self.vector_store vector_store # 向量数据库客户端 self.items: Dict[str, MemoryItem] {} # 内存缓存 def add(self, content: str, category: str general, importance: float 0.5) - MemoryItem: 添加长期记忆 item MemoryItem( idhashlib.md5(f{content}_{datetime.now().isoformat()}.encode()).hexdigest()[:12], contentcontent, memory_typeMemoryType.LONG_TERM, timestampdatetime.now(), importanceimportance, metadata{category: category}, ) self.items[item.id] item # 存储到向量数据库 if self.vector_store: self._store_to_vector_db(item) logger.info(f长期记忆添加: [{category}] {content[:50]}...) return item def search(self, query: str, top_k: int 5) - List[MemoryItem]: 语义检索长期记忆 Args: query: 查询文本 top_k: 返回最相关的 K 条记忆 if self.vector_store: return self._search_vector_db(query, top_k) # 降级方案关键词匹配 results [] for item in self.items.values(): if any(word in item.content for word in query.split()): results.append(item) results.sort(keylambda x: x.importance, reverseTrue) return results[:top_k] def update_importance(self, item_id: str, delta: float): 更新记忆重要性被访问时提升被忽略时降低 if item_id in self.items: self.items[item_id].importance max(0.0, min(1.0, self.items[item_id].importance delta)) self.items[item_id].access_count 1 self.items[item_id].last_accessed datetime.now() def _store_to_vector_db(self, item: MemoryItem): 存储到向量数据库 # 实际实现依赖具体的向量数据库如 ChromaDB、Pinecone pass def _search_vector_db(self, query: str, top_k: int) - List[MemoryItem]: 从向量数据库检索 # 实际实现依赖具体的向量数据库 return [] class ReflectiveMemory: 反思记忆从经验中提炼的策略和教训 存储什么有效、什么无效、何时用何法等元知识 这是最抽象的记忆层也是 Agent 自我进化的关键 def __init__(self): self.strategies: List[Dict] [] # 成功策略 self.failures: List[Dict] [] # 失败教训 self.patterns: List[Dict] [] # 行为模式 def add_success(self, task: str, strategy: str, outcome: str): 记录成功经验 entry { task: task, strategy: strategy, outcome: outcome, timestamp: datetime.now().isoformat(), } self.strategies.append(entry) logger.info(f成功经验记录: {task} → {strategy}) def add_failure(self, task: str, strategy: str, reason: str): 记录失败教训 entry { task: task, strategy: strategy, reason: reason, timestamp: datetime.now().isoformat(), } self.failures.append(entry) logger.info(f失败教训记录: {task} → {reason}) def get_relevant_strategy(self, task: str) - Optional[Dict]: 获取与当前任务相关的历史策略 # 简单的关键词匹配实际应使用语义检索 for strategy in reversed(self.strategies): if any(word in strategy[task] for word in task.split()): return strategy return None def should_avoid(self, task: str, strategy: str) - bool: 检查某个策略是否应该避免 for failure in self.failures: if ( any(word in failure[task] for word in task.split()) and failure[strategy] strategy ): return True return False def reflect(self, long_term_memory: LongTermMemory) - List[Dict]: 反思从长期记忆中提炼经验 定期执行将频繁出现的模式提炼为策略 insights [] # 分析成功策略的共性 if len(self.strategies) 3: strategy_counts {} for s in self.strategies: key s[strategy] strategy_counts[key] strategy_counts.get(key, 0) 1 for strategy, count in strategy_counts.items(): if count 3: insights.append({ type: reliable_strategy, strategy: strategy, success_count: count, insight: f策略 {strategy} 已成功 {count} 次可视为可靠策略, }) # 分析失败教训的共性 if len(self.failures) 2: failure_reasons {} for f in self.failures: key f[reason] failure_reasons[key] failure_reasons.get(key, 0) 1 for reason, count in failure_reasons.items(): if count 2: insights.append({ type: recurring_failure, reason: reason, failure_count: count, insight: f失败原因 {reason} 反复出现 {count} 次需要规避, }) return insights class AgentMemorySystem: Agent 完整记忆系统四层记忆的协调器 def __init__(self, vector_storeNone): self.working WorkingMemory(max_tokens4096) self.short_term ShortTermMemory(max_items50) self.long_term LongTermMemory(vector_storevector_store) self.reflective ReflectiveMemory() def process_interaction( self, user_input: str, agent_response: str, task_type: str general ): 处理一次交互更新各层记忆 # 1. 添加到工作记忆 self.working.add(user_input, roleuser) self.working.add(agent_response, roleassistant) # 2. 检查工作记忆是否溢出 overflow self.working.get_overflow_items() if overflow: self.short_term.add_from_working_memory(overflow) # 3. 判断是否需要持久化到长期记忆 importance self._estimate_importance(user_input, agent_response) if importance 0.7: self.long_term.add( contentfQ: {user_input}\nA: {agent_response}, categorytask_type, importanceimportance, ) def retrieve_context(self, current_input: str) - str: 检索与当前输入相关的上下文 context_parts [] # 1. 工作记忆当前对话上下文 context_parts.append(f 当前对话 \n{self.working.get_context()}) # 2. 短期记忆最近的交互摘要 recent self.short_term.get_recent(5) if recent: recent_text \n.join(f- {item.content} for item in recent) context_parts.append(f 近期摘要 \n{recent_text}) # 3. 长期记忆语义检索相关知识 relevant self.long_term.search(current_input, top_k3) if relevant: relevant_text \n.join(f- {item.content} for item in relevant) context_parts.append(f 相关知识 \n{relevant_text}) # 4. 反思记忆相关策略 strategy self.reflective.get_relevant_strategy(current_input) if strategy: context_parts.append( f 历史策略 \n f任务: {strategy[task]}\n f策略: {strategy[strategy]}\n f结果: {strategy[outcome]} ) return \n\n.join(context_parts) staticmethod def _estimate_importance(user_input: str, agent_response: str) - float: 估算交互的重要性 实际项目中应使用 LLM 评估 这里用简单的启发式规则 importance 0.3 # 包含关键信息 keywords [重要, 记住, 偏好, 喜欢, 不喜欢, always, never] if any(kw in user_input.lower() for kw in keywords): importance 0.3 # 交互较长可能包含复杂信息 if len(user_input) 200: importance 0.1 # 响应较长可能包含重要结论 if len(agent_response) 500: importance 0.1 return min(1.0, importance)四、边界分析与架构权衡记忆容量与检索效率的矛盾长期记忆存储越多语义检索越慢。向量数据库的 ANN 检索虽然快毫秒级但记忆条目超过百万级后检索质量下降。解决方案按类别分区索引、定期合并相似记忆、淘汰低重要性记忆。这就像 Tensor 的猫粮柜——放太多反而找不到想吃的。记忆一致性长期记忆中可能存在矛盾的信息用户昨天说喜欢 A今天说喜欢 B。需要设计记忆更新策略——覆盖旧记忆、标记冲突、或保留时间线。建议按时间戳保留最新记忆旧记忆降权而非删除。反思记忆的可靠性反思记忆是从有限经验中提炼的可能存在偏差——3 次成功不代表策略一定可靠。建议设置最低样本量阈值至少 5 次成功才标记为可靠策略并定期重新评估策略的有效性。隐私与安全长期记忆存储用户的个人信息和交互历史存在隐私泄露风险。建议敏感信息加密存储、设置记忆过期时间如 90 天后自动删除、提供用户查看和删除记忆的接口。五、总结Agent 记忆机制是从金鱼脑到长期记忆的进化之路——工作记忆是当下短期记忆是近期长期记忆是知识反思记忆是智慧。落地建议工作记忆容量 4-8K tokens超出时摘要压缩转移短期记忆按重要性淘汰保留关键事件长期记忆用向量数据库存储支持语义检索反思记忆定期从经验中提炼策略设置最低样本量阈值。记忆不是越多越好而是越精准越好——就像 Tensor 的记忆它不记得三天前吃的什么猫粮但永远记得零食柜的位置。好的记忆系统应该记住该记住的忘掉该忘掉的。