Module 8: Memory and Persistence

Durable Execution

Next →

Capsule overview

Durable execution means your agent survives crashes, restarts, and interruptions. If the process dies mid-run, the agent resumes exactly where it stopped — no repeated work, no lost data, no wasted spend.

In the previous capsules you set up checkpointing with MemorySaver and PostgresSaver. You learned that every step of the graph gets saved as a checkpoint. But there's one question we never answered: what happens when the process dies mid-execution?

The answer is durable execution. The checkpointer already has the progress saved up to the last completed node. When the process restarts and you use the same thread_id, LangGraph detects the existing checkpoint and resumes from where it left off. It doesn't repeat work. It doesn't lose data. It doesn't spend extra money.

This isn't a demo feature — it's the difference between a prototype that "works on my laptop" and a system you can put in production with confidence.

The scenario that makes it real

Picture your Research Assistant doing a long research run:

Research: "State of the art in AI Agents"
  → Step 1: Search Wikipedia         ($0.10 in API calls)  ✅ done
  → Step 2: Search arXiv             ($0.10 in API calls)  ✅ done
  → Step 3: Search news              ($0.10 in API calls)  ✅ done
  → Step 4: Search technical blogs   ($0.10 in API calls)  💥 PROCESS CRASH
  → Step 5: Synthesize report        ($0.10 in API calls)  ⏳ pending

Without durable execution:

  • The process restarts
  • There's no way to know steps 1-3 were already done
  • You redo EVERYTHING from the start: 5 steps × $0.10 = $0.50 wasted
  • The user waits another 5 minutes

With durable execution:

  • The process restarts
  • You use the same thread_id
  • LangGraph figures out: "I already got through step 3"
  • It resumes from step 4: $0.00 wasted
  • The user waits only 2 minutes (steps 4 and 5)

Multiply that by 100 users a day, research runs with 10+ steps, and APIs that cost $0.50+ per call. Durable execution isn't a "nice to have" — it's real money and real time you get back.

How durable execution works

The mechanism is surprisingly simple, because you already have every piece:

Normal run:
  Node 1 → checkpoint saved → Node 2 → checkpoint saved → Node 3 → checkpoint saved

Crash after Node 3:
  [process dies]

Restart with the same thread_id:
  LangGraph asks: "Is there a checkpoint for this thread_id?"
  → Yes: the Node 3 checkpoint
  → "Was the run complete?"
  → No: Nodes 4 and 5 were still pending
  → Resume from Node 4

Every checkpoint holds:

  • ✅ The graph's complete state at that point
  • ✅ Which node just ran
  • ✅ Which node comes next
  • ✅ The full history of messages and accumulated data

Implementation: checkpointing + thread management

Durable execution doesn't need new code. It's checkpointing — which you already know — used properly:

from dotenv import load_dotenv
load_dotenv()

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

