3. Checkpointing with MemorySaver

Overview

In the previous capsule you learned that an agent with MessagesState accumulates messages throughout a conversation — as long as the session is open, the history grows and the agent has "memory." But there's a fundamental problem: it all lives in a Python variable. If your program ends, if the process crashes, if you close the laptop — it all disappears. The agent goes back to being a blank slate.

Checkpointing solves this. Every time a node in the graph runs, LangGraph can automatically save a "snapshot" of the agent's complete state — messages, custom variables, metadata — into a persistent store. When the user comes back, the graph loads the last checkpoint and continues as if nothing had happened. It's not that the agent "remembers" the conversation — it literally restores its exact state.

In this capsule you'll work with MemorySaver, LangGraph's simplest checkpointer. It persists in RAM (a Python dictionary), which means it's instantaneous and requires no external infrastructure. It's perfect for development and testing. But it also means it's lost when you restart the process — in production you'll need something more robust (PostgresSaver, capsule 04). Here you'll understand the full mechanics of checkpointing: how it works, what gets saved, how to handle multiple sessions with thread_id, and — the key moment — how to interrupt a research run, close everything, come back, and have the agent continue exactly where it was.

What checkpointing is

The problem it solves

Imagine your Research Agent is halfway through a 6-step research run. It has already searched three sources, analyzed the results, and was about to synthesize the final report. The user closes the tab. Without checkpointing, those three research steps — the searches, the results, the partial analysis — disappear. When the user comes back, the agent has no idea what it was doing.

With checkpointing, every step of the research was saved automatically. When the user comes back with the same thread_id, the agent loads the exact state of step 3 and continues from there. It doesn't repeat searches, it doesn't lose context, it doesn't start from zero.

How it works internally

In LangGraph, a checkpoint is a serialized snapshot of the graph's complete state at a specific point in the execution. The mechanism is:

1. Before each node: LangGraph reads the current checkpoint from the store
2. The node runs: it modifies the state (adds messages, updates variables)
3. After each node: LangGraph serializes the new state and saves it as a new checkpoint
4. Each checkpoint has a unique ID: based on the thread_id and the position in the execution

Execution without checkpointing:
  START → agent_node → tool_node → agent_node → END
  (everything in Python memory, lost when it ends)

Execution with checkpointing:
  START → [checkpoint_0] → agent_node → [checkpoint_1] → tool_node → [checkpoint_2] → agent_node → [checkpoint_3] → END
  (each checkpoint gets saved in the store)

This means that if the process dies between checkpoint_2 and checkpoint_3, you can restore from checkpoint_2 and re-run only agent_node — you don't repeat tool_node.

What gets saved in each checkpoint

A checkpoint contains:

It's not just "saving messages." It's saving the agent's complete state, including any custom variable you defined in your state.

from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    sources_found: list[str]
    current_step: str
    confidence_score: float

# With checkpointing, EVERYTHING gets persisted:
# - messages: the complete conversation
# - sources_found: ["arxiv.org/...", "github.com/..."]
# - current_step: "analyzing"
# - confidence_score: 0.72

MemorySaver: setup and usage

Basic configuration

