6. 持久化机制和可恢复执行（下）：使用场景

6.5. 使用场景

6.5.1. 多轮对话

最基础的使用场景是多轮对话。前文示例已经实现了一个简单的多轮对话流程，这里不再展开。

6.5.2. 中断恢复

中断恢复需要结合 LangGraph 的中断机制理解，见下文。

6.5.3. 失败后恢复运行

6.5.3.1. 用法说明

任务失败的原因可能是偶发的外部因素，如网络波动、秘钥过期、第三方服务异常；也可能是程序自身的BUG。

如果是前者，故障发生后，修复外部问题，直接重试即可；如果是后者，需要更改代码修复BUG后，再基于历史检查点恢复运行。

要在失败后基于历史检查点恢复运行，需要满足以下条件：

启用检查点存储器
- 如果是 InMemorySaver，不要重建检查点存储器对象，否则历史检查点丢失，无法恢复
- 如果希望跨进程、服务重启后仍可恢复，应使用基于 SQLite、Postgres 等持久化检查点后端的存储器
再次运行时用 None 作为计算图的状态输入
传递的配置信息应包含 thread_id 而不能包含 checkpoint_id

此时，LangGraph 会根据 thread_id 从 checkpointer 读取该会话的最新检查点，并从该检查点继续推进。

如果某个超步中有多个并行任务，其中一部分任务成功，另一部分任务失败，那么已经成功完成的任务结果会作为 pending_writes 被保存。恢复运行时，已经成功完成的任务不会重复执行。

6.5.3.2. 示例

6.5.3.2.1. 模拟失败

本例使用如下结构：

text

START
  -> node_change_topic
      -> node_poem
      -> node_joke
  -> node_output
  -> END

运行时，node_change_topic 会把输入主题 "猫咪" 改写为 "猫咪: 布偶猫"，然后并行执行 node_poem 和 node_joke，最后由 node_output 汇总输出。

示例如下

python

from typing import TypedDict

from dotenv import  load_dotenv
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph,START,END
from loguru import logger
load_dotenv(override=True)

model = ChatDeepSeek(
    model = "deepseek-v4-flash",
    extra_body={
        "thinking":{
            "type":"disabled"
        }
    }
)

#1. 声明状态
class OverAllState(TypedDict):
    topic:str
    poem:str
    joke:str
    final_output:str

#1.1 输入状态
class InputState(TypedDict):
    topic:str

#1.2 输出状态
class OutputState(TypedDict):
    final_output:str

topics = ["布偶猫","狸花猫","金渐层"]
topic_index = 0
#2. 定义节点
def node_change_topic(state:InputState)->OverAllState:
    global topic_index
    logger.info("topic_index:{}",topic_index)
    sub_topic = topics[topic_index]
    topic_index += 1
    topic_index %= len(topics)

    return {
        "topic":f"{state["topic"]}:{sub_topic}"
    }
#2.1 同一个超步的位置 成功运行的节点
def node_poem(state:OverAllState) -> OverAllState:
    logger.info("node_poem正在执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于{topic}主题的七言绝句")]).content
    return {
        "poem":poem
    }
import  time
#2.2 同一个超步的位置 运行失败的节点
def node_joke(state:OverAllState) -> OverAllState:
    logger.info("node_joke正在执行")
    topic = state["topic"]
    time.sleep(5)
    raise Exception("人为抛异常")
    joke = model.invoke([HumanMessage(f"写一首关于{topic}主题的笑话")]).content
    return {
        "joke":joke
    }

