Graph Persistence

🧠 LangGraph: Graph Persistence

The Magic of Remembering

Imagine you’re building the world’s most amazing LEGO castle. You’ve spent hours placing each brick perfectly. But then—oh no!—it’s dinnertime, and you have to stop.

What if, when you came back, your castle was exactly where you left it? Every single brick in place. Every tower standing tall.

That’s what Graph Persistence does for your AI workflows. It saves everything so you can pause, come back, and continue like nothing happened.

🎯 What We’ll Learn

graph LR
    A["Graph Persistence"] --> B["Checkpointing"]
    A --> C["Checkpoint Savers"]
    A --> D["Memory vs Checkpointer"]
    A --> E["Graph Interrupts"]
    A --> F["Time Travel"]
    A --> G["Durable Execution"]

    style A fill:#667eea,color:#fff
    style B fill:#4ECDC4,color:#fff
    style C fill:#FF6B6B,color:#fff
    style D fill:#95E1D3,color:#333
    style E fill:#F38181,color:#fff
    style F fill:#AA96DA,color:#fff
    style G fill:#FCBAD3,color:#333

1️⃣ Graph Checkpointing

What Is It?

Think of checkpointing like pressing the SAVE button in a video game.

You’re playing a tough level. You beat the dragon. You find the treasure. Before the next challenge, you SAVE.

If something goes wrong later, you can load that save and start from that exact moment—with the dragon defeated and treasure in hand.

How It Works

Every time your graph finishes a step, LangGraph can take a snapshot. This snapshot captures:

• The current state (all your data)
• Which nodes ran (your progress)
• Everything the AI remembers (messages, context)

Simple Example

from langgraph.checkpoint import MemorySaver

# Create a "save file" system
checkpointer = MemorySaver()

# Build your graph with checkpointing
graph = workflow.compile(
    checkpointer=checkpointer
)

# Run with a unique ID (like a save slot)
config = {"configurable": {"thread_id": "game-1"}}
result = graph.invoke(input_data, config)

What happens:

1. Graph runs step by step
2. After each step → 📸 snapshot saved
3. You can come back anytime and continue!

2️⃣ Checkpoint Savers

Your Storage Options

Checkpointing needs somewhere to store those snapshots. Think of it like choosing where to keep your LEGO instruction book:

MemorySaver (The Whiteboard)

Fast and easy! But when you turn off your computer… poof! Gone.

from langgraph.checkpoint.memory import MemorySaver

saver = MemorySaver()

SqliteSaver (The Notebook)

Saves to a file. Still there tomorrow!

from langgraph.checkpoint.sqlite import SqliteSaver

saver = SqliteSaver.from_conn_string(
    "my_saves.db"
)

PostgresSaver (The Library)

Professional. Reliable. Perfect for real apps.

from langgraph.checkpoint.postgres import PostgresSaver

saver = PostgresSaver.from_conn_string(
    "postgresql://user:pass@host/db"
)

3️⃣ Memory Store vs Checkpointer

Two Different Jobs

This is where it gets interesting! LangGraph has two ways to remember things:

graph TD
    A["Remembering Stuff"] --> B["Checkpointer"]
    A --> C["Memory Store"]

    B --> D["Saves WHOLE<br/>conversation state"]
    C --> E["Saves SPECIFIC<br/>facts to recall later"]

    D --> F["Like saving<br/>your game"]
    E --> G["Like writing<br/>in your diary"]

    style B fill:#4ECDC4,color:#fff
    style C fill:#FF6B6B,color:#fff

Checkpointer = Game Save

• Saves everything at a moment in time
• Used to pause and resume
• One conversation = one save slot

Memory Store = Diary/Notes

• Saves specific facts to remember later
• “User’s name is Alex”
• “User prefers short answers”
• Can be shared across conversations!

Example: The Difference

# CHECKPOINTER: Saves the whole chat
# "Where did we leave off?"
checkpointer = MemorySaver()

# MEMORY STORE: Saves specific facts
# "What do I know about this user?"
from langgraph.store.memory import InMemoryStore
memory_store = InMemoryStore()

Real-life analogy:

• Checkpointer = Bookmarking your page in a book
• Memory Store = Writing notes about the book’s characters

4️⃣ Graph Persistence

The Full Picture

Graph Persistence is the big umbrella term. It means your graph can:

1. ✅ Stop at any point
2. ✅ Save its current state
3. ✅ Resume exactly where it stopped
4. ✅ Survive crashes and restarts

Why This Matters

Imagine your AI assistant is helping plan a vacation:

Step 1: User says "Plan a trip to Paris"
Step 2: AI searches for flights ✓
Step 3: AI searches for hotels ✓
--- Computer crashes! ---
Step 4: AI should book the best option

Without persistence: Start over from scratch. User frustrated. 😤

With persistence: Resume at Step 4. User happy. 🎉

Making It Work

# 1. Pick your saver
checkpointer = SqliteSaver.from_conn_string("app.db")

# 2. Compile with persistence
graph = workflow.compile(checkpointer=checkpointer)

# 3. Always use a thread_id
config = {"configurable": {"thread_id": "user-123"}}

