摘要：本文围绕 LangGraph 中基于检查点（Checkpoint）的状态更新与时间旅行能力展开，重点讲解
graph.update_state()的传值机制与参数含义。文章先说明上一节代码中 input 只能传 None 的局限，再结合官方文档与检查点记录截图，演示如何从指定节点（如节点 2）开始执行，并解释更新后会在旧检查点基础上创建新的分支而非覆盖原记录。随后介绍 config、values、as_node 三个参数的作用，以及如何借助 update_state 实现人机交互式的草稿修改。最后给出完整的 Python 示例代码，涵盖状态结构定义、归约函数（reducer）绑定、线性流程构建、检查点历史查看与分叉执行，帮助读者掌握在历史快照上修改状态并继续运行的实践方法。
内容参考于：图灵AI大模型全栈
如下图红框是上一节的代码，它的input传入的是None，这个input它没办法传值，只能传None，如果想传值接着往下看
LangChain官网说明
Use time-travel - Docs by LangChain
它传值是用的 graph.update_state() 这个函数，如下图是一个检查点记录的内容，现在要从2开始执行
它会变成下图的样子，黑色的是原本旧的检查点，它会在节点2开始创建一个新的分支（检查点），旧的检查点还是会保存，就是说它并不是在原本的检查点里改的，它是创建出新的检查点分支
它的参数
config：线程id（用来指定要更新哪一个检查点列表），还可以传递一个检查点id（checkpoint_id）
values：它是用来更新状态的值，它的值会传递给归并函数（reducer），如果没有设置归并函数那么会覆盖
as_node：通过它可以知道把更新的内容传递给谁
它可以进行人机交互，就是我们给用户一个草稿，如果用户觉得草稿不满意，可以编辑草稿
代码说明：
如下图红框，它的逻辑 就是从创建任务开始执行，然后再执行更新任务，然后再执行准备任务
如下图运行的效果图：注意它是倒着的，start是在最后一个
如下图红框，我们通过 update_state 对检查点进行了修改，可以通过它进行人机交互，用户通过对节点输入新的内容，然后我们通过 update_state 更新到节点对应的检查点里
代码
# ============================================================
# 模块导入
# ============================================================
from langgraph.graph import StateGraph, START, END # 状态图构建器、流程起止标记
from langgraph.checkpoint.memory import InMemorySaver # 内存版检查点持久化器
from typing_extensions import TypedDict # 结构化字典类型定义
from typing import Annotated # 类型元数据注解，用于绑定状态归约规则
from operator import add # Python标准运算符，此处作为列表拼接的归约函数
# ============================================================
# 定义全局任务状态结构
# ============================================================
class TaskState(TypedDict):
task_id: str # 任务唯一标识
title: str # 任务标题
assignee: str # 任务负责人
priority: int # 优先级数值
# 【核心机制】评论字段使用Annotated绑定add归约规则
# 原理：任何节点返回的comments列表，不会覆盖原有列表，而是自动拼接追加
# 效果：所有节点产生的评论会按执行顺序累加到总评论列表中
comments: Annotated[list, add]
status: str # 任务当前状态
# ============================================================
# 业务节点定义
# ============================================================
def create_task(state: TaskState):
"""节点1：创建任务，初始化任务状态和第一条评论"""
return {
"status": "进行中", # 更新状态为进行中
# 因为有add归约规则，这个单元素列表会自动追加到总评论列表
"comments": [f"任务 '{state['title']}' 已创建，分配给 {state['assignee']}"]
}
def update_task(state: TaskState):
"""节点2：更新任务状态"""
return {
"status": "已更新",
# 同样走归约规则，追加新评论
"comments": ["任务状态已更新"]
}
def plan_task(state: TaskState):
"""节点3：标记任务准备完成"""
return {
"status": "已准备",
"comments": [f"任务 '{state['title']}' 已准备"]
}
# ============================================================
# 构建线性任务流程
# ============================================================
workflow = StateGraph(state_schema=TaskState)
# 注册三个业务节点
workflow.add_node("create", create_task)
workflow.add_node("update", update_task)
workflow.add_node("complete", plan_task)
# 定义纯线性执行链路：START → create → update → complete → END
workflow.add_edge(START, "create")
workflow.add_edge("create", "update")
workflow.add_edge("update", "complete")
workflow.add_edge("complete", END)
# 绑定检查点持久化器
# 【机制】图每跑完一个节点，就自动把当前全量状态存为一个不可变的Checkpoint快照
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)
# ============================================================
# 第一次完整执行流程
# ============================================================
# 配置会话标识thread_id：同一thread_id下的所有检查点归为一组，不同会话完全隔离
config = {"configurable": {"thread_id": "task_001"}}
# 传入初始任务状态，从START跑完整条链路
result = graph.invoke({
"task_id": "T001",
"title": "开发注册功能",
"assignee": "张三",
"priority": 1,
"comments": [], # 初始评论为空列表
"status": "待分配"
}, config)
print("=== 任务创建完成 ===")
print(f"状态: {result['status']}")
print("评论:", result['comments'])
# ============================================================
# 查看历史检查点列表
# 【执行后会生成4个快照】
# 1. 初始状态快照：START刚启动，next=("create",)
# 2. create节点跑完：next=("update",) （我们后面要做分叉的目标点）
# 3. update节点跑完：next=("complete",)
# 4. complete节点跑完：next=() 流程结束
# ============================================================
history = list(graph.get_state_history(config))
for i, checkpoint in enumerate(history):
print("\n--------------------------------")
print("Checkpoint：", i + 1)
print("checkpoint_id：", checkpoint.config["configurable"]["checkpoint_id"])
# 长文本截断展示，避免控制台输出过长
strvalue = str(checkpoint.values)
if len(strvalue) > 200:
print("当前 State：", strvalue[:200])
else:
print("当前 State：", strvalue)
print("下一步执行：", checkpoint.next)
# ============================================================
# 找到分叉起点：create节点刚执行完，下一步准备跑update节点的快照
# 匹配条件：next == ("update",) —— 代表当前快照状态是create跑完的结果，下一步计划执行update
# ============================================================
before_joke = next(s for s in history if s.next == ("update",))
# ============================================================
# 【核心API演示】update_state：修改历史检查点，生成分支
# 【机制说明】
# 1. 不是直接修改原来的旧快照（旧快照永远保留，不会被覆盖）
# 2. 而是在指定的快照位置，应用传入的状态更新，生成一个全新的分叉快照
# 3. 更新内容会走State定义里的归约规则：comments字段是追加，不是覆盖整个列表
# 4. 返回值是新生成的分叉快照的config，后续invoke就基于这个新快照跑
# 【典型场景】人工介入纠错、审批通过后修改状态、节点执行结果不对时中途修改参数重跑
# ============================================================
fork_config = graph.update_state(
before_joke.config, # 要基于哪个历史快照做分叉
{
# 注意：传入的是单元素列表，因为有add归约规则
# 不是把原来的comments整个替换成这条内容，而是把这条追加到原有评论后面
"comments": ["任务，已经开发完成"]
}
)
# ============================================================
# 从修改后的分叉快照继续执行
# 【执行效果】
# 不会重跑前面的create节点，直接从next指向的update节点开始跑：
# 1. update节点执行：追加"任务状态已更新"评论
# 2. complete节点执行：追加"任务已准备"评论
# ============================================================
final_result = graph.invoke(None, fork_config)
print("\n\n")
print("#" * 70)
print("查看使用update_state之后的全部检查点")
print("#" * 70)
# 再次拉取历史检查点
# 【机制】新执行产生的检查点会追加到历史里，原来的旧时间线快照全部保留
# 相当于在原来的执行链上，多了一条改过评论的新时间线分支
new_history = list(graph.get_state_history(config))
for i, checkpoint in enumerate(new_history):
print("\n--------------------------------")
print("Checkpoint：", i + 1)
print("checkpoint_id：", checkpoint.config["configurable"]["checkpoint_id"])
strvalue = str(checkpoint.values)
if len(strvalue) > 200:
print("当前 State：", strvalue[:200])
else:
print("当前 State：", strvalue)
print("下一步执行：", checkpoint.next)
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/qq_36301061/article/details/169732062
