向量数据库 知识图谱 记忆管理 | 短期记忆 长期记忆 工作记忆 | LangChain Memory实战 | 完整项目代码 为什么需要Agent记忆无记忆Agent的问题# 传统Agent - 每次对话都是新的开始 agent create_agent(llm, tools) # 第一轮对话 response1 agent.run(我叫张三今年30岁) # 输出: 你好张三很高兴认识你 # 第二轮对话 response2 agent.run(我刚才说了什么) # 输出: 抱歉我不记得我们之前的对话 ❌问题❌ 无法记住用户信息❌ 多轮对话上下文丢失❌ 无法积累历史经验❌ 每次都要重新学习有记忆Agent的优势# 带记忆的Agent agent create_memory_agent(llm, tools, memory_system) # 第一轮对话 response1 agent.run(我叫张三今年30岁喜欢编程) # 输出: 你好张三很高兴认识一位30岁的程序员 # 第二轮对话第二天 response2 agent.run(你还记得我吗) # 输出: 当然记得你是张三30岁喜欢编程 ✅ # 第三轮对话 response3 agent.run(推荐一些适合我的技术书籍) # 输出: 基于你对编程的兴趣我推荐... ✅优势✅ 记住用户偏好和信息✅ 跨会话保持上下文✅ 积累历史经验和知识✅ 提供个性化服务️ 记忆系统架构三层记忆模型┌─────────────────────────────────────┐ │ 长期记忆 (Long-term Memory) │ │ - 向量数据库存储 │ │ - 用户画像 │ │ - 历史对话摘要 │ │ - 知识库 │ └──────────────┬──────────────────────┘ │ 检索/更新 ┌──────────────▼──────────────────────┐ │ 工作记忆 (Working Memory) │ │ - 当前任务上下文 │ │ - 临时变量 │ │ - 中间结果 │ └──────────────┬──────────────────────┘ │ 读取/写入 ┌──────────────▼──────────────────────┐ │ 短期记忆 (Short-term Memory) │ │ - 最近N轮对话 │ │ - 对话缓冲区 │ │ - 快速访问 │ └─────────────────────────────────────┘记忆类型详解记忆类型存储位置容量时效用途短期记忆内存有限10-20轮会话内保持对话连贯性工作记忆内存/Redis中等任务期间处理复杂任务长期记忆向量数据库无限永久用户画像、知识库️ 核心技术实现方案1LangChain Memory基础用法安装依赖pip install langchain langchain-openai chromadb redis1. ConversationBufferMemory对话缓冲记忆from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain from langchain_openai import ChatOpenAI # 创建LLM llm ChatOpenAI(modelgpt-4, temperature0.7) # 创建记忆 memory ConversationBufferMemory() # 创建对话链 conversation ConversationChain( llmllm, memorymemory, verboseTrue ) # 使用 conversation.predict(input我叫张三今年30岁) # AI: 你好张三很高兴认识你。 conversation.predict(input我喜欢编程和阅读) # AI: 太好了编程和阅读都是很好的爱好。 conversation.predict(input你还记得我叫什么吗) # AI: 当然记得你叫张三。特点✅ 简单直接✅ 保存完整对话历史❌ 随着对话增长token消耗大❌ 不适合长对话2. ConversationSummaryMemory对话摘要记忆from langchain.memory import ConversationSummaryMemory # 创建摘要记忆 memory ConversationSummaryMemory.from_llm( llmllm, max_token_limit1000 # 超过1000 token时生成摘要 ) conversation ConversationChain( llmllm, memorymemory, verboseTrue ) # 使用 conversation.predict(input今天天气不错) conversation.predict(input我打算去公园散步) # ... 更多对话 # 查看摘要 print(memory.buffer) # 输出: 用户提到今天天气不错计划去公园散步...特点✅ 节省token✅ 适合长对话❌ 可能丢失细节❌ 摘要质量依赖LLM3. ConversationBufferWindowMemory滑动窗口记忆from langchain.memory import ConversationBufferWindowMemory # 只保留最近5轮对话 memory ConversationBufferWindowMemory(k5) conversation ConversationChain( llmllm, memorymemory, verboseTrue ) # 第6轮对话时第1轮会被遗忘特点✅ 控制记忆大小✅ 平衡成本和效果❌ 会遗忘早期对话方案2向量数据库长期记忆核心架构from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.schema import Document import uuid from datetime import datetime class VectorMemory: 基于向量数据库的长期记忆系统 def __init__(self, collection_nameuser_memories): # 初始化嵌入模型 self.embeddings OpenAIEmbeddings() # 初始化向量数据库 self.vectorstore Chroma( collection_namecollection_name, embedding_functionself.embeddings, persist_directory./memory_db ) # 元数据存储 self.metadata_store {} def add_memory(self, user_id: str, content: str, memory_type: str fact) - str: 添加记忆 Args: user_id: 用户ID content: 记忆内容 memory_type: 记忆类型fact/opinion/preference/event Returns: 记忆ID memory_id str(uuid.uuid4()) # 创建文档 doc Document( page_contentcontent, metadata{ id: memory_id, user_id: user_id, type: memory_type, timestamp: datetime.now().isoformat() } ) # 添加到向量数据库 self.vectorstore.add_documents([doc]) # 保存元数据 self.metadata_store[memory_id] { user_id: user_id, type: memory_type, created_at: datetime.now().isoformat() } print(f✅ 记忆已保存: {content[:50]}...) return memory_id def search_memories(self, user_id: str, query: str, top_k: int 5) - list: 搜索相关记忆 Args: user_id: 用户ID query: 查询内容 top_k: 返回数量 Returns: 相关记忆列表 # 向量搜索 results self.vectorstore.similarity_search( query, ktop_k * 2 # 多取一些然后过滤 ) # 过滤出指定用户的记忆 user_memories [ { content: doc.page_content, metadata: doc.metadata } for doc in results if doc.metadata.get(user_id) user_id ] # 按相关性排序并返回top_k return user_memories[:top_k] def get_user_profile(self, user_id: str) - dict: 获取用户画像 Args: user_id: 用户ID Returns: 用户画像字典 # 搜索所有用户记忆 all_memories self.search_memories(user_id, , top_k100) # 分类整理 profile { facts: [], preferences: [], opinions: [], events: [] } for memory in all_memories: mem_type memory[metadata].get(type, fact) if mem_type in profile: profile[mem_type].append(memory[content]) return profile def forget_memory(self, memory_id: str) - bool: 删除记忆 Args: memory_id: 记忆ID Returns: 是否成功删除 try: self.vectorstore.delete(ids[memory_id]) del self.metadata_store[memory_id] print(f✅ 记忆已删除: {memory_id}) return True except Exception as e: print(f❌ 删除失败: {e}) return False # 使用示例 def example_vector_memory(): 向量记忆使用示例 # 创建记忆系统 memory VectorMemory(collection_namedemo_memories) # 添加记忆 user_id user_001 memory.add_memory(user_id, 我叫张三今年30岁, fact) memory.add_memory(user_id, 我喜欢编程和阅读, preference) memory.add_memory(user_id, 我认为Python是最好的编程语言, opinion) memory.add_memory(user_id, 昨天我去参加了技术大会, event) # 搜索记忆 print(\n 搜索编程相关记忆:) results memory.search_memories(user_id, 编程, top_k3) for i, mem in enumerate(results, 1): print(f{i}. {mem[content]}) # 获取用户画像 print(\n 用户画像:) profile memory.get_user_profile(user_id) for category, items in profile.items(): if items: print(f\n{category.upper()}:) for item in items: print(f - {item}) if __name__ __main__: example_vector_memory()方案3知识图谱记忆核心思想将记忆组织成图结构节点是实体边是关系。from typing import Dict, List, Set import json class KnowledgeGraphMemory: 基于知识图谱的记忆系统 def __init__(self): # 节点实体 - 属性 self.nodes: Dict[str, Dict[str, str]] {} # 边(源实体, 关系, 目标实体) self.edges: List[tuple] [] # 索引实体 - 相关边 self.entity_index: Dict[str, Set[int]] {} def add_entity(self, entity: str, attributes: Dict[str, str]): 添加实体节点 Args: entity: 实体名称 attributes: 属性字典 self.nodes[entity] attributes self.entity_index[entity] set() print(f✅ 添加实体: {entity}) def add_relation(self, source: str, relation: str, target: str): 添加关系边 Args: source: 源实体 relation: 关系类型 target: 目标实体 edge_id len(self.edges) self.edges.append((source, relation, target)) # 更新索引 if source not in self.entity_index: self.entity_index[source] set() if target not in self.entity_index: self.entity_index[target] set() self.entity_index[source].add(edge_id) self.entity_index[target].add(edge_id) print(f✅ 添加关系: {source} --[{relation}]-- {target}) def query_entity(self, entity: str) - Dict: 查询实体信息 Args: entity: 实体名称 Returns: 实体信息和相关关系 if entity not in self.nodes: return {error: f实体 {entity} 不存在} # 获取实体属性 attributes self.nodes[entity] # 获取相关关系 related_edges [] if entity in self.entity_index: for edge_id in self.entity_index[entity]: source, relation, target self.edges[edge_id] related_edges.append({ source: source, relation: relation, target: target }) return { entity: entity, attributes: attributes, relations: related_edges } def find_path(self, start: str, end: str, max_depth: int 3) - List: 查找两个实体之间的路径 Args: start: 起始实体 end: 目标实体 max_depth: 最大搜索深度 Returns: 路径列表 # BFS搜索 from collections import deque queue deque([(start, [start])]) visited {start} while queue: current, path queue.popleft() if len(path) max_depth: continue if current end: return path # 探索邻居节点 if current in self.entity_index: for edge_id in self.entity_index[current]: source, relation, target self.edges[edge_id] # 确定下一个节点 next_node target if source current else source if next_node not in visited: visited.add(next_node) queue.append((next_node, path [next_node])) return [] def export_graph(self) - Dict: 导出知识图谱 Returns: 图谱数据 return { nodes: self.nodes, edges: [ {source: s, relation: r, target: t} for s, r, t in self.edges ] } # 使用示例 def example_knowledge_graph(): 知识图谱记忆示例 # 创建知识图谱 kg KnowledgeGraphMemory() # 添加实体 kg.add_entity(张三, {age: 30, occupation: 程序员}) kg.add_entity(Python, {type: 编程语言, paradigm: 多范式}) kg.add_entity(机器学习, {field: 人工智能, difficulty: 中等}) kg.add_entity(李四, {age: 28, occupation: 数据科学家}) # 添加关系 kg.add_relation(张三, 喜欢, Python) kg.add_relation(张三, 学习, 机器学习) kg.add_relation(张三, 同事, 李四) kg.add_relation(李四, 擅长, Python) kg.add_relation(李四, 研究, 机器学习) # 查询实体 print(\n 查询张三的信息:) info kg.query_entity(张三) print(json.dumps(info, indent2, ensure_asciiFalse)) # 查找路径 print(\n 查找张三到机器学习的路径:) path kg.find_path(张三, 机器学习) print(f路径: { - .join(path)}) # 导出图谱 print(\n 知识图谱:) graph kg.export_graph() print(f节点数: {len(graph[nodes])}) print(f边数: {len(graph[edges])}) if __name__ __main__: example_knowledge_graph()方案4混合记忆系统生产级完整实现from langchain.memory import ConversationBufferWindowMemory from typing import List, Dict, Any import json class HybridMemorySystem: 混合记忆系统 结合短期记忆、工作记忆和长期记忆 def __init__(self, user_id: str, short_term_k: int 10): self.user_id user_id # 短期记忆滑动窗口 self.short_term_memory ConversationBufferWindowMemory(kshort_term_k) # 长期记忆向量数据库 self.long_term_memory VectorMemory( collection_namefuser_{user_id}_memories ) # 工作记忆临时存储 self.working_memory: Dict[str, Any] {} # 记忆统计 self.stats { short_term_count: 0, long_term_count: 0, total_interactions: 0 } def add_interaction(self, human_input: str, ai_response: str, save_to_long_term: bool True): 添加交互记录 Args: human_input: 用户输入 ai_response: AI回复 save_to_long_term: 是否保存到长期记忆 # 添加到短期记忆 self.short_term_memory.save_context( {input: human_input}, {output: ai_response} ) self.stats[short_term_count] 1 # 提取重要信息保存到长期记忆 if save_to_long_term: important_facts self._extract_important_facts(human_input) for fact in important_facts: self.long_term_memory.add_memory( self.user_id, fact[content], fact[type] ) self.stats[long_term_count] 1 self.stats[total_interactions] 1 def get_context(self, query: str) - str: 获取完整的上下文 Args: query: 当前查询 Returns: 格式化的上下文字符串 context_parts [] # 1. 短期记忆最近对话 short_term self.short_term_memory.load_memory_variables({}) if short_term.get(history): context_parts.append(【最近对话】) context_parts.append(short_term[history]) # 2. 长期记忆相关信息 long_term self.long_term_memory.search_memories( self.user_id, query, top_k3 ) if long_term: context_parts.append(\n【相关记忆】) for i, mem in enumerate(long_term, 1): context_parts.append(f{i}. {mem[content]}) # 3. 工作记忆临时信息 if self.working_memory: context_parts.append(\n【当前任务】) for key, value in self.working_memory.items(): context_parts.append(f- {key}: {value}) return \n.join(context_parts) def set_working_memory(self, key: str, value: Any): 设置工作记忆 self.working_memory[key] value def clear_working_memory(self): 清空工作记忆 self.working_memory.clear() def _extract_important_facts(self, text: str) - List[Dict]: 从文本中提取重要事实简化版 实际应用中可以使用LLM来提取 facts [] # 简单的规则提取 if 我叫 in text or 名字是 in text: facts.append({content: text, type: fact}) if 我喜欢 in text or 我爱 in text: facts.append({content: text, type: preference}) if 我认为 in text or 我觉得 in text: facts.append({content: text, type: opinion}) return facts def get_stats(self) - Dict: 获取记忆统计 return self.stats.copy() def export_memories(self) - Dict: 导出所有记忆 return { user_id: self.user_id, stats: self.get_stats(), working_memory: self.working_memory, long_term_memories: self.long_term_memory.get_user_profile( self.user_id ) } # 使用示例 def example_hybrid_memory(): 混合记忆系统示例 # 创建混合记忆系统 memory HybridMemorySystem(user_iduser_001) # 模拟对话 interactions [ (我叫张三今年30岁, 你好张三很高兴认识你。), (我喜欢编程和阅读, 编程和阅读都是很好的爱好), (我正在学习机器学习, 机器学习是个很有前景的领域), (你觉得Python怎么样, Python是一门优秀的编程语言。), ] # 添加交互 for human_input, ai_response in interactions: memory.add_interaction(human_input, ai_response) # 获取上下文 print( 查询编程相关上下文:) context memory.get_context(编程) print(context) # 设置工作记忆 memory.set_working_memory(current_task, 推荐编程书籍) # 再次获取上下文 print(\n\n 带工作记忆的上下文:) context memory.get_context(推荐书籍) print(context) # 查看统计 print(\n\n 记忆统计:) stats memory.get_stats() print(json.dumps(stats, indent2, ensure_asciiFalse)) # 导出记忆 print(\n\n 导出记忆:) exported memory.export_memories() print(json.dumps(exported, indent2, ensure_asciiFalse)) if __name__ __main__: example_hybrid_memory() 实战案例个性化助手系统架构用户输入 │ ▼ ┌──────────────┐ │ 记忆检索 │ ← 从长期记忆中检索相关信息 └──────┬───────┘ │ ▼ ┌──────────────┐ │ 上下文构建 │ ← 整合短期长期工作记忆 └──────┬───────┘ │ ▼ ┌──────────────┐ │ LLM推理 │ ← 生成个性化回复 └──────┬───────┘ │ ▼ ┌──────────────┐ │ 记忆更新 │ ← 提取新信息存入长期记忆 └──────┬───────┘ │ ▼ 输出回复完整实现from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate class PersonalizedAssistant: 个性化AI助手 def __init__(self, user_id: str): self.user_id user_id self.llm ChatOpenAI(modelgpt-4, temperature0.7) self.memory HybridMemorySystem(user_id) # 系统提示词 self.system_prompt ChatPromptTemplate.from_messages([ (system, 你是一个个性化的AI助手。 基于以下用户信息提供个性化服务 {context} 请 1. 利用用户的历史信息提供个性化建议 2. 保持对话的连贯性 3. 如果用户提到新的重要信息请记住 回答要友好、专业、个性化。), (human, {input}) ]) def chat(self, user_input: str) - str: 与用户对话 Args: user_input: 用户输入 Returns: AI回复 # 获取上下文 context self.memory.get_context(user_input) # 构建提示词 prompt self.system_prompt.format( contextcontext if context else 新用户暂无历史信息, inputuser_input ) # 调用LLM response self.llm.invoke(prompt).content # 保存到记忆 self.memory.add_interaction(user_input, response) return response def get_user_profile(self) - Dict: 获取用户画像 return self.memory.long_term_memory.get_user_profile(self.user_id) def clear_memory(self): 清空记忆 self.memory.clear_working_memory() print(✅ 工作记忆已清空) # 使用示例 def example_personalized_assistant(): 个性化助手示例 print( 个性化AI助手演示\n) print(*60) # 创建助手 assistant PersonalizedAssistant(user_iduser_001) # 模拟对话 conversations [ 你好我叫张三, 我今年30岁是一名软件工程师, 我喜欢Python和机器学习, 你能推荐一些学习资源吗, 我还对自然语言处理感兴趣, 你还记得我叫什么吗, 基于我的兴趣我应该先学什么 ] for i, user_input in enumerate(conversations, 1): print(f\n[对话 {i}]) print(f 用户: {user_input}) response assistant.chat(user_input) print(f AI: {response}) if i len(conversations): print(- * 60) # 显示用户画像 print(\n *60) print( 用户画像:) print(*60) profile assistant.get_user_profile() print(json.dumps(profile, indent2, ensure_asciiFalse)) if __name__ __main__: example_personalized_assistant() 性能优化技巧1. 记忆压缩def compress_memory(memories: List[str], max_length: int 500) - str: 压缩记忆保留关键信息 # 方法1提取关键词 keywords extract_keywords(memories) # 方法2生成摘要 summary generate_summary(memories) # 方法3重要性评分 scored_memories score_by_importance(memories) top_memories scored_memories[:10] return \n.join(top_memories)2. 记忆分层存储class TieredMemoryStorage: 分层记忆存储 def __init__(self): # L1: Redis热数据快速访问 self.hot_storage RedisMemory() # L2: 向量数据库温数据语义搜索 self.warm_storage VectorMemory() # L3: 文件系统/数据库冷数据归档 self.cold_storage FileStorage() def store(self, memory: Memory): 智能存储 if memory.importance 0.8: self.hot_storage.store(memory) self.warm_storage.store(memory) elif memory.importance 0.5: self.warm_storage.store(memory) else: self.cold_storage.store(memory)3. 记忆遗忘策略def forget_unimportant_memories(memory_system, threshold: float 0.3): 遗忘不重要的记忆 all_memories memory_system.get_all_memories() for memory in all_memories: # 计算重要性分数 importance calculate_importance( memory, factors[recency, frequency, user_feedback] ) if importance threshold: memory_system.forget(memory.id) print(f遗忘记忆: {memory.content[:50]}) 最佳实践总结1. 记忆系统设计原则原则说明实施方法分层存储不同热度数据用不同存储Redis 向量DB 文件智能检索只检索相关记忆向量相似度 元数据过滤定期清理避免记忆膨胀遗忘策略 压缩算法隐私保护保护用户数据加密存储 权限控制可解释性让用户知道记住了什么记忆查看 删除功能2. 记忆提取技巧# 使用LLM提取重要信息 def extract_memories_with_llm(conversation: str) - List[Dict]: 使用LLM提取记忆 prompt f 从以下对话中提取需要长期记住的重要信息 对话{conversation} 提取格式 - 事实类用户个人信息 - 偏好类用户喜好 - 观点类用户看法 - 事件类重要事件 只提取真正重要的信息不要提取琐碎内容。 response llm.invoke(prompt) return parse_extraction(response)3. 记忆评估指标指标说明目标值召回率相关记忆的检索比例 80%准确率检索到的记忆确实相关 90%响应时间记忆检索耗时 100ms存储效率单位信息的存储空间越小越好用户满意度用户对记忆功能的评分 4.0/5.0 常见问题解答Q1: 如何平衡记忆的成本和效果A:采用分层策略短期记忆滑动窗口10-20轮长期记忆只存重要信息用LLM筛选定期清理遗忘低价值记忆Q2: 如何处理隐私问题A:# 1. 用户同意 ask_permission(是否保存对话历史用于个性化服务) # 2. 数据加密 encrypted_memory encrypt(memory_data) # 3. 提供删除功能 user.can_delete_own_memories() # 4. 匿名化处理 anonymize_sensitive_info(memory)Q3: 记忆系统会影响响应速度吗A:优化策略异步检索记忆缓存热门记忆限制检索数量top_k3-5使用快速存储RedisQ4: 如何评估记忆系统的好坏A:# 自动化测试 test_cases [ (用户说名字, AI能记住名字), (隔天对话, AI还记得之前信息), (矛盾信息, AI能更新记忆) ] for input_text, expected in test_cases: result assistant.chat(input_text) assert_check(result, expected) 相关资源LangChain Memory文档Chroma向量数据库配套项目代码 总结Agent记忆系统是构建智能助手的核心技术通过✅三层记忆架构- 短期、工作、长期记忆协同✅向量数据库- 实现语义搜索和长期存储✅知识图谱- 结构化存储复杂关系✅混合系统- 结合多种技术的优势掌握了记忆系统的设计和实施你就能打造出真正记住用户的智能Agent下一步动手实践从简单的ConversationBufferMemory开始逐步构建复杂的记忆系统。专栏AI Agent实战专栏日期2026年5月11日系列AI Agent高级进阶系列第2篇