LangChain 中文文档

外观

概述 ​

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

  • 重放：从先前的检查点重试。
  • 派生（Fork）：从带有修改状态的前检查点分支，以探索替代路径。

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

重放 ​

使用先前检查点的配置调用图，以从该点重放。

WARNING

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

Replay

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

python
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict, NotRequired
from langchain_core.utils.uuid import uuid7

class State(TypedDict):
    topic: NotRequired[str]
    joke: NotRequired[str]

def generate_topic(state: State):
    return {"topic": "socks in the dryer"}

def write_joke(state: State):
    return {"joke": f"Why do {state['topic']} disappear? They elope!"}

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(checkpointer=checkpointer)
)

# 步骤 1：运行图
config = {"configurable": {"thread_id": str(uuid7())}}
result = graph.invoke({}, config)

# 步骤 2：找到要重放的检查点
history = list(graph.get_state_history(config))
# 历史记录按时间倒序排列
for state in history:
    print(f"next={state.next}, checkpoint_id={state.config['configurable']['checkpoint_id']}")

# 步骤 3：从特定检查点重放
# 找到 write_joke 之前的检查点
before_joke = next(s for s in history if s.next == ("write_joke",))
replay_result = graph.invoke(None, before_joke.config)
# write_joke 会重新执行（再次运行），generate_topic 不会

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

typescript
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 });

// 步骤 1：运行图
const config = { configurable: { thread_id: uuid7() } };
const result = await graph.invoke({}, config);

// 步骤 2：找到要重放的检查点
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}

// 步骤 3：从特定检查点重放
const beforeJoke = states.find((s) => s.next.includes("writeJoke"));
const replayResult = await graph.invoke(null, beforeJoke.config);
// writeJoke 会重新执行（再次运行），generateTopic 不会

派生（Fork） ​

派生（Fork）会从带有修改状态的过去检查点创建新分支。在先前检查点上调用 update_state 创建派生，然后用 None 调用 invoke 以继续执行。

Fork

WARNING

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

python
# 找到 write_joke 之前的检查点
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": "chickens"},
)

# 从派生点恢复——write_joke 使用新主题重新执行
fork_result = graph.invoke(None, fork_config)
print(fork_result["joke"])  # 一个关于鸡的笑话，而不是袜子
typescript
// 找到 writeJoke 之前的检查点
const states = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}
const beforeJoke = states.find((s) => s.next.includes("writeJoke"));

// 派生：更新状态以更改主题
const forkConfig = await graph.updateState(
  beforeJoke.config,
  { topic: "chickens" },
);

// 从派生点恢复——writeJoke 使用新主题重新执行
const forkResult = await graph.invoke(null, forkConfig);
console.log(forkResult.joke); // 一个关于鸡的笑话，而不是袜子

从特定节点 ​

当你调用 update_state 时，值会使用指定节点的写入器（包括 reducer）应用。检查点会记录该节点产生了这次更新，并从该节点的后继节点恢复执行。

默认情况下，LangGraph 会从检查点的版本历史推断 as_node。当从特定检查点派生时，这种推断几乎总是正确的。

在以下情况下请显式指定 as_node：

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

# 将此次更新视为由 generate_topic 产生。
# 执行在 write_joke（generate_topic 的后继节点）处恢复。
fork_config = graph.update_state(
    before_joke.config,
    values={"topic": "chickens"},
    as_node="generate_topic",
)
typescript
// 图：generateTopic -> writeJoke

// 将此次更新视为由 generateTopic 产生。
// 执行在 writeJoke（generateTopic 的后继节点）处恢复。
const forkConfig = await graph.updateState(
  beforeJoke.config,
  { topic: "chickens" },
  { asNode: "generateTopic" },
);

中断 ​

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

python
from langgraph.types import interrupt, Command

class State(TypedDict):
    value: list[str]

def ask_human(state: State):
    answer = interrupt("What is your name?")
    return {"value": [f"Hello, {answer}!"]}