def node_output(state:OverAllState) -> OutputState:
    logger.info("node_output正在执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于{topic}的七言绝句:{poem}\n 笑话:{joke}\n"
    return {
        "final_output":final_output
    }

#3. 构建图
builder = StateGraph(state_schema=OverAllState,input_schema=InputState,output_schema=OutputState)

#3.1 添加节点
builder.add_node("node_change_topic",node_change_topic)
builder.add_node("node_poem",node_poem)
builder.add_node("node_joke",node_joke)
builder.add_node("node_output",node_output)

#3.2 添加边
builder.add_edge(START,"node_change_topic")
builder.add_edge("node_change_topic","node_poem")
builder.add_edge("node_change_topic","node_joke")
builder.add_edge("node_poem","node_output")
builder.add_edge("node_joke","node_output")
builder.add_edge("node_output",END)

#4. 添加检查点后端
DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    from IPython.display import display
    display(graph)

    config = {
        "configurable":{
            "thread_id":"chapter03-05"
        }
    }

    res = graph.invoke({"topic":"猫"},config=config)
    print(res)

运行结果如下

json

2026-06-10 16:17:43.544 | INFO     | __main__:node_change_topic:45 - topic_idx: 0
2026-06-10 16:17:43.545 | INFO     | __main__:node_joke:65 - node_joke 已执行
2026-06-10 16:17:43.548 | INFO     | __main__:node_poem:54 - node_poem 已执行
>Traceback...
Exception: 人为中断
During task with name 'node_joke' and id 'edb59019-4808-ed51-68eb-d5fba8696b13'

本例在 node_joke 中抛出异常，从而模拟计算图执行失败的场景。

为了观察恢复执行时哪些节点会被重新执行，示例中额外定义了全局变量 topics 和 topic_idx。node_change_topic 每执行一次，都会从 topics 中取出一个新的子主题，作为状态字段 topic 的值。

如果恢复运行时 node_change_topic 没有再次执行，则最终结果仍然会使用第一次生成的主题，即 猫咪: 布偶猫。如果该节点被重新执行，则主题可能会变成 猫咪: 狸花猫 或 猫咪: 金渐层。

6.5.3.2.2. 查看历史检查点列表

示例如下

python

list(graph.get_state_history(config=config))

运行结果如下

json

[
    StateSnapshot(
        values={
            "topic": "猫咪: 布偶猫"
        },
        next=(
            "node_poem",
            "node_joke"
        ),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e60d-6881-8001-c717582e8893"
            }
        },
        metadata={
            "source": "loop",
            "step": 1,
            "parents": {}
        },
        created_at="2026-06-10T08:43:19.431987+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e608-6596-8000-ed45af11cea7"
            }
        },
        tasks=(
            PregelTask(
                id="66287e00-4228-a985-fc09-6791a14e26ce",
                name="node_poem",
                path=(
                    "__pregel_pull",
                    "node_poem"
                ),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "poem": "《戏题布偶猫》\n雪翼云裘自绝尘，琉璃碧眼若含春。\n行来莲步生兰气，卧处梨花是此身。"
                }
            ),
            PregelTask(
                id="b2632b70-562b-7d5c-0dae-0f83bd6c2912",
                name="node_joke",
                path=(
                    "__pregel_pull",
                    "node_joke"
                ),
                error="Exception('人为中断')",
                interrupts=(),
                state=None,
                result=None
            )
        ),
        interrupts=()
    ),

    StateSnapshot(
        values={
            "topic": "猫咪"
        },
        next=(
            "node_change_topic",
        ),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e608-6596-8000-ed45af11cea7"
            }
        },
        metadata={
            "source": "loop",
            "step": 0,
            "parents": {}
        },
        created_at="2026-06-10T08:43:19.429860+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e606-6a42-bfff-9f09d8a3aa96"
            }
        },
        tasks=(
            PregelTask(
                id="273fe47f-b4cc-93d0-0ab7-57a0db4bad56",
                name="node_change_topic",
                path=(
                    "__pregel_pull",
                    "node_change_topic"
                ),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "猫咪: 布偶猫"
                }
            ),
        ),
        interrupts=()
    ),

    StateSnapshot(
        values={},
        next=(
            "__start__",
        ),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e606-6a42-bfff-9f09d8a3aa96"
            }
        },
        metadata={
            "source": "input",
            "step": -1,
            "parents": {}
        },
        created_at="2026-06-10T08:43:19.429167+00:00",
        parent_config=None,
        tasks=(
            PregelTask(
                id="ec8e6e32-6969-6798-9c5f-aade41e9211a",
                name="__start__",
                path=(
                    "__pregel_pull",
                    "__start__"
                ),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "猫咪"
                }
            ),
        ),
        interrupts=()
    )
]

可以看到，失败后的检查点中（step=1）记录了当前超步相关任务的执行情况：

node_poem 执行成功，结果记录在 result 字段中。
node_joke 执行失败，result 为空，异常信息记录在 error 字段中。

本例中，node_poem 和 node_joke 属于同一个超步。虽然 node_joke 失败了，但 node_poem 已成功，其计算结果会被持久化机制保存下来。因此，后续恢复运行时，node_poem 不会重复执行。

6.5.3.2.3. 修复BUG之后恢复运行

修复 node_joke 中的人为异常后，重新编译计算图，并基于最新检查点恢复运行。

示例如下

python

def node_joke(state: InputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]

    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话")]).content

    return {
        "joke": joke
    }