class ResearchState(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    steps_completed: Annotated[list[str], operator.add]
    final_report: str

def search_wikipedia(state: ResearchState) -> dict:
    time.sleep(0.5)
    result = f"Wikipedia: encyclopedic information about '{state['topic']}'"
    return {
        "sources": [result],
        "steps_completed": ["wikipedia"],
    }

def search_arxiv(state: ResearchState) -> dict:
    time.sleep(0.5)
    result = f"arXiv: academic papers about '{state['topic']}'"
    return {
        "sources": [result],
        "steps_completed": ["arxiv"],
    }

def search_news(state: ResearchState) -> dict:
    time.sleep(0.5)
    result = f"News: recent coverage of '{state['topic']}'"
    return {
        "sources": [result],
        "steps_completed": ["news"],
    }

def synthesize(state: ResearchState) -> dict:
    report = f"Report on '{state['topic']}' based on {len(state['sources'])} sources:\n"
    for i, source in enumerate(state["sources"], 1):
        report += f"  {i}. {source}\n"
    report += f"Steps completed: {', '.join(state['steps_completed'])}"
    return {
        "final_report": report,
        "steps_completed": ["synthesis"],
    }

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("wikipedia", search_wikipedia)
graph_builder.add_node("arxiv", search_arxiv)
graph_builder.add_node("news", search_news)
graph_builder.add_node("synthesize", synthesize)

graph_builder.add_edge(START, "wikipedia")
graph_builder.add_edge("wikipedia", "arxiv")
graph_builder.add_edge("arxiv", "news")
graph_builder.add_edge("news", "synthesize")
graph_builder.add_edge("synthesize", END)

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

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

result = graph.invoke(
    {"topic": "AI agents", "sources": [], "steps_completed": [], "final_report": ""},
    config,
)
print("Run complete:")
print(f"Steps: {result['steps_completed']}")
print(f"Sources: {len(result['sources'])}")
print(f"Report:\n{result['final_report']}")
# Expected output:
# Run complete:
# Steps: ['wikipedia', 'arxiv', 'news', 'synthesis']
# Sources: 3
# Report:
# Report on 'AI agents' based on 3 sources:
#   1. Wikipedia: encyclopedic information about 'AI agents'
#   2. arXiv: academic papers about 'AI agents'
#   3. News: recent coverage of 'AI agents'
# Steps completed: wikipedia, arxiv, news, synthesis

The key: MemorySaver() + thread_id = durable execution. Every node that completes writes a checkpoint automatically. If the process dies, the next invocation with the same thread_id finds the checkpoint and carries on.

Resume from checkpoint: the mechanics of restarting

When you invoke a graph with a thread_id that already has checkpoints, LangGraph runs this evaluation:

graph.invoke(input, {"configurable": {"thread_id": "research-001"}})

  1. Is there a checkpoint for "research-001"?
     → No: run from the start (a new execution)
     → Yes: continue to step 2

  2. Did the previous run finish (reach END)?
     → Yes: this is a new invocation on an existing thread
     → No: the previous run was interrupted → RESUME

Let's see how to inspect a thread's state and figure out where it stopped:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    progress: Annotated[list[str], operator.add]
    result: str

def step_1(state: State) -> dict:
    return {"progress": ["step_1_done"]}

def step_2(state: State) -> dict:
    return {"progress": ["step_2_done"]}

def step_3(state: State) -> dict:
    return {"progress": ["step_3_done"], "result": "Task complete."}

graph_builder = StateGraph(State)
graph_builder.add_node("step_1", step_1)
graph_builder.add_node("step_2", step_2)
graph_builder.add_node("step_3", step_3)

graph_builder.add_edge(START, "step_1")
graph_builder.add_edge("step_1", "step_2")
graph_builder.add_edge("step_2", "step_3")
graph_builder.add_edge("step_3", END)

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

config = {"configurable": {"thread_id": "task-resume-demo"}}

result = graph.invoke(
    {"task": "Process data", "progress": [], "result": ""},
    config,
)

current_state = graph.get_state(config)
print(f"Current state: {current_state.values['progress']}")
print(f"Next node: {current_state.next}")
print(f"Result: {current_state.values['result']}")
# Expected output:
# Current state: ['step_1_done', 'step_2_done', 'step_3_done']
# Next node: ()
# Result: Task complete.

When current_state.next is an empty tuple (), the run finished normally. If it contains a node name, the run was interrupted right there — and you can resume it.

Simulating a crash and verifying the resume

To make the value concrete, let's simulate an interruption. We'll use LangGraph's interrupt to pause the run (a controlled crash), and then verify that the resume works:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    steps_done: Annotated[list[str], operator.add]
    cost_usd: float

def expensive_step_1(state: State) -> dict:
    """Step 1: costs $0.10 in API calls."""
    print("  Running step 1 (cost: $0.10)...")
    return {
        "steps_done": ["step_1"],
        "cost_usd": state.get("cost_usd", 0) + 0.10,
    }

def expensive_step_2(state: State) -> dict:
    """Step 2: costs $0.10 in API calls."""
    print("  Running step 2 (cost: $0.10)...")
    return {
        "steps_done": ["step_2"],
        "cost_usd": state.get("cost_usd", 0) + 0.10,
    }

def crash_point(state: State) -> dict:
    """Simulates a crash: the process gets interrupted here."""
    print("  💥 Interruption at step 3...")
    interrupt("Simulating a process crash")
    return {
        "steps_done": ["step_3"],
        "cost_usd": state.get("cost_usd", 0) + 0.10,
    }

def final_step(state: State) -> dict:
    """Step 4: produces the final result."""
    print("  Running step 4 (cost: $0.10)...")
    return {
        "steps_done": ["step_4"],
        "cost_usd": state.get("cost_usd", 0) + 0.10,
    }

graph_builder = StateGraph(State)
graph_builder.add_node("step_1", expensive_step_1)
graph_builder.add_node("step_2", expensive_step_2)
graph_builder.add_node("step_3", crash_point)
graph_builder.add_node("step_4", final_step)

graph_builder.add_edge(START, "step_1")
graph_builder.add_edge("step_1", "step_2")
graph_builder.add_edge("step_2", "step_3")
graph_builder.add_edge("step_3", "step_4")
graph_builder.add_edge("step_4", END)

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

config = {"configurable": {"thread_id": "crash-demo-001"}}

print("=== FIRST RUN (interrupted at step 3) ===")
result = graph.invoke(
    {"task": "Long research run", "steps_done": [], "cost_usd": 0.0},
    config,
)

state_after_crash = graph.get_state(config)
print(f"\nState after the crash:")
print(f"  Steps completed: {state_after_crash.values['steps_done']}")
print(f"  Accumulated cost: ${state_after_crash.values['cost_usd']:.2f}")
print(f"  Next node: {state_after_crash.next}")

print("\n=== RESUME (picks up where it stopped) ===")
result = graph.invoke(Command(resume="continue"), config)

state_after_resume = graph.get_state(config)
print(f"\nState after the resume:")
print(f"  Steps completed: {state_after_resume.values['steps_done']}")
print(f"  Accumulated cost: ${state_after_resume.values['cost_usd']:.2f}")
print(f"  Next node: {state_after_resume.next}")
# Expected output:
# === FIRST RUN (interrupted at step 3) ===
#   Running step 1 (cost: $0.10)...
#   Running step 2 (cost: $0.10)...
#   💥 Interruption at step 3...
#
# State after the crash:
#   Steps completed: ['step_1', 'step_2']
#   Accumulated cost: $0.20
#   Next node: ('step_3',)
#
# === RESUME (picks up where it stopped) ===
#   Running step 3 (cost: $0.10)...
#   Running step 4 (cost: $0.10)...
#
# State after the resume:
#   Steps completed: ['step_1', 'step_2', 'step_3', 'step_4']
#   Accumulated cost: $0.40
#   Next node: ()

Look at what happened:

  1. The first run completed steps 1 and 2, then got interrupted at step 3
  2. The state records $0.20 spent and 2 steps done
  3. next says ('step_3',) — the graph knows exactly where to pick up
  4. The resume runs steps 3 and 4 without repeating steps 1 and 2
  5. Total cost: $0.40 (no repetition) instead of $0.80 (if we redid everything)

Long-running agents

Some agents run for minutes or hours. A Research Assistant analyzing 50 papers, an agent processing 1000 database records, or a generation pipeline building 20 sections of a document. These are the agents that benefit most from durable execution:

from dotenv import load_dotenv
load_dotenv()

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

class BatchState(TypedDict):
    items_to_process: list[str]
    processed: Annotated[list[dict], operator.add]
    current_batch: int
    total_batches: int

def process_batch(state: BatchState) -> dict:
    """Processes one batch of items. Each batch is a checkpoint."""
    batch_num = state["current_batch"]
    batch_size = 3
    start_idx = batch_num * batch_size
    end_idx = min(start_idx + batch_size, len(state["items_to_process"]))

    items = state["items_to_process"][start_idx:end_idx]

    results = []
    for item in items:
        time.sleep(0.1)
        results.append({
            "item": item,
            "result": f"Processed: {item}",
            "batch": batch_num,
        })

    return {
        "processed": results,
        "current_batch": batch_num + 1,
    }

def should_continue(state: BatchState) -> str:
    if state["current_batch"] >= state["total_batches"]:
        return "done"
    return "process"

def finalize(state: BatchState) -> dict:
    return {}

graph_builder = StateGraph(BatchState)
graph_builder.add_node("process", process_batch)
graph_builder.add_node("done", finalize)

graph_builder.add_edge(START, "process")
graph_builder.add_conditional_edges(
    "process", should_continue,
    {"process": "process", "done": "done"},
)
graph_builder.add_edge("done", END)

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

items = [f"item_{i}" for i in range(12)]
config = {"configurable": {"thread_id": "batch-001"}}

result = graph.invoke(
    {
        "items_to_process": items,
        "processed": [],
        "current_batch": 0,
        "total_batches": 4,
    },
    config,
)

print(f"Items processed: {len(result['processed'])}")
print(f"Batches completed: {result['current_batch']}")
for item in result["processed"]:
    print(f"  Batch {item['batch']}: {item['result']}")
# Expected output:
# Items processed: 12
# Batches completed: 4
#   Batch 0: Processed: item_0
#   Batch 0: Processed: item_1
#   Batch 0: Processed: item_2
#   Batch 1: Processed: item_3
#   Batch 1: Processed: item_4
#   Batch 1: Processed: item_5
#   Batch 2: Processed: item_6
#   Batch 2: Processed: item_7
#   Batch 2: Processed: item_8
#   Batch 3: Processed: item_9
#   Batch 3: Processed: item_10
#   Batch 3: Processed: item_11

The key pattern: the graph uses a loop with a conditional edge that processes one batch per iteration. Every iteration produces a checkpoint. If the process dies after batch 2, restarting with the same thread_id resumes from batch 3 — the 6 items already processed aren't redone.

Idempotent nodes: designing for safe re-execution

There's an important edge case: what if the process dies while a node is running, before its checkpoint gets written? In that case the node will re-run on restart. That's why your nodes need to be idempotent — safe to run more than once without duplicating side effects.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    processed_ids: Annotated[list[str], operator.add]
    results: Annotated[list[str], operator.add]

def idempotent_process(state: State) -> dict:
    """Idempotent node: checks whether it already ran before acting."""
    task_id = "api-call-001"

    if task_id in state.get("processed_ids", []):
        print(f"  ⏭️  {task_id} already processed, skipping...")
        return {"processed_ids": [], "results": []}

    print(f"  🔄 Processing {task_id}...")
    result = f"Result of {task_id}"

    return {
        "processed_ids": [task_id],
        "results": [result],
    }

def non_idempotent_process(state: State) -> dict:
    """NON-idempotent node: every run adds a duplicate."""
    task_id = "api-call-002"
    print(f"  🔄 Processing {task_id} (no duplicate check)...")
    return {
        "processed_ids": [task_id],
        "results": [f"Result of {task_id}"],
    }

graph_builder = StateGraph(State)
graph_builder.add_node("safe", idempotent_process)
graph_builder.add_node("unsafe", non_idempotent_process)

graph_builder.add_edge(START, "safe")
graph_builder.add_edge("safe", "unsafe")
graph_builder.add_edge("unsafe", END)

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

config = {"configurable": {"thread_id": "idempotent-demo"}}

print("=== First run ===")
result = graph.invoke(
    {"task": "demo", "processed_ids": [], "results": []},
    config,
)
print(f"Results: {result['results']}")
print(f"Processed IDs: {result['processed_ids']}")
# Expected output:
# === First run ===
#   🔄 Processing api-call-001...
#   🔄 Processing api-call-002 (no duplicate check)...
# Results: ['Result of api-call-001', 'Result of api-call-002']
# Processed IDs: ['api-call-001', 'api-call-002']

Idempotency patterns

OperationNot idempotentIdempotent
API callAlways callsChecks whether you already have the result
DB insertINSERT (duplicates)INSERT ... ON CONFLICT DO NOTHING
Send an emailAlways sendsChecks an email_sent flag in the state
Generate a fileAlways overwritesChecks whether the file already exists
Charge a paymentAlways chargesUses the payment API's idempotency_key

The rule: if a node has side effects (sends data, charges money, mutates databases), make it idempotent. Nodes that only transform internal state don't need extra protection, because the checkpoint handles them.

Automatic retry at the graph level

LangGraph lets you configure retry policies directly on the nodes. This complements durable execution — a node that fails on a transient error gets retried automatically before the graph treats it as a failure:

from dotenv import load_dotenv
load_dotenv()

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

CALL_COUNT = 0

class State(TypedDict):
    query: str
    result: str
    attempts: int

def flaky_api_call(state: State) -> dict:
    """Simulates an API that fails intermittently."""
    global CALL_COUNT
    CALL_COUNT += 1

    if CALL_COUNT <= 2:
        raise ConnectionError(f"API timeout (attempt #{CALL_COUNT})")

    return {
        "result": f"Data fetched successfully on attempt #{CALL_COUNT}",
        "attempts": CALL_COUNT,
    }

graph_builder = StateGraph(State)
graph_builder.add_node(
    "api_call",
    flaky_api_call,
    retry={"max_attempts": 4, "delay": 0.5, "multiplier": 2.0},
)

graph_builder.add_edge(START, "api_call")
graph_builder.add_edge("api_call", END)

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

CALL_COUNT = 0
config = {"configurable": {"thread_id": "retry-demo"}}
result = graph.invoke({"query": "test", "result": "", "attempts": 0}, config)
print(f"Result: {result['result']}")
print(f"Attempts needed: {result['attempts']}")
# Expected output:
# Result: Data fetched successfully on attempt #3
# Attempts needed: 3

The retry parameter on add_node configures:

  • ✅ max_attempts: how many times to try (4 = 1 original + 3 retries)
  • ✅ delay: base wait between attempts (in seconds)
  • ✅ multiplier: exponential backoff factor (2.0 = the delay doubles each time)

This is automatic — you don't write retry logic in every node. LangGraph handles it. If the node still fails after all the retries, then the error propagates so your error handling can catch it.

Checking the checkpoint before expensive operations

An advanced pattern: before running an expensive operation (a $1 API call, 10 minutes of processing), check whether the result already exists in the state:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    cache: dict
    results: Annotated[list[str], operator.add]
    cost_saved: float

def expensive_analysis(state: State) -> dict:
    """Expensive operation: $1.00 per run. Checks the cache first."""
    cache_key = f"analysis_{state['topic']}"
    cache = state.get("cache", {})

    if cache_key in cache:
        print(f"  💰 Cache hit for '{cache_key}' — saved $1.00")
        return {
            "results": [cache[cache_key]],
            "cost_saved": state.get("cost_saved", 0) + 1.00,
        }

    print(f"  🔄 Running the expensive analysis for '{state['topic']}'...")
    result = f"Deep analysis of '{state['topic']}': 15 papers reviewed, 3 trends identified."

    new_cache = {**cache, cache_key: result}
    return {
        "results": [result],
        "cache": new_cache,
        "cost_saved": state.get("cost_saved", 0),
    }

def secondary_analysis(state: State) -> dict:
    print(f"  🔄 Secondary analysis...")
    return {"results": [f"Complementary analysis of '{state['topic']}'"]}

graph_builder = StateGraph(State)
graph_builder.add_node("expensive", expensive_analysis)
graph_builder.add_node("secondary", secondary_analysis)

graph_builder.add_edge(START, "expensive")
graph_builder.add_edge("expensive", "secondary")
graph_builder.add_edge("secondary", END)

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

config = {"configurable": {"thread_id": "cost-aware-001"}}

print("=== First run ===")
result1 = graph.invoke(
    {"topic": "transformer architectures", "cache": {}, "results": [], "cost_saved": 0},
    config,
)
print(f"Results: {len(result1['results'])}")
print(f"Cost saved: ${result1['cost_saved']:.2f}")

config2 = {"configurable": {"thread_id": "cost-aware-002"}}

print("\n=== Second run (with the cache from the first) ===")
result2 = graph.invoke(
    {
        "topic": "transformer architectures",
        "cache": result1["cache"],
        "results": [],
        "cost_saved": 0,
    },
    config2,
)
print(f"Results: {len(result2['results'])}")
print(f"Cost saved: ${result2['cost_saved']:.2f}")
# Expected output:
# === First run ===
#   🔄 Running the expensive analysis for 'transformer architectures'...
#   🔄 Secondary analysis...
# Results: 2
# Cost saved: $0.00
#
# === Second run (with the cache from the first) ===
#   💰 Cache hit for 'analysis_transformer architectures' — saved $1.00
#   🔄 Secondary analysis...
# Results: 2
# Cost saved: $1.00

The cache lives in the graph's state. If you're using PostgresSaver, the cache gets persisted along with the checkpoint. The next run with the same data can skip the expensive operation entirely.

From MemorySaver to PostgresSaver: real durability

MemorySaver is perfect for development and testing, but it has one fundamental limitation: it lives in memory. If the process restarts, the checkpoints are gone. For real durable execution in production, you need PostgresSaver:

# Development: MemorySaver (fast, no dependencies)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# Production: PostgresSaver (durable, survives restarts)
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:password@localhost:5432/mydb"
)

