LangGraph实战：用SQLite和InMemoryStore给你的AI助手加上短期与长期记忆（附完整代码）

LangGraph实战构建具备双记忆系统的AI助手在构建对话系统时让AI记住用户偏好和历史交互是提升体验的关键。LangGraph作为新兴的AI编排框架其独特之处在于提供了分层记忆管理机制——通过SQLite实现短期记忆持久化结合InMemoryStore构建长期知识库。本文将手把手教你实现一个能记住用户披萨偏好的智能助手。1. 理解LangGraph的双层记忆架构现代对话系统需要处理两种不同类型的记忆短期记忆保存当前对话的上下文如最近8条消息长期记忆存储跨会话的用户偏好和重要事实如用户喜欢意大利披萨LangGraph通过两种技术组件实现这一架构记忆类型存储方案典型应用场景技术实现示例短期记忆状态检查点对话历史维护SqliteSaver长期记忆可检索存储用户偏好记忆InMemoryStore这种分离设计带来三个显著优势上下文隔离避免不同对话线程间的记忆污染性能优化短期记忆高频更新长期记忆按需检索成本控制仅将重要信息存入长期存储2. 环境准备与依赖安装开始前确保Python≥3.12环境并安装必要依赖pip install langgraph langchain-openai langgraph-checkpoint-sqlite项目结构建议如下/pizza_assistant ├── memory_demo.py # 主程序 ├── memory.sqlite # SQLite数据库文件 └── requirements.txt关键依赖说明langgraph-checkpoint-sqlite提供SQLite-backed的检查点存储langchain-openai可选用于语义检索的嵌入模型3. 实现短期记忆系统短期记忆的核心是维护对话状态。我们通过TypedDict定义状态结构from typing import TypedDict, Annotated, List from operator import add class FlowState(TypedDict, totalFalse): user_id: str messages: Annotated[List[str], add] # 自动合并消息列表 retrieved: List[str] # 长期记忆检索结果3.1 配置SQLite检查点import sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver def build_app(db_pathmemory.sqlite): conn sqlite3.connect(db_path, check_same_threadFalse) saver SqliteSaver(conn) # 构建图时会传入这个saver graph StateGraph(FlowState) ... return graph.compile(checkpointersaver)这段代码创建了自动持久化的对话状态管理器具有以下特性每次状态变更自动保存到SQLite支持通过thread_id恢复历史对话内置压缩机制防止数据膨胀3.2 对话历史修剪策略为防止上下文窗口爆炸需要实现消息修剪逻辑MAX_MSG 8 # 保留最近8条消息 def trim_messages(state: FlowState): messages state.get(messages, []) if len(messages) MAX_MSG: return state # 保留最新消息并生成摘要 kept messages[-MAX_MSG:] summary f历史压缩已省略{len(messages)-MAX_MSG}条早期消息 return {**state, messages: kept, summary: summary}4. 构建长期记忆系统长期记忆的核心是支持语义检索的存储系统。我们使用InMemoryStorefrom langgraph.store.memory import InMemoryStore store InMemoryStore( index{ embed: OpenAIEmbeddings(), # 语义嵌入模型 dims: 768, # 向量维度 fields: [text] # 被索引字段 } )4.1 记忆写入机制当用户表达重要偏好时将其存入长期记忆def save_preference(user_id: str, utterance: str): key fpref:{hash(utterance)} # 生成唯一键 store.put( namespace(user, user_id), # 按用户隔离 keykey, value{text: utterance, type: preference} )4.2 语义检索实现def search_memory(user_id: str, query: str, limit3): return list(store.search( namespace(user, user_id), queryquery, limitlimit ))检索过程会自动将查询文本向量化计算与存储内容的相似度返回最相关的记忆项5. 完整工作流集成现在将两层记忆系统整合到对话流程中graph LR A[用户输入] -- B(更新短期记忆) B -- C{是否需要长期记忆} C --|是| D[语义检索] C --|否| E[生成响应] D -- E E -- F{是否重要信息} F --|是| G[存入长期记忆] F --|否| H[结束]对应代码实现from langgraph.graph import StateGraph, START, END builder StateGraph(FlowState) # 定义节点 builder.add_node(ingest, process_input) builder.add_node(retrieve, retrieve_memories) builder.add_node(generate, generate_response) # 构建流程 builder.add_edge(START, ingest) builder.add_edge(ingest, retrieve) builder.add_edge(retrieve, generate) builder.add_edge(generate, END) # 编译应用 app builder.compile( checkpointersaver, # 短期记忆 storestore # 长期记忆 )6. 实战测试启动对话测试# 第一次对话表达偏好 result1 app.invoke( {user_id: alice}, config{configurable: {thread_id: t1, say: 我爱意大利披萨}} ) # 第二次对话利用记忆 result2 app.invoke( {user_id: alice}, config{configurable: {thread_id: t2, say: 推荐晚餐吃什么}} ) print(result2[retrieved]) # 输出: [我爱意大利披萨]7. 高级优化技巧7.1 混合检索策略def hybrid_search(user_id, query): # 先尝试语义搜索 results semantic_search(user_id, query) if not results: # 回退到关键词匹配 results keyword_search(user_id, query) return results7.2 记忆加权机制def save_with_weight(user_id, key, value, weight1.0): store.put( namespace(user, user_id), keykey, value{**value, _weight: weight}, metadata{last_accessed: datetime.now()} )7.3 定期记忆整理def cleanup_memory(user_id): old_items store.scan( namespace(user, user_id), filterlambda x: x.metadata.last_accessed (NOW - timedelta(days30)) ) for item in old_items: store.delete(namespace, item.key)8. 生产环境建议持久化方案将InMemoryStore替换为Redis或PostgreSQL嵌入模型使用本地化模型如bge-small避免API依赖监控指标记忆命中率检索延迟存储增长趋势测试策略记忆一致性测试跨会话上下文保持测试压力测试模拟长时间对话# 监控装饰器示例 def monitor_memory(func): def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) latency time.time() - start monitor.gauge(memory_op_latency, latency) return result return wrapper这套实现不仅适用于披萨推荐场景经过适当调整可应用于个性化购物助手医疗咨询机器人教育领域的自适应学习系统关键是要根据具体场景调整记忆的存储策略和检索方式。比如电商场景可能需要增加产品ID关联而教育场景则需要知识点关联度计算。