builder = StateGraph(state_schema=OverAllState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_change_topic", node_change_topic)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_change_topic")
builder.add_edge("node_change_topic", "node_poem")
builder.add_edge("node_change_topic", "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

new_graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(new_graph)

res = new_graph.invoke(None, config=config)
print(res)

这里需要注意两点：

重新编译计算图时，仍然使用原来的 checkpointer 对象。
恢复运行时，输入参数传入 None，并且配置中只传入 thread_id。

运行结果如下

json

2026-06-10 16:37:12.985 | INFO     | __main__:node_joke:2 - node_joke 已执行
2026-06-10 16:37:18.878 | INFO     | __main__:node_output:76 - node_output 已执行
{
    'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《戏题布偶猫》\n雪翼云裘自绝尘，琉璃碧眼若含春。\n行来莲步生兰气，卧处梨花是此身。\n笑话：\n一只布偶猫走进咖啡馆，优雅地跳上吧台，对老板说：“给我来一杯最贵的猫屎咖啡。”  \n老板愣住：“可……您本身就是猫啊？”  \n布偶猫舔了舔爪子，翻了个白眼：“所以呢？我自己产的屎，你们人类不是喝得挺香吗？今天我倒要尝尝，你们拿我的‘周边产品’能泡出什么花样来。”'
}

运行结果中可以观察到：

node_joke 被重新执行。
node_output 被执行。
node_poem 没有被重新执行。

最终输出中的七言绝句和历史检查点中 node_poem.result 记录的内容一致，说明 node_poem 的执行结果被成功复用了。

6.5.4. Time Travel

Time Travel 直译为 时间旅行，在当前场景下太过生硬，从技术实现和使用场景来看，译为 检查点回溯 更为合理。

所谓检查点回溯，是指基于某个历史检查点，重新执行后续流程，或者在该检查点基础上修改状态并创建新的执行分支。

检查点回溯有两种形式，根据是否更改历史状态区分：

Replay：检查点重放，回到某个历史检查点，沿着原先的执行路径重新执行后续节点。
Fork：检查点分叉，回到某个历史检查点，修改状态，从该位置创建一条新的执行分支。

二者的共同点是：

检查点之前的节点不会重新执行，检查点之后的节点会重新执行。

二者的区别是：

Replay 不修改历史状态；Fork 会基于历史检查点应用新的状态更新，并创建新的检查点分支。

6.5.4.1. Replay

Replay 模式和失败恢复很接近，但二者的触发方式和语义不同。

失败恢复通常基于最新检查点继续运行，配置中只包含 thread_id，不包含 checkpoint_id。
Replay 则是显式传入某个历史检查点的配置，配置中包含 checkpoint_id。LangGraph 会据此从该历史检查点开始重放后续步骤。

需要注意：

Replay 不是简单读取历史缓存，而是重新执行该检查点之后的节点。

因此，检查点之后的 LLM 调用、API 请求、工具调用、中断 等都会重新触发，最终结果可能和原始结果不同。

6.5.4.1.1. 正常运行的任务

本例仍然使用如下结构：

json

START
  -> node_change_topic
      -> node_poem
      -> node_joke
  -> node_output
  -> END

示例如下

python

from typing import TypedDict

from dotenv import  load_dotenv
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph,START,END
from loguru import logger
load_dotenv(override=True)

model = ChatDeepSeek(
    model = "deepseek-v4-flash",
    extra_body={
        "thinking":{
            "type":"disabled"
        }
    }
)

#1. 声明状态
class OverAllState(TypedDict):
    topic:str
    poem:str
    joke:str
    final_output:str

#1.1 输入状态
class InputState(TypedDict):
    topic:str

#1.2 输出状态
class OutputState(TypedDict):
    final_output:str

topics = ["布偶猫","狸花猫","金渐层"]
topic_index = 0
#2. 定义节点
def node_change_topic(state:InputState)->OverAllState:
    global topic_index
    logger.info("topic_index:{}",topic_index)
    sub_topic = topics[topic_index]
    topic_index += 1
    topic_index %= len(topics)

    return {
        "topic":f"{state["topic"]}:{sub_topic}"
    }
#2.1 同一个超步的位置 成功运行的节点
def node_poem(state:OverAllState) -> OverAllState:
    logger.info("node_poem正在执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于{topic}主题的七言绝句")]).content
    return {
        "poem":poem
    }
import  time
#2.2 同一个超步的位置 运行失败的节点
def node_joke(state:OverAllState) -> OverAllState:
    logger.info("node_joke正在执行")
    topic = state["topic"]
    # time.sleep(5)
    # raise Exception("人为抛异常")
    joke = model.invoke([HumanMessage(f"写一首关于{topic}主题的笑话")]).content
    return {
        "joke":joke
    }