MemorySaver is LangGraph's simplest checkpointer. You configure it in one line and it requires no external infrastructure:

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from typing import TypedDict, Annotated

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(AgentState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# The key line: compile with a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

That's it. From now on, every run of the graph automatically saves a checkpoint after each node. You don't have to call any "save" function manually — LangGraph does it for you.

Invoking with thread_id

For the checkpointer to know where to save and where to restore from, you need a thread_id in the config:

config = {"configurable": {"thread_id": "session-001"}}

result_1 = app.invoke(
    {"messages": [("user", "My name is Carlos and I'm researching RAG")]},
    config
)
print(result_1["messages"][-1].content)
# "Hi Carlos! Sure, RAG is a fascinating topic..."

result_2 = app.invoke(
    {"messages": [("user", "What was I researching again?")]},
    config
)
print(result_2["messages"][-1].content)
# "You're researching RAG (Retrieval-Augmented Generation)..."

Without the thread_id, you get an error. The checkpointer needs to know where to save the state. Think of thread_id as a filename — each thread is a different file with a different conversation.

What happens internally on each invoke

When you call app.invoke(input, config) with a checkpointer configured:

1. LangGraph looks for the most recent checkpoint for thread_id="session-001"
2. If it exists: loads the saved state and MERGES the new input
3. If it doesn't: creates a new state with the provided input
4. Runs the graph node by node
5. After EACH node: saves a new checkpoint
6. When it finishes: the last checkpoint is the final state

Step 2 is fundamental. When you send {"messages": [("user", "second question")]}, LangGraph doesn't start with a one-message list. It loads the complete state from the previous checkpoint (which has all the earlier messages), applies the add_messages reducer to append the new message, and runs the graph with the full history.

MemorySaver with a ReAct agent

The most common pattern: an agent with tools and checkpointing.

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def web_search(query: str) -> str:
    """Search the web for up-to-date information."""
    return f"Results for '{query}': LangGraph 1.0 supports native checkpointing with MemorySaver and PostgresSaver."

@tool
def save_note(content: str) -> str:
    """Save a note for future reference."""
    return f"Note saved: {content}"

model = init_chat_model("openai:gpt-4.1-mini")
memory = MemorySaver()

agent = create_react_agent(
    model,
    [web_search, save_note],
    checkpointer=memory
)

config = {"configurable": {"thread_id": "research-session-1"}}

result = agent.invoke(
    {"messages": [("user", "Look up information about checkpointing in LangGraph and save a note with the most important parts")]},
    config
)

print(result["messages"][-1].content)

Notice: create_react_agent takes checkpointer directly as a parameter. You don't need to compile manually. The agent keeps the whole conversation — including tool calls and their results — between invocations.

Thread IDs: independent sessions

One thread = one conversation

The thread_id is the central concept of checkpointing. Each thread_id represents a completely independent line of conversation. Two different threads don't share state, don't share messages, don't see each other.

memory = MemorySaver()
agent = create_react_agent(model, [web_search], checkpointer=memory)

# Thread 1: conversation about RAG
config_1 = {"configurable": {"thread_id": "thread-rag"}}
agent.invoke({"messages": [("user", "Explain what RAG is")]}, config_1)

# Thread 2: conversation about MCP
config_2 = {"configurable": {"thread_id": "thread-mcp"}}
agent.invoke({"messages": [("user", "Explain what MCP is")]}, config_2)

# Thread 1 knows nothing about MCP
result = agent.invoke(
    {"messages": [("user", "What were we talking about?")]},
    config_1
)
print(result["messages"][-1].content)
# "We were talking about RAG..."

# Thread 2 knows nothing about RAG
result = agent.invoke(
    {"messages": [("user", "What were we talking about?")]},
    config_2
)
print(result["messages"][-1].content)
# "We were talking about MCP..."

Patterns for assigning thread_id

The practical question: how do you generate thread_ids? It depends on your use case:

import uuid

def thread_per_user(user_id: str) -> dict:
    return {"configurable": {"thread_id": f"user-{user_id}"}}

def thread_per_session() -> dict:
    return {"configurable": {"thread_id": f"session-{uuid.uuid4()}"}}

def thread_per_task(user_id: str, task: str) -> dict:
    return {"configurable": {"thread_id": f"{task}-{user_id}"}}

config_carlos = thread_per_user("carlos-42")
config_new_session = thread_per_session()
config_rag_research = thread_per_task("carlos-42", "rag-deep-dive")

Multiple simultaneous sessions

A single MemorySaver can handle hundreds of threads at the same time. Each thread is independent:

memory = MemorySaver()
agent = create_react_agent(model, [web_search], checkpointer=memory)

users = ["alice", "bob", "carlos"]
questions = [
    "What is a vector database?",
    "How does fine-tuning work?",
    "What is prompt engineering?"
]

for user, question in zip(users, questions):
    config = {"configurable": {"thread_id": f"user-{user}"}}
    result = agent.invoke({"messages": [("user", question)]}, config)
    print(f"[{user}] {result['messages'][-1].content[:80]}...")

# Each user can continue their conversation independently
config_alice = {"configurable": {"thread_id": "user-alice"}}
result = agent.invoke(
    {"messages": [("user", "Give me more detail on what you explained")]},
    config_alice
)
# Alice keeps talking about vector databases, not fine-tuning or prompt engineering

Interrupting and resuming

The "wow" moment

This is the demo that makes the power of checkpointing tangible. You're going to do exactly this:

1. Start a multi-step research run
2. Interrupt it halfway through (simulating a crash or a closed session)
3. Come back with the same thread_id
4. Watch the agent continue exactly where it was

Setup: a research agent with custom state

First, build an agent with visible steps so you can verify the resumption:

from dotenv import load_dotenv
load_dotenv()

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage, AIMessage
from typing import TypedDict, Annotated

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    research_steps_completed: int
    sources: list[str]
    findings: list[str]
    status: str

model = init_chat_model("openai:gpt-4.1-mini")

def search_sources(state: ResearchState) -> dict:
    step = state.get("research_steps_completed", 0) + 1
    source = f"source-{step}.example.com"
    finding = f"Finding from step {step}: relevant information about the research topic."
    
    print(f"  [Step {step}] Searching sources... found: {source}")
    
    return {
        "messages": [AIMessage(content=f"I searched the source {source} and found: {finding}")],
        "research_steps_completed": step,
        "sources": state.get("sources", []) + [source],
        "findings": state.get("findings", []) + [finding],
        "status": "researching"
    }

def analyze_findings(state: ResearchState) -> dict:
    n_findings = len(state.get("findings", []))
    print(f"  [Analysis] Analyzing {n_findings} findings...")
    
    summary = f"Analysis of {n_findings} sources completed. Confidence: high."
    return {
        "messages": [AIMessage(content=summary)],
        "status": "analyzed"
    }

def synthesize_report(state: ResearchState) -> dict:
    sources = state.get("sources", [])
    findings = state.get("findings", [])
    print(f"  [Synthesis] Generating report with {len(sources)} sources...")
    
    report = f"FINAL REPORT\n\nSources consulted: {len(sources)}\n"
    for s, f in zip(sources, findings):
        report += f"\n- {s}: {f}"
    report += f"\n\nStatus: completed"
    
    return {
        "messages": [AIMessage(content=report)],
        "status": "completed"
    }

def route_research(state: ResearchState) -> str:
    steps = state.get("research_steps_completed", 0)
    if steps < 3:
        return "search_more"
    elif state.get("status") != "analyzed":
        return "analyze"
    else:
        return "synthesize"

graph = StateGraph(ResearchState)
graph.add_node("search", search_sources)
graph.add_node("analyze", analyze_findings)
graph.add_node("synthesize", synthesize_report)

graph.add_edge(START, "search")
graph.add_conditional_edges("search", route_research, {
    "search_more": "search",
    "analyze": "analyze",
    "synthesize": "synthesize"
})
graph.add_edge("analyze", "synthesize")
graph.add_edge("synthesize", END)

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

Full demo: interrupt and resume

Now run the complete research to see the normal flow:

config = {"configurable": {"thread_id": "research-demo-1"}}

print("=== Full execution ===")
result = app.invoke(
    {"messages": [("user", "Research AI agents in production")]},
    config
)

print(f"\nFinal state:")
print(f"  Steps completed: {result['research_steps_completed']}")
print(f"  Sources: {result['sources']}")
print(f"  Status: {result['status']}")
print(f"  Total messages: {len(result['messages'])}")

Now the real demo — interrupting halfway. Use stream to process node by node and stop after the second step:

config_interrupt = {"configurable": {"thread_id": "research-demo-2"}}

print("=== Execution with an interruption ===")
steps_seen = 0
for step in app.stream(
    {"messages": [("user", "Research checkpointing in LangGraph")]},
    config_interrupt,
    stream_mode="updates"
):
    steps_seen += 1
    for node_name, update in step.items():
        print(f"  Node executed: {node_name}")
    
    if steps_seen >= 2:
        print("\n  *** INTERRUPTION: simulating a crash ***")
        break

# Check what got saved
state = app.get_state(config_interrupt)
print(f"\nState saved after the interruption:")
print(f"  Steps completed: {state.values.get('research_steps_completed', 0)}")
print(f"  Sources found: {state.values.get('sources', [])}")
print(f"  Status: {state.values.get('status', 'unknown')}")
print(f"  Messages: {len(state.values.get('messages', []))}")

Now the key part — resuming. Call invoke with the same thread_id and None as the input:

print("\n=== Resumption ===")
# invoke with None picks up from the last checkpoint
result = app.invoke(None, config_interrupt)

print(f"\nFinal state after resuming:")
print(f"  Steps completed: {result['research_steps_completed']}")
print(f"  Sources: {result['sources']}")
print(f"  Status: {result['status']}")
print(f"  Total messages: {len(result['messages'])}")

What you'll see: the agent does not repeat the first two search steps. It continues straight from where it was — it does the third search, analyzes, and synthesizes. The intermediate checkpoints preserved all the progress.

Why invoke(None, config) works

When you pass None as the input:

1. LangGraph loads the last checkpoint for that thread_id
2. It checks which node the graph stopped at
3. It continues execution from that point, using the saved state
4. It doesn't need new input because the state already contains all the necessary information

This is what makes checkpointing truly powerful for agents — it isn't just "saving the conversation." It's saving the execution progress and being able to resume it.

Inspecting checkpoints

get_state: the current state

get_state gives you the most recent checkpoint for a thread_id. It's like taking a snapshot of where the agent is right now:

config = {"configurable": {"thread_id": "research-demo-1"}}
state = app.get_state(config)

print(f"State values:")
for key, value in state.values.items():
    if key == "messages":
        print(f"  messages: {len(value)} messages")
    else:
        print(f"  {key}: {value}")

print(f"\nMetadata:")
print(f"  Checkpoint ID: {state.config['configurable']['checkpoint_id']}")
print(f"  Thread ID: {state.config['configurable']['thread_id']}")
print(f"  Next node(s): {state.next}")

The state.next field is particularly useful: it tells you which node would run if you continued the graph. If it's empty (()), the graph finished. If it has a value (for example ("search",)), the graph stopped before running that node — you can resume it.

get_state_history: the complete timeline

get_state_history gives you every checkpoint of a thread, in reverse chronological order (most recent first):

config = {"configurable": {"thread_id": "research-demo-1"}}

print("=== Checkpoint history ===\n")
for i, state in enumerate(app.get_state_history(config)):
    step = state.values.get("research_steps_completed", 0)
    status = state.values.get("status", "initial")
    n_msgs = len(state.values.get("messages", []))
    next_nodes = state.next
    
    print(f"Checkpoint {i}:")
    print(f"  Steps completed: {step}")
    print(f"  Status: {status}")
    print(f"  Messages: {n_msgs}")
    print(f"  Next node: {next_nodes}")
    print(f"  Checkpoint ID: {state.config['configurable']['checkpoint_id']}")
    print()

Typical output (for a run of 3 searches + analysis + synthesis):

Checkpoint 0:                          ← Most recent (final report)
  Steps completed: 3
  Status: completed
  Messages: 6
  Next node: ()

Checkpoint 1:                          ← After the analysis
  Steps completed: 3
  Status: analyzed
  Messages: 5
  Next node: ('synthesize',)

Checkpoint 2:                          ← After the 3rd search
  Steps completed: 3
  Status: researching
  Messages: 4
  Next node: ('analyze',)

Checkpoint 3:                          ← After the 2nd search
  Steps completed: 2
  Status: researching
  Messages: 3
  Next node: ('search',)

Checkpoint 4:                          ← After the 1st search
  Steps completed: 1
  Status: researching
  Messages: 2
  Next node: ('search',)

Checkpoint 5:                          ← Initial state
  Steps completed: 0
  Status:
  Messages: 1
  Next node: ('search',)

This is the foundation of time-travel debugging (capsule 06). Each checkpoint is a point you can go back to and re-run from.

Deep inspection of the messages in a checkpoint

For debugging, sometimes you need to see exactly which messages live in a specific checkpoint:

config = {"configurable": {"thread_id": "research-demo-1"}}

for i, state in enumerate(app.get_state_history(config)):
    if i == 3:  # Checkpoint after the 2nd search
        print(f"=== Checkpoint {i}: after the 2nd search ===\n")
        for j, msg in enumerate(state.values.get("messages", [])):
            msg_type = msg.__class__.__name__
            content_preview = msg.content[:100] if msg.content else "(no content)"
            print(f"  [{j}] {msg_type}: {content_preview}")
        
        print(f"\n  Custom state:")
        print(f"    sources: {state.values.get('sources', [])}")
        print(f"    findings: {state.values.get('findings', [])}")
        print(f"    steps: {state.values.get('research_steps_completed', 0)}")
        break

Using a specific checkpoint to resume

You can restore from any checkpoint, not just the last one:

config = {"configurable": {"thread_id": "research-demo-1"}}

# Find the checkpoint after the first search
target_checkpoint = None
for state in app.get_state_history(config):
    if state.values.get("research_steps_completed", 0) == 1:
        target_checkpoint = state.config
        break

if target_checkpoint:
    print(f"Resuming from checkpoint: {target_checkpoint['configurable']['checkpoint_id']}")
    result = app.invoke(None, target_checkpoint)
    print(f"Result: {result['research_steps_completed']} steps, status: {result['status']}")

This runs the graph from the point where only 1 search had been completed — it redoes searches 2 and 3, the analysis, and the synthesis. It's like "rewinding" the agent's execution to an earlier point.

MemorySaver's limitations

What MemorySaver is and what it isn't

MemorySaver stores checkpoints in a Python dictionary inside the current process. That has direct implications:

When to use MemorySaver

MemorySaver is the right choice in these scenarios:

Local development:

memory = MemorySaver()
agent = create_react_agent(model, tools, checkpointer=memory)
# Perfect for iterating fast, trying changes, debugging

Automated testing:

def test_agent_remembers_context():
    memory = MemorySaver()
    agent = create_react_agent(model, tools, checkpointer=memory)
    config = {"configurable": {"thread_id": "test-1"}}
    
    agent.invoke({"messages": [("user", "My name is Carlos")]}, config)
    result = agent.invoke({"messages": [("user", "What's my name?")]}, config)
    
    assert "Carlos" in result["messages"][-1].content
    # Each test creates its own MemorySaver — total isolation

Prototypes and demos:

# To show off the checkpointing feature without setting up Postgres
memory = MemorySaver()
# "Look, the agent remembers" — works perfectly in a 30-minute demo

When NOT to use MemorySaver

Production with real users:

# WRONG for production
memory = MemorySaver()  # If the server restarts, everyone loses their history

# RIGHT for production
from langgraph.checkpoint.postgres import PostgresSaver
postgres = PostgresSaver.from_conn_string("postgresql://...")

Multi-process applications:

# WRONG: each Gunicorn/Uvicorn worker has its own MemorySaver
# Worker 1: MemorySaver() → has Alice's checkpoints
# Worker 2: MemorySaver() → does NOT have Alice's checkpoints
# If Alice switches workers, she loses her conversation

Data you can't afford to lose:

# If the user spent 20 minutes on a research run with the agent,
# losing that to a server restart is unacceptable.
# MemorySaver = acceptable to lose data
# PostgresSaver = durable data

The natural progression

The transition from MemorySaver to a durable store is one of the cleanest progressions in LangGraph. The code barely changes:

# Development: MemorySaver
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# Production: PostgresSaver (capsule 04)
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@host/db")
checkpointer.setup()

# The rest of the code is IDENTICAL
agent = create_react_agent(model, tools, checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-123"}}
result = agent.invoke({"messages": [("user", "Hello")]}, config)

Develop with MemorySaver, deploy with PostgresSaver. The interface is the same — only the import line and the initialization change.

Connection with the project

Module 6 — Research Agent with persistent memory

In capsule 08 you'll add checkpointing to the Research Agent. What you learned here is the direct foundation:

• Thread per research session: Each of the user's research runs will have a unique thread_id. If the user says "research RAG" and then "research MCP," those are two independent threads with two separate research runs
• Resume capability: If the research gets interrupted (timeout, crash, the user closes the session), it resumes with invoke(None, config) exactly where it was — without repeating searches or losing findings
• Custom state: The Research Agent will have a state that includes sources, findings, analysis_step, and confidence_score — all persisted by the checkpointer
• Progress inspection: When the user comes back, you can use get_state to show them a summary of where they left off before resuming

Upcoming capsules in this module

Troubleshooting

Problem 1: "No checkpoint found" when trying to continue a conversation

Symptom: The agent doesn't remember anything from the previous session. Every invocation starts from zero.

Cause: The thread_id doesn't match between the first and second invocation. It could be a typo, a regenerated ID, or a different MemorySaver.

Fix: Make sure you're using exactly the same thread_id and the same MemorySaver object:

memory = MemorySaver()
agent = create_react_agent(model, tools, checkpointer=memory)

# WRONG: different thread_ids
config_1 = {"configurable": {"thread_id": "session-1"}}
config_2 = {"configurable": {"thread_id": "session_1"}}  # underscore vs hyphen
agent.invoke({"messages": [("user", "Hello")]}, config_1)
agent.invoke({"messages": [("user", "Do you remember me?")]}, config_2)  # Doesn't remember

# RIGHT: the exact same thread_id
config = {"configurable": {"thread_id": "session-1"}}
agent.invoke({"messages": [("user", "Hello")]}, config)
agent.invoke({"messages": [("user", "Do you remember me?")]}, config)  # It does remember

Problem 2: The state is lost when you restart the script

Symptom: Checkpoints work within one run of the script, but if you close it and run it again, everything is gone.

Cause: MemorySaver stores in RAM. When the process ends, the memory is freed.

Fix: This is MemorySaver's expected behavior. If you need persistence between runs, use PostgresSaver (capsule 04). For development, you can seed the state with test data at the start of the script:

memory = MemorySaver()
agent = create_react_agent(model, tools, checkpointer=memory)
config = {"configurable": {"thread_id": "dev-session"}}

# "Seed" the conversation for fast development
agent.invoke(
    {"messages": [
        ("user", "My name is Carlos and I'm researching RAG"),
        ("assistant", "Hi Carlos! Got it, you're researching RAG.")
    ]},
    config
)
# Now you can continue from here without repeating the setup every time

Problem 3: invoke(None, config) raises an error

Symptom: When you try to resume with app.invoke(None, config), you get an error like ValueError or the graph doesn't run anything.

Cause: The graph finished its execution (state.next is empty). There's nothing to resume — the run already completed all the nodes.

Fix: Check the state before trying to resume:

config = {"configurable": {"thread_id": "my-thread"}}
state = app.get_state(config)

if state.next:
    print(f"Resuming from node: {state.next}")
    result = app.invoke(None, config)
else:
    print("The execution already finished. Send a new message to continue the conversation.")
    result = app.invoke(
        {"messages": [("user", "your new message here")]},
        config
    )

Problem 4: Duplicated messages when resuming

Symptom: After resuming, repeated messages show up in the history. The agent seems to have an "echo."

Cause: You're passing messages that already exist in the checkpoint as new input. The add_messages reducer appends them again.

Fix: When resuming, pass None (to continue the execution) or a new message (to add to the conversation). Never pass messages that are already in the history:

config = {"configurable": {"thread_id": "my-thread"}}

# WRONG: re-sending a message that's already in the checkpoint
# result = app.invoke({"messages": [("user", "a message I already sent")]}, config)

# RIGHT option 1: resume the pending execution
result = app.invoke(None, config)

# RIGHT option 2: send a new message
result = app.invoke(
    {"messages": [("user", "a completely new message")]},
    config
)

Problem 5: MemorySaver eats a lot of memory with many threads

Symptom: The Python process grows in RAM usage as more threads pile up. After hundreds of conversations, the consumption is noticeable.

Cause: MemorySaver never deletes checkpoints. Each thread accumulates all of its historical checkpoints in memory.

Fix: For development, restart the MemorySaver periodically. For production, use PostgresSaver, where you can configure retention and cleanup:

# In development: create a fresh MemorySaver when needed
memory = MemorySaver()

# If you need to clean up manually, just create a new one
memory = MemorySaver()  # The previous checkpoints get garbage-collected

# In production: PostgresSaver with retention policies (capsule 04)

Exercises

Exercise 1: Conversation with basic memory (Easy)

Create a simple chatbot with MemorySaver. Make three invocations with the same thread_id: in the first one say your name, in the second ask something unrelated, and in the third ask "what's my name?" Verify that the agent remembers.

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from typing import TypedDict, Annotated

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

model = init_chat_model("openai:gpt-4.1-mini")

# Build the graph, compile it with MemorySaver,
# and make the 3 invocations with the same thread_id
# Your code here...

def chatbot(state: ChatState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "test-memory"}}

# Invocation 1: introduce yourself
result_1 = app.invoke({"messages": [("user", "Hi, my name is Valentina")]}, config)
print(f"1: {result_1['messages'][-1].content[:80]}")

# Invocation 2: an unrelated question
result_2 = app.invoke({"messages": [("user", "How many planets does the solar system have?")]}, config)
print(f"2: {result_2['messages'][-1].content[:80]}")

# Invocation 3: check the memory
result_3 = app.invoke({"messages": [("user", "What's my name?")]}, config)
print(f"3: {result_3['messages'][-1].content[:80]}")

assert "Valentina" in result_3["messages"][-1].content
print("\n✓ The agent remembers the name across 3 invocations")

Exercise 2: Independent sessions (Easy)

Create two threads with different thread_ids. In the "physics" thread talk about quantum physics. In the "cooking" thread talk about recipes. Then verify that asking "what are we talking about?" in each thread returns the right topic.

# Use the same agent but with two different configs
# Verify that the threads are completely independent
# Your code here...

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from typing import TypedDict, Annotated

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: ChatState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

config_physics = {"configurable": {"thread_id": "physics"}}
config_cooking = {"configurable": {"thread_id": "cooking"}}

# Initialize each thread with its topic
app.invoke({"messages": [("user", "Let's talk about quantum mechanics and the uncertainty principle")]}, config_physics)
app.invoke({"messages": [("user", "Let's talk about how to make authentic pasta carbonara")]}, config_cooking)

# Verify the independence
result_physics = app.invoke({"messages": [("user", "What are we talking about?")]}, config_physics)
result_cooking = app.invoke({"messages": [("user", "What are we talking about?")]}, config_cooking)

print(f"Physics thread: {result_physics['messages'][-1].content[:100]}")
print(f"Cooking thread: {result_cooking['messages'][-1].content[:100]}")

# The physics thread talks about quantum mechanics, not pasta
# The cooking thread talks about pasta carbonara, not physics
print("\n✓ The threads are completely independent")

Exercise 3: Inspect the checkpoint history (Medium)

Build an agent that takes exactly 3 steps (use a graph with a counter in the state). After the full run, use get_state_history to list all the checkpoints. For each one, print: the counter value, how many messages there are, and which node comes next. Figure out how many checkpoints were created and why.

# Hint: a linear graph with 3 nodes generates more checkpoints than you'd expect
# (there are intermediate checkpoints between nodes)
# Your code here...

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from typing import TypedDict, Annotated

class CounterState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    counter: int

def step_one(state: CounterState) -> dict:
    return {
        "messages": [AIMessage(content="Step 1 completed")],
        "counter": state.get("counter", 0) + 1
    }

def step_two(state: CounterState) -> dict:
    return {
        "messages": [AIMessage(content="Step 2 completed")],
        "counter": state.get("counter", 0) + 1
    }

def step_three(state: CounterState) -> dict:
    return {
        "messages": [AIMessage(content="Step 3 completed")],
        "counter": state.get("counter", 0) + 1
    }

graph = StateGraph(CounterState)
graph.add_node("step_one", step_one)
graph.add_node("step_two", step_two)
graph.add_node("step_three", step_three)

graph.add_edge(START, "step_one")
graph.add_edge("step_one", "step_two")
graph.add_edge("step_two", "step_three")
graph.add_edge("step_three", END)

memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "counter-test"}}