The change is one line. All of your graph code stays identical. The nodes neither know nor care which checkpointer you're using.

MemorySaver                          PostgresSaver
  ├── Fast (in-memory)                 ├── Durable (disk/network)
  ├── No dependencies                   ├── Requires PostgreSQL
  ├── Lost on restart                   ├── Survives restarts
  ├── Perfect for dev/test              ├── Perfect for production
  └── Single process                    └── Multi-process/multi-server

With PostgresSaver:

  • ✅ The process can die and restart → the checkpoints live in PostgreSQL
  • ✅ Multiple servers can reach the same threads
  • ✅ You can inspect checkpoints directly in the database
  • ✅ PostgreSQL backups = a backup of your agents' entire state

Design patterns for durable agents

Pattern 1: Explicit progress tracking

Always include a progress field in your state so you know exactly where the agent is:

from typing import TypedDict, Annotated
import operator

class DurableState(TypedDict):
    task: str
    steps_completed: Annotated[list[str], operator.add]
    total_steps: int
    current_step: int
    result: str

Pattern 2: Atomic operations per node

Each node should do one well-defined thing. If a node does 3 operations and fails on the second, all 3 get re-run. Split them into independent nodes:

# ❌ One node doing too much
def do_everything(state):
    a = call_api_a()    # $0.50
    b = call_api_b()    # $0.50 ← fails here
    c = process(a, b)   # never runs
    return {"result": c}