def node_output(state:OverAllState) -> OutputState:
    logger.info("node_output正在执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于{topic}的七言绝句:{poem}\n 笑话:{joke}\n"
    return {
        "final_output":final_output
    }

#3. 构建图
builder = StateGraph(state_schema=OverAllState,input_schema=InputState,output_schema=OutputState)

#3.1 添加节点
builder.add_node("node_change_topic",node_change_topic)
builder.add_node("node_poem",node_poem)
builder.add_node("node_joke",node_joke)
builder.add_node("node_output",node_output)

#3.2 添加边
builder.add_edge(START,"node_change_topic")
builder.add_edge("node_change_topic","node_poem")
builder.add_edge("node_change_topic","node_joke")
builder.add_edge("node_poem","node_output")
builder.add_edge("node_joke","node_output")
builder.add_edge("node_output",END)

#4. 添加检查点后端
DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    from IPython.display import display
    display(graph)

    config = {
        "configurable":{
            "thread_id":"chapter03-08"
        }
    }

    res = graph.invoke({"topic":"猫咪"},config=config)
    print(res)

运行结果如下

json

2026-06-12 14:32:42.019 | INFO     | __main__:node_change_topic:45 - topic_idx: 0
2026-06-12 14:32:42.022 | INFO     | __main__:node_joke:63 - node_joke 已执行
2026-06-12 14:32:42.026 | INFO     | __main__:node_poem:54 - node_poem 已执行
2026-06-12 14:32:43.585 | INFO     | __main__:node_output:72 - node_output 已执行
2026-06-12 14:32:43.588 | INFO     | __main__:<module>:102 - res: 
{
    'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。\n笑话：\n一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
}

6.5.4.1.2. 查看历史检查点列表

示例如下

python

history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

运行结果如下

json

[
    StateSnapshot(
        values={
            'topic': '猫咪: 布偶猫',
            'poem': '《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。',
            'joke': '一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”',
            'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。\n笑话：\n一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
        },
        next=(),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-4ad3-67ef-8003-01d021de9403'
            }
        },
        metadata={
            'source': 'loop',
            'step': 3,
            'parents': {}
        },
        created_at='2026-06-12T06:32:43.586547+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-4acf-62cf-8002-aed204b5115d'
            }
        },
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'topic': '猫咪: 布偶猫',
            'poem': '《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。',
            'joke': '一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
        },
        next=('node_output',),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-4acf-62cf-8002-aed204b5115d'
            }
        },
        metadata={
            'source': 'loop',
            'step': 2,
            'parents': {}
        },
        created_at='2026-06-12T06:32:43.584774+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3be4-6523-8001-dd86f05c34b8'
            }
        },
        tasks=(
            PregelTask(
                id='b9042203-0ddf-f638-aeeb-e0501601381a',
                name='node_output',
                path=('__pregel_pull', 'node_output'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。\n笑话：\n一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
                }
            ),
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'topic': '猫咪: 布偶猫'
        },
        next=('node_poem', 'node_joke'),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3be4-6523-8001-dd86f05c34b8'
            }
        },
        metadata={
            'source': 'loop',
            'step': 1,
            'parents': {}
        },
        created_at='2026-06-12T06:32:42.020569+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bde-6373-8000-9750b4be0665'
            }
        },
        tasks=(
            PregelTask(
                id='eab6bf51-4986-a60f-753c-e493b78c56b8',
                name='node_poem',
                path=('__pregel_pull', 'node_poem'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'poem': '《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。'
                }
            ),
            PregelTask(
                id='a228a7bf-4c3b-2ad3-039f-12ac0107c87b',
                name='node_joke',
                path=('__pregel_pull', 'node_joke'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'joke': '一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
                }
            )
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'topic': '猫咪'
        },
        next=('node_change_topic',),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bde-6373-8000-9750b4be0665'
            }
        },
        metadata={
            'source': 'loop',
            'step': 0,
            'parents': {}
        },
        created_at='2026-06-12T06:32:42.018069+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bdc-60a4-bfff-d4fdcf6e6a9e'
            }
        },
        tasks=(
            PregelTask(
                id='e0034c08-dd19-0ae9-7059-f8414745d212',
                name='node_change_topic',
                path=('__pregel_pull', 'node_change_topic'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'topic': '猫咪: 布偶猫'
                }
            ),
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={},
        next=('__start__',),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bdc-60a4-bfff-d4fdcf6e6a9e'
            }
        },
        metadata={
            'source': 'input',
            'step': -1,
            'parents': {}
        },
        created_at='2026-06-12T06:32:42.017182+00:00',
        parent_config=None,
        tasks=(
            PregelTask(
                id='1a1ac312-152a-dcc3-2aa7-c2588f651448',
                name='__start__',
                path=('__pregel_pull', '__start__'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'topic': '猫咪'
                }
            ),
        ),
        interrupts=()
    )
]

6.5.4.1.3. 重放

get_state_history() 返回的是历史检查点列表，并且默认按照时间倒序排列。

history_checkpoints 结构如下：

text

history_checkpoints[0] -> 最终完成后的检查点
						next=()
history_checkpoints[1] -> node_output 执行前的检查点
						next=('node_output',)
history_checkpoints[2] -> node_poem、node_joke 执行前的检查点
						next=('node_poem', 'node_joke')
history_checkpoints[3] -> node_change_topic 执行前的检查点
						next=('node_change_topic',)
history_checkpoints[4] -> __start__ 执行前的输入检查点
						next=('__start__',)

因此，如果希望从写诗写笑话开始重放，选择 next == ('node_poem', 'node_joke') 的检查点即可。

示例如下

python

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    config = {
        "configurable":{
            "thread_id":"chapter03-08"
        }
    }
    # 获取到检查点历史
    history_checkpoints = list(graph.get_state_history(config=config))

    new_checkpoint = None
    #next = ('node_poem', 'node_joke')
    for checkpoint in history_checkpoints:
        if checkpoint.next == ('node_poem', 'node_joke'):
            new_checkpoint = checkpoint
            break

    # 如果想要实现replay的效果 状态填写None  config填写为之前某一个检查点的config
    res = graph.invoke(None,config=new_checkpoint.config)
    print(res)

运行结果如下

json

