使用服务器 API 进行时间旅行 - 以编程方式使用文档

LangGraph 提供了 **时间旅行** 功能允许从之前的检查点恢复执行，可以重放相同状态或修改状态以探索替代方案。在所有情况下，从过去的执行点恢复都会在历史记录中产生一个新的分支。

要使用 LangSmith Deployment API（通过 LangGraph SDK）进行时间旅行：

1. **运行图** 使用初始输入通过 LangGraph SDK的 client.runs.wait 或 client.runs.stream API。 2. **识别现有线程中的检查点**：使用 client.threads.get_history 方法来检索特定 thread_id 的执行历史并定位所需的 checkpoint_id. 或者，在要暂停执行断点的节点之前设置一个断点。然后可以找到该断点之前记录的最新检查点。 3. **（可选）修改图状态**：使用 client.threads.update_state 方法来修改图在该检查点的状态，并从替代状态恢复执行。 4. **从检查点恢复执行**：使用 client.runs.wait 或 client.runs.stream API，并指定 None 输入和适当的 thread_id 和 checkpoint_id.

在工作流中使用时间旅行

Example graph

    from typing_extensions import TypedDict, NotRequired
    from langgraph.graph import StateGraph, START, END
    from langchain.chat_models import init_chat_model
    from langgraph.checkpoint.memory import InMemorySaver

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

    model = init_chat_model(
        "claude-sonnet-4-6",
        temperature=0,
    )

    def generate_topic(state: State):
        """LLM call to generate a topic for the joke"""
        msg = model.invoke("Give me a funny topic for a joke")
        return {"topic": msg.content}

    def write_joke(state: State):
        """LLM call to write a joke based on the topic"""
        msg = model.invoke(f"Write a short joke about {state['topic']}")
        return {"joke": msg.content}

    # Build workflow
    builder = StateGraph(State)

    # Add nodes
    builder.add_node("generate_topic", generate_topic)
    builder.add_node("write_joke", write_joke)

    # Add edges to connect nodes
    builder.add_edge(START, "generate_topic")
    builder.add_edge("generate_topic", "write_joke")

    # Compile
    graph = builder.compile()

1. 运行图

Python

    from langgraph_sdk import get_client
    client = get_client(url=)

    # Using the graph deployed with the name "agent"
    assistant_id = "agent"

    # create a thread
    thread = await client.threads.create()
    thread_id = thread["thread_id"]

    # Run the graph
    result = await client.runs.wait(
        thread_id,
        assistant_id,
        input={}
    )

JavaScript

    const client = new Client({ apiUrl:  });

    // Using the graph deployed with the name "agent"
    const assistantID = "agent";

    // create a thread
    const thread = await client.threads.create();
    const threadID = thread["thread_id"];

    // Run the graph
    const result = await client.runs.wait(
      threadID,
      assistantID,
      { input: {}}
    );

cURL

创建线程：

    curl --request POST \
    --url /threads \
    --header 'Content-Type: application/json' \
    --data '{}'

运行图：

    curl --request POST \
    --url /threads//runs/wait \
    --header 'Content-Type: application/json' \
    --data "{
      \"assistant_id\": \"agent\",
      \"input\": {}
    }"

2. 识别检查点

Python

    # The states are returned in reverse chronological order.
    states = await client.threads.get_history(thread_id)
    selected_state = states[1]
    print(selected_state)

JavaScript

    // The states are returned in reverse chronological order.
    const states = await client.threads.getHistory(threadID);
    const selectedState = states[1];
    console.log(selectedState);

cURL

    curl --request GET \
    --url /threads//history \
    --header 'Content-Type: application/json'

<a id="optional"></a> ### 3. 更新状态

@[update_state将创建一个新的检查点。新检查点将与同一线程关联，但具有新的检查点 ID。

Python

    new_config = await client.threads.update_state(
        thread_id,
        {"topic": "chickens"},
        checkpoint_id=selected_state["checkpoint_id"]
    )
    print(new_config)

JavaScript

    const newConfig = await client.threads.updateState(
      threadID,
      {
        values: { "topic": "chickens" },
        checkpointId: selectedState["checkpoint_id"]
      }
    );
    console.log(newConfig);

cURL

    curl --request POST \
    --url /threads//state \
    --header 'Content-Type: application/json' \
    --data "{
      \"assistant_id\": \"agent\",
      \"checkpoint_id\": ,
      \"values\": {\"topic\": \"chickens\"}
    }"

4. 从检查点恢复执行

Python

    await client.runs.wait(
        thread_id,
        assistant_id,
        input=None,
        checkpoint_id=new_config["checkpoint_id"]
    )

JavaScript

    await client.runs.wait(
      threadID,
      assistantID,
      {
        input: null,
        checkpointId: newConfig["checkpoint_id"]
      }
    );

cURL

    curl --request POST \
    --url /threads//runs/wait \
    --header 'Content-Type: application/json' \
    --data "{
      \"assistant_id\": \"agent\",
      \"checkpoint_id\":
    }"

了解更多

* **LangGraph 时间旅行指南**：了解更多关于在 LangGraph 中使用时间旅行的信息。