Checkpointing with MemorySaver

Capsule overview

Your Research Agent processes 5 research sources. It decomposes the query, searches the web, analyzes papers, pulls data from news, and synthesizes a report. The process takes 5 minutes and burns API tokens at every step. After processing 3 sources — crash. The process dies. Without checkpointing: you redo all 5 sources from scratch. Another 5 minutes. Double the API cost. The user waits all over again.

With checkpointing: the agent resumes from source 4. One minute. No repeated work. No duplicated cost.

Checkpointing saves your graph's full state after every node. It's an automatic snapshot: which nodes ran, what state they had, what messages were exchanged. If the process gets interrupted, you don't start from zero — you pick up from the last checkpoint.

In the previous capsule you saw short-term memory: how the graph keeps message history during a conversation. That solves continuity within a session. Checkpointing solves something different: the durability of that session. If the process dies, the memory dies with it — unless you have checkpointing.

MemorySaver is LangGraph's simplest checkpointer: it stores checkpoints in RAM. It's perfect for development and testing. No database, no configuration. One line of code and your graph has checkpointing. In the next capsule you'll migrate to PostgresSaver for production — and you'll see the change is literally one line.

The problem: state that disappears

Without checkpointing, every invocation of your graph is independent. The graph doesn't know it ran before. It has no prior context. Every graph.invoke() starts with an empty state.

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()

result1 = graph.invoke({"messages": [("user", "What is RAG?")]})
print(result1["messages"][-1].content)
# Output: "RAG (Retrieval-Augmented Generation) is a technique that..."

result2 = graph.invoke({"messages": [("user", "Give me examples")]})
print(result2["messages"][-1].content)
# Output: "Examples of what? I have no prior context..."

The second invoke knows nothing about the first. "Give me examples" has no referent — the agent doesn't know you were talking about RAG two seconds ago. Every invocation is a brand-new conversation from scratch.

MemorySaver: checkpointing in one line

MemorySaver stores your graph's state in memory after every node. You turn it on in two steps: create the checkpointer and pass it in at compile time.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user_001"}}

result1 = graph.invoke({"messages": [("user", "What is RAG?")]}, config)
print(result1["messages"][-1].content)
# Output: "RAG (Retrieval-Augmented Generation) is a technique that..."

result2 = graph.invoke({"messages": [("user", "Give me examples")]}, config)
print(result2["messages"][-1].content)
# Output: "Here are some RAG examples: 1) A chatbot that queries
#          internal documentation... 2) A legal assistant that looks up
#          relevant case law..."

Three changes from the version without checkpointing:

1. checkpointer = MemorySaver() — you create the checkpointer
2. graph_builder.compile(checkpointer=checkpointer) — you pass it in at compile time
3. config = {"configurable": {"thread_id": "user_001"}} — you identify the thread

Now the second invoke knows that "examples" refers to RAG. The checkpointer saved the full state after the first turn (user message + model response), and the second turn received it automatically.

How it works under the hood

When you compile with a checkpointer, LangGraph intercepts each node's execution and saves a snapshot of the state:

Execution with checkpointing:

[START] → Initial state saved as Checkpoint 0
   ↓
[chatbot] runs → Updated state saved as Checkpoint 1
   ↓
[END] → Final state saved as Checkpoint 2

Every checkpoint holds:
- The graph's complete state at that point
- The name of the node that produced that state
- A timestamp
- The parent checkpoint's ID (so you can walk the history)

The important part: a checkpoint stores the complete state, not a diff. Checkpoint 2 has all the messages, not just the last one. That means you can resume from any checkpoint without needing the ones before it.

A two-turn flow

Turn 1: "What is RAG?"
  1. Checkpointer looks for a checkpoint for thread "user_001" → none
  2. Node "chatbot" runs → model response
  3. Checkpointer saves: [user_msg, ai_msg]

Turn 2: "Give me examples"
  1. Checkpointer looks for a checkpoint → FINDS [user_msg_1, ai_msg_1]
  2. MERGE: previous state + new input = [user_msg_1, ai_msg_1, user_msg_2]
  3. Node "chatbot" runs with full context → coherent response
  4. Checkpointer saves: [user_msg_1, ai_msg_1, user_msg_2, ai_msg_2]

