使用时间旅行 - Docs by LangChain中文

概览

LangGraph 通过检查点支持时间旅行：

重放：从之前的检查点重试。
分叉：从之前的检查点分支出来，并修改状态以探索替代路径。

两者都是通过从之前的检查点恢复来工作的。检查点之前的节点不会重新执行（结果已经保存）。检查点之后的节点会重新执行，包括任何 LLM 调用、API 请求和中断（这些可能会产生不同的结果）。

重放

使用之前检查点的配置调用图，即可从该点开始重放。

重放会重新执行节点——它不只是从缓存中读取。LLM 调用、API 请求和中断会再次触发，并且可能返回不同的结果。从最终检查点（没有 next 节点）重放不会产生任何操作。

使用 getStateHistory 找到你想从中重放的检查点，然后使用该检查点的配置调用 invoke：

import { v7 as uuid7 } from "uuid";
import { StateGraph, MemorySaver, START } from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
  topic: Annotation<string>(),
  joke: Annotation<string>(),
});

function generateTopic(state: typeof StateAnnotation.State) {
  return { topic: "socks in the dryer" };
}

function writeJoke(state: typeof StateAnnotation.State) {
  return { joke: `Why do ${state.topic} disappear? They elope!` };
}

const checkpointer = new MemorySaver();
const graph = new StateGraph(StateAnnotation)
  .addNode("generateTopic", generateTopic)
  .addNode("writeJoke", writeJoke)
  .addEdge(START, "generateTopic")
  .addEdge("generateTopic", "writeJoke")
  .compile({ checkpointer });

// Step 1: Run the graph
const config = { configurable: { thread_id: uuid7() } };
const result = await graph.invoke({}, config);

// Step 2: Find a checkpoint to replay from
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}

// Step 3: Replay from a specific checkpoint
const beforeJoke = states.find((s) => s.next.includes("writeJoke"));
const replayResult = await graph.invoke(null, beforeJoke.config);
// writeJoke re-executes (runs again), generateTopic does not

分叉

分叉会基于过去的检查点创建一个带有修改后状态的新分支。对之前的检查点调用 update_state 来创建分叉，然后使用 None 调用 invoke 继续执行。

update_state 不会回滚线程。它会创建一个从指定点分支出来的新检查点。原始执行历史会保持不变。

// Find checkpoint before writeJoke
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}
const beforeJoke = states.find((s) => s.next.includes("writeJoke"));

// Fork: update state to change the topic
const forkConfig = await graph.updateState(
  beforeJoke.config,
  { topic: "chickens" },
);

// Resume from the fork — writeJoke re-executes with the new topic
const forkResult = await graph.invoke(null, forkConfig);
console.log(forkResult.joke); // A joke about chickens, not socks

从特定节点开始

当你调用 update_state 时，值会使用指定节点的写入器进行应用（包括 reducers）。检查点会记录该节点生成了此次更新，并且执行会从该节点的后继节点继续。默认情况下，LangGraph 会根据检查点的版本历史推断 as_node。从特定检查点分叉时，这种推断几乎总是正确的。在以下情况下需要显式指定 as_node：

并行分支：多个节点在同一步中更新了状态，而 LangGraph 无法确定哪个节点是最后更新的（InvalidUpdateError）。
没有执行历史：在一个新线程上设置状态（常见于测试）。
跳过节点：将 as_node 设置为更靠后的节点，让图认为该节点已经运行过。

// graph: generateTopic -> writeJoke

// Treat this update as if generateTopic produced it.
// Execution resumes at writeJoke (the successor of generateTopic).
const forkConfig = await graph.updateState(
  beforeJoke.config,
  { topic: "chickens" },
  { asNode: "generateTopic" },
);

中断

如果你的图使用 interrupt 来实现人在回路中工作流，那么在时间旅行期间，中断总是会被重新触发。包含该中断的节点会重新执行，而 interrupt() 会暂停并等待新的 Command(resume=...)。

import { interrupt, Command } from "@langchain/langgraph";

function askHuman(state: { value: string[] }) {
  const answer = interrupt("What is your name?");
  return { value: [`Hello, ${answer}!`] };
}

function finalStep(state: { value: string[] }) {
  return { value: ["Done"] };
}

// ... build graph with checkpointer ...

// First run: hits interrupt
await graph.invoke({ value: [] }, config);
// Resume with answer
await graph.invoke(new Command({ resume: "Alice" }), config);