def final_step(state: State):
    return {"value": ["Done"]}

graph = (
    StateGraph(State)
    .add_node("ask_human", ask_human)
    .add_node("final_step", final_step)
    .add_edge(START, "ask_human")
    .add_edge("ask_human", "final_step")
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

# 首次运行：命中 interrupt
graph.invoke({"value": []}, config)
# 用答案恢复
graph.invoke(Command(resume="Alice"), config)

# 从 ask_human 之前重放
history = list(graph.get_state_history(config))
before_ask = [s for s in history if s.next == ("ask_human",)][-1]

replay_result = graph.invoke(None, before_ask.config)
# 在 interrupt 处暂停——等待新的 Command(resume=...)

# 从 ask_human 之前派生
fork_config = graph.update_state(before_ask.config, {"value": ["forked"]})
fork_result = graph.invoke(None, fork_config)
# 在 interrupt 处暂停——等待新的 Command(resume=...)

# 用不同的答案恢复派生的 interrupt
graph.invoke(Command(resume="Bob"), fork_config)
# 结果：{"value": ["forked", "Hello, Bob!", "Done"]}
typescript
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"] };
}

// ... 使用检查点构建图 ...

// 首次运行：命中 interrupt
await graph.invoke({ value: [] }, config);
// 用答案恢复
await graph.invoke(new Command({ resume: "Alice" }), config);

// 从 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);
// 在 interrupt 处暂停——等待新的 Command({ resume: ... })

// 从 askHuman 之前派生
const forkConfig = await graph.updateState(beforeAsk.config, { value: ["forked"] });
const forkResult = await graph.invoke(null, forkConfig);
// 在 interrupt 处暂停——等待新的 Command({ resume: ... })

// 用不同的答案恢复派生的 interrupt
await graph.invoke(new Command({ resume: "Bob" }), forkConfig);
// 结果：{ value: ["forked", "Hello, Bob!", "Done"] }

多个中断 ​

如果你的图在多个点收集输入（例如多步表单），你可以从各中断之间派生，从而更改后面的答案，而无需重新询问前面的问题。

python
def ask_name(state):
    name = interrupt("What is your name?")
    return {"value": [f"name:{name}"]}

def ask_age(state):
    age = interrupt("How old are you?")
    return {"value": [f"age:{age}"]}

# 图：ask_name -> ask_age -> final
# 完成两个 interrupt 之后：

# 从两个 interrupt 之间派生（在 ask_name 之后、ask_age 之前）
history = list(graph.get_state_history(config))
between = [s for s in history if s.next == ("ask_age",)][-1]

fork_config = graph.update_state(between.config, {"value": ["modified"]})
result = graph.invoke(None, fork_config)
# ask_name 的结果被保留（"name:Alice"）
# ask_age 在 interrupt 处暂停——等待新的答案
typescript
// 从两个 interrupt 之间派生（在 askName 之后、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 的结果被保留（"name:Alice"）
// askAge 在 interrupt 处暂停——等待新的答案

子图 ​

使用子图进行时间旅行取决于子图是否拥有自己的检查点。这决定了你能够从何种粒度的检查点进行时间旅行。

继承的检查点（默认） ​

默认情况下，子图继承父图的检查点。父图将整个子图视为一个单一超步——整个子图执行只有父级一个检查点。从子图之前进行时间旅行会从头重新执行它。

在默认的子图中，你无法时间旅行到节点之间的某个点——你只能从父级进行时间旅行。

python
# 没有自己检查点的子图（默认）
subgraph = (
    StateGraph(State)
    .add_node("step_a", step_a)       # 包含 interrupt()
    .add_node("step_b", step_b)       # 包含 interrupt()
    .add_edge(START, "step_a")
    .add_edge("step_a", "step_b")
    .compile()  # 没有检查点——继承自父图
)