Steps 1-2 of turn 2 are where the magic happens: the checkpointer recovers the previous state and merges it with the new input. The chatbot node receives the whole conversation, not just the last message.

thread_id: isolation between conversations

The thread_id is mandatory once you use a checkpointer. It's the identifier that separates conversations. Without it, LangGraph doesn't know where to save the checkpoint or where to read it from.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config_alice = {"configurable": {"thread_id": "alice_session_001"}}
config_bob = {"configurable": {"thread_id": "bob_session_001"}}

graph.invoke({"messages": [("user", "I'm Alice. I research RAG.")]}, config_alice)
graph.invoke({"messages": [("user", "I'm Bob. I research fine-tuning.")]}, config_bob)

result_alice = graph.invoke({"messages": [("user", "What am I researching?")]}, config_alice)
print(f"Alice: {result_alice['messages'][-1].content}")
# Output: "You're researching RAG..."

result_bob = graph.invoke({"messages": [("user", "What am I researching?")]}, config_bob)
print(f"Bob: {result_bob['messages'][-1].content}")
# Output: "You're researching fine-tuning..."

Each thread_id has its own independent chain of checkpoints. Alice and Bob never mix. This is the foundation of multi-user support: in production, every user (or every session of a user) gets their own thread_id.

Common thread_id patterns

Inspecting checkpoints: seeing the graph's state

Checkpoints aren't only for continuity — they're a debugging tool. You can inspect the graph's exact state at any point in its execution.

get_state: see the current checkpoint

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "debug_session"}}
graph.invoke({"messages": [("user", "What is prompt engineering?")]}, config)
graph.invoke({"messages": [("user", "Give me 3 techniques")]}, config)

state_snapshot = graph.get_state(config)

print(f"Number of messages: {len(state_snapshot.values['messages'])}")
print(f"Last node executed: {state_snapshot.next}")
print(f"Config: {state_snapshot.config}")
print(f"Checkpoint ID: {state_snapshot.config['configurable']['checkpoint_id']}")

for msg in state_snapshot.values["messages"]:
    role = "User" if msg.type == "human" else "AI"
    preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
    print(f"  [{role}] {preview}")
# Expected output:
# Number of messages: 4
# Last node executed: ()
# Config: {'configurable': {'thread_id': 'debug_session', 'checkpoint_id': '...'}}
# Checkpoint ID: 1ef8a...
#   [User] What is prompt engineering?
#   [AI] Prompt engineering is the discipline of designing effective instructions...
#   [User] Give me 3 techniques
#   [AI] 1) Zero-shot prompting: give the instruction with no examples...

get_state() returns a StateSnapshot with:

• values: the graph's complete state
• next: tuple of pending nodes (empty if it finished)
• config: the configuration, including the checkpoint_id
• metadata: extra information about the checkpoint
• parent_config: a reference to the previous checkpoint

get_state_history: walking the full history

history = list(graph.get_state_history(config))

print(f"Total checkpoints: {len(history)}")
for i, snapshot in enumerate(history):
    n_msgs = len(snapshot.values.get("messages", []))
    created = snapshot.metadata.get("created_by", "unknown")
    step = snapshot.metadata.get("step", "?")
    print(f"  Checkpoint {i}: step={step}, messages={n_msgs}, created_by={created}")
# Expected output:
# Total checkpoints: 5
#   Checkpoint 0: step=3, messages=4, created_by=loop
#   Checkpoint 1: step=2, messages=3, created_by=loop
#   Checkpoint 2: step=1, messages=2, created_by=loop
#   Checkpoint 3: step=0, messages=1, created_by=loop
#   Checkpoint 4: step=-1, messages=0, created_by=system

The history comes back in reverse chronological order: most recent first. Each checkpoint corresponds to one step in the graph's execution. Checkpoint 4 (step=-1) is the initial state, before any node runs.

Resuming from a specific checkpoint (time-travel)

