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。

​
在工作流程中使用时间旅行

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-5-20250929",
    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
  • JavaScript
  • cURL
from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_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={}
)

​
2. Identify a checkpoint

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

​
3. Update the state

update_state will create a new checkpoint. The new checkpoint will be associated with the same thread, but a new checkpoint ID.
  • Python
  • JavaScript
  • cURL
new_config = await client.threads.update_state(
    thread_id,
    {"topic": "chickens"},
    checkpoint_id=selected_state["checkpoint_id"]
)
print(new_config)

​
4. Resume execution from the checkpoint

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

​
Learn more

Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.