每执行完一个图节点LangGraph 自动保存一份当前完整状态就是快照。每条快照包含全量状态、父快照 ID、时间戳、自定义业务元数据。Checkpointer存储器快照存放载体分三类MemorySaver内存存储本地测试临时使用进程关闭数据丢失RedisSaverRedis 持久化中小型线上项目首选PostgresSaver数据库持久化大规模企业业务thread_id 会话隔离每一条独立对话 / 任务分配唯一thread_id不同线程快照完全隔离互不干扰。回滚机制指定checkpoint_id即可加载历史快照不删除原有链路生成新分支完全可逆最小可运行图 自动生成快照使用内置MemorySaver内存存储无需中间件适合本地测试。from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage # 1. 定义基础图 builder StateGraph(MessagesState) # 简单节点回复用户消息 def chat_node(state: MessagesState): last_msg state[messages][-1].content return {messages: [AIMessage(contentf收到你的消息{last_msg})]} builder.add_node(chat, chat_node) builder.add_edge(START, chat) builder.add_edge(chat, END) # 2. 挂载检查点存储器核心开启快照 checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) # 3. 会话唯一标识thread_id thread_config {configurable: {thread_id: test-thread-001}} # 4. 多轮执行自动生成多条快照 graph.invoke({messages: [HumanMessage(第一步初始化对话)]}, thread_config) graph.invoke({messages: [HumanMessage(第二步发送第二个请求)]}, thread_config) graph.invoke({messages: [HumanMessage(第三步发送第三个请求)]}, thread_config) # 5. 查询全部历史快照 history list(graph.get_state_history(thread_config)) print(f总快照数量{len(history)}) # 倒序输出[0]最新[-1]最早 for idx, snap in enumerate(history): msg_cnt len(snap.values[messages]) print(f[{idx}] 快照ID{snap.id[:16]}... | 消息数{msg_cnt} | 父快照ID{snap.parent_id[:16] if snap.parent_id else 无})输出执行后会输出 4 条快照初始空状态 三轮执行后列表按时间倒序排列最新快照在最前面。基础回滚跳转到历史快照继续执行回滚核心在 config 中传入checkpoint_id指定目标快照执行会生成新分支原有历史保留。# 承接上面代码继续执行 history list(graph.get_state_history(thread_config)) # 选取倒数第二个快照第二轮执行后的状态 target_snap history[-2] rollback_config { configurable: { thread_id: test-thread-001, checkpoint_id: target_snap.id } } # 加载回滚后的状态 rollback_state graph.get_state(rollback_config) print(f\n回滚完成当前消息数{len(rollback_state.values[messages])}) # 从历史快照继续运行生成新分支快照 new_result graph.invoke( {messages: [HumanMessage(从历史快照重新执行)]}, rollback_config ) print(回滚后新执行结果, new_result[messages][-1].content) # 再次查看历史会多出一条新快照分支 new_history list(graph.get_state_history(thread_config)) print(f回滚后快照总数{len(new_history)})进阶 Redis 持久化快照内存存储进程销毁即丢失生产环境使用RedisSaver持久化快照多服务共享历史。Redis 初始化与持久化图from langgraph.checkpoint.redis import RedisSaver from langgraph.graph import StateGraph, START, END, MessagesState from langchain_core.messages import HumanMessage, AIMessage # 1. 初始化Redis检查点 redis_checkpointer RedisSaver() redis_checkpointer.setup(redis://localhost:6379/0) # 2. 构建相同对话图 builder StateGraph(MessagesState) def chat_node(state: MessagesState): last_msg state[messages][-1].content return {messages: [AIMessage(contentfRedis持久化响应{last_msg})]} builder.add_node(chat, chat_node) builder.add_edge(START, chat) builder.add_edge(chat, END) graph builder.compile(checkpointerredis_checkpointer) # 会话配置 thread_cfg {configurable: {thread_id: redis-thread-001}} # 多轮执行持久化快照 graph.invoke({messages: [HumanMessage(Redis测试1)]}, thread_cfg) graph.invoke({messages: [HumanMessage(Redis测试2)]}, thread_cfg) # 查询Redis存储的历史快照 history list(graph.get_state_history(thread_cfg)) print(Redis持久化快照列表) for snap in history: print(f快照ID{snap.id[:16]} | 消息数{len(snap.values[messages])})给快照添加自定义元数据业务标记通过update_state给快照附加标签、用户 ID、步骤标识方便筛选历史快照# 承接上面Redis代码 meta_cfg {configurable: {thread_id: redis-thread-001}} # 更新状态并写入自定义元数据 graph.update_state( meta_cfg, {}, metadata{ label: 业务流程-步骤2完成, step_name: chat_finish, user_id: 10086 } ) # 获取最新快照查看元数据 latest_snap graph.get_state(meta_cfg) print(快照自定义元数据, latest_snap.metadata)高阶实战场景异常自动回滚重试故障自愈from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage # 自定义状态继承消息状态增加错误计数、成功快照ID class ResilientState(TypedDict): messages: list error_count: int last_success_checkpoint: str # 模拟会随机报错的危险工具 import random def risky_tool(): if random.random() 0.6: raise Exception(工具调用超时/LLM格式异常) return 工具执行成功获取数据完成 def safe_exec_node(state: ResilientState): thread_id retry-thread-001 cfg {configurable: {thread_id: thread_id}} try: res risky_tool() # 执行成功记录当前快照ID current_snap graph.get_state(cfg) return { messages: state[messages] [AIMessage(res)], error_count: 0, last_success_checkpoint: current_snap.id } except Exception as e: err_cnt state.get(error_count, 0) 1 if err_cnt 3: # 重试耗尽终止流程 return { messages: state[messages] [AIMessage(f重试3次全部失败{str(e)})], error_count: err_cnt } # 回滚到上一个成功快照 history list(graph.get_state_history(cfg)) parent_id history[1].parent_id # 取上一级父快照 rollback_cfg {configurable: {thread_id: thread_id, checkpoint_id: parent_id}} graph.get_state(rollback_cfg) return { messages: state[messages] [AIMessage(f执行失败第{err_cnt}次重试)], error_count: err_cnt } # 构建容错图 builder StateGraph(ResilientState) builder.add_node(safe_exec, safe_exec_node) builder.add_edge(START, safe_exec) builder.add_edge(safe_exec, END) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) # 启动流程 run_cfg {configurable: {thread_id: retry-thread-001}} result graph.invoke({messages: [HumanMessage(执行危险工具)], error_count: 0}, run_cfg) print(最终流程输出, result[messages][-1].content)多分支策略对比from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage # 基础对话图 builder StateGraph(MessagesState) def strategy_a(state): return {messages: [AIMessage(策略A精简快速方案得分80)]} def strategy_b(state): return {messages: [AIMessage(策略B精细完整方案得分95)]} builder.add_node(A, strategy_a) builder.add_node(B, strategy_b) builder.add_edge(START, A) builder.add_edge(A, END) builder.add_edge(START, B) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) thread_id branch-compare-001 base_cfg {configurable: {thread_id: thread_id}} # 1. 执行到分支基准点 graph.invoke({messages: [HumanMessage(业务需求基准)]}, base_cfg) base_snap graph.get_state(base_cfg).id # 2. 分支A不回滚直接执行 res_a graph.invoke({messages: []}, base_cfg) score_a 80 # 3. 回滚到基准快照执行分支B rollback_cfg {configurable: {thread_id: thread_id, checkpoint_id: base_snap}} res_b graph.invoke({messages: []}, rollback_cfg) score_b 95 # 对比择优 if score_a score_b: print(最优策略A, res_a[messages][-1].content) else: print(最优策略B, res_b[messages][-1].content)Postgres 限制最大快照数量from langgraph.checkpoint.postgres import PostgresSaver from sqlalchemy import create_engine engine create_engine(postgresql://user:pwd127.0.0.1:5432/agent_db) # 仅保留每个thread最近20条快照自动清理更早历史 pg_checkpointer PostgresSaver(engine, max_history20) pg_checkpointer.setup()定时清理过期快照import asyncio from datetime import datetime, timedelta async def clean_expired_checkpoints(checkpointer): 删除7天前所有快照保留最近50条 expire_time datetime.now() - timedelta(days7) # 遍历全部thread批量删除过期快照 all_threads checkpointer.list_threads() for tid in all_threads: history list(checkpointer.list_checkpoints(tid)) # 过滤过期快照并删除 pass完整可运行整合代码 运行前置本地启动 Redis 运行功能LangGraph Redis快照 历史查询 回滚重试 元数据标记 依赖langgraph1.2.7、langgraph-checkpoint-redis、redis-py from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.checkpoint.redis import RedisSaver from langchain_core.messages import HumanMessage, AIMessage # 1. 初始化Redis存储 # 连接本地Redis无密码默认端口6379 redis_saver RedisSaver() redis_saver.setup(redis://localhost:6379/0) # 2. 构建基础对话图 graph_builder StateGraph(MessagesState) def chat_handler(state: MessagesState) - dict: user_input state[messages][-1].content resp AIMessage(f已处理{user_input}) return {messages: [resp]} graph_builder.add_node(chat, chat_handler) graph_builder.add_edge(START, chat) graph_builder.add_edge(chat, END) # 绑定Redis检查点 agent_graph graph_builder.compile(checkpointerredis_saver) # 3. 会话配置 THREAD_ID full-demo-thread-001 base_config {configurable: {thread_id: THREAD_ID}} # 4. 多轮执行生成快照 print( 执行多轮对话自动生成快照 ) agent_graph.invoke({messages: [HumanMessage(第一轮创建任务)]}, base_config) agent_graph.invoke({messages: [HumanMessage(第二轮补充参数)]}, base_config) agent_graph.invoke({messages: [HumanMessage(第三轮提交执行)]}, base_config) # 给最新快照添加业务元数据 agent_graph.update_state( base_config, {}, metadata{step: task_submit, user_id: 999, tag: 任务提交完成} ) # 5. 查询全部历史快照 history_snapshots list(agent_graph.get_state_history(base_config)) print(f\n 共生成 {len(history_snapshots)} 条快照 ) for idx, snap in enumerate(history_snapshots): msg_count len(snap.values[messages]) meta snap.metadata print(f[{idx}] ID:{snap.id[:20]} 消息数:{msg_count} 元数据:{meta}) # 6. 回滚到第二轮快照倒数第二个 target_snap history_snapshots[-2] rollback_config { configurable: { thread_id: THREAD_ID, checkpoint_id: target_snap.id } } rollback_state agent_graph.get_state(rollback_config) print(f\n 回滚完成 ) print(f目标快照ID{target_snap.id[:20]}) print(f回滚后消息条数{len(rollback_state.values[messages])}) # 7. 从回滚快照重新执行生成新分支 new_run_result agent_graph.invoke( {messages: [HumanMessage(回滚后重新提交任务修正参数)]}, rollback_config ) print(f\n 回滚后新执行输出 ) print(new_run_result[messages][-1].content) # 再次查看快照验证新增分支 new_history list(agent_graph.get_state_history(base_config)) print(f\n回滚后快照总数更新为{len(new_history)})学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】