You can resume not just from the latest checkpoint — but from any of them. Grab the checkpoint you want from get_state_history(), take its .config, and pass that config to graph.invoke():

history = list(graph.get_state_history(config))
old_checkpoint = history[-3]  # checkpoint after turn 1
result = graph.invoke(
    {"messages": [("user", "New question")]},
    old_checkpoint.config  # resume from that point
)

This is time-travel debugging: you rewind to a specific point and create a branch. The original history stays intact. Invoking again from the old checkpoint creates a new branch. Exercises 4 and 5 in this capsule implement it in detail.

Checkpointing in multi-node graphs

Checkpointing pays off more the more nodes you have. Every node produces a checkpoint, so if the process is interrupted between nodes, you don't lose the work of the earlier ones.

from typing import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class ResearchState(TypedDict):
    query: str
    sources_searched: list[str]
    findings: list[str]
    report: str
    status: str


def decompose_query(state: ResearchState) -> dict:
    print("[1/4] Decomposing query...")
    return {"status": "decomposed"}


def search_sources(state: ResearchState) -> dict:
    sources = ["web", "papers", "news"]
    findings = []
    for source in sources:
        print(f"[2/4] Searching {source}...")
        findings.append(f"Finding from {source} about '{state['query']}'")
    return {
        "sources_searched": sources,
        "findings": findings,
        "status": "searched"
    }


def analyze_findings(state: ResearchState) -> dict:
    print(f"[3/4] Analyzing {len(state['findings'])} findings...")
    return {"status": "analyzed"}


def generate_report(state: ResearchState) -> dict:
    print("[4/4] Generating report...")
    report = f"Report: {len(state['findings'])} findings from {len(state['sources_searched'])} sources"
    return {"report": report, "status": "completed"}


builder = StateGraph(ResearchState)
builder.add_node("decompose", decompose_query)
builder.add_node("search", search_sources)
builder.add_node("analyze", analyze_findings)
builder.add_node("report", generate_report)
builder.add_edge(START, "decompose")
builder.add_edge("decompose", "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "research_001"}}
result = graph.invoke(
    {"query": "State of the art in RAG", "sources_searched": [], "findings": [], "report": "", "status": ""},
    config
)
# Output:
# [1/4] Decomposing query...
# [2/4] Searching web...
# [2/4] Searching papers...
# [2/4] Searching news...
# [3/4] Analyzing 3 findings...
# [4/4] Generating report...

history = list(graph.get_state_history(config))
print(f"Checkpoints created: {len(history)}")
# Expected output: Checkpoints created: 6
# (1 per node + 1 input + 1 system)

6 checkpoints for 4 nodes. If the process had died between search and analyze, you could pick up from the checkpoint with status=searched — without redoing the decomposition or the searches.

What a checkpoint contains

Every checkpoint stores enough information to rebuild the graph's exact state:

The critical limitation: MemorySaver is volatile

MemorySaver stores everything in the process's RAM. That means:

Python process alive:
  ✅ Checkpoints available
  ✅ You can resume conversations
  ✅ Full history accessible

Python process restarts (deploy, crash, Ctrl+C):
  ❌ ALL checkpoints are gone
  ❌ Every conversation starts from zero
  ❌ The entire history disappears

This is NOT a defect — it's a design decision. MemorySaver is built for:

For production you need a durable checkpointer. In the next capsule you'll implement PostgresSaver — and you'll see the change is a single line of code. Everything you learned here about thread_id, get_state(), get_state_history(), and the checkpointing flow works exactly the same. Only the storage location changes.

Troubleshooting

Problem 1: "RunnableConfigurableFieldsSpec" or an error when invoking without config

Symptom: A cryptic error when you call graph.invoke({"messages": [...]}) without passing a config. Cause: Once you compile with a checkpointer, the thread_id is mandatory. Fix: Always pass config = {"configurable": {"thread_id": "some_id"}} as the second argument to invoke().

Problem 2: The agent doesn't remember previous turns

Symptom: Every invoke feels like a fresh conversation. Cause 1: You're using a different thread_id on each invoke. Cause 2: The checkpointer wasn't passed at compile time. Fix: Check that (1) the same thread_id is used on both invokes and (2) graph_builder.compile(checkpointer=checkpointer) actually includes the checkpointer.

Problem 3: Checkpoints disappear after restarting the script

Symptom: You run the script, it works. You run it again, the agent remembers nothing. Cause: MemorySaver is in-memory. When the Python process exits, the checkpoints go with it. Fix: This is expected MemorySaver behavior. For persistence across restarts, use PostgresSaver (capsule 04).

Problem 4: Conversations bleeding between users

Symptom: One user gets another user's context. Cause: You're reusing the same thread_id for different users. Fix: Use a unique thread_id per user or per session. A safe pattern: f"user_{user_id}_session_{session_id}".

Problem 5: The history grows without bound and the model fails

Symptom: After many turns, the model throws a context window error. Cause: The checkpointer stores every message. After 50+ turns, the message list exceeds the model's limit. Fix: Apply message trimming before calling the model. Capsule 02 of this module covers trim_messages() for exactly this.

Exercises

Exercise 1: Chatbot with basic memory (Easy)

Build a graph with a single chatbot node that uses MessagesState and MemorySaver. Make 3 invocations with the same thread_id: (1) "My name is Carlos", (2) "What's my name?", (3) "How many messages have we exchanged?". Check that the model remembers the name and can count the turns.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "carlos_001"}}

