LangGraph provides time travel functionality to resume execution from a prior checkpoint — either replaying the same state or modifying it to explore alternatives. In all cases, resuming past execution produces a new fork in the history.
Run the graph with initial inputs using LangGraph SDK's or APIs.
Identify a checkpoint in an existing thread: Use method to retrieve the execution history for a specific thread_id and locate the desired checkpoint_id.
Alternatively, set a breakpoint before the node(s) where you want execution to pause. You can then find the most recent checkpoint recorded up to that breakpoint.
(Optional) modify the graph state: Use the method to modify the graph’s state at the checkpoint and resume execution from alternative state.
Resume execution from the checkpoint: Use the or APIs with an input of None and the appropriate thread_id and checkpoint_id.
fromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,ENDfromlangchain.chat_modelsimportinit_chat_modelfromlanggraph.checkpoint.memoryimportInMemorySaverclassState(TypedDict):topic:NotRequired[str]joke:NotRequired[str]llm=init_chat_model("anthropic:claude-3-7-sonnet-latest",temperature=0,)defgenerate_topic(state:State):"""LLM call to generate a topic for the joke"""msg=llm.invoke("Give me a funny topic for a joke")return{"topic":msg.content}defwrite_joke(state:State):"""LLM call to write a joke based on the topic"""msg=llm.invoke(f"Write a short joke about {state['topic']}")return{"joke":msg.content}# Build workflowbuilder=StateGraph(State)# Add nodesbuilder.add_node("generate_topic",generate_topic)builder.add_node("write_joke",write_joke)# Add edges to connect nodesbuilder.add_edge(START,"generate_topic")builder.add_edge("generate_topic","write_joke")# Compilegraph=builder.compile()
fromlanggraph_sdkimportget_clientclient=get_client(url=<DEPLOYMENT_URL>)# Using the graph deployed with the name "agent"assistant_id="agent"# create a threadthread=awaitclient.threads.create()thread_id=thread["thread_id"]# Run the graphresult=awaitclient.runs.wait(thread_id,assistant_id,input={})
import{Client}from"@langchain/langgraph-sdk";constclient=newClient({apiUrl:<DEPLOYMENT_URL>});// Using the graph deployed with the name "agent"constassistantID="agent";// create a threadconstthread=awaitclient.threads.create();constthreadID=thread["thread_id"];// Run the graphconstresult=awaitclient.runs.wait(threadID,assistantID,{input:{}});
# The states are returned in reverse chronological order.states=awaitclient.threads.get_history(thread_id)selected_state=states[1]print(selected_state)
// The states are returned in reverse chronological order.conststates=awaitclient.threads.getHistory(threadID);constselectedState=states[1];console.log(selectedState);