无状态的 AI 助手每次对话都从零开始这是当前应用体验差的核心原因之一。本文系统性地拆解 Agent 记忆系统的工程实现从短期工作记忆到长期知识库构建有真实记忆的 AI Agent。记忆系统的四个层次人类记忆是分层的有即时工作记忆当前任务有情景记忆具体事件有语义记忆通用知识有程序记忆技能习惯。AI Agent 的记忆系统同样需要分层设计。四层记忆架构┌─────────────────────────────────────┐│ 工作记忆In-Context Memory │ 当前对话 任务上下文├─────────────────────────────────────┤│ 情景记忆Episodic Memory │ 历史交互摘要├─────────────────────────────────────┤│ 语义记忆Semantic Memory │ 用户偏好 领域知识├─────────────────────────────────────┤│ 程序记忆Procedural Memory │ 工具使用经验 错误教训└─────────────────────────────────────┘大多数 Agent 应用只实现了工作记忆即对话历史缺少后三层才是用户体验差距的根本原因。## 工作记忆管理工作记忆就是 LLM 的 Context Window。管理的核心挑战是如何在有限的上下文窗口内保留最重要的信息。### 滑动窗口策略最简单的方案只保留最近 N 轮对话pythonclass SlidingWindowMemory: def __init__(self, max_turns: int 20): self.max_turns max_turns self.messages [] def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) # 超出窗口时删除最旧的用户-助手对保留system prompt while len(self.messages) self.max_turns * 2: # 找到第一个非system消息对并删除 for i, msg in enumerate(self.messages): if msg[role] user: self.messages.pop(i) # 删除用户消息 self.messages.pop(i) # 删除对应的助手消息 break def get_context(self) - list: return self.messages.copy()### 摘要压缩策略当对话历史超出限制时对旧内容进行摘要压缩pythonfrom openai import OpenAIclient OpenAI()class SummarizingMemory: def __init__(self, max_tokens: int 4000, compression_threshold: int 3000): self.messages [] self.summary self.max_tokens max_tokens self.compression_threshold compression_threshold def estimate_tokens(self, messages: list) - int: # 粗略估算每个字符约0.5个token total_chars sum(len(m[content]) for m in messages) return int(total_chars * 0.5) def compress_if_needed(self): if self.estimate_tokens(self.messages) self.compression_threshold: # 压缩较旧的一半消息 old_messages self.messages[:len(self.messages)//2] old_text \n.join([f{m[role]}: {m[content]} for m in old_messages]) response client.chat.completions.create( modelgpt-4o-mini, messages[{ role: user, content: f请将以下对话历史压缩为简洁摘要200字以内保留关键信息\n\n{old_text} }] ) self.summary response.choices[0].message.content self.messages self.messages[len(self.messages)//2:] def get_context(self) - list: context [] if self.summary: context.append({ role: system, content: f对话历史摘要{self.summary} }) context.extend(self.messages) return context## 情景记忆记住发生过什么情景记忆存储的是具体事件的记录。对于 Agent 来说这意味着记住用户做过什么、说过什么、达成过什么结论。### 情景记忆的数据结构pythonfrom dataclasses import dataclassfrom datetime import datetimefrom typing import Optional, Listdataclassclass Episode: episode_id: str timestamp: datetime summary: str # 本次交互的简要描述 key_facts: List[str] # 提取的关键事实 user_intent: str # 用户意图标签 outcome: str # 结果成功/失败/进行中 tags: List[str] # 主题标签便于检索 def to_searchable_text(self) - str: return f{self.summary} { .join(self.key_facts)} { .join(self.tags)}### 情景记忆的写入每次对话结束后自动提取并存储情景pythonasync def extract_and_store_episode(conversation: list, user_id: str): 从对话中提取情景记忆 conversation_text \n.join([ f{m[role]}: {m[content]} for m in conversation ]) extraction_prompt f 分析以下对话提取情景记忆。以JSON格式返回 {{ summary: 一句话描述这次交互, key_facts: [事实1, 事实2], user_intent: 用户意图分类, outcome: 成功/失败/进行中, tags: [主题标签] }} 对话内容 {conversation_text} response await llm.complete(extraction_prompt) episode_data json.loads(response) episode Episode( episode_idgenerate_id(), timestampdatetime.now(), **episode_data ) # 存入向量数据库用于语义检索 await vector_store.upsert( idepisode.episode_id, vectorawait embed(episode.to_searchable_text()), metadataepisode.__dict__ )## 语义记忆用户画像与知识积累语义记忆存储的是结构化的用户偏好和领域知识它是 AI 个性化体验的核心。### 用户画像系统pythonclass UserProfile: def __init__(self, user_id: str): self.user_id user_id self.preferences {} # 偏好设置 self.expertise {} # 专业领域和水平 self.communication_style {} # 沟通风格偏好 self.context {} # 工作/生活背景 def update_preference(self, key: str, value, confidence: float 1.0): 更新用户偏好带置信度权重 if key not in self.preferences: self.preferences[key] {value: value, confidence: confidence, count: 1} else: # 增量更新置信度 current self.preferences[key] new_count current[count] 1 if current[value] value: new_confidence min(1.0, current[confidence] 0.1) else: new_confidence max(0.1, current[confidence] - 0.2) if new_confidence 0.3: current[value] value # 偏好已改变 self.preferences[key] { value: current[value], confidence: new_confidence, count: new_count } def to_system_prompt(self) - str: 生成个性化 System Prompt 片段 lines [关于用户的了解] for key, pref in self.preferences.items(): if pref[confidence] 0.7: lines.append(f- {key}: {pref[value]}) if self.communication_style: style self.communication_style lines.append(f- 偏好的沟通风格: {style.get(tone, 专业)}, {style.get(detail_level, 适中)}) return \n.join(lines)### 知识库更新当 Agent 从用户处获得新知识时应将其存入语义记忆pythonasync def update_knowledge_base(user_id: str, conversation: list): 从对话中提取并更新知识库 extraction_prompt 从以下对话中提取值得记住的用户信息 1. 用户的明确偏好如我喜欢/不喜欢... 2. 用户的背景信息工作、项目、技术栈 3. 用户提到的重要约束或要求 4. 用户的沟通风格偏好 只提取明确表达的信息不要推断。以JSON数组返回每项包含{type, key, value}。 knowledge_items await extract_from_conversation(conversation, extraction_prompt) profile await load_user_profile(user_id) for item in knowledge_items: profile.update_preference(item[key], item[value]) await save_user_profile(user_id, profile)## 程序记忆学习工具使用经验程序记忆存储 Agent 在工具调用和任务执行中积累的经验——哪些方法有效哪些方法失败过。pythondataclassclass ToolExperience: tool_name: str use_case: str # 使用场景描述 parameters_pattern: dict # 成功的参数模式 success_rate: float common_errors: list # 常见错误及解决方案 best_practices: list # 最佳实践class ProceduralMemory: def __init__(self): self.tool_experiences {} def record_success(self, tool_name: str, params: dict, context: str): 记录成功的工具调用 key f{tool_name}:{context} if key not in self.tool_experiences: self.tool_experiences[key] ToolExperience( tool_nametool_name, use_casecontext, parameters_patternparams, success_rate1.0, common_errors[], best_practices[] ) else: exp self.tool_experiences[key] exp.success_rate exp.success_rate * 0.9 1.0 * 0.1 # EMA def record_failure(self, tool_name: str, params: dict, error: str, context: str): 记录失败的工具调用及错误信息 key f{tool_name}:{context} if key in self.tool_experiences: exp self.tool_experiences[key] exp.success_rate exp.success_rate * 0.9 0.0 * 0.1 if error not in exp.common_errors: exp.common_errors.append(error) def get_advice(self, tool_name: str, context: str) - str: 获取工具使用建议 key f{tool_name}:{context} if key not in self.tool_experiences: return exp self.tool_experiences[key] advice [] if exp.success_rate 0.8: advice.append(f参考历史成功模式: {exp.parameters_pattern}) if exp.common_errors: advice.append(f注意避免: {, .join(exp.common_errors[:3])}) return \n.join(advice)## 记忆检索找到最相关的记忆存储是基础检索才是关键。Agent 在响应时需要快速找到最相关的历史记忆。pythonclass MemoryRetriever: def __init__(self, vector_store, user_profile_store): self.vector_store vector_store self.user_profile user_profile_store async def retrieve_relevant_memories( self, current_query: str, user_id: str, top_k: int 5 ) - dict: 检索与当前查询最相关的记忆 # 1. 情景记忆检索向量相似度 query_embedding await embed(current_query) episodes await self.vector_store.search( vectorquery_embedding, filter{user_id: user_id}, limittop_k ) # 2. 用户画像全量加载较小 profile await self.user_profile.load(user_id) # 3. 组合为上下文 context_parts [] if profile: context_parts.append(f用户背景\n{profile.to_system_prompt()}) if episodes: episode_texts [] for ep in episodes: episode_texts.append( f[{ep.timestamp.strftime(%Y-%m-%d)}] {ep.summary} ) context_parts.append( f相关历史记录\n \n.join(episode_texts) ) return { context: \n\n.join(context_parts), episodes: episodes, profile: profile }## 记忆遗忘机制无限累积的记忆会降低检索质量。合理的遗忘机制必不可少pythonclass MemoryDecay: 基于艾宾浩斯遗忘曲线的记忆衰减 staticmethod def calculate_retention(days_since_created: float, initial_importance: float 1.0) - float: 计算记忆保留率 import math # 艾宾浩斯公式R e^(-t/S)S为稳定性 stability initial_importance * 10 # 重要性越高遗忘越慢 retention math.exp(-days_since_created / stability) return retention async def cleanup_old_memories(self, user_id: str, threshold: float 0.1): 清理保留率低于阈值的记忆 all_episodes await self.load_all_episodes(user_id) now datetime.now() to_delete [] for episode in all_episodes: days_old (now - episode.timestamp).days retention self.calculate_retention(days_old, episode.importance) if retention threshold: to_delete.append(episode.episode_id) if to_delete: await self.vector_store.delete(to_delete) print(f清理了 {len(to_delete)} 条过期记忆)## 生产部署考量存储选择- 向量数据库Qdrant/Pinecone/Weaviate存储情景记忆的嵌入向量- PostgreSQL/MongoDB存储结构化用户画像- Redis工作记忆缓存TTL 控制自动过期隐私与安全- 记忆内容应加密存储- 为用户提供查看和删除自己记忆的接口- 明确告知用户哪些信息被记忆扩展性- 用户量大时记忆检索需要按用户 ID 分片- 考虑记忆冷热分层近期热数据 vs 历史冷数据## 总结Agent 记忆系统是从工具到伙伴的关键进化。四层记忆架构工作记忆 情景记忆 语义记忆 程序记忆缺一不可但工程落地可以分阶段推进1.Phase 1实现工作记忆管理摘要压缩2.Phase 2构建用户画像语义记忆核心3.Phase 3引入情景记忆历史对话检索4.Phase 4加入程序记忆工具经验积累每个 Phase 都能带来可感知的用户体验提升是值得持续投入的工程方向。
Agent记忆系统工程：让AI真正记住重要的事
无状态的 AI 助手每次对话都从零开始这是当前应用体验差的核心原因之一。本文系统性地拆解 Agent 记忆系统的工程实现从短期工作记忆到长期知识库构建有真实记忆的 AI Agent。记忆系统的四个层次人类记忆是分层的有即时工作记忆当前任务有情景记忆具体事件有语义记忆通用知识有程序记忆技能习惯。AI Agent 的记忆系统同样需要分层设计。四层记忆架构┌─────────────────────────────────────┐│ 工作记忆In-Context Memory │ 当前对话 任务上下文├─────────────────────────────────────┤│ 情景记忆Episodic Memory │ 历史交互摘要├─────────────────────────────────────┤│ 语义记忆Semantic Memory │ 用户偏好 领域知识├─────────────────────────────────────┤│ 程序记忆Procedural Memory │ 工具使用经验 错误教训└─────────────────────────────────────┘大多数 Agent 应用只实现了工作记忆即对话历史缺少后三层才是用户体验差距的根本原因。## 工作记忆管理工作记忆就是 LLM 的 Context Window。管理的核心挑战是如何在有限的上下文窗口内保留最重要的信息。### 滑动窗口策略最简单的方案只保留最近 N 轮对话pythonclass SlidingWindowMemory: def __init__(self, max_turns: int 20): self.max_turns max_turns self.messages [] def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) # 超出窗口时删除最旧的用户-助手对保留system prompt while len(self.messages) self.max_turns * 2: # 找到第一个非system消息对并删除 for i, msg in enumerate(self.messages): if msg[role] user: self.messages.pop(i) # 删除用户消息 self.messages.pop(i) # 删除对应的助手消息 break def get_context(self) - list: return self.messages.copy()### 摘要压缩策略当对话历史超出限制时对旧内容进行摘要压缩pythonfrom openai import OpenAIclient OpenAI()class SummarizingMemory: def __init__(self, max_tokens: int 4000, compression_threshold: int 3000): self.messages [] self.summary self.max_tokens max_tokens self.compression_threshold compression_threshold def estimate_tokens(self, messages: list) - int: # 粗略估算每个字符约0.5个token total_chars sum(len(m[content]) for m in messages) return int(total_chars * 0.5) def compress_if_needed(self): if self.estimate_tokens(self.messages) self.compression_threshold: # 压缩较旧的一半消息 old_messages self.messages[:len(self.messages)//2] old_text \n.join([f{m[role]}: {m[content]} for m in old_messages]) response client.chat.completions.create( modelgpt-4o-mini, messages[{ role: user, content: f请将以下对话历史压缩为简洁摘要200字以内保留关键信息\n\n{old_text} }] ) self.summary response.choices[0].message.content self.messages self.messages[len(self.messages)//2:] def get_context(self) - list: context [] if self.summary: context.append({ role: system, content: f对话历史摘要{self.summary} }) context.extend(self.messages) return context## 情景记忆记住发生过什么情景记忆存储的是具体事件的记录。对于 Agent 来说这意味着记住用户做过什么、说过什么、达成过什么结论。### 情景记忆的数据结构pythonfrom dataclasses import dataclassfrom datetime import datetimefrom typing import Optional, Listdataclassclass Episode: episode_id: str timestamp: datetime summary: str # 本次交互的简要描述 key_facts: List[str] # 提取的关键事实 user_intent: str # 用户意图标签 outcome: str # 结果成功/失败/进行中 tags: List[str] # 主题标签便于检索 def to_searchable_text(self) - str: return f{self.summary} { .join(self.key_facts)} { .join(self.tags)}### 情景记忆的写入每次对话结束后自动提取并存储情景pythonasync def extract_and_store_episode(conversation: list, user_id: str): 从对话中提取情景记忆 conversation_text \n.join([ f{m[role]}: {m[content]} for m in conversation ]) extraction_prompt f 分析以下对话提取情景记忆。以JSON格式返回 {{ summary: 一句话描述这次交互, key_facts: [事实1, 事实2], user_intent: 用户意图分类, outcome: 成功/失败/进行中, tags: [主题标签] }} 对话内容 {conversation_text} response await llm.complete(extraction_prompt) episode_data json.loads(response) episode Episode( episode_idgenerate_id(), timestampdatetime.now(), **episode_data ) # 存入向量数据库用于语义检索 await vector_store.upsert( idepisode.episode_id, vectorawait embed(episode.to_searchable_text()), metadataepisode.__dict__ )## 语义记忆用户画像与知识积累语义记忆存储的是结构化的用户偏好和领域知识它是 AI 个性化体验的核心。### 用户画像系统pythonclass UserProfile: def __init__(self, user_id: str): self.user_id user_id self.preferences {} # 偏好设置 self.expertise {} # 专业领域和水平 self.communication_style {} # 沟通风格偏好 self.context {} # 工作/生活背景 def update_preference(self, key: str, value, confidence: float 1.0): 更新用户偏好带置信度权重 if key not in self.preferences: self.preferences[key] {value: value, confidence: confidence, count: 1} else: # 增量更新置信度 current self.preferences[key] new_count current[count] 1 if current[value] value: new_confidence min(1.0, current[confidence] 0.1) else: new_confidence max(0.1, current[confidence] - 0.2) if new_confidence 0.3: current[value] value # 偏好已改变 self.preferences[key] { value: current[value], confidence: new_confidence, count: new_count } def to_system_prompt(self) - str: 生成个性化 System Prompt 片段 lines [关于用户的了解] for key, pref in self.preferences.items(): if pref[confidence] 0.7: lines.append(f- {key}: {pref[value]}) if self.communication_style: style self.communication_style lines.append(f- 偏好的沟通风格: {style.get(tone, 专业)}, {style.get(detail_level, 适中)}) return \n.join(lines)### 知识库更新当 Agent 从用户处获得新知识时应将其存入语义记忆pythonasync def update_knowledge_base(user_id: str, conversation: list): 从对话中提取并更新知识库 extraction_prompt 从以下对话中提取值得记住的用户信息 1. 用户的明确偏好如我喜欢/不喜欢... 2. 用户的背景信息工作、项目、技术栈 3. 用户提到的重要约束或要求 4. 用户的沟通风格偏好 只提取明确表达的信息不要推断。以JSON数组返回每项包含{type, key, value}。 knowledge_items await extract_from_conversation(conversation, extraction_prompt) profile await load_user_profile(user_id) for item in knowledge_items: profile.update_preference(item[key], item[value]) await save_user_profile(user_id, profile)## 程序记忆学习工具使用经验程序记忆存储 Agent 在工具调用和任务执行中积累的经验——哪些方法有效哪些方法失败过。pythondataclassclass ToolExperience: tool_name: str use_case: str # 使用场景描述 parameters_pattern: dict # 成功的参数模式 success_rate: float common_errors: list # 常见错误及解决方案 best_practices: list # 最佳实践class ProceduralMemory: def __init__(self): self.tool_experiences {} def record_success(self, tool_name: str, params: dict, context: str): 记录成功的工具调用 key f{tool_name}:{context} if key not in self.tool_experiences: self.tool_experiences[key] ToolExperience( tool_nametool_name, use_casecontext, parameters_patternparams, success_rate1.0, common_errors[], best_practices[] ) else: exp self.tool_experiences[key] exp.success_rate exp.success_rate * 0.9 1.0 * 0.1 # EMA def record_failure(self, tool_name: str, params: dict, error: str, context: str): 记录失败的工具调用及错误信息 key f{tool_name}:{context} if key in self.tool_experiences: exp self.tool_experiences[key] exp.success_rate exp.success_rate * 0.9 0.0 * 0.1 if error not in exp.common_errors: exp.common_errors.append(error) def get_advice(self, tool_name: str, context: str) - str: 获取工具使用建议 key f{tool_name}:{context} if key not in self.tool_experiences: return exp self.tool_experiences[key] advice [] if exp.success_rate 0.8: advice.append(f参考历史成功模式: {exp.parameters_pattern}) if exp.common_errors: advice.append(f注意避免: {, .join(exp.common_errors[:3])}) return \n.join(advice)## 记忆检索找到最相关的记忆存储是基础检索才是关键。Agent 在响应时需要快速找到最相关的历史记忆。pythonclass MemoryRetriever: def __init__(self, vector_store, user_profile_store): self.vector_store vector_store self.user_profile user_profile_store async def retrieve_relevant_memories( self, current_query: str, user_id: str, top_k: int 5 ) - dict: 检索与当前查询最相关的记忆 # 1. 情景记忆检索向量相似度 query_embedding await embed(current_query) episodes await self.vector_store.search( vectorquery_embedding, filter{user_id: user_id}, limittop_k ) # 2. 用户画像全量加载较小 profile await self.user_profile.load(user_id) # 3. 组合为上下文 context_parts [] if profile: context_parts.append(f用户背景\n{profile.to_system_prompt()}) if episodes: episode_texts [] for ep in episodes: episode_texts.append( f[{ep.timestamp.strftime(%Y-%m-%d)}] {ep.summary} ) context_parts.append( f相关历史记录\n \n.join(episode_texts) ) return { context: \n\n.join(context_parts), episodes: episodes, profile: profile }## 记忆遗忘机制无限累积的记忆会降低检索质量。合理的遗忘机制必不可少pythonclass MemoryDecay: 基于艾宾浩斯遗忘曲线的记忆衰减 staticmethod def calculate_retention(days_since_created: float, initial_importance: float 1.0) - float: 计算记忆保留率 import math # 艾宾浩斯公式R e^(-t/S)S为稳定性 stability initial_importance * 10 # 重要性越高遗忘越慢 retention math.exp(-days_since_created / stability) return retention async def cleanup_old_memories(self, user_id: str, threshold: float 0.1): 清理保留率低于阈值的记忆 all_episodes await self.load_all_episodes(user_id) now datetime.now() to_delete [] for episode in all_episodes: days_old (now - episode.timestamp).days retention self.calculate_retention(days_old, episode.importance) if retention threshold: to_delete.append(episode.episode_id) if to_delete: await self.vector_store.delete(to_delete) print(f清理了 {len(to_delete)} 条过期记忆)## 生产部署考量存储选择- 向量数据库Qdrant/Pinecone/Weaviate存储情景记忆的嵌入向量- PostgreSQL/MongoDB存储结构化用户画像- Redis工作记忆缓存TTL 控制自动过期隐私与安全- 记忆内容应加密存储- 为用户提供查看和删除自己记忆的接口- 明确告知用户哪些信息被记忆扩展性- 用户量大时记忆检索需要按用户 ID 分片- 考虑记忆冷热分层近期热数据 vs 历史冷数据## 总结Agent 记忆系统是从工具到伙伴的关键进化。四层记忆架构工作记忆 情景记忆 语义记忆 程序记忆缺一不可但工程落地可以分阶段推进1.Phase 1实现工作记忆管理摘要压缩2.Phase 2构建用户画像语义记忆核心3.Phase 3引入情景记忆历史对话检索4.Phase 4加入程序记忆工具经验积累每个 Phase 都能带来可感知的用户体验提升是值得持续投入的工程方向。