graph = (
    StateGraph(State)
    .add_node("subgraph_node", subgraph)
    .add_edge(START, "subgraph_node")
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

# 完成两个 interrupt
graph.invoke({"value": []}, config)            # 命中 step_a 的 interrupt
graph.invoke(Command(resume="Alice"), config)  # 命中 step_b 的 interrupt
graph.invoke(Command(resume="30"), config)     # 完成

# 从子图之前进行时间旅行
history = list(graph.get_state_history(config))
before_sub = [s for s in history if s.next == ("subgraph_node",)][-1]

fork_config = graph.update_state(before_sub.config, {"value": ["forked"]})
result = graph.invoke(None, fork_config)
# 整个子图从头重新执行
# 无法时间旅行到 step_a 与 step_b 之间的某个点
typescript
// 没有自己检查点的子图（默认）
const subgraph = new StateGraph(StateAnnotation)
  .addNode("stepA", stepA)       // 包含 interrupt()
  .addNode("stepB", stepB)       // 包含 interrupt()
  .addEdge(START, "stepA")
  .addEdge("stepA", "stepB")
  .compile();  // 没有检查点——继承自父图

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

// 完成两个 interrupt
await graph.invoke({ value: [] }, config);
await graph.invoke(new Command({ resume: "Alice" }), config);
await graph.invoke(new Command({ resume: "30" }), config);

// 从子图之前进行时间旅行
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);
// 整个子图从头重新执行
// 无法时间旅行到 stepA 与 stepB 之间的某个点

子图检查点 ​

在子图上设置 checkpointer=True，让它拥有自己的检查点历史。这会在子图内部的每个步骤创建检查点，让你能够从其中的特定点进行时间旅行——例如，在两个中断之间。

使用 get_state 并传入 subgraphs=True 来访问子图自己的检查点配置，然后从它派生：

python
# 拥有自己检查点的子图
subgraph = (
    StateGraph(State)
    .add_node("step_a", step_a)       # 包含 interrupt()
    .add_node("step_b", step_b)       # 包含 interrupt()
    .add_edge(START, "step_a")
    .add_edge("step_a", "step_b")
    .compile(checkpointer=True)  # 自己的检查点历史
)

graph = (
    StateGraph(State)
    .add_node("subgraph_node", subgraph)
    .add_edge(START, "subgraph_node")
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

# 运行到 step_a 的 interrupt
graph.invoke({"value": []}, config)

# 恢复 step_a -> 命中 step_b 的 interrupt
graph.invoke(Command(resume="Alice"), config)

# 获取子图自己的检查点（在 step_a 与 step_b 之间）
parent_state = graph.get_state(config, subgraphs=True)
sub_config = parent_state.tasks[0].state.config

# 从子图检查点派生
fork_config = graph.update_state(sub_config, {"value": ["forked"]})
result = graph.invoke(None, fork_config)
# step_b 重新执行，step_a 的结果被保留
typescript
// 拥有自己检查点的子图
const subgraph = new StateGraph(StateAnnotation)
  .addNode("stepA", stepA)       // 包含 interrupt()
  .addNode("stepB", stepB)       // 包含 interrupt()
  .addEdge(START, "stepA")
  .addEdge("stepA", "stepB")
  .compile({ checkpointer: true });  // 自己的检查点历史

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

// 运行到 stepA 的 interrupt，然后恢复 -> 命中 stepB 的 interrupt
await graph.invoke({ value: [] }, config);
await graph.invoke(new Command({ resume: "Alice" }), config);

// 获取子图自己的检查点（在 stepA 与 stepB 之间）
const parentState = await graph.getState(config, { subgraphs: true });
const subConfig = parentState.tasks[0].state.config;

// 从子图检查点派生
const forkConfig = await graph.updateState(subConfig, { value: ["forked"] });
const result = await graph.invoke(null, forkConfig);
// stepB 重新执行，stepA 的结果被保留

有关配置子图检查点的更多信息，请参阅子图持久化。