# If api_b fails, next time api_a runs again ($0.50 wasted)

# ✅ Each node is one atomic operation
def call_a(state):
    return {"data_a": call_api_a()}  # checkpoint right after this

def call_b(state):
    return {"data_b": call_api_b()}  # if it fails, only this repeats

def process(state):
    return {"result": process(state["data_a"], state["data_b"])}

Pattern 3: Checkpoint verification

Before an expensive operation, check whether it already completed:

def smart_node(state):
    if state.get("expensive_result"):
        return {}
    result = expensive_operation()
    return {"expensive_result": result}

Pattern 4: Graceful shutdown

Design your agent so it can stop cleanly at any point:

import signal

shutdown_requested = False

def handle_shutdown(signum, frame):
    global shutdown_requested
    shutdown_requested = True
    print("Shutdown requested, finishing after the current node...")

signal.signal(signal.SIGTERM, handle_shutdown)
signal.signal(signal.SIGINT, handle_shutdown)

The checkpoint gets written after every node, so a shutdown between nodes always leaves the state consistent.

Troubleshooting

Problem 1: "The graph repeats every step on restart"

Symptom: You use the same thread_id but the graph runs from the beginning.

Cause: You're using MemorySaver and the process restarted (the in-memory data was lost).

Fix: Use PostgresSaver for real durability:

