LangGraph 长期记忆管理实战：从基础配置到语义搜索

1. LangGraph长期记忆管理入门想象一下你正在和一个健忘的朋友聊天。每次见面他都要重新问你的名字、喜好这种体验肯定让人抓狂。LangGraph的长期记忆功能就是为了解决这个问题而生的它能让AI系统记住用户的信息和偏好就像一个有心的老朋友一样。长期记忆的核心是Store接口它就像一个智能文件柜。每个用户都有自己的文件夹命名空间里面可以存放各种文件键值对。比如你可以为每个用户创建一个独立命名空间存储他们的语言偏好、历史对话摘要等。与短期记忆不同长期记忆是跨会话持久化的。短期记忆只保存当前对话的上下文而长期记忆则像一个永不重置的数据库。这种设计让AI应用能提供真正个性化的体验无论用户何时回来都能记住他们的习惯。2. 基础配置实战2.1 初始化内存存储我们先从最简单的内存存储开始。虽然生产环境建议使用数据库存储但内存存储非常适合快速原型开发。from langgraph.store.memory import InMemoryStore # 创建内存存储实例 store InMemoryStore() # 定义用户命名空间类似文件夹 user_namespace user_profiles这个InMemoryStore虽然简单但已经具备了长期记忆的核心功能。它使用Python字典在内存中保存数据重启应用后数据会丢失所以只适合开发和测试。2.2 存储用户信息让我们实现一个保存用户信息的函数。这个函数会被AI代理调用将用户数据存入长期记忆。from typing import TypedDict from langgraph.config import get_store class UserProfile(TypedDict): name: str language: str preferences: dict def save_user_profile(profile: UserProfile, config: RunnableConfig) - str: 保存用户档案到长期记忆 store get_store() # 获取存储实例 user_id config[configurable].get(user_id) if not user_id: return 错误缺少用户ID # 将用户信息存入指定命名空间 store.put((user_namespace,), user_id, profile) return 用户信息保存成功这个函数有几个关键点使用TypedDict定义数据结构确保类型安全从配置中获取用户ID作为存储键使用put方法将数据存入指定命名空间2.3 读取用户信息存储之后我们需要能读取这些信息。下面是一个读取函数示例def get_user_profile(config: RunnableConfig) - dict: 从长期记忆中读取用户档案 store get_store() user_id config[configurable].get(user_id) if not user_id: return {error: 缺少用户ID} # 从存储中获取数据 item store.get((user_namespace,), user_id) return item.value if item else {}这个读取函数会返回完整的用户档案字典。如果用户不存在则返回空字典。在实际应用中你可能需要添加更多错误处理逻辑。3. 高级数据操作3.1 更新和删除数据长期记忆不只是简单的存储还需要支持更新和删除操作。def update_user_profile(updates: dict, config: RunnableConfig) - str: 更新用户档案 current get_user_profile(config) if not current: return 用户不存在 # 合并更新 current.update(updates) save_user_profile(current, config) return 更新成功 def delete_user_profile(config: RunnableConfig) - str: 删除用户档案 store get_store() user_id config[configurable].get(user_id) if not user_id: return 错误缺少用户ID # 从存储中删除 store.delete((user_namespace,), user_id) return 删除成功更新操作采用了先读取-修改-保存的模式确保数据一致性。删除操作则直接移除指定键的数据。3.2 批量操作当需要处理大量数据时批量操作能显著提高效率。def batch_save_profiles(profiles: dict, config: RunnableConfig) - str: 批量保存用户档案 store get_store() user_id config[configurable].get(user_id) if not user_id: return 错误缺少用户ID # 批量写入 with store.batch() as batch: for key, profile in profiles.items(): batch.put((user_namespace,), f{user_id}_{key}, profile) return f成功保存{len(profiles)}条记录批量操作通过上下文管理器实现所有操作会在with块结束时一起提交减少IO开销。4. 语义搜索功能4.1 配置语义搜索LangGraph最强大的功能之一是语义搜索。与普通键值查找不同它能根据内容含义找到相关信息。首先需要配置嵌入模型from langchain_community.embeddings import HuggingFaceEmbeddings # 初始化嵌入模型 embeddings HuggingFaceEmbeddings(model_nameall-MiniLM-L6-v2) # 创建支持语义搜索的存储 semantic_store InMemoryStore( index{ embed: embeddings.embed_query, # 嵌入函数 dims: 384 # 向量维度 } )这个配置让存储能够为每个值生成向量表示后续可以基于向量相似度进行搜索。4.2 存储可搜索数据要使数据可搜索需要存储包含语义信息的结构化数据user_preferences { preferences: { food: 喜欢辣味食物特别是川菜, hobbies: 热爱徒步旅行和摄影, reading: 偏好科幻小说最爱《三体》系列 } } semantic_store.put( (user_123, preferences), # 命名空间 profile, # 键 user_preferences # 值 )注意我们使用了更复杂的命名空间结构第一级是用户ID第二级是数据类型。4.3 执行语义搜索现在可以执行真正的语义搜索了# 搜索与户外活动相关的偏好 results semantic_store.search( (user_123, preferences), query户外活动, limit1 ) print(results[0].value) # 输出{preferences: {hobbies: 热爱徒步旅行和摄影}}即使查询词户外活动没有直接出现在数据中系统也能找到相关的徒步旅行信息。这是因为嵌入模型理解了这些概念之间的语义关联。4.4 混合搜索策略结合精确过滤和语义搜索可以获得更好效果# 同时使用过滤器和语义查询 results semantic_store.search( (user_123, preferences), filter{type: hobbies}, # 精确过滤 query周末活动建议, # 语义查询 limit2 )这种混合策略先用过滤器缩小范围再在结果中做语义搜索兼顾了准确性和相关性。5. 生产环境最佳实践5.1 持久化存储配置内存存储不适合生产环境我们需要更可靠的解决方案。以PostgreSQL为例from langgraph.store.postgres import PostgresStore # 配置PostgreSQL存储 production_store PostgresStore( connection_stringpostgresql://user:passwordlocalhost:5432/langgraph, index{ embed: embeddings.embed_query, dims: 384 } )PostgresStore支持所有内存存储的功能同时提供持久化和更好的性能。它还支持连接池和事务适合高并发场景。5.2 缓存策略为平衡性能与实时性可以实现多级缓存from langgraph.store.cached import CachedStore # 创建带缓存的存储 cached_store CachedStore( primaryproduction_store, # 主存储 cacheInMemoryStore(), # 缓存层 ttl300 # 缓存5分钟 )这种设计将热点数据缓存在内存中减少数据库查询同时通过TTL保证数据不会太旧。5.3 监控与维护生产环境需要监控存储健康状况def check_store_health(store) - dict: 检查存储健康状况 try: # 测试读写 test_key (healthcheck, str(time.time())) store.put(test_key, test, {value: test}) item store.get(test_key, test) store.delete(test_key, test) return { status: healthy, latency: time.time() - start_time, error: None } except Exception as e: return { status: unhealthy, error: str(e) }定期运行这样的健康检查可以在问题影响用户前及时发现。6. 性能优化技巧6.1 批量操作优化当处理大量数据时正确的批量操作方式可以提升数倍性能# 低效方式 for item in large_dataset: store.put(namespace, item.id, item.data) # 高效方式 with store.batch() as batch: for item in large_dataset: batch.put(namespace, item.id, item.data)批处理减少了网络往返和事务开销特别适合初始化大量数据。6.2 查询优化对于读取密集型应用可以优化查询模式# 同时获取多个键的数据减少IO def get_multiple_profiles(user_ids: list) - dict: return { uid: store.get((user_namespace,), uid) for uid in user_ids }另外合理设计命名空间结构也能提高查询效率。例如将频繁一起查询的数据放在相同命名空间。6.3 索引设计对于自定义存储良好的索引设计至关重要# 创建带索引的存储 indexed_store PostgresStore( connection_stringpostgresql://user:passwordlocalhost:5432/langgraph, index{ embed: embeddings.embed_query, dims: 384, extra_indexes: [ (created_at, DESC), # 时间索引 (user_id, BTREE) # 用户ID索引 ] } )适当的索引可以加速特定查询模式但要注意索引也会增加写入开销。7. 实际应用案例7.1 个性化推荐系统长期记忆非常适合构建个性化推荐def get_recommendations(user_id: str, context: dict) - list: 获取个性化推荐 # 读取用户偏好 profile store.get((users, user_id), profile) # 语义搜索相关项目 items store.search( (catalog,), filter{category: profile.preferred_category}, querycontext.get(current_interest, ), limit5 ) return [item.value for item in items]这个例子结合了精确过滤用户偏好的类别和语义搜索当前兴趣提供精准推荐。7.2 多轮对话管理在对话系统中长期记忆可以维持对话一致性def handle_dialog(turn: dict) - dict: 处理对话轮次 # 读取对话历史摘要 summary store.get( (dialogs, turn[user_id]), summary ) # 更新上下文 context build_context(turn, summary) # 生成响应 response generate_response(context) # 保存重要信息到长期记忆 if should_remember(turn): store.put( (users, turn[user_id]), fmemory_{time.time()}, extract_memory(turn) ) return response这种模式让AI能记住对话中的重要信息并在后续对话中引用。7.3 用户行为分析长期记忆可以存储和分析用户行为模式def analyze_behavior(user_id: str) - dict: 分析用户行为模式 # 获取近期活动 activities store.list( (users, user_id, activities), limit100 ) # 使用LLM分析模式 analysis llm.analyze( f分析以下用户活动模式\n{activities} ) # 保存分析结果 store.put( (users, user_id), behavior_profile, analysis ) return analysis这种分析可以帮助AI更好地预测用户需求提供更智能的服务。