2026-07-07 18:31:32.019 | INFO     | __main__:node_joke:59 - node_joke正在执行
2026-07-07 18:31:32.023 | INFO     | __main__:node_poem:50 - node_poem正在执行
2026-07-07 18:31:38.158 | INFO     | __main__:node_output:69 - node_output正在执行
{'final_output': '关于猫咪:狸花猫的七言绝句:《狸花猫》\n玄文斑驳隐花光，夜半巡檐气自昂。\n踏雪无痕轻似梦，一窗明月一炉香。\n\n注：我的创作思路是捕捉狸花猫的神秘与灵动。首句“玄文斑驳”描绘其毛发纹理，如暗夜中的花纹；次句“夜半巡檐”展现其夜行习性，气度不凡。后两句以“踏雪无痕”喻其轻盈，结句“明月炉香”营造温馨画面，暗合狸花猫守护家园的意象，使全诗动静相生，虚实相映。\n 笑话:# 狸花猫的烦恼\n\n有只狸花猫去宠物医院看病，医生问它：“哪里不舒服？”\n\n狸花猫叹了口气说：“医生，我总觉得自己不够‘花’。”\n\n医生疑惑：“你可是标准的狸花猫啊，条纹多漂亮！”\n\n狸花猫委屈地说：“可是隔壁那只三花猫，人家身上有黑、白、橘三种颜色，比我缤纷多了！我整天就这一身灰黑条纹，太单调了。”\n\n医生笑了：“那你想要怎么办？”\n\n狸花猫眼睛一亮：“医生，您能不能给我开点染色剂？我想在背上加几块橘色斑点，尾巴再来点白的，耳朵尖染成黑的…”\n\n医生打断它：“等等，那样你就变成‘四不像’了！”\n\n狸花猫思考了一会，突然骄傲地挺起胸膛：“你说得对！我这一身条纹可是老祖宗传下来的迷彩服，抓老鼠时特别管用！三花猫再好看，还得靠我帮忙抓家里的老鼠呢！”\n\n医生欣慰地说：“这就对了。”\n\n狸花猫走出诊室，正好遇到一只胖橘猫。胖橘猫嘲笑它：“哟，这不是那只会抓老鼠却不会卖萌的狸花猫吗？”\n\n狸花猫淡定地回了一句：“我会抓老鼠，你会什么？”\n\n胖橘猫想了想，尴尬地说：“我…我会吃。”\n\n狸花猫笑道：“那不就成了？你负责吃，我负责抓，咱们各有各的活法！”\n\n—— **适合自己的，才是最好的。** 🐱\n'}

本例中：

node_poem 和 node_joke 被重新执行。
node_output 被重新执行。

由于没有执行change_topic,所以可以一直写同一只猫的内容

6.5.4.2. Fork

6.5.4.2.1. 用法说明

Fork 依赖状态图的 update_state() 方法

示例如下

python

change_input_config = graph.update_state(
    config=before_router_config.config,
    values={"user_input": "帮我写一个关于布偶猫的笑话"},
    as_node=START
)

其中：

config：历史检查点的配置信息，通常包含 thread_id、checkpoint_ns 和 checkpoint_id。LangGraph 会据此检索历史检查点
values：要应用到该检查点上的状态更新
as_node：指定这次状态更新应当被视为哪个节点产生的输出。
LangGraph 会按照该节点的写入逻辑，将 values 写入状态通道，并据此决定后续从哪些节点继续执行。
基于分叉检查点运行时，并不是重新执行 as_node 本身，而是从它的后继节点继续推进
如 START 所属超步编号为 0，则恢复时从编号为 1 的超步开始运行。

注意

update_state() 不会修改原来的历史检查点，而是基于某个历史检查点创建一个新的检查点。这个新检查点相当于一条新的执行分支。

6.5.4.2.2. 准备检查点历史

6.5.4.2.2.1. 准备运行图并运行一次

示例如下

python

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from loguru import logger
from dotenv import load_dotenv

load_dotenv(override=True)


model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    username: str
    user_input: str
    output: str

def router_node(state: OverAllState) -> StructuredOutputState:
    logger.info("路由节点已执行")
    user_input = state["user_input"]
    res = model_with_structure.invoke(
        [
            HumanMessage(user_input)
        ]
    )
    logger.info("路由结果-res: {}", res)

    return res

def router(state: StructuredOutputState) -> Literal["node_poem", "node_joke", "node_default"]:
    logger.info("路由函数已执行")
    if state["mode"] == "poem":
        logger.info("路由至 node_poem 节点")
        return "node_poem"
    elif state["mode"] == "joke":
        logger.info("路由至 node_joke 节点")
        return "node_joke"

    logger.info("路由至兜底节点")
    return "node_default"

def node_poem(state: StructuredOutputState) -> OverAllState:
    logger.info(f"node_poem 已执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于 {topic} 的七言绝句，不要赏析，只给诗句即可")]).content

    return {
        "output": poem
    }

def node_joke(state: StructuredOutputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]
    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话，尽可能简单，不超过一百字")]).content

    return {
        "output": joke
    }

def node_default(state: StructuredOutputState) -> OverAllState:
    logger.info(f"node_default 已执行")

    return {
        "output": "无法处理的任务类型"
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("router_node", router_node)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_default", node_default)

builder.add_edge(START, "router_node")
builder.add_conditional_edges(
    "router_node",
    router,
    path_map=["node_poem", "node_joke", "node_default"]
)
builder.add_edge("node_poem", END)
builder.add_edge("node_joke", END)
builder.add_edge("node_default", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "123"}}
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