# MemorySaver loses data on restart
checkpointer = MemorySaver()  # ← development only

# PostgresSaver persists across restarts
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

Problem 2: "The agent processes duplicate items after a crash"

Symptom: On restart, the agent reprocesses items it had already handled.

Cause: The node isn't idempotent — it doesn't check whether the work was already done.

Fix: Add an idempotency check:

def process_item(state):
    item_id = state["current_item_id"]
    if item_id in state.get("processed_ids", []):
        return {}  # already processed
    result = do_work(item_id)
    return {"processed_ids": [item_id], "results": [result]}

Problem 3: "The checkpoints take up too much space in PostgreSQL"

Symptom: The checkpoints table grows out of control.

Cause: Every invocation of every thread writes multiple checkpoints.

Fix: Implement periodic cleanup of old threads:

# Keep only the last N checkpoints per thread
# or delete threads that completed more than X days ago
# This is handled at the database level with SQL queries

Problem 4: "The automatic retry never fires"

Symptom: The node fails and the graph errors out immediately.

Cause: The retry configuration isn't on add_node.

Fix: Check that retry is configured correctly:

# ❌ No retry
graph_builder.add_node("my_node", my_func)

# ✅ With retry
graph_builder.add_node(
    "my_node", my_func,
    retry={"max_attempts": 3, "delay": 1.0, "multiplier": 2.0},
)