r1 = graph.invoke({"messages": [("user", "My name is Carlos")]}, config)
print(f"Turn 1: {r1['messages'][-1].content}")

r2 = graph.invoke({"messages": [("user", "What's my name?")]}, config)
print(f"Turn 2: {r2['messages'][-1].content}")

r3 = graph.invoke({"messages": [("user", "How many messages have we exchanged in this conversation?")]}, config)
print(f"Turn 3: {r3['messages'][-1].content}")

state = graph.get_state(config)
print(f"\nTotal messages in the checkpoint: {len(state.values['messages'])}")
assert len(state.values["messages"]) == 6, "There should be 6 messages (3 user + 3 AI)"
print("✅ The checkpoint holds the whole conversation")
# Expected output:
# Turn 1: Hi Carlos! Nice to meet you...
# Turn 2: Your name is Carlos...
# Turn 3: We've had 3 exchanges (6 messages in total)...
# Total messages in the checkpoint: 6
# ✅ The checkpoint holds the whole conversation

Exercise 2: Multi-user isolation (Easy)

Build a chatbot with MemorySaver. Use two different thread_ids ("user_A" and "user_B"). In thread A, tell the model "My favorite language is Python". In thread B, say "My favorite language is Rust". Then ask each thread "What's my favorite language?" and check that the answers are correct and don't bleed.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config_a = {"configurable": {"thread_id": "user_A"}}
config_b = {"configurable": {"thread_id": "user_B"}}

graph.invoke({"messages": [("user", "My favorite language is Python")]}, config_a)
graph.invoke({"messages": [("user", "My favorite language is Rust")]}, config_b)

result_a = graph.invoke({"messages": [("user", "What's my favorite language?")]}, config_a)
result_b = graph.invoke({"messages": [("user", "What's my favorite language?")]}, config_b)

print(f"User A: {result_a['messages'][-1].content}")
print(f"User B: {result_b['messages'][-1].content}")

assert "python" in result_a["messages"][-1].content.lower()
assert "rust" in result_b["messages"][-1].content.lower()
print("\n✅ The conversations are properly isolated")
# Expected output:
# User A: Your favorite language is Python...
# User B: Your favorite language is Rust...
# ✅ The conversations are properly isolated

Exercise 3: Inspect the checkpoint history (Medium)

Build a graph with 3 sequential nodes (analyze → enrich → summarize), each adding a field to the state. Use get_state_history() to verify that each checkpoint carries progressively more data.

from typing import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class PipelineState(TypedDict):
    input_text: str
    analysis: str
    enrichment: str
    summary: str


