Agent系列（二）-记忆系统的设计

目录一、为什么需要记忆系统二、记忆系统的分层2.1 第一层工作记忆Working Memory2.2 第二层短期记忆Short-term Memory2.3 第三层长期记忆Long-term Memory2.4 第四层永久记忆Persistent Memory / Knowledge三、记忆分层对比及协同机制3.1 四层记忆分层对比3.2 四层记忆的协同机制一、为什么需要记忆系统持续性期望实现跨会话保持上下文个性化记住用户偏好和历史自我进化积累经验提升能力效率减少获取重复信息二、记忆系统的分层Agent的记忆系统分为四层——工作记忆即时推理、短期记忆当前会话、长期记忆跨会话积累、永久记忆结构化知识。各层职责明确协同工作共同支撑Agent的持续对话和个性化服务能力。2.1 第一层工作记忆Working Memory定义Agent当前推理过程中的即时工作区生命周期当前推理步骤毫秒级~秒级存储位置进程内存容量极小几个关键变量存储内容1当前任务的中间状态2思考链Thought Chain3工具调用的输入输出4临时计算结果5决策过程中的关键判断class WorkingMemory: 工作记忆Agent推理时的即时工作台 def __init__(self): self.variables {} # 变量存储 self.thought_chain [] # 思考链 self.step_context {} # 当前步骤上下文 def set_var(self, key, value): 设置变量 self.variables[key] value def get_var(self, key, defaultNone): 获取变量 return self.variables.get(key, default) def think(self, thought): 记录思考步骤 self.thought_chain.append({ thought: thought, step: len(self.thought_chain) 1 }) def get_thought_chain_text(self): 获取思考链文本 return \n.join([ f[思考{i}] {t[thought]} for i, t in enumerate(self.thought_chain, 1) ]) def set_step_context(self, step_name, context): 设置当前步骤的上下文 self.step_context[step_name] context def get_step_context(self, step_name): 获取某步骤的上下文 return self.step_context.get(step_name) def clear(self): 清空工作记忆新任务开始时 self.variables {} self.thought_chain [] self.step_context {}2.2 第二层短期记忆Short-term Memory定义当前会话周期内的对话上下文生命周期当前会话分钟级~小时级存储位置内存 / Redis会话存储容量有限通常10-20轮对话存储内容1当前会话的所有对话轮次2当前任务的中间结果3未完成的后续事项4当前会话中建立的临时结论过期处理1会话结束自动清除2重要内容摘要后转入长期记忆3超容量时删除最旧的轮次class ShortTermMemory: 短期记忆基于轮次的上下文管理 def __init__(self, max_turns10): self.max_turns max_turns self.turns [] # 对话轮次列表 def add_turn(self, user_input, agent_output, metadataNone): 添加一轮对话 turn { user: user_input, agent: agent_output, metadata: metadata or {}, timestamp: datetime.now().isoformat() } self.turns.append(turn) # 超容量时删除最旧的轮次 if len(self.turns) self.max_turns: self.turns.pop(0) def get_full_context(self): 获取完整上下文 return \n.join([ f用户: {t[user]}\n助手: {t[agent]} for t in self.turns ]) def get_recent_turns(self, n3): 获取最近N轮对话 return self.turns[-n:] def get_turn_count(self): 获取对话轮数 return len(self.turns) def summarize_to_long_term(self, llm): 将会话摘要后存入长期记忆 if len(self.turns) 0: return conversation self.get_full_context() summary llm.invoke(f 请对以下对话进行简短摘要100字以内 {conversation} 摘要 ) return summary def clear(self): 清空短期记忆 self.turns []2.3 第三层长期记忆Long-term Memory定义跨会话积累的重要信息和经验生命周期会话之间持续天级~月级存储位置向量数据库Pinecone/Milvus/Weaviate容量大可存储多年数据存储内容1重要会话的摘要2用户偏好和习惯3达成的重要结论和决策4解决方案和经验积累5问题和应对的处理方式检索方式1向量相似度检索语义检索2关键词检索BM253时间衰减越近的记忆权重越高class LongTermMemory: 长期记忆基于向量数据库的跨会话记忆 def __init__(self, vector_store, user_id): self.vector_store vector_store self.user_id user_id def store(self, content, memory_typegeneral, importance1.0): 存储记忆到长期记忆 Args: content: 记忆内容文本 memory_type: 记忆类型preference/decision/experience等 importance: 重要程度0-1用于后续清理时参考 vector self.embed(content) self.vector_store.insert({ id: f{self.user_id}_{uuid.uuid4()}, values: vector, text: content, metadata: { user_id: self.user_id, type: memory_type, importance: importance, created_at: datetime.now().isoformat() } }) def retrieve(self, query, top_k5, memory_typesNone): 从长期记忆中检索相关内容 Args: query: 查询文本 top_k: 返回数量 memory_types: 只检索特定类型的记忆 query_vector self.embed(query) # 构建过滤条件 filter_cond {user_id: self.user_id} if memory_types: filter_cond[type] {$in: memory_types} results self.vector_store.query( vectorquery_vector, top_ktop_k, filterfilter_cond, include_metadataTrue ) # 应用时间衰减 return self.apply_time_decay(results) def apply_time_decay(self, results): 应用时间衰减越新的记忆权重越高 now datetime.now() scored_results [] for r in results: created_at datetime.fromisoformat(r[metadata][created_at]) age_days (now - created_at).days # 指数衰减半衰期30天 decay math.exp(-age_days / 30) final_score r[score] * decay scored_results.append({ content: r[text], original_score: r[score], decay_score: decay, final_score: final_score, metadata: r[metadata] }) # 按最终分数排序 scored_results.sort(keylambda x: x[final_score], reverseTrue) return scored_results def get_user_preferences(self): 获取用户偏好专门检索偏好类记忆 return self.retrieve(用户偏好, memory_types[preference], top_k10) def get_relevant_experiences(self, problem): 获取相关经验用于解决新问题 return self.retrieve(problem, memory_types[experience], top_k5)2.4 第四层永久记忆Persistent Memory / Knowledge定义持久化的结构化知识包括用户画像、系统配置、专业知识生命周期永久直到显式删除存储位置结构化数据库PostgreSQL/MongoDB容量理论无限需有效管理存储内容1用户画像用户基本信息、偏好设置、会员等级、标签、账户状态2领域知识产品知识库、业务流程规范、常见问题解答、专业领域知识3Agent经验成功的解决方案模式、失败的教训优化记录class PersistentMemory: 永久记忆结构化的持久化存储 def __init__(self, db): self.db db # 数据库连接 # 用户画像 def get_user_profile(self, user_id): 获取用户画像 user self.db.users.find_one({user_id: user_id}) if user: return { basic_info: user.get(basic_info, {}), preferences: user.get(preferences, {}), tags: user.get(tags, []), membership_level: user.get(membership_level, 普通), created_at: user.get(created_at), updated_at: user.get(updated_at) } return self._create_empty_profile(user_id) def update_user_profile(self, user_id, updates): 更新用户画像 self.db.users.update_one( {user_id: user_id}, { $set: { **updates, updated_at: datetime.now().isoformat() } }, upsertTrue ) def add_user_tag(self, user_id, tag): 添加用户标签 self.db.users.update_one( {user_id: user_id}, { $addToSet: {tags: tag}, $set: {updated_at: datetime.now().isoformat()} } ) # 知识库 def save_knowledge(self, category, question, answer, metadataNone): 保存知识条目 self.db.knowledge.insert_one({ category: category, question: question, answer: answer, metadata: metadata or {}, created_at: datetime.now().isoformat(), usage_count: 0, last_used: None }) def search_knowledge(self, query, categoryNone, top_k5): 搜索知识库 filter_cond {} if category: filter_cond[category] category # 简单实现基于关键词匹配 results self.db.knowledge.find({ $or: [ {question: {$regex: query}}, {answer: {$regex: query}} ], **filter_cond }).limit(top_k) return list(results) # Agent经验 def save_agent_experience(self, agent_id, experience): 保存Agent经验 self.db.agent_experiences.insert_one({ agent_id: agent_id, experience: experience, created_at: datetime.now().isoformat() }) def get_agent_experiences(self, agent_id, limit20): 获取Agent经验 experiences self.db.agent_experiences.find( {agent_id: agent_id} ).sort(created_at, -1).limit(limit) return list(experiences)三、记忆分层对比及协同机制3.1 四层记忆分层对比维度工作记忆短期记忆长期记忆永久记忆生命周期单次推理当前会话跨会话永久存储位置进程内存内存/Redis向量数据库结构化DB容量极小小10-20轮大无限访问频率最高高中低检索方式直接访问顺序遍历向量检索条件查询过期策略随任务清除会话结束/摘要时间衰减手动删除类比工作台短期记忆长期记忆知识库数据特征即时状态对话记录经验积累结构化画像3.2 四层记忆的协同机制四层记忆协同目标让Agent在每次对话时都能获取完整的上下文协同原则热数据优先当前任务状态 会话上下文 历史经验按需下沉重要信息逐层沉淀不重要信息及时清理容量平衡各层有容量限制通过压缩/清理维持平衡统一输出最终整合为LLM可直接使用的prompt四层记忆写入流程每一轮推理后立即写入Working Memory轮次结束时写入对话记录立即写入Short-term Memory会话结束/重要交互时写入Long-term Memory识别到明确偏好/决策时写入Persistent Memory四层记忆读取流程每一轮对话依次获取工作记忆、短期记忆、中期记忆、长期记忆最终构建统一prompt给到LLM最后我们以一个「订机票场景记忆协同」的案例来说明一下四层记忆的协同机制┌─────────────────────────────────────────────────────────────┐ │ 订机票场景记忆协同 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 【第一轮对话】 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 用户: 我想订明天去北京的机票 │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ │ │ 写入流程: │ │ • Working: task订机票, destination北京, date明天 │ │ • Short: [用户: 我想订明天去北京的机票] │ │ • Long: 不写入非重要交互 │ │ • Persistent: 不写入非偏好/决策 │ │ ↓ │ │ 读取流程: │ │ • Working: 提供当前任务状态 │ │ • Short: 空第一次对话 │ │ • Long: 检索相关历史可能找到之前的订票经验 │ │ • Persistent: 获取用户画像 │ │ ↓ │ │ Context整合: │ │ 用户VIP会员之前偏好国航不坐红眼航班常用出发地... │ │ ↓ │ │ LLM回复 写入Short │ │ │ └─────────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────────┐ │ │ │ 【第二轮对话】 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 用户: 对帮我看看国航的 │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ │ │ 写入流程: │ │ • Working: airline国航更新状态 │ │ • Short: 添加第二轮对话 │ │ • Long: 不写入 │ │ • Persistent: 不写入 │ │ ↓ │ │ 读取流程: │ │ • Working: 提供当前任务状态订机票、国航 │ │ • Short: 提供第一轮对话上下文 │ │ • Long: 检索国航相关的历史经验 │ │ • Persistent: 获取用户画像中的偏好 │ │ ↓ │ │ Context整合: │ │ 当前订机票国航明天北京 │ │ 会话用户想订明天去北京的机票 │ │ 偏好用户是VIP偏好国航不坐红眼航班 │ │ ↓ │ │ LLM回复 写入Short │ │ │ └─────────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────────┐ │ │ │ 【第三轮对话用户确认偏好】 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 用户: 我喜欢靠窗的位置记得存一下 │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ │ │ 写入流程: │ │ • Working: seat_preference靠窗 │ │ • Short: 添加第三轮对话 │ │ • Long: 存储偏好经验喜欢靠窗位置 │ │ • Persistent: ★ 写入用户画像重要偏好 │ │ │ 更新 persistent storage: │ │ │ preferences.seat 靠窗 │ │ ↓ │ │ 读取流程: │ │ • Working: 当前状态 │ │ • Short: 完整会话 │ │ • Long: 检索相关偏好历史 │ │ • Persistent: ★ 获取完整用户画像 │ │ │ 包括刚才保存的靠窗偏好 │ │ ↓ │ │ Context整合: │ │ 用户VIP偏好国航不坐红眼航班喜欢靠窗位置 │ │ 常用出发地...完整画像 │ │ ↓ │ │ LLM回复 写入各层记忆 │ │ │ └─────────────────────────────────────────────────────────────┘