Production-Grade State Management

You've wired PostgreSQL as your LangGraph checkpointer. Checkpoints survive process restarts, conversations persist across sessions, and your agents have memory. Now comes the hard part: making state management reliable, performant, and observable at production scale.

This document codifies our team's standards for managing LangGraph state with PostgreSQL. Every recommendation is opinionated, actionable, and grounded in the framework's persistence architecture. Treat this as a living standard — not a suggestion list.

---

What "Production-Grade" Means for State Management

A production-grade state management layer is not just one that saves data. It is a system that behaves predictably under load, survives infrastructure failures, and provides visibility into what is happening. Reliability across five dimensions:

Dimension	What It Means	What Happens Without It
Connection resilience	Database connections are pooled, recycled, and recovered	Connection exhaustion under traffic spikes
Checkpoint hygiene	Old checkpoints are pruned, orphaned threads are cleaned up	Unbounded table growth, slowing all lookups
State discipline	Only necessary data is serialized and stored	Bloating checkpoints, slow serialization, high storage costs
Thread governance	Thread IDs are structured, tenant-aware, and bounded	Namespace collisions, multi-tenant data leaks
Observability	Checkpoint latency, table sizes, and pool health are monitored	Silent degradation until users complain

These five dimensions map directly to the sections below.

---

Connection Pooling Strategies

Every checkpoint read and write hits PostgreSQL. Without connection pooling, each graph execution opens and closes a database connection — and at scale, that is the bottleneck.

Why Connection Pooling Matters for LangGraph Agents

A single agent conversation can produce dozens of checkpoints (one per super-step). If you have 100 concurrent conversations, each checkpoint operation opens a new connection. PostgreSQL has a default connection limit of 100, and you will hit it fast.

Connection pooling solves three problems:

1. Connection reuse — open connections once, reuse them across requests.
2. Backpressure — limit concurrent database connections to prevent overwhelming the server.
3. Lifecycle management — detect and replace broken connections automatically.

Using asyncpg Connection Pools

LangGraph's PostgresSaver and AsyncPostgresSaver work with psycopg (Psycopg 3) connections. For production, wrap them in a pool:

Sync Pool (psycopg ConnectionPool)

from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://langgraph_user:password@postgres.internal:5432/langgraph"

