durable-state-patterns

Durable State Patterns

Build agents that remember state across failures, restarts, and sessions.

Why Durable State?

Without durability:

• Long-running agents lose progress on crash
• Users can't resume conversations after timeout
• No audit trail of agent decisions
• Expensive recomputation on every restart

With durability:

• Resume from any checkpoint
• Survive infrastructure failures
• Debug by replaying history
• Share state across instances

LangGraph Checkpointing

Basic Setup

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# In-memory (development only)
memory_checkpointer = MemorySaver()

# SQLite (single instance)
sqlite_checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# PostgreSQL (production, multi-instance)
postgres_checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/db"
)

# Compile with checkpointer
app = workflow.compile(checkpointer=postgres_checkpointer)

Thread-Based State

from uuid import uuid4

# Each conversation gets a unique thread_id
thread_id = str(uuid4())

# First invocation
config = {"configurable": {"thread_id": thread_id}}
result1 = app.invoke(
    {"messages": [{"role": "user", "content": "My name is Alice"}]},
    config
)

# Later invocation (same thread = same state)
result2 = app.invoke(
    {"messages": [{"role": "user", "content": "What's my name?"}]},
    config
)
# Agent remembers: "Your name is Alice"

# Different thread = fresh state
other_config = {"configurable": {"thread_id": str(uuid4())}}
result3 = app.invoke(
    {"messages": [{"role": "user", "content": "What's my name?"}]},
    other_config
)
# Agent doesn't know: "I don't have that information"

State Schema Design

Versioned State

from typing import TypedDict, Annotated
import operator

class AgentStateV1(TypedDict):
    """Version 1 of agent state."""
    messages: Annotated[list, operator.add]
    user_id: str

class AgentStateV2(TypedDict):
    """Version 2 with preferences."""
    messages: Annotated[list, operator.add]
    user_id: str
    preferences: dict  # New field
    state_version: int  # Track version

def migrate_v1_to_v2(old_state: AgentStateV1) -> AgentStateV2:
    """Migrate old state to new schema."""
    return {
        **old_state,
        "preferences": {},  # Default value
        "state_version": 2
    }