Going back in time
The checkpointer does not keep only the latest state. It keeps every one, so you can look at any moment of a past run, and start again from it.
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
class State(TypedDict):
topic: str
joke: strfrom langgraph.checkpoint.memory import InMemorySaver
builder = StateGraph(State)
builder.add_node("pick_topic", lambda s: {"topic": "cats"})
builder.add_node("write_joke", lambda s: {"joke": f"a joke about {s['topic']}"})
builder.add_edge(START, "pick_topic")
builder.add_edge("pick_topic", "write_joke")
graph = builder.compile(checkpointer=InMemorySaver())config = {"configurable": {"thread_id": "jokes"}}
print(graph.invoke({"topic": "", "joke": ""}, config)["joke"])Every moment of that run
Lesson 20 used get_state for the latest one. There is a second method that gives you all of them.
for saved in graph.get_state_history(config):
print(saved.next, saved.values)That gives you every saved moment of the thread, newest first, and each one knows what was about to run at the time.
Four moments. The last line is the empty state before anything ran, and the first is the finished one with nothing left to do.
Start again from the middle
Each of those carries its own config, and handing that config back to invoke picks the run up from there.
before_joke = next(s for s in graph.get_state_history(config) if s.next == ("write_joke",))
changed = graph.update_state(before_joke.config, {"topic": "dogs"})
print(graph.invoke(None, changed)["joke"])The first line finds the moment after the topic was picked and before the joke was written. update_state writes a change into that moment and hands back a config pointing at the new branch. Then None as the input means there is nothing new to add, so the config alone decides where to carry on from.
That run did not start from the beginning. It picked up the moment before write_joke, with the topic changed by hand, and only that node ran again.
What it is actually for
- Understanding a bad run. Look at the state at every step instead of guessing which node spoiled it.
- Trying a different answer. Change one value at the point things went wrong and see what would have happened.
- Undo. A user regrets a turn, so you go back to before it and carry on.
- Print
saved.config['configurable']['checkpoint_id']in the history loop. - Fork from before
pick_topicinstead and see both nodes run again. - Call
get_state_historyagain afterwards and count the moments now.
Little by little, you're building something great.