app.invoke({"messages": [("user", "Run the 3 steps")]}, config)

print("=== Checkpoint History ===\n")
checkpoints = list(app.get_state_history(config))
for i, state in enumerate(checkpoints):
    counter = state.values.get("counter", 0)
    n_msgs = len(state.values.get("messages", []))
    next_node = state.next
    checkpoint_id = state.config["configurable"]["checkpoint_id"]
    
    print(f"Checkpoint {i}:")
    print(f"  Counter: {counter}")
    print(f"  Messages: {n_msgs}")
    print(f"  Next node: {next_node}")
    print(f"  ID: {checkpoint_id[:20]}...")
    print()

print(f"Total checkpoints: {len(checkpoints)}")
print("Checkpoints get created: one before START, one after each node, and one at the end.")

Exercise 4: Interrupt and resume a ReAct agent (Hard)

Create a ReAct agent with tools and MemorySaver. Use stream to run it step by step. Interrupt after the first tool call (the agent searched but hasn't answered). Then resume with invoke(None, config) and verify that the agent produces the final answer without repeating the search.

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def research(topic: str) -> str:
    """Research a topic in depth."""
    return f"Research on {topic}: detailed data and a complete analysis of the topic."

# Create the agent with a checkpointer
# Use stream to run it step by step
# Interrupt after the first tool call
# Resume and verify it doesn't repeat the search
# Your code here...

model = init_chat_model("openai:gpt-4.1-mini")
memory = MemorySaver()

agent = create_react_agent(model, [research], checkpointer=memory)
config = {"configurable": {"thread_id": "interrupt-test"}}

print("=== Execution with an interruption ===\n")
tool_call_seen = False

for step in agent.stream(
    {"messages": [("user", "Research the MCP protocol")]},
    config,
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"Node: {node_name}")
        if "messages" in update:
            for msg in update["messages"]:
                if hasattr(msg, "tool_calls") and msg.tool_calls:
                    print(f"  → Tool calls: {[tc['name'] for tc in msg.tool_calls]}")
                    tool_call_seen = True
                elif hasattr(msg, "content") and msg.content:
                    print(f"  → {msg.content[:80]}...")
    
    if tool_call_seen:
        print("\n*** INTERRUPTION after the tool call ***")
        break

# Check the saved state
state = agent.get_state(config)
print(f"\nState after the interruption:")
print(f"  Messages: {len(state.values.get('messages', []))}")
print(f"  Next node: {state.next}")

# Resume
print("\n=== Resumption ===\n")
result = agent.invoke(None, config)

print(f"Final messages: {len(result['messages'])}")
print(f"Answer: {result['messages'][-1].content[:150]}...")

# Verify it didn't repeat the tool call
tool_calls_total = sum(
    len(m.tool_calls) for m in result["messages"]
    if hasattr(m, "tool_calls") and m.tool_calls
)
print(f"\nTotal tool calls across the whole conversation: {tool_calls_total}")
print("✓ The agent continued from the checkpoint without repeating the search")

Exercise 5: Multi-thread system with an activity summary (Hard)

Build a system that handles 3 simultaneous threads (simulating 3 users). Each user has 2 exchanges. At the end, implement a summarize_all_threads function that uses get_state for each thread and generates an executive summary: thread_id, number of messages, the user's last message, the agent's last message.

# Your system must:
# 1. Create 3 threads with different conversations
# 2. Do 2 exchanges per thread
# 3. Implement summarize_all_threads()
# 4. Present the summary as a table
# Your code here...

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage, HumanMessage
from typing import TypedDict, Annotated

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: ChatState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

users = {
    "user-alice": [
        "I'm learning about RAG. What components do I need?",
        "Which vector database do you recommend to start with?"
    ],
    "user-bob": [
        "I need to implement an agent with LangGraph. Where do I start?",
        "How do I add tools to the agent?"
    ],
    "user-carlos": [
        "I want to understand checkpointing in LangGraph.",
        "What's the difference between MemorySaver and PostgresSaver?"
    ]
}

# Run the conversations
for user_id, questions in users.items():
    config = {"configurable": {"thread_id": user_id}}
    for question in questions:
        result = app.invoke({"messages": [("user", question)]}, config)
        print(f"[{user_id}] Q: {question[:50]}... → A: {result['messages'][-1].content[:50]}...")

# Summary function
def summarize_all_threads(app, thread_ids: list[str]):
    print(f"\n{'='*80}")
    print(f"{'Thread ID':<15} {'Msgs':>5} {'Last user msg':<30} {'Last answer':<30}")
    print(f"{'-'*80}")
    
    for tid in thread_ids:
        config = {"configurable": {"thread_id": tid}}
        state = app.get_state(config)
        messages = state.values.get("messages", [])
        
        last_user = ""
        last_ai = ""
        for msg in reversed(messages):
            if isinstance(msg, HumanMessage) and not last_user:
                last_user = msg.content[:28]
            elif hasattr(msg, "content") and msg.content and not isinstance(msg, HumanMessage) and not last_ai:
                last_ai = msg.content[:28]
            if last_user and last_ai:
                break
        
        print(f"{tid:<15} {len(messages):>5} {last_user:<30} {last_ai:<30}")

summarize_all_threads(app, list(users.keys()))

================================================================================
Thread ID        Msgs Last user msg                  Last answer
--------------------------------------------------------------------------------
user-alice          6 Which vector database do y...  To get started, I'd recomm...
user-bob            6 How do I add tools to the ...  To add tools, use the...
user-carlos         6 What's the difference betw...  The main difference is...

Summary

In this capsule you learned:

• What checkpointing is: LangGraph automatically saves a snapshot of the agent's complete state after each node. Each checkpoint includes messages, custom variables, metadata, and a reference to the previous checkpoint
• MemorySaver as the development checkpointer: you configure it in one line (MemorySaver()), it requires no infrastructure, and it's instantaneous. Ideal for development, testing, and demos. It's lost when the process restarts
• Thread IDs as independent sessions: each thread_id is an isolated checkpoint space. Two different threads never share state. You can have hundreds of simultaneous threads in a single MemorySaver
• Interrupting and resuming: the central demo of this capsule — start a multi-step execution, interrupt it, and resume it with invoke(None, config) from the last checkpoint. The agent continues exactly where it was without repeating work
• Checkpoint inspection: get_state() to see the current state, get_state_history() to see every checkpoint of a thread. Each checkpoint shows which node comes next (state.next) and the complete state at that point
• Clear limitations: MemorySaver lives in RAM, doesn't survive restarts, doesn't scale to multiple processes, and accumulates memory over time. For production, you need PostgresSaver

Next capsule: PostgresSaver — Durable Persistence. Same concept, same interface, but with a real database behind it. The checkpoints survive restarts, crashes, and deployments. It's the transition from "works on my machine" to "works in production."

Additional resources

1. LangGraph Persistence Concepts — Complete conceptual documentation on checkpointing, threads, and stores
2. LangGraph MemorySaver Reference — API reference for the in-memory checkpointer
3. LangGraph How-to: Add Persistence — Step-by-step tutorial for adding checkpointing to a graph
4. LangGraph How-to: Manage Conversation History — Patterns for handling history with checkpointing
5. LangGraph How-to: Time Travel — Navigating and restoring historical checkpoints (a preview of capsule 06)