def analyze(state: PipelineState) -> dict:
    return {"analysis": f"Analysis of: {state['input_text'][:30]}"}

def enrich(state: PipelineState) -> dict:
    return {"enrichment": f"Data for: {state['analysis'][:30]}"}

def summarize(state: PipelineState) -> dict:
    return {"summary": f"Summary: {state['analysis'][:20]} + {state['enrichment'][:20]}"}


builder = StateGraph(PipelineState)
builder.add_node("analyze", analyze)
builder.add_node("enrich", enrich)
builder.add_node("summarize", summarize)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "enrich")
builder.add_edge("enrich", "summarize")
builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "pipeline_001"}}
graph.invoke({"input_text": "LangGraph for agents", "analysis": "", "enrichment": "", "summary": ""}, config)

history = list(graph.get_state_history(config))
print(f"Total checkpoints: {len(history)}\n")
for i, snap in enumerate(history):
    filled = [k for k, v in snap.values.items() if v]
    print(f"  Checkpoint {i}: fields={filled}")
# Expected output:
# Total checkpoints: 5
#   Checkpoint 0: fields=['input_text', 'analysis', 'enrichment', 'summary']
#   Checkpoint 1: fields=['input_text', 'analysis', 'enrichment']
#   ...

assert len(history) == 5
print("✅ Checkpoint progression verified")

Exercise 4: Time-travel — branch from an earlier checkpoint (Medium)

Use the graph from Exercise 3. Get the checkpoint after analyze (before enrich). Invoke from that checkpoint with graph.invoke(None, checkpoint.config). Verify that enrich and summarize re-run while analysis is preserved.

from typing import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class PipelineState(TypedDict):
    input_text: str
    analysis: str
    enrichment: str
    summary: str

def analyze(state: PipelineState) -> dict:
    return {"analysis": f"Analysis of: {state['input_text'][:30]}"}

def enrich(state: PipelineState) -> dict:
    return {"enrichment": f"Enriched: {state['analysis'][:30]}"}

def summarize(state: PipelineState) -> dict:
    return {"summary": f"Summary: {state['analysis'][:20]} + {state['enrichment'][:20]}"}


builder = StateGraph(PipelineState)
for name, fn in [("analyze", analyze), ("enrich", enrich), ("summarize", summarize)]:
    builder.add_node(name, fn)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "enrich")
builder.add_edge("enrich", "summarize")
builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "time_travel_ex"}}
original = graph.invoke(
    {"input_text": "LangGraph for agents", "analysis": "", "enrichment": "", "summary": ""}, config
)

history = list(graph.get_state_history(config))
after_analyze = next(s for s in history if s.values.get("analysis") and not s.values.get("enrichment"))

branched = graph.invoke(None, after_analyze.config)

assert branched["analysis"] == original["analysis"], "Analysis must be preserved"
assert branched["summary"] != "", "Summary must be regenerated"
print(f"Analysis preserved: {branched['analysis']}")
print(f"Enrichment regenerated: {branched['enrichment']}")
print("✅ Time-travel branching successful")
# Expected output:
# Analysis preserved: Analysis of: LangGraph for agents
# Enrichment regenerated: Enriched: Analysis of: LangGraph fo
# ✅ Time-travel branching successful

Exercise 5: Simulate a crash and recovery (Advanced)

Build a graph with 4 sequential nodes (step_1 → step_2 → step_3 → step_4) where each node appends its name to a steps_completed list in the state. Run the graph normally and keep the result. Then simulate a "crash" by grabbing the checkpoint after step_2 and resuming from there. Verify that the resumed run completes step_3 and step_4 without repeating step_1 and step_2.

import operator
from typing import TypedDict, Annotated
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class CrashState(TypedDict):
    task: str
    steps_completed: Annotated[list[str], operator.add]
    output: str


def step_1(state: CrashState) -> dict:
    print("[step_1] Running...")
    return {"steps_completed": ["step_1"]}


def step_2(state: CrashState) -> dict:
    print("[step_2] Running...")
    return {"steps_completed": ["step_2"]}


