LangGraph 持久化详解

LangGraph 持久化详解一、持久化概述持久化Persistence是 LangGraph 的核心能力之一它允许图状态在执行过程中被保存和恢复。LangGraph 提供了两种互补的持久化机制机制作用存储范围记忆类型Checkpointer保存图状态快照单个线程Thread短期记忆Store保存应用数据跨线程长期记忆核心价值人机交互支持中断-恢复工作流对话记忆跨多轮对话保持上下文时间旅行回溯到历史状态重新执行容错恢复故障后从检查点恢复执行二、Checkpointer检查点记录器2.1 核心概念Thread线程线程是检查点的组织单元每个线程有一个唯一的thread_id。线程包含了一系列运行的累积状态。config{configurable:{thread_id:user-123}}graph.invoke({input:hello},config)Checkpoint检查点检查点是在每个超级步骤Super-step保存的图状态快照由StateSnapshot对象表示。超级步骤说明对于顺序图START → A → B → END会创建4个检查点每个节点执行后都会保存一个检查点并行节点在同一超级步中执行2.2 内置实现InMemorySaver内存检查点fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointerInMemorySaver()graphbuilder.compile(checkpointercheckpointer)# 使用config{configurable:{thread_id:1}}resultgraph.invoke({foo:},config)注意InMemorySaver将检查点存储在 RAM 中进程重启后数据丢失仅适用于开发和测试。SqliteSaverSQLite 检查点fromlanggraph.checkpoint.sqliteimportSqliteSaver# 同步版本checkpointerSqliteSaver.from_conn_string(checkpoints.sqlite)graphbuilder.compile(checkpointercheckpointer)# 异步版本fromlanggraph.checkpoint.sqlite.aioimportAsyncSqliteSaverasyncwithAsyncSqliteSaver.from_conn_string(checkpoints.sqlite)ascheckpointer:graphbuilder.compile(checkpointercheckpointer)resultawaitgraph.ainvoke({foo:},config)PostgresSaverPostgreSQL 检查点fromlanggraph.checkpoint.postgresimportPostgresSaverfrompsycopg_poolimportConnectionPool DB_URIpostgresql://postgres:postgreslocalhost:5432/postgres# 使用连接池withConnectionPool(conninfoDB_URI,max_size20)aspool:checkpointerPostgresSaver(pool)checkpointer.setup()# 首次使用需要初始化数据库graphbuilder.compile(checkpointercheckpointer)resultgraph.invoke({foo:},config)# 使用连接字符串withPostgresSaver.from_conn_string(DB_URI)ascheckpointer:graphbuilder.compile(checkpointercheckpointer)2.3 检查点器接口所有检查点器都实现BaseCheckpointSaver接口方法说明put存储检查点及其配置和元数据put_writes存储中间写入待处理写入get_tuple获取检查点元组用于graph.get_state()list列出符合配置的检查点用于graph.get_state_history()2.4 StateSnapshot 字段graph.get_state()返回的StateSnapshot包含以下字段字段类型说明valuesdict此检查点处的通道状态值nexttuple下一个要执行的节点名称空元组表示图已结束configdict包含thread_id、checkpoint_ns、checkpoint_idmetadatadict执行元数据source、writes、stepcreated_atstr检查点创建时间ISO 8601parent_configdict | None上一个检查点的配置taskstuple此步骤要执行的任务列表三、状态管理3.1 获取状态# 获取最新状态config{configurable:{thread_id:1}}stategraph.get_state(config)# 获取特定检查点config{configurable:{thread_id:1,checkpoint_id:1ef663ba-28fe-6528-8002-5a559208592c}}stategraph.get_state(config)3.2 获取状态历史config{configurable:{thread_id:1}}historygraph.get_state_history(config)# 遍历历史forcheckpointinhistory:print(fStep:{checkpoint.metadata[step]})print(fValues:{checkpoint.values})print(fNext:{checkpoint.next})3.3 更新状态使用update_state编辑图状态会创建新的检查点config{configurable:{thread_id:1}}# 更新状态graph.update_state(config,values{foo:updated_value},as_nodenode_a# 可选指定更新来源节点)关键点更新不会修改原始检查点而是创建新检查点如果字段有 reducer 函数值会通过 reducer 处理as_node参数影响接下来执行哪个节点四、时间旅行Time Travel时间旅行允许从历史检查点重新执行图是 LangGraph 的强大调试能力。4.1 重放Replay从特定检查点重新执行后续步骤# 从 checkpoint_id 重放config{configurable:{thread_id:1,checkpoint_id:1ef663ba-28fe-6528-8002-5a559208592c}}resultgraph.invoke(None,config)# 传入 None 触发重放重放行为检查点之前的步骤被跳过已保存检查点之后的步骤重新执行LLM 调用、API 请求、中断会重新触发4.2 分叉Fork通过update_state创建状态分叉# 获取历史historylist(graph.get_state_history(config))oldest_checkpointhistory[-1]# 从旧检查点分叉fork_configoldest_checkpoint.config graph.update_state(fork_config,values{foo:forked_value},as_nodenode_a)# 继续执行会创建新的执行路径resultgraph.invoke(None,fork_config)五、人机交互Human-in-the-Loop持久化是实现人机交互工作流的基础。5.1 中断点配置在编译图时配置中断点graphbuilder.compile(checkpointercheckpointer,interrupt_before[human_review_node],# 进入节点前中断interrupt_after[ai_generate_node]# 离开节点后中断)5.2 中断-恢复流程# 1. 执行图会在中断点暂停config{configurable:{thread_id:1}}resultgraph.invoke({input:user query},config)# 2. 检查状态stategraph.get_state(config)print(state.next)# (human_review_node,)# 3. 人工审核并更新状态graph.update_state(config,values{approval:True,feedback:looks good},as_nodehuman_review_node)# 4. 恢复执行resultgraph.invoke(None,config)5.3 待处理写入Pending Writes当超级步中的节点执行失败时LangGraph 会保存成功节点的写入# 恢复时已成功节点的写入会被保留# 无需重新执行成功的节点resultgraph.invoke(None,config)六、Store存储Store 用于跨线程持久化应用数据实现长期记忆。6.1 基本使用fromlanggraph.store.memoryimportInMemoryStore storeInMemoryStore()# 编译图时同时使用 checkpointer 和 storegraphbuilder.compile(checkpointercheckpointer,storestore)6.2 在节点中访问 Storefromlanggraph.store.baseimportBaseStoredefmy_node(state,config,*,store:BaseStore):user_idconfig[configurable][user_id]namespace(user_preferences,user_id)# 读取存储的数据memoriesstore.search(namespace,queryfavorite color)# 写入数据store.put(namespace,preference_1,{data:blue is favorite color})return{response:fHello{user_id}!}6.3 跨线程记忆示例fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.store.memoryimportInMemoryStore checkpointerInMemorySaver()storeInMemoryStore()# 节点函数defchatbot(state,config,*,store:BaseStore):user_idconfig[configurable][user_id]namespace(memories,user_id)# 获取用户记忆memoriesstore.search(namespace)memory_text\n.join([m.value[data]forminmemories])# 检查是否需要记住新信息last_messagestate[messages][-1].contentifrememberinlast_message.lower():store.put(namespace,str(uuid.uuid4()),{data:User prefers dark mode})# 使用记忆生成回复responsemodel.invoke([{role:system,content:fUser memories:\n{memory_text}},*state[messages]])return{messages:[response]}# 编译图graphbuilder.compile(checkpointercheckpointer,storestore)# 线程1用户告诉机器人记住偏好config1{configurable:{thread_id:thread-1,user_id:user-123}}graph.invoke({messages:[(user,Remember that I like blue)]},config1)# 线程2仍然可以访问用户记忆config2{configurable:{thread_id:thread-2,user_id:user-123}}graph.invoke({messages:[(user,Whats my favorite color?)]},config2)七、序列化检查点器需要对状态值进行序列化和反序列化。7.1 JsonPlusSerializerLangGraph 默认使用JsonPlusSerializer支持多种类型LangChain 和 LangGraph 基本类型日期时间和枚举Pydantic 模型自定义类fromlanggraph.checkpoint.serde.jsonplusimportJsonPlusSerializer serdeJsonPlusSerializer()checkpointerInMemorySaver(serdeserde)7.2 EncryptedSerializer加密序列化支持加密持久化状态fromlanggraph.checkpoint.serde.encryptedimportEncryptedSerializer# 从环境变量读取密钥encrypted_serdeEncryptedSerializer.from_pycryptodome_aes()# 或直接指定密钥encrypted_serdeEncryptedSerializer.from_pycryptodome_aes(keyyour-secret-key)checkpointerInMemorySaver(serdeencrypted_serde)八、Checkpointer 对比特性InMemorySaverSqliteSaverPostgresSaver存储位置RAM本地文件数据库持久性❌ 进程重启丢失✅ 文件持久✅ 数据库持久异步支持✅✅✅并发性能低中高适用场景开发测试本地开发生产环境安装命令内置pip install langgraph-checkpoint-sqlitepip install langgraph-checkpoint-postgres九、最佳实践9.1 选择合适的检查点器# 开发环境checkpointerInMemorySaver()# 本地开发/小型项目checkpointerSqliteSaver.from_conn_string(checkpoints.sqlite)# 生产环境checkpointerPostgresSaver.from_conn_string(postgresql://...)checkpointer.setup()9.2 管理检查点大小长期对话会导致检查点无限增长定期清理# 获取状态历史historygraph.get_state_history(config)# 删除旧检查点保留最近N个forcheckpointinhistory[10:]:# 保留前10个checkpointer.delete_thread(config[configurable][thread_id])9.3 使用 Reducer 处理状态更新fromtypingimportAnnotatedfromoperatorimportaddclassState(TypedDict):messages:Annotated[list,add]# 消息列表使用 add reducercontext:str# 普通字段直接覆盖# update_state 时messages 会追加而不是覆盖graph.update_state(config,values{messages:[new_message]})9.4 子图持久化为子图配置独立的 checkpointer# 子图使用独立 checkpointersubgraph_builderStateGraph(SubState)subgraphsubgraph_builder.compile(checkpointerTrue)# 或共享父图的 checkpointersubgraphsubgraph_builder.compile(checkpointerparent_checkpointer)十、实战示例多轮对话机器人fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.store.memoryimportInMemoryStorefromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessagefromlangchain_openaiimportChatOpenAIclassChatState(TypedDict):messages:Annotated[list[BaseMessage],append]# 模型modelChatOpenAI(modelgpt-4)# 节点defchatbot(state:ChatState):responsemodel.invoke(state[messages])return{messages:[response]}# 构建图builderStateGraph(ChatState)builder.add_node(chatbot,chatbot)builder.add_edge(START,chatbot)builder.add_edge(chatbot,END)# 持久化配置checkpointerInMemorySaver()storeInMemoryStore()graphbuilder.compile(checkpointercheckpointer,storestore)# 多轮对话config{configurable:{thread_id:user-001}}# 第一轮result1graph.invoke({messages:[HumanMessage(contentHi, Im Alice)]},config)print(result1[messages][-1].content)# 第二轮自动保持上下文result2graph.invoke({messages:[HumanMessage(contentWhats my name?)]},config)print(result2[messages][-1].content)# Your name is Alice# 查看状态历史historylist(graph.get_state_history(config))print(fTotal checkpoints:{len(history)})十一、总结LangGraph 的持久化机制为构建强大的 AI 应用提供了基础Checkpointer保存线程级别的图状态快照InMemorySaver开发测试SqliteSaver本地开发PostgresSaver生产环境Store跨线程持久化应用数据用户偏好和记忆共享知识库状态管理get_state获取当前状态get_state_history查看执行历史update_state编辑状态并分叉高级特性时间旅行从历史状态重放人机交互中断-恢复工作流容错恢复待处理写入保留