Exercises

Exercise 1: Basic durable execution (Easy)

Build a graph with 4 sequential nodes representing the steps of a research run. Use MemorySaver. Run the graph and then inspect the final state with get_state() to verify that every step completed. The state should include steps_completed and total_cost.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    steps_completed: Annotated[list[str], operator.add]
    total_cost: float

def search(state: State) -> dict:
    return {
        "steps_completed": ["search"],
        "total_cost": state.get("total_cost", 0) + 0.05,
    }

def analyze(state: State) -> dict:
    return {
        "steps_completed": ["analyze"],
        "total_cost": state.get("total_cost", 0) + 0.10,
    }

def synthesize(state: State) -> dict:
    return {
        "steps_completed": ["synthesize"],
        "total_cost": state.get("total_cost", 0) + 0.15,
    }

def format_report(state: State) -> dict:
    return {
        "steps_completed": ["format"],
        "total_cost": state.get("total_cost", 0) + 0.02,
    }

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("synthesize", synthesize)
graph_builder.add_node("format", format_report)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "synthesize")
graph_builder.add_edge("synthesize", "format")
graph_builder.add_edge("format", END)

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

config = {"configurable": {"thread_id": "exercise-1"}}
result = graph.invoke(
    {"topic": "durable execution", "steps_completed": [], "total_cost": 0.0},
    config,
)

state = graph.get_state(config)
print(f"Steps completed: {state.values['steps_completed']}")
print(f"Total cost: ${state.values['total_cost']:.2f}")
print(f"Run complete: {state.next == ()}")
# Expected output:
# Steps completed: ['search', 'analyze', 'synthesize', 'format']
# Total cost: $0.32
# Run complete: True

Exercise 2: Crash simulation with interrupt (Easy)

Build a 3-node graph where the second node uses interrupt() to simulate a crash. Inspect the state after the interruption (which steps completed, which node is next). Then resume the run with Command(resume=...).

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    data: str
    log: Annotated[list[str], operator.add]

def step_a(state: State) -> dict:
    return {"log": ["step_a done"]}

def step_b(state: State) -> dict:
    interrupt("Simulating a crash in step_b")
    return {"log": ["step_b done"]}

def step_c(state: State) -> dict:
    return {"log": ["step_c done"], "data": "final result"}

graph_builder = StateGraph(State)
graph_builder.add_node("a", step_a)
graph_builder.add_node("b", step_b)
graph_builder.add_node("c", step_c)

graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", "c")
graph_builder.add_edge("c", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "crash-exercise"}}

print("=== Initial run (gets interrupted) ===")
result = graph.invoke({"data": "", "log": []}, config)

state = graph.get_state(config)
print(f"Log: {state.values['log']}")
print(f"Next node: {state.next}")

print("\n=== Resume ===")
result = graph.invoke(Command(resume="ok"), config)

state = graph.get_state(config)
print(f"Log: {state.values['log']}")
print(f"Next node: {state.next}")
print(f"Result: {state.values['data']}")
# Expected output:
# === Initial run (gets interrupted) ===
# Log: ['step_a done']
# Next node: ('b',)
#
# === Resume ===
# Log: ['step_a done', 'step_b done', 'step_c done']
# Next node: ()
# Result: final result

Exercise 3: Batch processing with a loop (Medium)

Build a graph that processes a list of 10 items in batches of 3. The graph should use a loop (conditional edge) that repeats the processing node until every batch is done. Use checkpointing and print the progress on each batch.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    items: list[str]
    processed: Annotated[list[str], operator.add]
    batch_num: int
    batch_size: int

def process_batch(state: State) -> dict:
    batch_num = state["batch_num"]
    batch_size = state["batch_size"]
    start = batch_num * batch_size
    end = min(start + batch_size, len(state["items"]))

    batch_items = state["items"][start:end]
    results = [f"✅ {item}" for item in batch_items]

    print(f"  Batch {batch_num}: processed {len(results)} items ({start}-{end-1})")

    return {
        "processed": results,
        "batch_num": batch_num + 1,
    }

