01Real-world example

A checkpointer saves the state after every step. Each thread_id keeps its own history.


compile(checkpointer=InMemorySaver()) saves the state for the input and after every step.thread_id picks which saved state to continue: nightly-regression and smoke never see each other's runs.get_state() returns the latest snapshot; get_state_history() returns every checkpoint.InMemorySaver forgets when the process exits. A SQLite or Postgres saver keeps the history across restarts.cd chapter_18_LangGraph/src/chapters
python 006_Checkpointer_Memory.py
nightly-regression remembers: ['run #1', 'run #2', 'run #3']
smoke is separate: ['run #1']
Work in chapter_18_LangGraph/src/chapters. Change one thing, run it, and compare with the expected result.
Add a release thread and invoke it twice.
['run #1', 'run #2']; nightly still remembers three runs.After three invokes on one thread, how many entries does get_state_history() return?
START and record_run).Compile the same graph without a checkpointer and invoke it twice with the same config.
['run #1'] both times. No checkpointer, no memory.The exact file from the course repo, plus the output it printed when this page was written.
"""006 - Memory with a checkpointer.
Compile with a checkpointer and pass a thread_id. Each thread keeps its own
state between invoke() calls, like separate test sessions. Swap InMemorySaver
for a SQLite or Postgres saver and it survives restarts.
"""
import operator
from typing import Annotated, TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
class State(TypedDict):
runs: Annotated[list[str], operator.add]
def record_run(state: State) -> dict:
return {"runs": [f"run #{len(state.get('runs', [])) + 1}"]}
builder = StateGraph(State)
builder.add_node("record_run", record_run)
builder.add_edge(START, "record_run")
builder.add_edge("record_run", END)
app = builder.compile(checkpointer=InMemorySaver())
if __name__ == "__main__":
nightly = {"configurable": {"thread_id": "nightly-regression"}}
smoke = {"configurable": {"thread_id": "smoke"}}
app.invoke({"runs": []}, nightly)
app.invoke({"runs": []}, nightly)
out = app.invoke({"runs": []}, nightly)
print("nightly-regression remembers:", out["runs"])
out = app.invoke({"runs": []}, smoke)
print("smoke is separate: ", out["runs"])
snap = app.get_state(nightly)
print("\nget_state(nightly).values:", snap.values)
print("checkpoints saved for nightly:", len(list(app.get_state_history(nightly))))
nightly-regression remembers: ['run #1', 'run #2', 'run #3']
smoke is separate: ['run #1']