LangGraph 09：时光旅行

发表于 2026-08-19 更新于 2026-08-25 分类于 ai

基于 Checkpoint 做错误恢复、replay 与 fork，从历史快照续跑或改输入走出新时间线。

前言

有了 Checkpoint，图就不再是「跑完即弃」：可以从历史快照续跑、改输入再分叉，甚至在并行失败后只重跑出错分支。
本文只讲检查点时间线上的操作：错误恢复、invoke(None) 回放、update_state 分叉。
跨线程长期记忆见下一篇《LangGraph 10：Store 与 Context》。
聊天模型对接 火山方舟 Coding Plan。
下文需要 Python 3.12+，依赖用 uv 管理。

概要

本文只在同一条检查点时间线上操作。
先讲出错后如何恢复。
再用 invoke(None) 做 replay 回放。
最后用 update_state 改输入或跳过节点，走出 fork 分叉。
跨线程长期记忆留给下一篇 Store。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明依赖。

uv init langgraph-timetravel
cd langgraph-timetravel
uv venv --python 3.12
uv add "langchain>=1.0,<2.0" langchain-openai langchain-core langgraph python-dotenv rich

在项目根目录创建 .env：

1 2	OPENAI_API_KEY=你的火山方舟 API Key OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

不要把 .env 提交进 Git。
Saver 与 thread_id 的基本用法见《LangGraph 08：Checkpoint 持久化》。

示例

错误恢复思路

并行图中，同一超步里可能一边成功、一边抛错。
启用 checkpointer 后，成功节点的结果往往已写入检查点；修好出错逻辑后，对同一 Saver 与 thread_id 再 invoke(None, config=...)，可从中断处续跑。

下面用可变开关模拟：第一次 node_joke 失败；关掉开关后空输入续跑。
请保持同一个 InMemorySaver 实例；换进程或新进程时，内存检查点不会自动保留，需要 Postgres 等持久化后端。

import os
from typing import TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.3,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

flags = {"fail_joke": True}


class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str


class InputState(TypedDict):
    topic: str


class OutputState(TypedDict):
    final_output: str


def node_change_topic(state: InputState) -> OverAllState:
    return {"topic": f"{state['topic']}:布偶猫"}


def node_poem(state: OverAllState) -> OverAllState:
    poem = model.invoke(
        [HumanMessage(f"写一首关于{state['topic']}的七言绝句，只输出诗句。")]
    ).content
    return {"poem": poem}


def node_joke(state: OverAllState) -> OverAllState:
    if flags["fail_joke"]:
        raise RuntimeError("人为抛异常：笑话节点失败")
    joke = model.invoke(
        [HumanMessage(f"写一个关于{state['topic']}的笑话，只输出正文。")]
    ).content
    return {"joke": joke}