// Replay from before askHuman
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}
const beforeAsk = states.filter((s) => s.next.includes("askHuman")).pop();

const replayResult = await graph.invoke(null, beforeAsk.config);
// Pauses at interrupt — waiting for new Command({ resume: ... })

// Fork from before askHuman
const forkConfig = await graph.updateState(beforeAsk.config, { value: ["forked"] });
const forkResult = await graph.invoke(null, forkConfig);
// Pauses at interrupt — waiting for new Command({ resume: ... })

// Resume the forked interrupt with a different answer
await graph.invoke(new Command({ resume: "Bob" }), forkConfig);
// Result: { value: ["forked", "Hello, Bob!", "Done"] }

多个中断

如果你的图在多个位置收集输入（例如，多步骤表单），你可以从这些中断之间的位置进行分叉，从而修改后面的回答，而不必重新询问前面的问题。

// Fork from BETWEEN the two interrupts (after askName, before askAge)
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}
const between = states.filter((s) => s.next.includes("askAge")).pop();

const forkConfig = await graph.updateState(between.config, { value: ["modified"] });
const result = await graph.invoke(null, forkConfig);
// askName result preserved ("name:Alice")
// askAge pauses at interrupt — waiting for new answer

子图

使用子图进行时间旅行时，具体行为取决于子图是否有自己的检查点保存器。这决定了你可以从多细的检查点粒度开始时间旅行。

Inherited checkpointer (default)
Subgraph checkpointer

默认情况下，子图会继承父图的检查点保存器。父图会将整个子图视为一个单一超步——整个子图执行过程中，在父级别只有一个检查点。从子图之前的位置进行时间旅行时，会从头重新执行该子图。你无法时间旅行到默认子图中节点之间的某个位置——你只能从父级别进行时间旅行。

// Subgraph without its own checkpointer (default)
const subgraph = new StateGraph(StateAnnotation)
  .addNode("stepA", stepA)       // Has interrupt()
  .addNode("stepB", stepB)       // Has interrupt()
  .addEdge(START, "stepA")
  .addEdge("stepA", "stepB")
  .compile();  // No checkpointer — inherits from parent

const graph = new StateGraph(StateAnnotation)
  .addNode("subgraphNode", subgraph)
  .addEdge(START, "subgraphNode")
  .compile({ checkpointer });

// Complete both interrupts
await graph.invoke({ value: [] }, config);
await graph.invoke(new Command({ resume: "Alice" }), config);
await graph.invoke(new Command({ resume: "30" }), config);

// Time travel from before the subgraph
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}
const beforeSub = states.filter((s) => s.next.includes("subgraphNode")).pop();

const forkConfig = await graph.updateState(beforeSub.config, { value: ["forked"] });
const result = await graph.invoke(null, forkConfig);
// The entire subgraph re-executes from scratch
// You cannot time travel to a point between stepA and stepB

在子图上设置 checkpointer=True，为它提供自己的检查点历史。这会在子图内部的每一步创建检查点，让你可以从子图内部的特定位置开始时间旅行——例如，在两个中断之间。使用带有 subgraphs=True 的 get_state 来访问子图自己的检查点配置，然后从它进行分叉：

// Subgraph with its own checkpointer
const subgraph = new StateGraph(StateAnnotation)
  .addNode("stepA", stepA)       // Has interrupt()
  .addNode("stepB", stepB)       // Has interrupt()
  .addEdge(START, "stepA")
  .addEdge("stepA", "stepB")
  .compile({ checkpointer: true });  // Own checkpoint history

const graph = new StateGraph(StateAnnotation)
  .addNode("subgraphNode", subgraph)
  .addEdge(START, "subgraphNode")
  .compile({ checkpointer });

// Run until stepA interrupt, then resume -> hits stepB interrupt
await graph.invoke({ value: [] }, config);
await graph.invoke(new Command({ resume: "Alice" }), config);

// Get the subgraph's own checkpoint (between stepA and stepB)
const parentState = await graph.getState(config, { subgraphs: true });
const subConfig = parentState.tasks[0].state.config;

// Fork from the subgraph checkpoint
const forkConfig = await graph.updateState(subConfig, { value: ["forked"] });
const result = await graph.invoke(null, forkConfig);
// stepB re-executes, stepA's result is preserved

更多关于配置子图检查点保存器的内容，请参阅子图持久化。

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

​概览

​重放

​分叉

​从特定节点开始

​中断

​多个中断

​子图

概览

重放

分叉

从特定节点开始

中断

多个中断

子图