res = graph.invoke({
    "username": "小黄",
    "user_input": "帮我写一首关于布偶猫的七言绝句"
}, config=config)
print(res)

执行逻辑如下：

router_node 根据用户输入提取结构化结果，例如 {"topic": "布偶猫", "mode": "poem"}。
router 根据 mode 决定后续分支。
如果 mode == "poem"，进入 node_poem。
如果 mode == "joke"，进入 node_joke。
如果无法识别，则进入 node_default。

运行结果如下

json

2026-06-11 14:26:48.353 | INFO     | __main__:router_node:31 - 路由节点已执行
2026-06-11 14:26:49.618 | INFO     | __main__:router_node:38 - 路由结果-res: {'topic': '布偶猫', 'mode': 'poem'}
2026-06-11 14:26:49.619 | INFO     | __main__:router:43 - 路由函数已执行
2026-06-11 14:26:49.619 | INFO     | __main__:router:45 - 路由至 node_poem 节点
2026-06-11 14:26:49.620 | INFO     | __main__:node_poem:55 - node_poem 已执行
{
    'username': '小黄', 
    'user_input': '帮我写一首关于布偶猫的七言绝句', 
    'output': '《布偶猫》\n蓝眼澄明玉作团，云踪雪影卧雕栏。\n无端一跃梅花落，犹带娇声唤月寒。'
}

6.5.4.2.2.2. 获取历史检查点列表

示例如下

python

history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

运行结果如下

json

[
    StateSnapshot(
        values={
            "username": "小黄",
            "user_input": "帮我写一首关于布偶猫的七言绝句",
            "output": (
                "《布偶猫》\n"
                "蓝眼澄明玉作团，云踪雪影卧雕栏。\n"
                "无端一跃梅花落，犹带娇声唤月寒。"
            ),
            "topic": "布偶猫",
            "mode": "poem",
        },
        next=(),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-8784-65e9-8002-e21663d55c94",
            }
        },
        metadata={
            "source": "loop",
            "step": 2,
            "parents": {},
        },
        created_at="2026-06-11T06:26:51.611071+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-7488-66c4-8001-f46b7c2d9a4e",
            }
        },
        tasks=(),
        interrupts=(),
    ),

    StateSnapshot(
        values={
            "username": "小黄",
            "user_input": "帮我写一首关于布偶猫的七言绝句",
            "topic": "布偶猫",
            "mode": "poem",
        },
        next=("node_poem",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-7488-66c4-8001-f46b7c2d9a4e",
            }
        },
        metadata={
            "source": "loop",
            "step": 1,
            "parents": {},
        },
        created_at="2026-06-11T06:26:49.620436+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-6871-6073-8000-ddcbe680da21",
            }
        },
        tasks=(
            PregelTask(
                id="cb3a8901-3a87-d0b0-3bab-d1adda5563d8",
                name="node_poem",
                path=("__pregel_pull", "node_poem"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "output": (
                        "《布偶猫》\n"
                        "蓝眼澄明玉作团，云踪雪影卧雕栏。\n"
                        "无端一跃梅花落，犹带娇声唤月寒。"
                    )
                },
            ),
        ),
        interrupts=(),
    ),

    StateSnapshot(
        values={
            "username": "小黄",
            "user_input": "帮我写一首关于布偶猫的七言绝句",
        },
        next=("router_node",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-6871-6073-8000-ddcbe680da21",
            }
        },
        metadata={
            "source": "loop",
            "step": 0,
            "parents": {},
        },
        created_at="2026-06-11T06:26:48.352564+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-686b-66c7-bfff-24e0e8d44f4c",
            }
        },
        tasks=(
            PregelTask(
                id="fb88436e-b163-6308-6341-7d8368e041f9",
                name="router_node",
                path=("__pregel_pull", "router_node"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "布偶猫",
                    "mode": "poem",
                },
            ),
        ),
        interrupts=(),
    ),

    StateSnapshot(
        values={},
        next=("__start__",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-686b-66c7-bfff-24e0e8d44f4c",
            }
        },
        metadata={
            "source": "input",
            "step": -1,
            "parents": {},
        },
        created_at="2026-06-11T06:26:48.350273+00:00",
        parent_config=None,
        tasks=(
            PregelTask(
                id="a1b5e5fb-ee60-fd7b-68bc-475fa9cb66e2",
                name="__start__",
                path=("__pregel_pull", "__start__"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "username": "小黄",
                    "user_input": "帮我写一首关于布偶猫的七言绝句",
                },
            ),
        ),
        interrupts=(),
    ),
]

6.5.4.2.2.3. 获取 `router_node` 之前的检查点

示例如下

router_node 的上一个检查点的 next 字段为 ('router_node',)，据此筛选检查点

python

before_router_checkpoint = next(h for h in history_checkpoints if h.next == ('router_node',))
before_router_checkpoint

运行结果如下

json