def node_output(state: OverAllState) -> OutputState:
    return {
        "final_output": (
            f"主题:{state['topic']}\n诗:{state['poem']}\n笑话:{state['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)

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

try:
    graph.invoke({"topic": "猫"}, config=config)
except RuntimeError as exc:
    rprint(f"首次失败: {exc}")
    snap = graph.get_state(config)
    rprint("已写入 keys:", [k for k in snap.values if snap.values.get(k)])
    rprint("next:", snap.next)

flags["fail_joke"] = False
rprint(graph.invoke(None, config=config))

排障流程通常是：get_state_history → 定位失败超步 → 修逻辑或数据 → invoke(None)。
跨进程恢复请改用 Postgres 等持久化 Saver，而不是新建空的 InMemorySaver。

Replay 回放

回放：选中历史里某个检查点的 config，再 invoke(None, config=该检查点)。
输入填 None 表示「不要新用户输入，从该快照的 next 继续」。

import os
from typing import TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.3,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str


class InputState(TypedDict):
    topic: str


class OutputState(TypedDict):
    final_output: str


topics = ["布偶猫", "狸花猫", "金渐层"]
topic_index = 0


def node_change_topic(state: InputState) -> OverAllState:
    global topic_index
    sub = topics[topic_index % len(topics)]
    topic_index += 1
    return {"topic": f"{state['topic']}:{sub}"}


def node_poem(state: OverAllState) -> OverAllState:
    poem = model.invoke(
        [HumanMessage(f"写一首关于{state['topic']}的七言绝句，只输出诗句。")]
    ).content
    return {"poem": poem}


def node_joke(state: OverAllState) -> OverAllState:
    joke = model.invoke(
        [HumanMessage(f"写一个关于{state['topic']}的笑话，只输出正文。")]
    ).content
    return {"joke": joke}


def node_output(state: OverAllState) -> OutputState:
    return {
        "final_output": (
            f"主题:{state['topic']}\n诗:{state['poem']}\n笑话:{state['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)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "replay-demo"}}
rprint(graph.invoke({"topic": "猫咪"}, config=config))

history = list(graph.get_state_history(config=config))
target = next(h for h in history if h.next == ("node_poem", "node_joke"))
rprint("回放到:", target.next)
rprint(graph.invoke(None, config=target.config))

回放会再次执行 next 指向的节点，适合复现或对比生成结果。
带副作用的节点（写库、扣费）回放前要自己做幂等。

Fork 分叉

分叉：在某一历史点用 update_state 改状态，再 invoke(None) 走出另一条时间线。
as_node 表示「这些更新视为刚从该节点执行完」，从而影响后续 next。

import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class OverAllState(TypedDict):
    username: str
    user_input: str
    topic: str
    mode: Literal["poem", "joke"]
    output: str


def router_node(state: OverAllState) -> OverAllState:
    text = state["user_input"]
    if "笑话" in text:
        return {"topic": "荷花", "mode": "joke"}
    return {"topic": "荷花", "mode": "poem"}


def route(state: OverAllState) -> Literal["node_poem", "node_joke"]:
    return "node_joke" if state.get("mode") == "joke" else "node_poem"


def node_poem(state: OverAllState) -> OverAllState:
    out = model.invoke(
        [HumanMessage(f"写一首关于{state['topic']}的七言绝句，只输出诗句。")]
    ).content
    return {"output": out}


def node_joke(state: OverAllState) -> OverAllState:
    out = model.invoke(
        [HumanMessage(f"写一个关于{state['topic']}的笑话，只输出正文。")]
    ).content
    return {"output": out}


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_edge(START, "router_node")
builder.add_conditional_edges(
    "router_node", route, path_map=["node_poem", "node_joke"]
)
builder.add_edge("node_poem", END)
builder.add_edge("node_joke", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "fork-demo"}}
rprint(
    graph.invoke(
        {"username": "小王", "user_input": "写一首关于荷花的诗"},
        config=config,
    )
)

history = list(graph.get_state_history(config=config))
before_router = next(h for h in history if h.next == ("router_node",))

# 改输入后从 START 视角分叉
forked = graph.update_state(
    config=before_router.config,
    values={"user_input": "帮我写一个荷花的笑话"},
    as_node=START,
)
rprint(graph.invoke(None, config=forked))

# 跳过路由：假装 router 已产出 topic/mode
skip_router = graph.update_state(
    config=before_router.config,
    values={"topic": "狸花猫", "mode": "joke"},
    as_node="router_node",
)
rprint(graph.invoke(None, config=skip_router))

as_node=START 适合「改用户输入再整段重跑」。
as_node="router_node" 适合「跳过路由、直接指定下游所需字段」。

验证

恢复：第一次失败后关掉开关再 invoke(None)，应补上笑话并产出 final_output。
Replay：从 next == ("node_poem", "node_joke") 的检查点空输入续跑，应再次生成诗与笑话。
Fork：改 user_input 含「笑话」后，输出应走笑话分支。

总结

恢复 / 回放：同一 checkpointer + invoke(None, config=检查点)。
分叉：update_state(..., as_node=...) 再续跑，走出新时间线。
下一篇换一条记忆轴：跨线程 Store 与请求级 Context，见《LangGraph 10：Store 与 Context》。