def step_3(state: CrashState) -> dict:
    print("[step_3] Running...")
    return {"steps_completed": ["step_3"]}


def step_4(state: CrashState) -> dict:
    print("[step_4] Running...")
    all_steps = state["steps_completed"] + ["step_4"]
    return {"steps_completed": ["step_4"], "output": f"Completed: {', '.join(all_steps)}"}


builder = StateGraph(CrashState)
for name, fn in [("step_1", step_1), ("step_2", step_2), ("step_3", step_3), ("step_4", step_4)]:
    builder.add_node(name, fn)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", "step_4")
builder.add_edge("step_4", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "crash_sim"}}
full_result = graph.invoke({"task": "test", "steps_completed": [], "output": ""}, config)
print(f"\nFull run: {full_result['steps_completed']}")

history = list(graph.get_state_history(config))
after_step_2 = None
for snap in history:
    completed = snap.values.get("steps_completed", [])
    if "step_2" in completed and "step_3" not in completed:
        after_step_2 = snap
        break

assert after_step_2 is not None, "No post-step_2 checkpoint found"
print(f"\n--- Simulating crash and recovery from step_2 ---")
print(f"State at the checkpoint: steps_completed={after_step_2.values['steps_completed']}")

recovered = graph.invoke(None, after_step_2.config)
print(f"\nRecovered steps: {recovered['steps_completed']}")
print(f"Output: {recovered['output']}")

assert "step_1" in recovered["steps_completed"]
assert "step_2" in recovered["steps_completed"]
assert "step_3" in recovered["steps_completed"]
assert "step_4" in recovered["steps_completed"]
print("\n✅ Recovery successful: step_3 and step_4 ran without repeating step_1 and step_2")
# Expected output:
# [step_1] Running...
# [step_2] Running...
# [step_3] Running...
# [step_4] Running...
#
# Full run: ['step_1', 'step_2', 'step_3', 'step_4']
#
# --- Simulating crash and recovery from step_2 ---
# State at the checkpoint: steps_completed=['step_1', 'step_2']
# [step_3] Running...
# [step_4] Running...
#
# Recovered steps: ['step_1', 'step_2', 'step_3', 'step_4']
# Output: Completed: step_1, step_2, step_3, step_4
#
# ✅ Recovery successful: step_3 and step_4 ran without repeating step_1 and step_2

Summary

In this capsule you learned:

• Checkpointing saves the graph's complete state after every node. It isn't a diff, it's a snapshot. If the process is interrupted, you can resume from the last checkpoint without losing prior work
• MemorySaver is the checkpointer for development. One line (checkpointer = MemorySaver()), no external dependencies, no configuration. Perfect for prototyping and testing
• thread_id is mandatory and it's the isolation mechanism. Each thread has its own chain of checkpoints. Multiple users = multiple thread_ids = completely separate histories
• get_state() and get_state_history() are debugging tools. You can see the graph's exact state at any point in its run — which messages existed, which nodes ran, what data it held
• Time-travel: you can resume from any checkpoint, not only the latest. That creates branches in the history — useful for debugging and for exploring alternative paths
• MemorySaver is volatile: if the process restarts, every checkpoint is gone. That's expected and acceptable in development. For production, you need durable persistence

Next capsule: Persistence with PostgresSaver and Redis — moving from MemorySaver to PostgresSaver is a one-line change. Everything else (thread_id, get_state, get_state_history) works exactly the same. The payoff: your checkpoints survive crashes, restarts, and deploys.

Further reading

1. LangGraph — Persistence — Official docs on checkpointing in LangGraph. Covers MemorySaver, PostgresSaver, and the overall architecture of the persistence system
2. LangGraph — How to add thread-level persistence — Step-by-step guide to implementing checkpointing with thread_id
3. LangGraph — How to manage conversation history — Strategies for keeping message history under control when you use checkpointing
4. LangGraph — Time travel — Official guide to time-travel debugging: navigating checkpoints, replaying, and forking executions
5. LangGraph — MemorySaver Reference — API reference for MemorySaver with every available method

Module 8 — LangChain & LangGraph: From Chains to Agents