StateSnapshot(
    values={
        "username": "小黄",
        "user_input": "帮我写一首关于布偶猫的七言绝句",
    },
    next=("router_node",),
    config={
        "configurable": {
            "thread_id": "123",
            "checkpoint_ns": "",
            "checkpoint_id": "1f1655e8-6871-6073-8000-ddcbe680da21",
        }
    },
    metadata={
        "source": "loop",
        "step": 0,
        "parents": {},
    },
    created_at="2026-06-11T06:26:48.352564+00:00",
    parent_config={
        "configurable": {
            "thread_id": "123",
            "checkpoint_ns": "",
            "checkpoint_id": "1f1655e8-686b-66c7-bfff-24e0e8d44f4c",
        }
    },
    tasks=(
        PregelTask(
            id="fb88436e-b163-6308-6341-7d8368e041f9",
            name="router_node",
            path=("__pregel_pull", "router_node"),
            error=None,
            interrupts=(),
            state=None,
            result={
                "topic": "布偶猫",
                "mode": "poem",
            },
        ),
    ),
    interrupts=(),
)

该检查点表示：

用户输入已经写入状态，下一步将要执行 router_node。

从这个检查点分叉，本文采用两种方案（还可以有别的方案）：

修改输入，让 router_node 重新执行。见 6.5.4.2.3. 节
直接伪造 router_node 的输出，从而跳过 router_node。见 6.5.4.2.4. 节

6.5.4.2.3. 只改状态

6.5.4.2.3.1. 更新配置

示例如下

python

change_input_config = graph.update_state(
    config=before_router_checkpoint.config,
    values={"user_input": "帮我写一个关于布偶猫的笑话"},
    as_node=START
)
change_input_config

这里把 as_node 设置为 START，表示：

把这次状态更新视为 START 的输出。

因此，后续运行时会从 START 的后继节点继续推进，也就是重新执行 router_node。

上述代码运行结果如下

json

{
    "configurable": {
        "thread_id": "123",
        "checkpoint_ns": "",
        "checkpoint_id": "1f1655f8-86fb-6dd1-8001-9e7808dfaf1a",
    }
}

执行 update_state() 后会返回一个新的配置，其中包含新的 checkpoint_id。这个新的 checkpoint_id 不同于原有历史检查点，说明 update_state() 创建了一个新的检查点分支。

需要注意：

update_state() 返回的是新检查点的配置，不是新状态本身。

新的状态已经被记录在检查点后端，运行时生效。可以执行以下代码查看新建的检查点快照：

python

graph.get_state(change_input_config)

此处不再演示。

6.5.4.2.3.2. 重新运行

示例如下

python

graph.invoke(None, change_input_config)

运行结果如下

json

2026-06-11 14:13:44.267 | INFO     | __main__:router_node:31 - 路由节点已执行
2026-06-11 14:13:45.222 | INFO     | __main__:router_node:38 - 路由结果-res: {'topic': '布偶猫', 'mode': 'joke'}
2026-06-11 14:13:45.223 | INFO     | __main__:router:43 - 路由函数已执行
2026-06-11 14:13:45.223 | INFO     | __main__:router:48 - 路由至 node_joke 节点
2026-06-11 14:13:45.225 | INFO     | __main__:node_joke:64 - node_joke 已执行
{
    "username": "小黄",
    "user_input": "帮我写一个关于布偶猫的笑话",
    "output": (
        "一只布偶猫照镜子，问：“魔镜魔镜，谁是世界上最美的猫？”  \n"
        "镜子没说话。  \n"
        "猫急了：“快说呀！不然我卖萌把你甜碎。”  \n"
        "镜子终于开口：“……我死机了，重启中。”"
    ),
}

由运行结果可知：

user_input 已经从“帮我写一首关于布偶猫的七言绝句”变成“帮我写一个关于布偶猫的笑话”。而 username 没有被更新，仍然保持原值 "小黄"。说明 values 是 状态更新 而非完整的 状态替换。
router_node 开始的节点全部重新运行。由于新的输入被识别为笑话任务，因此后续进入 node_joke。

6.5.4.2.4. 改状态且跳过某些节点

6.5.4.2.4.1. 更新配置

如果不希望重新执行 router_node，而是直接指定 router_node 的输出，可以把 as_node 设置为 "router_node"。

示例如下

python

skip_router_config = graph.update_state(
    config=before_router_checkpoint.config,
    values={"topic": "狸花猫", "mode": "笑话"},
    as_node="router_node"
)
skip_router_config

此处把 {"topic": "狸花猫", "mode": "笑话"} 当作 router_node 的输出。
因此，后续运行时，router_node 本身不会被重新执行，从它的后继节点继续推进。
由于 router_node 挂载了条件边：
python
```
builder.add_conditional_edges(
    "router_node",
    router,
    path_map=["node_poem", "node_joke", "node_default"]
)
```
所以当 update_state() 把 values 视为 router_node 的输出时，条件路由函数 router 也会根据这次更新后的状态计算后续分支。
这里把 mode 设置为 "笑话"，router 函数命中兜底分支 node_default。

运行结果如下

json

2026-06-11 14:14:02.223 | INFO     | __main__:router:43 - 路由函数已执行
2026-06-11 14:14:02.223 | INFO     | __main__:router:51 - 路由至兜底节点
{
    "configurable": {
        "thread_id": "123",
        "checkpoint_ns": "",
        "checkpoint_id": "1f1655cb-de15-6c4e-8001-ca03b011e993",
    }
}