# Create the pool once at application startup
pool = ConnectionPool(
    conninfo=DB_URI,
    min_size=3,         # Minimum idle connections kept warm
    max_size=10,        # Maximum connections under load
    timeout=5,          # Seconds to wait when pool is exhausted
    max_idle=300,       # Close idle connections after 5 minutes
    open=True,          # Open immediately instead of lazily
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()

# Reuse this checkpointer instance across all requests.
# The pool manages connection lifecycle automatically.
app = graph_builder.compile(checkpointer=checkpointer)

Async Pool (AsyncConnectionPool)

from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DB_URI = "postgresql://langgraph_user:password@postgres.internal:5432/langgraph"

async def create_checkpointer() -> AsyncPostgresSaver:
    """Create an async checkpointer with a connection pool."""
    pool = AsyncConnectionPool(
        conninfo=DB_URI,
        min_size=5,
        max_size=20,
        timeout=5,
        max_idle=300,
        open=True,
    )
    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.setup()
    return checkpointer

# At application startup
checkpointer = await create_checkpointer()
app = graph_builder.compile(checkpointer=checkpointer)

Pool Sizing Recommendations

Pool size depends on your concurrency profile. Use this formula as a starting point:

max_size = (2 * concurrent_graph_executions) + 1
min_size = concurrent_graph_executions / 2

Deployment Profile	min_size	max_size	Rationale
Low traffic (< 10 concurrent)	2	5	Small pool, quick recovery
Medium traffic (10–50 concurrent)	5	15	Balance between reuse and headroom
High traffic (50–200 concurrent)	10	30	Larger pool, monitor closely
Very high traffic (200+ concurrent)	20	50	Consider PgBouncer in front

Never set max_size beyond your PostgreSQL max_connections

Check your PostgreSQL configuration:

SHOW max_connections;

If your server has max_connections = 100 and you run three application instances, each pool should have max_size <= 30 to leave room for admin connections and other services.

Connection Lifecycle Management

The golden rule: create the pool once, reuse it everywhere. Do not create a new pool or checkpointer per request.

# Application startup (e.g., FastAPI lifespan, Celery init, etc.)
APP_STATE: dict = {}

async def startup():
    """Initialize shared resources once at startup."""
    APP_STATE["checkpointer"] = await create_checkpointer()
    APP_STATE["graph"] = graph_builder.compile(
        checkpointer=APP_STATE["checkpointer"]
    )

async def shutdown():
    """Clean up resources on shutdown."""
    checkpointer = APP_STATE.get("checkpointer")
    if checkpointer:
        await checkpointer.close()

Connection Pool Factory Pattern

For applications that need multiple checkpointers (e.g., separate databases per tenant), use a factory:

from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from functools import lru_cache


class CheckpointerFactory:
    """Manages a pool of checkpointers keyed by database."""

    def __init__(self):
        self._pools: dict[str, AsyncConnectionPool] = {}
        self._checkpointers: dict[str, AsyncPostgresSaver] = {}

    async def get_checkpointer(self, db_key: str, uri: str) -> AsyncPostgresSaver:
        """Return or create a checkpointer for the given database."""
        if db_key not in self._checkpointers:
            pool = AsyncConnectionPool(
                conninfo=uri,
                min_size=2,
                max_size=10,
                timeout=5,
                open=True,
            )
            checkpointer = AsyncPostgresSaver(pool)
            await checkpointer.setup()

            self._pools[db_key] = pool
            self._checkpointers[db_key] = checkpointer

        return self._checkpointers[db_key]

    async def close_all(self):
        """Close all pools and checkpointers."""
        for cp in self._checkpointers.values():
            await cp.close()
        for pool in self._pools.values():
            await pool.close()


# Usage
factory = CheckpointerFactory()

async def handle_request(tenant_id: str):
    checkpointer = await factory.get_checkpointer(
        db_key=f"tenant_{tenant_id}",
        uri=get_tenant_db_uri(tenant_id),
    )
    app = graph_builder.compile(checkpointer=checkpointer)
    # ... execute graph ...

---

Checkpoint Lifecycle Management

Checkpoints accumulate. A thread with 50 conversation turns has 50+ checkpoints. Across thousands of threads, this becomes a data management problem.

Checkpoint Retention Policies

Define a retention policy before checkpoints accumulate. Our standard:

Policy	Retention Period	Trigger
Active threads	Keep all checkpoints	Thread was active in last 7 days
Completed threads	Keep last 10 checkpoints	Graph reached END state
Stale threads	Delete entirely	No activity for 30 days
Error threads	Keep 5 checkpoints	Graph failed with unhandled exception

Archiving Old Checkpoints

Instead of deleting immediately, archive checkpoints older than the retention window to a separate table or schema:

-- Archive checkpoints older than 30 days to a separate table
CREATE TABLE IF NOT EXISTS checkpoints_archive AS
SELECT * FROM checkpoints
WHERE created_at < NOW() - INTERVAL '30 days'
AND NOT EXISTS (
    SELECT 1 FROM checkpoints c2
    WHERE c2.thread_id = checkpoints.thread_id
    AND c2.created_at > NOW() - INTERVAL '7 days'
);

-- Delete archived checkpoints from the main table
DELETE FROM checkpoint_writes
WHERE (thread_id, checkpoint_ns, checkpoint_id) IN (
    SELECT thread_id, checkpoint_ns, checkpoint_id
    FROM checkpoints_archive
);

DELETE FROM checkpoint_blobs
WHERE (thread_id, checkpoint_ns, checkpoint_id) IN (
    SELECT thread_id, checkpoint_ns, checkpoint_id
    FROM checkpoints_archive
);

DELETE FROM checkpoints
WHERE (thread_id, checkpoint_ns, checkpoint_id) IN (
    SELECT thread_id, checkpoint_ns, checkpoint_id
    FROM checkpoints_archive
);

Cleaning Up Orphaned Threads

Orphaned threads are threads with checkpoints but no associated application context (e.g., user sessions that expired). Clean them up periodically:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from datetime import datetime, timedelta


async def cleanup_orphaned_threads(
    checkpointer: AsyncPostgresSaver,
    max_age_days: int = 30,
) -> int:
    """Delete threads with no activity beyond the retention window.

    Returns the number of threads deleted.
    """
    cutoff = datetime.now() - timedelta(days=max_age_days)

    # Get all thread IDs older than the cutoff
    async with checkpointer.pool.connection() as conn:
        rows = await conn.fetch(
            """
            SELECT DISTINCT thread_id
            FROM checkpoints
            WHERE created_at < %s
            """,
            cutoff,
        )

    deleted_count = 0
    for row in rows:
        await checkpointer.delete_thread(row["thread_id"])
        deleted_count += 1

    return deleted_count

Schedule this as a cron job or Celery beat task:

# In your task scheduler
from celery import Celery

celery = Celery()

@celery.task(schedule=crontab(hour=3, minute=0))  # Daily at 3 AM
def daily_checkpoint_cleanup():
    """Remove orphaned threads older than 30 days."""
    count = cleanup_orphaned_threads(checkpointer, max_age_days=30)
    logger.info(f"Cleaned up {count} orphaned threads")

Database Maintenance

PostgreSQL checkpoint tables are write-heavy. Regular maintenance keeps performance stable:

-- Run weekly on checkpoint tables
VACUUM ANALYZE checkpoints;
VACUUM ANALYZE checkpoint_blobs;
VACUUM ANALYZE checkpoint_writes;

For heavy-write workloads, consider VACUUM (VERBOSE, ANALYZE) during low-traffic windows and monitor the pg_stat_user_tables view for bloat:

-- Check table bloat on checkpoint tables
SELECT
    relname AS table_name,
    n_dead_tup AS dead_tuples,
    n_live_tup AS live_tuples,
    CASE WHEN n_live_tup > 0
        THEN round(100.0 * n_dead_tup / (n_dead_tup + n_live_tup), 1)
        ELSE 0
    END AS dead_tuple_pct
FROM pg_stat_user_tables
WHERE relname LIKE 'checkpoint%';

Monitoring Checkpoint Table Sizes

Track checkpoint table growth as a leading indicator of retention policy violations:

-- Checkpoint table sizes in human-readable format
SELECT
    relname AS table_name,
    pg_size_pretty(pg_total_relation_size(oid)) AS total_size,
    pg_size_pretty(pg_relation_size(oid)) AS data_size,
    pg_size_pretty(pg_total_relation_size(oid) - pg_relation_size(oid)) AS index_size
FROM pg_class
WHERE relname LIKE 'checkpoint%';

---

State Schema Design

The state schema is the contract between your graph nodes and the checkpointing system. Poor schema design leads to bloated checkpoints, slow serialization, and wasted storage.

Keeping State Small

Every field in your state is serialized and written to PostgreSQL at every super-step. Store only what is shared across nodes:

from typing_extensions import TypedDict, Annotated
import operator


# GOOD: Minimal state — only shared data
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    search_results: Annotated[list, operator.add]
    step_count: int


# BAD: State overloaded with node-local data
class BloatedState(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    search_results: Annotated[list, operator.add]
    step_count: int
    raw_api_response: str          # Only used in one node
    intermediate_embeddings: list  # Only used in one node
    full_document_text: str        # Only used in one node
    debug_trace: dict              # Never needed in production

Rule of thumb: If a value is only read and written by a single node, keep it as a local variable inside that node function, not in state.

Using Reducers Wisely

Reducers control how state fields are updated. Choose them deliberately:

Reducer	Behavior	Best For
operator.add	Appends to list	Message history, accumulated results
operator.or_	Merges dicts (last-write-wins)	Config overrides, metadata
Custom function	Any logic	Complex merge strategies
No reducer (default)	Replaces entirely	Single-value fields like query

from typing_extensions import TypedDict, Annotated
import operator


def merge_search_results(existing: list, incoming: list) -> list:
    """Merge search results, deduplicating by source URL."""
    if not existing:
        return incoming
    seen_urls = {item.get("url") for item in existing}
    return existing + [item for item in incoming if item.get("url") not in seen_urls]


class ResearchState(TypedDict):
    messages: Annotated[list, operator.add]           # Append messages
    query: str                                        # Replace on each update
    results: Annotated[list, merge_search_results]    # Dedup merge
    metadata: Annotated[dict, operator.or_]           # Merge dicts

When to Truncate Message History in State

Unbounded message growth is the most common state bloat problem. Truncate proactively:

from langchain.messages import AnyMessage, trim_messages


def trim_message_history(state: AgentState) -> dict:
    """Trim message history to prevent unbounded growth.

    Keeps the last 20 messages plus the system message,
    and summarizes older messages.
    """
    messages = state.get("messages", [])
    if len(messages) <= 20:
        return {}  # No trimming needed

    # Keep system message + last 20 messages
    system_messages = [m for m in messages if m.type == "system"]
    non_system = [m for m in messages if m.type != "system"]
    trimmed = system_messages + non_system[-20:]

    return {"messages": trimmed}


# Wire it as a node that runs before LLM calls
graph.add_node("trim_history", trim_message_history)
graph.add_edge("trim_history", "call_llm")

For more sophisticated trimming, use LangChain's built-in trim_messages:

from langchain.messages import trim_messages

trimmer = trim_messages(
    messages,
    strategy="last",
    token_counter=len,  # Or use a real tokenizer
    max_tokens=4000,
    allow_partial=False,
    start_on="human",
)

Separating Hot State from Cold State

Not all state needs the same durability or access pattern:

State Type	Examples	Storage
Hot (active conversation)	Current messages, step count, routing state	PostgreSQL checkpointer
Warm (user preferences)	Language preference, tone, known facts	LangGraph Store (PostgresStore)
Cold (archived history)	Full conversation logs, audit trails	Separate PostgreSQL table or object storage

from langgraph.store.postgres import PostgresStore

# Use a Store for cross-thread data that does not belong in checkpoint state
store = PostgresStore.from_conn_string(DB_URI)
await store.setup()

# Store user preferences separately from conversation state
await store.aput(
    namespace=("users", user_id, "preferences"),
    key="default",
    value={
        "language": "en",
        "tone": "professional",
        "max_results": 5,
    },
)

# Read preferences in a node without bloating checkpoint state
async def load_user_preferences(state: AgentState, config: RunnableConfig) -> dict:
    user_id = config["metadata"].get("user_id")
    if not user_id:
        return {}

    item = await store.aget(("users", user_id, "preferences"), "default")
    if not item:
        return {}

    return {"user_preferences": item.value}

State Serialization Performance

LangGraph serializes state using msgpack. Large state objects slow down every checkpoint write. Monitor serialization time:

import time
from langgraph.checkpoint.base import Serialization


def measure_serialization(state: dict, serde: Serialization) -> float:
    """Measure how long it takes to serialize the current state."""
    start = time.perf_counter()
    serialized = serde.dumps(state)
    elapsed = time.perf_counter() - start
    size_kb = len(serialized) / 1024
    logger.debug(f"State serialization: {elapsed:.3f}s, {size_kb:.1f}KB")
    return elapsed

Target: Under 10ms per serialization, under 100KB per checkpoint. If you exceed these, reduce state size.

---

Thread Management

Threads are the unit of isolation in LangGraph. At scale, thread management becomes a governance problem.

Thread ID Strategies

Thread IDs must be unique, bounded in length, and meaningful:

Simple UUID (Default)

import uuid

thread_id = str(uuid.uuid4())
# "550e8400-e29b-41d4-a716-446655440000"
# Always under 255 characters, globally unique

Structured ID (Multi-Tenant)

import uuid

def make_thread_id(tenant_id: str, user_id: str) -> str:
    """Create a structured thread ID with tenant and user context.

    Format: {tenant_id}:{user_id}:{short_uuid}
    Total length is kept under 255 characters.
    """
    short_uuid = uuid.uuid4().hex[:12]
    return f"{tenant_id}:{user_id}:{short_uuid}"

# Example: "acme:usr_9f8e7d:3a1b2c4d5e6f"
thread_id = make_thread_id("acme", "usr_9f8e7d")

Hashed ID (Privacy-Focused)

import hashlib
import uuid

def make_hashed_thread_id(tenant_id: str, session_id: str) -> str:
    """Create a thread ID that embeds tenant info but hashes user data.

    Useful when thread IDs appear in logs but should not expose PII.
    """
    raw = f"{tenant_id}:{session_id}:{uuid.uuid4()}"
    hash_part = hashlib.sha256(raw.encode()).hexdigest()[:20]
    return f"{tenant_id}:{hash_part}"

# Example: "acme:a3f2b1c9d8e7f6a5b4c3"

Always keep thread_id under 255 characters

The PostgreSQL column has a length limit. If you construct composite IDs, verify they stay within bounds:

assert len(thread_id) < 255, f"thread_id too long: {len(thread_id)} chars"

Multi-Tenant Checkpointing Patterns

When serving multiple tenants, isolate their checkpoint data:

# Pattern 1: Thread ID scoping (simplest)
# Embed tenant_id in the thread_id itself
thread_id = f"tenant_{tenant_id}_{uuid.uuid4().hex[:12]}"

# Pattern 2: Separate database per tenant (strongest isolation)
tenant_db_uri = get_tenant_database_uri(tenant_id)
checkpointer = await factory.get_checkpointer(tenant_id, tenant_db_uri)

# Pattern 3: Schema-level isolation (PostgreSQL schemas)
# Each tenant gets its own schema within the same database
async with checkpointer.pool.connection() as conn:
    await conn.execute(f"SET search_path TO tenant_{tenant_id}, public")

Pattern	Isolation	Complexity	Best For
Thread ID scoping	Logical only	Low	Single-tenant databases, moderate scale
Separate database	Physical	Medium	Enterprise tenants, compliance requirements
Schema isolation	Logical + physical	Medium	Many tenants, shared infrastructure

Namespace Usage for Subgraphs

Checkpoint namespaces (checkpoint_ns) identify which graph or subgraph a checkpoint belongs to. Use them deliberately:

# Root graph checkpoints have empty namespace: ""
# Subgraph checkpoints have structured namespace: "node_name:uuid"
# Nested subgraphs: "outer_node:uuid|inner_node:uuid"

# Access namespace from within a node:
from langchain_core.runnables import RunnableConfig

def inspect_namespace(state: AgentState, config: RunnableConfig) -> dict:
    ns = config["configurable"]["checkpoint_ns"]
    if ns == "":
        graph_level = "root"
    else:
        graph_level = f"subgraph: {ns}"
    logger.info(f"Executing in namespace: {graph_level}")
    return {}

Thread Cleanup and TTL Strategies

Set a time-to-live for threads to prevent indefinite accumulation:

from datetime import datetime, timedelta


async def expire_inactive_threads(
    checkpointer: AsyncPostgresSaver,
    inactive_hours: int = 72,  # 3 days
) -> int:
    """Delete threads that have been inactive beyond the TTL.

    Returns the number of threads deleted.
    """
    cutoff = datetime.now() - timedelta(hours=inactive_hours)

    async with checkpointer.pool.connection() as conn:
        # Find threads with no recent checkpoints
        stale_threads = await conn.fetch(
            """
            SELECT DISTINCT thread_id
            FROM checkpoints
            WHERE thread_id NOT IN (
                SELECT DISTINCT thread_id
                FROM checkpoints
                WHERE created_at > %s
            )
            """,
            cutoff,
        )

    deleted = 0
    for row in stale_threads:
        await checkpointer.delete_thread(row["thread_id"])
        deleted += 1

    logger.info(f"Expired {deleted} inactive threads (TTL: {inactive_hours}h)")
    return deleted

---

Performance Tuning

Checkpointing adds latency to every super-step. Optimize the path from state mutation to disk.

Indexing Strategies for Checkpoint Tables

PostgresSaver.setup() creates default indexes. For high-traffic deployments, verify and augment them:

-- Default indexes created by setup() are on (thread_id, checkpoint_ns).
-- Add these for common query patterns:

-- Fast lookup of latest checkpoint per thread
CREATE INDEX IF NOT EXISTS idx_checkpoints_latest
ON checkpoints (thread_id, checkpoint_ns, checkpoint_id DESC);

-- Fast filtering by metadata (e.g., user_id, environment)
CREATE INDEX IF NOT EXISTS idx_checkpoints_metadata_user
ON checkpoints USING GIN ((metadata)->'user_id');

-- Fast cleanup queries by creation time
CREATE INDEX IF NOT EXISTS idx_checkpoints_created_at
ON checkpoints (created_at);

-- Fast blob lookups by thread and checkpoint
CREATE INDEX IF NOT EXISTS idx_checkpoint_blobs_lookup
ON checkpoint_blobs (thread_id, checkpoint_ns, checkpoint_id, channel);

GIN Indexes for Metadata Queries

If you frequently filter checkpoints by metadata fields, use a GIN index on the JSONB metadata column. This is significantly faster than sequential scans for high-cardinality metadata queries.

Query Optimization for Checkpoint Lookups

The most common checkpoint query is "get the latest checkpoint for a thread." Ensure it is fast:

-- This is the query pattern PostgresSaver uses internally.
-- The index on (thread_id, checkpoint_ns, checkpoint_id DESC) makes it O(1).
SELECT *
FROM checkpoints
WHERE thread_id = 'some-thread-id'
  AND checkpoint_ns = ''
ORDER BY checkpoint_id DESC
LIMIT 1;

If you see slow checkpoint lookups, run EXPLAIN (ANALYZE, BUFFERS) on the query to verify the index is used.

Batch Checkpointing vs. Per-Step Checkpointing

LangGraph checkpoints after every super-step by default. For high-throughput graphs, consider reducing checkpoint frequency:

# Default: checkpoint every step (most durable)
app = graph.compile(checkpointer=checkpointer)

# Reduced: checkpoint only at graph exit (less durable, faster)
# Note: This means you lose intermediate state if the process crashes mid-execution.
# Use only when graph execution is fast (< 5 seconds) and crash recovery
# from the beginning is acceptable.

Tradeoff:

Mode	Durability	Latency	Crash Recovery
Per-step (default)	Full — recover to last step	Higher (checkpoint per step)	Resume from last super-step
Exit-only	Partial — recover to last completed run	Lower (one checkpoint per run)	Restart from beginning

For most production agents, keep per-step checkpointing. The durability is worth the cost.

Durability Mode Selection

LangGraph checkpointers support different durability semantics. Choose based on your tolerance for data loss:

Mode	Behavior	When to Use
"sync" (default)	Checkpoint written and confirmed before returning	Production default — full durability
"async"	Checkpoint queued for background write	High-throughput, tolerate minor loss
"exit"	Checkpoint only at graph completion	Stateless-like behavior, minimum overhead

# Explicit sync mode (the default)
checkpointer = PostgresSaver(pool)

# If you need to tune durability, configure at the graph level
# Note: durability mode is a framework-level setting, not checkpointer-specific

Caching Strategies

Reduce database reads by caching checkpoint lookups within a single request:

from functools import lru_cache


class CachedCheckpointer:
    """Thin wrapper that caches checkpoint reads within a request lifecycle."""

    def __init__(self, base_checkpointer: AsyncPostgresSaver):
        self._base = base_checkpointer
        self._cache: dict[tuple[str, str], dict] = {}

    async def aget_tuple(self, config: dict) -> tuple:
        cache_key = (
            config.get("configurable", {}).get("thread_id"),
            config.get("configurable", {}).get("checkpoint_ns"),
        )
        if cache_key in self._cache:
            return self._cache[cache_key]

        result = await self._base.aget_tuple(config)
        self._cache[cache_key] = result
        return result

    async def aput(self, config: dict, checkpoint: dict, metadata: dict) -> dict:
        # Writes always go through — invalidate cache on write
        self._cache.clear()
        return await self._base.aput(config, checkpoint, metadata)

    # Delegate other methods to the base checkpointer
    async def aclose(self):
        await self._base.aclose()

Cache Invalidation

Cache checkpoint reads only within a single request lifecycle. Invalidate the cache on any write. Never cache checkpoint data across requests — the underlying state may have changed.

---

Disaster Recovery

Agent state is business-critical data. When the database goes down, you need a plan.

Database Backups and Restore Procedures

PostgreSQL provides two backup mechanisms:

# Full logical backup (pg_dump)
pg_dump -h postgres.internal -U langgraph_user -d langgraph -F custom -f langgraph_backup.dump

# Incremental backup via WAL archiving (continuous)
# Configure in postgresql.conf:
# wal_level = replica
# archive_mode = on
# archive_command = 'cp %p /backup/wal/%f'

Restore procedure:

# Restore from logical backup
pg_restore -h postgres.internal -U langgraph_user -d langgraph --clean --if-exists langgraph_backup.dump

# Verify restore
psql -h postgres.internal -U langgraph_user -d langgraph -c \
  "SELECT count(*) FROM checkpoints;"

Point-in-Time Recovery for Checkpoints

Configure WAL archiving for point-in-time recovery:

-- In postgresql.conf
wal_level = replica
max_wal_size = 1GB
min_wal_size = 80MB
archive_mode = on
archive_command = 'test ! -f /backup/wal/%f && cp %p /backup/wal/%f'

Recover to a specific point in time:

# Restore to 1 hour ago
pg_basebackup -h postgres.internal -D /tmp/restore -U replicator
# Then in recovery.conf:
# restore_command = 'cp /backup/wal/%f %p'
# recovery_target_time = '2025-01-15 14:00:00'

Multi-Region Replication Considerations

For multi-region deployments, PostgreSQL replication has implications for checkpoint consistency:

Replication Mode	Checkpoint Consistency	Latency Impact
Synchronous	Strong — all regions agree	Higher (waits for all regions)
Asynchronous	Eventual — may lose recent checkpoints	Lower (primary writes immediately)
Read replicas	Read-only — checkpoint writes go to primary only	Reads can be offloaded

Recommendation: Use synchronous replication between regions for checkpoint writes. For read-heavy workloads (e.g., listing thread history), route reads to replicas.

Testing Recovery Procedures

Test your recovery plan quarterly:

import asyncio
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver


async def test_recovery(checkpointer: AsyncPostgresSaver, thread_id: str) -> bool:
    """Verify that checkpoint data can be read and state can be restored."""
    try:
        # Attempt to read the latest checkpoint
        config = {"configurable": {"thread_id": thread_id}}
        state = await checkpointer.aget_tuple(config)

        if state is None:
            logger.warning(f"No checkpoint found for thread {thread_id}")
            return False

        # Verify state is deserializable
        checkpoint = state.checkpoint
        assert checkpoint is not None
        assert isinstance(checkpoint, dict)

        logger.info(f"Recovery test passed for thread {thread_id}")
        return True

    except Exception as exc:
        logger.error(f"Recovery test failed: {exc}")
        return False


# Run against a known-good thread
async def run_recovery_test():
    checkpointer = await create_checkpointer()
    success = await test_recovery(checkpointer, "known-good-thread-id")
    if not success:
        raise RuntimeError("Disaster recovery test failed")

---

Monitoring and Alerting

You cannot manage what you do not measure. Instrument your checkpointing layer.

Key Metrics to Track

Metric	How to Measure	Alert Threshold
Checkpoint write latency	Time between node completion and checkpoint confirmed	P99 > 500ms
Checkpoint read latency	Time to load state at graph start	P99 > 200ms
Checkpoint table size	pg_total_relation_size('checkpoints')	Growth > 10% per day
Checkpoint blob size	pg_total_relation_size('checkpoint_blobs')	Growth > 15% per day
Connection pool usage	Active connections / max pool size	> 80% sustained
Connection pool wait time	Time waiting for a connection	P99 > 2s
Orphaned thread count	Threads with no activity > 30 days	> 1000
Checkpoint errors	Failed checkpoint writes	Any non-zero rate

Setting Up Alerts for State Anomalies

Use your observability platform (Datadog, Prometheus + Alertmanager, etc.):

import time
import logging
from prometheus_client import Gauge, Histogram, Counter

# Prometheus metrics
checkpoint_write_latency = Histogram(
    "langgraph_checkpoint_write_seconds",
    "Checkpoint write latency",
)
checkpoint_read_latency = Histogram(
    "langgraph_checkpoint_read_seconds",
    "Checkpoint read latency",
)
checkpoint_table_size = Gauge(
    "langgraph_checkpoint_table_bytes",
    "Checkpoint table size in bytes",
    labelnames=["table"],
)
checkpoint_errors = Counter(
    "langgraph_checkpoint_errors_total",
    "Total checkpoint errors",
    labelnames=["type"],
)
pool_active_connections = Gauge(
    "langgraph_pool_active_connections",
    "Active database connections in pool",
)


class MonitoredCheckpointer:
    """Wrapper that instruments checkpoint operations."""

    def __init__(self, base: AsyncPostgresSaver):
        self._base = base

    async def aput(self, config, checkpoint, metadata):
        start = time.perf_counter()
        try:
            result = await self._base.aput(config, checkpoint, metadata)
            checkpoint_write_latency.observe(time.perf_counter() - start)
            return result
        except Exception as exc:
            checkpoint_errors.labels(type="write").inc()
            raise

    async def aget_tuple(self, config):
        start = time.perf_counter()
        try:
            result = await self._base.aget_tuple(config)
            checkpoint_read_latency.observe(time.perf_counter() - start)
            return result
        except Exception as exc:
            checkpoint_errors.labels(type="read").inc()
            raise

    async def aclose(self):
        await self._base.aclose()

Health Check Patterns

Expose a health endpoint that verifies checkpointing infrastructure:

from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/health/checkpointer")
async def checkpointer_health():
    """Verify that the checkpointer can read and write."""
    try:
        # Lightweight write test
        test_thread = f"health-check-{int(time.time())}"
        config = {"configurable": {"thread_id": test_thread}}

        # Write a minimal checkpoint
        await checkpointer.aput(
            config=config,
            checkpoint={"_health": True},
            metadata={"source": "health-check"},
        )

        # Read it back
        result = await checkpointer.aget_tuple(config)
        if result is None:
            raise HTTPException(status_code=503, detail="Checkpointer read failed")

        # Clean up
        await checkpointer.delete_thread(test_thread)

        return {"status": "healthy", "component": "checkpointer"}

    except Exception as exc:
        logger.error(f"Checkpointer health check failed: {exc}")
        raise HTTPException(status_code=503, detail=f"Checkpointer unhealthy: {exc}")

---

Summary

Area	Standard
Connection Pooling	Pool created once at startup. min_size matches expected concurrency; max_size is 2-3x min_size. Never create per-request pools.
Checkpoint Lifecycle	Retention policy defined and enforced. Orphaned threads cleaned up daily. Tables vacuumed weekly.
State Schema	Only shared data in state. Reducers chosen deliberately. Message history truncated proactively. Hot/cold state separated.
Thread Management	Thread IDs structured and under 255 characters. Multi-tenant isolation via ID scoping or separate databases. TTL enforced.
Performance	Indexes verified and augmented. Serialization under 10ms, checkpoints under 100KB. Per-step checkpointing retained for durability.
Disaster Recovery	Backups tested quarterly. WAL archiving enabled. Recovery procedures documented and run as code.
Monitoring	Checkpoint latency, table sizes, pool usage, and error rates instrumented. Health endpoint verifies read/write capability.

Follow these standards and your PostgreSQL-backed LangGraph checkpointing will be reliable at scale. Break them only with a documented reason.

---

Further Reading

• PostgreSQL Checkpointing & Memory Setup — initial setup, schema creation, and basic configuration.
• LangGraph Best Practices — agent design patterns, error handling, and testing.
• LangGraph Official Docs — Persistence — checkpointers, stores, and fault tolerance.
• LangGraph Official Docs — Stores — cross-thread long-term memory patterns.
• LangSmith Observability — tracing, debugging, and evaluation of checkpointed runs.

Production Checklist

Before deploying state management to production, verify:

• Connection pool is created once at startup and shared across requests.
• Pool sizing matches expected concurrency (max_size <= 1/3 of PostgreSQL max_connections).
• checkpointer.setup() has been called and tables exist with proper indexes.
• Connection string uses SSL (sslmode=require for Azure Database).
• LANGGRAPH_STRICT_MSGPACK=true is set for serialization safety.
• Thread IDs are under 255 characters and structured for multi-tenancy.
• A checkpoint retention policy is defined and a cleanup job is scheduled.
• Checkpoint table sizes are monitored with alerts on abnormal growth.
• Message history is truncated to prevent unbounded state growth.
• Database backups are configured and recovery procedures are tested.
• A health check endpoint verifies checkpointer read/write capability.
• Prometheus metrics (or equivalent) track checkpoint latency and pool usage.