def should_continue(state: State) -> str:
    total_batches = -(-len(state["items"]) // state["batch_size"])
    if state["batch_num"] >= total_batches:
        return "done"
    return "next_batch"

def finalize(state: State) -> dict:
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("process", process_batch)
graph_builder.add_node("done", finalize)

graph_builder.add_edge(START, "process")
graph_builder.add_conditional_edges(
    "process", should_continue,
    {"next_batch": "process", "done": "done"},
)
graph_builder.add_edge("done", END)

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

items = [f"doc_{i}" for i in range(10)]
config = {"configurable": {"thread_id": "batch-exercise"}}

result = graph.invoke(
    {"items": items, "processed": [], "batch_num": 0, "batch_size": 3},
    config,
)

print(f"\nTotal processed: {len(result['processed'])}")
print(f"Batches run: {result['batch_num']}")
for item in result["processed"]:
    print(f"  {item}")
# Expected output:
#   Batch 0: processed 3 items (0-2)
#   Batch 1: processed 3 items (3-5)
#   Batch 2: processed 3 items (6-8)
#   Batch 3: processed 1 items (9-9)
#
# Total processed: 10
# Batches run: 4
#   ✅ doc_0
#   ✅ doc_1
#   ...
#   ✅ doc_9

Exercise 4: Idempotent node with a check (Medium)

Build a graph with a node that simulates sending an email. The node must be idempotent: if email_sent is already True in the state, it doesn't send again. Run the graph twice with the same thread and verify that the email is only "sent" once.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    recipient: str
    email_sent: bool
    send_count: int
    log: str

def prepare_email(state: State) -> dict:
    return {"log": f"Email prepared for {state['recipient']}"}

def send_email(state: State) -> dict:
    if state.get("email_sent", False):
        print(f"  ⏭️  Email to {state['recipient']} already sent. Skipping.")
        return {}

    print(f"  📧 Sending email to {state['recipient']}...")
    return {
        "email_sent": True,
        "send_count": state.get("send_count", 0) + 1,
        "log": f"Email sent to {state['recipient']}",
    }

def confirm(state: State) -> dict:
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("prepare", prepare_email)
graph_builder.add_node("send", send_email)
graph_builder.add_node("confirm", confirm)

graph_builder.add_edge(START, "prepare")
graph_builder.add_edge("prepare", "send")
graph_builder.add_edge("send", "confirm")
graph_builder.add_edge("confirm", END)

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

config = {"configurable": {"thread_id": "email-idempotent"}}

print("=== First run ===")
result1 = graph.invoke(
    {"recipient": "user@email.com", "email_sent": False, "send_count": 0, "log": ""},
    config,
)
print(f"Email sent: {result1['email_sent']}")
print(f"Times sent: {result1['send_count']}")

config2 = {"configurable": {"thread_id": "email-idempotent-2"}}

print("\n=== Second run (simulates a re-run) ===")
result2 = graph.invoke(
    {
        "recipient": "user@email.com",
        "email_sent": result1["email_sent"],
        "send_count": result1["send_count"],
        "log": "",
    },
    config2,
)
print(f"Email sent: {result2['email_sent']}")
print(f"Times sent: {result2['send_count']}")
# Expected output:
# === First run ===
#   📧 Sending email to user@email.com...
# Email sent: True
# Times sent: 1
#
# === Second run (simulates a re-run) ===
#   ⏭️  Email to user@email.com already sent. Skipping.
# Email sent: True
# Times sent: 1

Exercise 5: Automatic retry configured per node (Medium)

Build a graph with 2 nodes: one stable (always works) and one unstable (fails the first 2 times). Configure automatic retry on the unstable node with max_attempts=4. Verify that the graph completes successfully without any manual error handling.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

UNSTABLE_CALLS = 0

class State(TypedDict):
    input: str
    stable_result: str
    unstable_result: str

def stable_node(state: State) -> dict:
    return {"stable_result": f"Stable analysis of '{state['input']}'"}

def unstable_node(state: State) -> dict:
    global UNSTABLE_CALLS
    UNSTABLE_CALLS += 1

    if UNSTABLE_CALLS <= 2:
        raise ConnectionError(f"Unstable service: failure #{UNSTABLE_CALLS}")

    return {"unstable_result": f"External data fetched (attempt #{UNSTABLE_CALLS})"}

graph_builder = StateGraph(State)
graph_builder.add_node("stable", stable_node)
graph_builder.add_node(
    "unstable",
    unstable_node,
    retry={"max_attempts": 4, "delay": 0.2, "multiplier": 2.0},
)

graph_builder.add_edge(START, "stable")
graph_builder.add_edge("stable", "unstable")
graph_builder.add_edge("unstable", END)

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

UNSTABLE_CALLS = 0
config = {"configurable": {"thread_id": "retry-exercise"}}
result = graph.invoke(
    {"input": "test query", "stable_result": "", "unstable_result": ""},
    config,
)
print(f"Stable: {result['stable_result']}")
print(f"Unstable: {result['unstable_result']}")
print(f"Attempts on the unstable node: {UNSTABLE_CALLS}")
# Expected output:
# Stable: Stable analysis of 'test query'
# Unstable: External data fetched (attempt #3)
# Attempts on the unstable node: 3

Exercise 6: Full durable pipeline with cost tracking (Advanced)

Build a research graph with 5 sequential nodes. Each node has a different simulated cost. Use interrupt() to simulate a crash after node 3. Show the accumulated cost before and after the crash. Resume the run and verify that the total cost is correct (no duplicated costs). Include an execution_log field that records each step with a timestamp.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    cost_usd: float
    results: Annotated[list[str], operator.add]
    execution_log: Annotated[list[dict], operator.add]

def make_step(name: str, cost: float, crash: bool = False):
    def node(state: State) -> dict:
        if crash:
            interrupt(f"Simulated crash in {name}")

        log_entry = {
            "step": name,
            "cost": cost,
            "timestamp": time.time(),
            "cumulative_cost": state.get("cost_usd", 0) + cost,
        }

        return {
            "results": [f"{name}: done (${cost:.2f})"],
            "cost_usd": state.get("cost_usd", 0) + cost,
            "execution_log": [log_entry],
        }
    return node

graph_builder = StateGraph(State)
graph_builder.add_node("search", make_step("search", 0.05))
graph_builder.add_node("fetch", make_step("fetch_papers", 0.15))
graph_builder.add_node("analyze", make_step("analyze", 0.25))
graph_builder.add_node("crash_point", make_step("deep_analysis", 0.30, crash=True))
graph_builder.add_node("report", make_step("generate_report", 0.20))

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "fetch")
graph_builder.add_edge("fetch", "analyze")
graph_builder.add_edge("analyze", "crash_point")
graph_builder.add_edge("crash_point", "report")
graph_builder.add_edge("report", END)

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