可以看到：

update_state() 创建了一个新的检查点，返回的 skip_router_config 指向这个新检查点。
日志中出现了：
text
```
路由函数已执行
路由至兜底节点
```
这说明 router_node 的节点函数没有被重新执行，但与 router_node 条件边相关的路由函数被执行了。
update_state() 底层会根据 as_node 找到对应图节点的 writers。
对于 router_node，writers 包括状态写入器和条件边对应的路由逻辑。
执行 update_state() 时，传入的 values 会被当作 router_node 的输出，交给该节点的 writers 处理。
状态写入器会把 values 转换为对状态 Channel 的写入；条件边对应的路由逻辑会根据当前状态计算后继分支，并生成类似 branch:to:node_default 这样的触发通道写入。
随后，update_state() 会把这些写入应用到检查点中，并保存一个新的检查点。
因此，状态更新和路由结果已经记录在 skip_router_config 指向的新检查点中。后续再调用：
python
```
graph.invoke(None, skip_router_config)
```
时，运行图会基于这个新检查点继续执行，被触发的后继节点也会继续运行。

综上，graph.update_state() 的作用可以理解为：

不执行指定节点的节点函数本身，而是把传入的 values 当作该节点已经产生的输出，执行该节点对应的 writers，应用状态写入和可能存在的路由写入，然后生成一个新的检查点。

因此，在本例中可以看到：router_node 的节点逻辑没有被执行，但它挂载的路由函数被执行，状态更新也已经生效。

6.5.4.2.4.2. 重新运行

示例如下

python

graph.invoke(None, skip_router_config)

运行结果如下

json

2026-06-11 16:49:42.501 | INFO     | __main__:node_default:73 - node_default 已执行
{
  "username": "小黄",
  "user_input": "帮我写一首关于布偶猫的七言绝句",
  "output": "无法处理的任务类型"
}

和分析一致，触发了兜底分支。

这种方式常用于测试场景。例如：

跳过不稳定的大模型路由节点。
手动指定路由结果，测试不同分支。
修改历史状态，验证后续节点能否正常运行。
基于同一个历史检查点探索多条可能的执行路径。

不过在正式业务流程中，应谨慎使用这种方式。因为它是在伪造某个节点的输出，如果伪造的状态不符合后续节点的预期，可能导致逻辑混乱，甚至抛出运行时异常。

6. 持久化机制和可恢复执行（下）：使用场景 ​

6.5. 使用场景 ​

6.5.1. 多轮对话 ​

6.5.2. 中断恢复 ​

6.5.3. 失败后恢复运行 ​

6.5.3.1. 用法说明 ​

6.5.3.2. 示例 ​

6.5.3.2.1. 模拟失败 ​

6.5.3.2.2. 查看历史检查点列表 ​

6.5.3.2.3. 修复BUG之后恢复运行 ​

6.5.4. Time Travel ​

6.5.4.1. Replay ​

6.5.4.1.1. 正常运行的任务 ​

6.5.4.1.2. 查看历史检查点列表 ​

6.5.4.1.3. 重放 ​

6.5.4.2. Fork ​

6.5.4.2.1. 用法说明 ​

6.5.4.2.2. 准备检查点历史 ​

6.5.4.2.2.1. 准备运行图并运行一次 ​

6.5.4.2.2.2. 获取历史检查点列表 ​

6.5.4.2.2.3. 获取 router_node 之前的检查点 ​

6.5.4.2.3. 只改状态 ​

6.5.4.2.3.1. 更新配置 ​

6.5.4.2.3.2. 重新运行 ​

6.5.4.2.4. 改状态且跳过某些节点 ​

6.5.4.2.4.1. 更新配置 ​

6.5.4.2.4.2. 重新运行 ​

6. 持久化机制和可恢复执行（下）：使用场景

6.5. 使用场景

6.5.1. 多轮对话

6.5.2. 中断恢复

6.5.3. 失败后恢复运行

6.5.3.1. 用法说明

6.5.3.2. 示例

6.5.3.2.1. 模拟失败

6.5.3.2.2. 查看历史检查点列表

6.5.3.2.3. 修复BUG之后恢复运行

6.5.4. Time Travel

6.5.4.1. Replay

6.5.4.1.1. 正常运行的任务

6.5.4.1.2. 查看历史检查点列表

6.5.4.1.3. 重放

6.5.4.2. Fork

6.5.4.2.1. 用法说明

6.5.4.2.2. 准备检查点历史

6.5.4.2.2.1. 准备运行图并运行一次

6.5.4.2.2.2. 获取历史检查点列表

6.5.4.2.2.3. 获取 `router_node` 之前的检查点

6.5.4.2.3. 只改状态

6.5.4.2.3.1. 更新配置

6.5.4.2.3.2. 重新运行

6.5.4.2.4. 改状态且跳过某些节点

6.5.4.2.4.1. 更新配置

6.5.4.2.4.2. 重新运行