持久化【Persistence】之人机协作改造from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command, interrupt # 1. 定义状态 class State(TypedDict): amount: int # 金额 status: str # pending, approved, rejected log: list # 2. 定义节点检查钱数 def check_amount(state): amount state[amount] log_entry f检查金额: ${amount} print(f {log_entry}) if amount 1000: # 返回部分状态标记为待处理 response interrupt({ reason: 金额过大需要人工审批 }) return {status: response, log: [log_entry]} else: return {status: auto_approved, log: [log_entry, 自动批准]} # 2.2 定义节点继续执行执法 def process_payment(state): print(f ✅ 执行支付: ${state[amount]} (状态: {state[status]})) return {log: state[log] [支付完成]} # 3. 构建图 builder StateGraph(State) builder.add_node(check, check_amount) builder.add_node(pay, process_payment) builder.add_edge(START, check) builder.add_conditional_edges( check, lambda s: pay if s[status] approved else END, # 如果 pending 先停在这里 ) builder.add_edge(pay, END) # 启用检查点 memory InMemorySaver() # 关键点在 pay 节点之前中断。 # 这意味着 check 运行完后如果下一步是 pay图会暂停不会执行 pay graph builder.compile(checkpointermemory) config {configurable: {thread_id: approval_demo}} print(\n 步骤 1: 提交大额交易 (触发中断) ) # 初始状态 initial_state {amount: 300, status: new, log: []} # 运行图它会在 pay 之前停下来 result graph.invoke(initial_state, config) if result.get(__interrupt__): # 人工输入继续执行 print(result.get(__interrupt__)) input_userinput(输入 approved 或者 reject:) result graph.invoke(Command(resumeinput_user), config) print(f完成操作最终日志流程: {result[log]}) else: print(f不需要审批直接付款)返回值是用来修改state的也就是只有返回的方式来进行return {status: response, log: [log_entry]}时间旅行from langgraph.graph import StateGraph, START from langgraph.checkpoint.memory import InMemorySaver from typing_extensions import TypedDict, NotRequired import uuid # 1 状态 class State(TypedDict): topic: NotRequired[str] # 主题 joke: NotRequired[str] # 笑话内容 # 2 节点生成主题 def generate_topic(state: State): return {topic: 烘干机里的袜子} # 3 节点编写笑话 def write_joke(state: State): return {joke: f为什么{state[topic]}会消失因为它们私奔了} checkpointer InMemorySaver() graph ( StateGraph(State) .add_node(generate_topic, generate_topic) .add_node(write_joke, write_joke) .add_edge(START, generate_topic) .add_edge(generate_topic, write_joke) .compile(checkpointercheckpointer) ) # 第一步运行图 config {configurable: {thread_id: str(uuid.uuid4())}} result graph.invoke({}, config) print(result[joke]) print() # 找到写笑话检查点 history list(graph.get_state_history(config)) before_joke next(s for s in history if s.next (write_joke,)) # 分支修改检查点状态更换笑话主题 fork_config graph.update_state( before_joke.config, values{topic: 小鸡}, # values会修改 state状态--》烘干机里的袜子 修改为 小鸡 ) # 从新分支继续执行——「编写笑话」节点基于新主题重新执行 fork_result graph.invoke(None, fork_config) print(fork_result[joke]) #以下代码before_joke next(s for s in history if s.next (write_joke,))next()函数可以提效并且减小消耗优势类型具体表现仅“加速”无法覆盖的场景1. 计算效率找到首个匹配项即停止遍历避免冗余计算若历史记录极长如10万条跳过99.99%的无效检查2. 内存控制零中间列表生成直接返回原对象引用大规模历史记录下避免内存溢出OOM3. 语义精确性明确表达“只需第一个匹配项”的意图代码可读性提升避免歧义如history[0]可能取错位置基于指定节点创建分支# 1 调用update_state方法时状态值的更新会遵循指定节点的写入规则。新检查点会记录该节点为状态更新的来源且执行流程会从该节点的后续节点开始恢复。# 2 默认情况下LangGraph 会从检查点的版本历史中自动推断状态更新的关联节点as_node基于指定检查点分支时该自动推断几乎始终有效。在以下场景中需要显式指定as_node并行分支多个节点在同一步骤中更新了状态LangGraph 无法判断最后执行的节点会抛出InvalidUpdateError异常无执行历史在全新的会话中初始化状态测试场景中常用跳过节点将as_node设为后续节点让图认为该节点已执行完毕from langgraph.graph import StateGraph, START from langgraph.checkpoint.memory import InMemorySaver from typing_extensions import TypedDict, NotRequired import uuid # 1 状态 class State(TypedDict): topic: NotRequired[str] # 主题 joke: NotRequired[str] # 笑话 # 2 节点生成主题 def generate_topic(state: State): return {topic: 烘干机里的袜子} # 3 节点 生成笑话 def write_joke(state: State): return {joke: f为什么{state[topic]}会消失因为它们私奔了} checkpointer InMemorySaver() graph ( StateGraph(State) .add_node(generate_topic, generate_topic) .add_node(write_joke, write_joke) .add_edge(START, generate_topic) .add_edge(generate_topic, write_joke) .compile(checkpointercheckpointer) ) # 第一步运行图 config {configurable: {thread_id: str(uuid.uuid4())}} result graph.invoke({}, config) print(result[joke]) print() # 在写笑话之前找到检查点 history list(graph.get_state_history(config)) before_joke next(s for s in history if s.next (write_joke,)) # 图的执行流程生成主题 - 编写笑话 # 将本次状态更新标记为「生成主题」节点的执行结果 # 执行流程会从「生成主题」的后续节点编写笑话开始恢复 fork_config graph.update_state( before_joke.config, values{topic: 小鸭}, as_nodegenerate_topic, # 做时间旅行认为generate_topic节点执行完了 ) fork_result graph.invoke(None, fork_config) print(fork_result[joke]) # 小鸭的笑话而不是袜子分支的本质含义就是重新恢复到之前某个检出点保存的状态快照然后重新执行这个检出点之后的流程从而衍生出一条新的执行状态就类似重新生成一个边。检查点当你运行 graph.invoke() 时每执行完一个节点或者遇到中断LangGraph 就会自动保存一个检查点。这个检查点包含1. 当前状态State比如你之前代码里的 {topic: 猫, joke: ...}。2. 元数据Metadata当前执行到了哪一步、是谁触发的。3. 下一步去哪Next引擎计算出的下一步要执行的节点比如 (write_joke,)。4. 唯一标识Config包含 thread_id哪个对话和 checkpoint_id具体哪一帧。查点带来的三大超能力正是因为有了检查点机制LangGraph 才能实现你之前问到的那些高级功能1.支持中断与恢复Human-in-the-loop原理遇到 interrupt() 时图暂停此时引擎把当前状态打成一个检查点存起来。等你输入 Command(resumeJustin) 后引擎找到这个检查点把数据读出来继续往下跑。2.时间旅行Time Travel原理因为每一步都存了检查点你可以通过 get_state_history() 把历史存档全部拉出来找到过去任意一帧的 checkpoint_id直接穿越回去。3.分支Fork原理基于某个历史检查点修改里面的数据比如把“猫”改成“鸭子”然后让引擎基于这个修改后的存档生成一条全新的时间线。故障恢复from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver # 仅调试用 from typing_extensions import TypedDict import random # 1. 定义状态 class TaskState(TypedDict): task_id: str step1: bool False # False表示没执行,True表示执行了 step2: bool False step3: bool False # 2. 定义节点模拟随机崩溃 def step1(state: TaskState): print(执行步骤1) return {step1: True} def step2(state: TaskState): # 模拟50%概率崩溃 if random.random() 0.5: print(执行步骤2崩溃了) raise RuntimeError(服务崩溃) print(执行步骤2没崩溃) return {step2: True} def step3(state: TaskState): print(执行步骤3) return {step3: True} # 3. 初始化检查点 checkpointer InMemorySaver() # 4. 构建图 builder StateGraph(TaskState) builder.add_node(step1, step1) builder.add_node(step2, step2) builder.add_node(step3, step3) builder.add_edge(START, step1) builder.add_edge(step1, step2) builder.add_edge(step2, step3) builder.add_edge(step3, END) graph builder.compile(checkpointercheckpointer) # 5. 模拟崩溃与恢复 config {configurable: {thread_id: task_thread_789}} print( 第一次执行可能崩溃 ) try: graph.invoke({task_id: TASK789}, configconfig) except RuntimeError as e: print(f执行失败{e}) # 恢复执行 print(\n 恢复执行 ) result graph.invoke(None, configconfig) print(\n 恢复执行结果 ) print(f步骤1完成{result[step1]}) print(f步骤2完成{result[step2]}) print(f步骤3完成{result[step3]})系统崩溃之后步骤如何实现加入检查点之后就可以实现这种断点重连加入检查点之后LangGraph 就自动具备了断点重连的能力。你不需要自己去写任何保存进度。