config = {"configurable": {"thread_id": "durable-pipeline"}}

print("=== Initial run (crash after analyze) ===")
result = graph.invoke(
    {"topic": "AI safety", "cost_usd": 0.0, "results": [], "execution_log": []},
    config,
)

state = graph.get_state(config)
print(f"Steps completed: {len(state.values['results'])}")
print(f"Accumulated cost: ${state.values['cost_usd']:.2f}")
print(f"Next node: {state.next}")
for entry in state.values["execution_log"]:
    print(f"  ${entry['cost']:.2f} — {entry['step']}")

print("\n=== Resume (picks up at deep_analysis) ===")
result = graph.invoke(Command(resume="continue"), config)

state = graph.get_state(config)
print(f"Steps completed: {len(state.values['results'])}")
print(f"Total cost: ${state.values['cost_usd']:.2f}")
print(f"Next node: {state.next}")
print(f"\nFull log:")
for entry in state.values["execution_log"]:
    print(f"  ${entry['cost']:.2f} — {entry['step']} (cumulative: ${entry['cumulative_cost']:.2f})")
# Expected output:
# === Initial run (crash after analyze) ===
# Steps completed: 3
# Accumulated cost: $0.45
# Next node: ('crash_point',)
#   $0.05 — search
#   $0.15 — fetch_papers
#   $0.25 — analyze
#
# === Resume (picks up at deep_analysis) ===
# Steps completed: 5
# Total cost: $0.95
# Next node: ()
#
# Full log:
#   $0.05 — search (cumulative: $0.05)
#   $0.15 — fetch_papers (cumulative: $0.20)
#   $0.25 — analyze (cumulative: $0.45)
#   $0.30 — deep_analysis (cumulative: $0.75)
#   $0.20 — generate_report (cumulative: $0.95)

Summary

In this capsule you learned:

  • Durable execution = checkpointing + thread management — it isn't a separate feature, it's the correct use of what you already knew. Every completed node writes a checkpoint, and when you restart with the same thread_id, the graph resumes where it stopped
  • The value is tangible and measurable: a 5-step research run at $0.10 each that crashes at step 3 costs $0.00 extra with durable execution vs $0.50 without it. Multiply by 100 users a day
  • Idempotent nodes are mandatory for operations with side effects — a node that charges a payment or sends an email must check whether it already ran before acting
  • Automatic retry at the node level handles transient errors with no manual code — retry={"max_attempts": 3, "delay": 1.0} on add_node is all you need
  • From MemorySaver to PostgresSaver is one line — the change is trivial but the payoff is huge: real durability that survives process restarts
  • Batch processing with loops is the natural pattern for long-running agents — every loop iteration is a checkpoint, every processed batch is saved progress
  • Key design patterns: explicit progress tracking, atomic operations per node, checkpoint verification before expensive operations, and graceful shutdown to keep the state consistent

Next capsule: Time-Travel Debugging — walking your agent's full state history, rewinding to any step, and understanding exactly why your agent made each decision.

Further reading

  1. LangGraph Persistence — Persistence and checkpointing concepts in LangGraph
  2. LangGraph Checkpointers — MemorySaver, PostgresSaver, and other backends
  3. How to use LangGraph's built-in retry policy — Automatic retry at the node level
  4. LangGraph Interrupt — Breakpoints and interruptions for execution control
  5. PostgresSaver Setup — Configuring PostgreSQL as the checkpoint backend
  6. Idempotency Patterns — AWS: Making retries safe with idempotent APIs

Module 8 — LangChain & LangGraph: From Chains to Agents

← PreviousPersistence with PostgresSaver and RedisNext →Time-Travel Debugging