# 4. Run (or resume) anytime
result = graph.invoke(user_input, config)

5️⃣ Graph Interrupts

Pause for Humans

Sometimes your AI needs to stop and ask before continuing. Like a kid asking “Can I have a cookie?” before taking one.

graph TD
    A["Start"] --> B["AI thinks"]
    B --> C{Need approval?}
    C -->|Yes| D["⏸️ INTERRUPT"]
    D --> E["Human reviews"]
    E --> F["Human says OK"]
    F --> G["Continue"]
    C -->|No| G
    G --> H["Done!"]

    style D fill:#FF6B6B,color:#fff

How to Interrupt

from langgraph.graph import END, StateGraph

def needs_approval(state):
    if state["action"] == "delete_everything":
        # This is dangerous! Pause here.
        return "__interrupt__"
    return "continue"

# Add the interrupt point
graph.add_conditional_edges(
    "check_action",
    needs_approval,
    {
        "__interrupt__": END,
        "continue": "do_action"
    }
)

Real Example: Email Sending

def send_email_node(state):
    # Before sending, let human review
    if not state.get("human_approved"):
        # Interrupt! Wait for approval.
        return {
            "status": "waiting_for_approval",
            "draft_email": state["email"]
        }
    # Human said OK, send it!
    send_email(state["email"])
    return {"status": "sent"}

6️⃣ Time Travel Debugging

Go Back in Time!

Made a mistake? No problem. LangGraph can rewind to any previous checkpoint.

Think of it like having a time machine for your AI:

Checkpoint 1: User asked about weather ←── Go here!
Checkpoint 2: AI searched for info
Checkpoint 3: AI gave wrong city 😬

Instead of starting over, you travel back to Checkpoint 1 and try again.

How to Time Travel

# Get all checkpoints for a thread
all_checkpoints = list(
    checkpointer.list(config)
)

# See what we have
for cp in all_checkpoints:
    print(f"ID: {cp.id}")
    print(f"State: {cp.values}")

# Pick one and go back!
past_config = {
    "configurable": {
        "thread_id": "game-1",
        "checkpoint_id": "abc123"  # Old save
    }
}

# Resume from that point
result = graph.invoke(new_input, past_config)

Why This Is Amazing

7️⃣ Durable Execution

Survive Anything

Durable Execution means your graph will finish no matter what.

• Server restarts? ✅ Keeps going
• Network hiccup? ✅ Retries automatically
• Power outage? ✅ Resumes after reboot

The Secret: Persistence + Retries

from langgraph.checkpoint.postgres import PostgresSaver

# Use a reliable database
saver = PostgresSaver.from_conn_string(DATABASE_URL)

# Graph survives restarts
graph = workflow.compile(
    checkpointer=saver,
    retry_policy={
        "max_attempts": 3,
        "wait_time": 5  # seconds
    }
)

How It Works

graph TD
    A["Graph runs step"] --> B{Success?}
    B -->|Yes| C["Save checkpoint"]
    C --> D["Next step"]
    B -->|No| E{Retries left?}
    E -->|Yes| F["Wait & retry"]
    F --> A
    E -->|No| G["Save error state"]
    G --> H["Alert human"]

    style C fill:#4ECDC4,color:#fff
    style G fill:#FF6B6B,color:#fff

Real World Example

Imagine booking a flight:

def book_flight(state):
    try:
        # Try to book
        result = airline_api.book(state["flight"])
        return {"booking": result, "status": "success"}
    except NetworkError:
        # Don't worry! Graph will retry
        raise  # LangGraph handles this

# If server restarts mid-booking,
# graph resumes and tries again!

🎁 Putting It All Together

Here’s how all these pieces work as a team:

graph TD
    A["Your LangGraph App"] --> B["Checkpointer"]
    B --> C["Saves state after each step"]

    A --> D["Memory Store"]
    D --> E["Remembers user facts"]

    A --> F["Interrupts"]
    F --> G["Pauses for human input"]

    C --> H["Enables time travel"]
    C --> I["Enables durable execution"]

    H --> J["Debug by rewinding"]
    I --> K["Survives crashes"]

    style A fill:#667eea,color:#fff
    style B fill:#4ECDC4,color:#fff
    style D fill:#FF6B6B,color:#fff
    style F fill:#AA96DA,color:#fff

🚀 Quick Start Code

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

# 1. Define your state
class State(TypedDict):
    messages: list
    step: int

# 2. Build your graph
workflow = StateGraph(State)
# ... add nodes and edges ...

# 3. Add persistence
checkpointer = MemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# 4. Run with thread_id
config = {"configurable": {"thread_id": "my-chat"}}
result = graph.invoke({"messages": []}, config)

# 5. Resume anytime with same thread_id
result = graph.invoke({"messages": ["Hello!"]}, config)

🎯 Key Takeaways

🌟 You Did It!

You now understand how LangGraph remembers everything. Your AI apps can:

• ⏸️ Pause and resume anytime
• 🔄 Recover from crashes gracefully
• ⏪ Travel back in time to debug
• 🛑 Stop for human approval when needed
• 🧠 Remember user preferences forever

This is what makes AI apps feel magical and reliable!

Go build something amazing! 🚀