4. PostgresSaver: Durable Persistence

Overview

In the previous capsule you implemented checkpointing with MemorySaver. It works perfectly for development: every agent invocation persists its state, you can resume interrupted conversations, and the thread_id isolates sessions. But MemorySaver has a fatal flaw for production: it lives in the process's memory. If the server restarts, if you deploy a new version, if the process crashes at 3am — all the state disappears. Every conversation, every checkpoint, every half-finished research run. Gone.

PostgresSaver solves this with a surprisingly simple change: the same API as MemorySaver, but the checkpoints get written to PostgreSQL instead of a dictionary in RAM. The agent doesn't know the difference. Your code changes one line — literally — and you go from "persistence that survives the session" to "persistence that survives restarts, deploys, crashes and server migrations." It's the difference between a prototype and a production system.

This capsule covers the complete setup: installing the package, spinning up PostgreSQL with Docker, configuring the connection string, creating the saver, and compiling the graph. It also covers the async variants for FastAPI applications, cleanup strategies to keep the database from growing indefinitely, and clear criteria for choosing between MemorySaver, PostgresSaver and Redis based on your use case.

From MemorySaver to PostgresSaver

The same interface, a different backend

LangGraph designed its checkpointing system with a brilliant abstraction: every checkpointer implements the same BaseCheckpointSaver interface. That means the graph doesn't know — or care — where the checkpoints get stored. It just calls .put() to save and .get() to retrieve. The backend is an implementation detail.

This is what you have today with MemorySaver:

from langgraph.checkpoint.memory import MemorySaver

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

And this is PostgresSaver:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/agents_db"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    agent = graph.compile(checkpointer=checkpointer)

The rest of your code — the nodes, the edges, the invocations with thread_id, the checkpoint inspection — doesn't change at all. The config is still:

config = {"configurable": {"thread_id": "user-123"}}
result = agent.invoke({"messages": [HumanMessage(content="Hi")]}, config)

What changes under the hood

With MemorySaver, the checkpoints live in a Python dict — lost on restart. With PostgresSaver, each checkpoint gets serialized as jsonb and written as a row in PostgreSQL. The table has an index on thread_id (retrieving the last checkpoint is O(1)) and a parent_checkpoint_id that lets you reconstruct the full chain for time-travel debugging (capsule 06).

PostgreSQL Setup

Docker Compose for local development

You don't need to install PostgreSQL on your machine. Docker Compose gives you a local Postgres in seconds:

# docker-compose.yml
services:
  postgres:
    image: postgres:16
    container_name: agents_postgres
    environment:
      POSTGRES_USER: agents
      POSTGRES_PASSWORD: agents_secret
      POSTGRES_DB: agents_db
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Bring it up:

docker compose up -d

Verify it's running:

docker compose ps
# NAME              STATUS    PORTS
# agents_postgres   Up        0.0.0.0:5432->5432/tcp

Connection string format

The PostgreSQL connection string follows this format:

postgresql://USER:PASSWORD@HOST:PORT/DATABASE

For the Docker Compose above:

postgresql://agents:agents_secret@localhost:5432/agents_db

In production, never hardcode the connection string. Use environment variables:

import os

DB_URI = os.environ["DATABASE_URL"]

# .env
DATABASE_URL=postgresql://agents:agents_secret@localhost:5432/agents_db

For production with SSL (cloud services like Supabase, Neon, Railway), add ?sslmode=require at the end of the connection string. The full URI is in the service's dashboard — copy it and put it in your environment variable.

Configuring PostgresSaver

Installation

PostgresSaver lives in a separate package from LangGraph:

pip install langgraph-checkpoint-postgres

This package depends on psycopg (PostgreSQL's driver for Python 3). It gets installed automatically as a dependency.

The synchronous saver

For scripts and synchronous applications:

from dotenv import load_dotenv
load_dotenv()

import os
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = os.environ["DATABASE_URL"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    
    agent = graph.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "research-session-001"}}
    result = agent.invoke(
        {"messages": [("user", "Research RAG patterns in 2025")]},
        config
    )
    print(result["messages"][-1].content)

The .setup() method creates the necessary tables in PostgreSQL if they don't exist. It's idempotent — you can call it multiple times with no problem. You only need to call it once (at application startup), not on every request.

The async saver (for FastAPI)

If your application uses async (FastAPI, aiohttp, etc.), use the async variant:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DB_URI = os.environ["DATABASE_URL"]

async def create_agent():
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        await checkpointer.setup()
        agent = graph.compile(checkpointer=checkpointer)
        return agent

Integration with FastAPI

The production pattern uses FastAPI's lifespan to manage the connection's lifecycle:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langchain_core.messages import HumanMessage
import os

checkpointer = None
agent = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global checkpointer, agent
    db_uri = os.environ["DATABASE_URL"]
    checkpointer = AsyncPostgresSaver.from_conn_string(db_uri)
    await checkpointer.setup()
    agent = graph.compile(checkpointer=checkpointer)
    yield
    await checkpointer.conn.close()

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=message)]},
        config
    )
    return {"response": result["messages"][-1].content}

The checkpointer gets initialized once at startup and reused across requests — you don't open a new connection on every invocation.

Migrating from MemorySaver

The change is literally one line

If your current code is:

from langgraph.checkpoint.memory import MemorySaver

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

The migration is:

from langgraph.checkpoint.postgres import PostgresSaver
import os

DB_URI = os.environ["DATABASE_URL"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    agent = graph.compile(checkpointer=checkpointer)

That's it. No other change in your code. The nodes, edges, conditional routing, invocations — all the same.

State format compatibility

The state format is identical regardless of the backend. There's no data migration or transformation. What you can't do is migrate existing MemorySaver checkpoints — they live in RAM and are lost when the process closes. The migration is "from here on out": the new checkpoints go to Postgres.

Pattern: a backend selector by environment

A useful pattern for using MemorySaver in development and PostgresSaver in production:

import os
from langgraph.checkpoint.memory import MemorySaver

def create_checkpointer():
    env = os.environ.get("ENVIRONMENT", "development")
    
    if env == "production":
        from langgraph.checkpoint.postgres import PostgresSaver
        db_uri = os.environ["DATABASE_URL"]
        checkpointer = PostgresSaver.from_conn_string(db_uri)
        checkpointer.setup()
        return checkpointer
    else:
        return MemorySaver()

checkpointer = create_checkpointer()
agent = graph.compile(checkpointer=checkpointer)

In development, you don't need Docker or Postgres — MemorySaver is enough. In CI/CD, you can use MemorySaver for unit tests and PostgresSaver for integration tests. In production, always PostgresSaver.

When to Use Each Backend

Decision table

*Redis checkpoint support is available via langgraph-checkpoint-redis.

Quick selection guide

MemorySaver: development, testing, prototypes, CI/CD. Zero setup, doesn't survive restarts.

PostgresSaver: general production, multi-instance, auditing, < 10k checkpoints/hour.

Redis: high-throughput (> 10k/hr), latency < 1ms, ephemeral sessions with TTL.

The natural progression

Development         Staging             Production
┌─────────────┐    ┌──────────────┐    ┌──────────────────┐
│ MemorySaver │───→│ PostgresSaver│───→│ PostgresSaver     │
│ (zero setup)│    │ (Docker)     │    │ (managed DB)      │
└─────────────┘    └──────────────┘    └──────────────────┘
                                              │
                                              │ If throughput > 10k/hr
                                              ▼
                                       ┌──────────────────┐
                                       │ Redis             │
                                       │ (high speed)      │
                                       └──────────────────┘

PostgresSaver handles hundreds of checkpoints per second — it covers 95% of cases. Redis is for thousands of concurrent users generating checkpoints at every step.

Performance and Scalability

Connection pooling

In production with multiple concurrent requests, each request needs a connection to Postgres. Without a pool, each request opens and closes a connection — slow and expensive. With a pool, the connections get reused:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@host:5432/dbname"

with PostgresSaver.from_conn_string(
    DB_URI,
    pool_size=20,
) as checkpointer:
    checkpointer.setup()
    agent = graph.compile(checkpointer=checkpointer)

Rule of thumb for pool_size: your server's number of workers × 2. If your FastAPI runs with 4 workers, pool_size=8 is a good starting point.

Checkpoint size

Each checkpoint serializes the complete state. A state with 20 messages weighs ~5 KB. A graph with 4 nodes and 1,000 active sessions generates ~60 MB — nothing for PostgreSQL. But at 100,000 sessions with long histories, that's 6 GB. That's when you need cleanup.

Cleanup strategies

Manual TTL (Time-to-Live)

PostgresSaver has no built-in TTL. Implement cleanup with a cron job:

import psycopg

def cleanup_old_checkpoints(db_uri: str, days: int = 30):
    """Delete checkpoints older than N days."""
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                DELETE FROM checkpoints 
                WHERE created_at < NOW() - INTERVAL '%s days'
            """, (days,))
            deleted = cur.rowcount
            conn.commit()
    return deleted

deleted = cleanup_old_checkpoints(DB_URI, days=30)
print(f"Deleted {deleted} old checkpoints")

Limiting checkpoints per thread, and monitoring

If you only need the recent checkpoints, clean up the old ones per thread. Combine it with monitoring so you know when to clean:

def keep_latest_per_thread(db_uri: str, keep: int = 5):
    """Keep only the N most recent checkpoints per thread."""
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                DELETE FROM checkpoints 
                WHERE checkpoint_id NOT IN (
                    SELECT checkpoint_id FROM (
                        SELECT checkpoint_id,
                               ROW_NUMBER() OVER (
                                   PARTITION BY thread_id 
                                   ORDER BY created_at DESC
                               ) as rn
                        FROM checkpoints
                    ) ranked
                    WHERE rn <= %s
                )
            """, (keep,))
            deleted = cur.rowcount
            conn.commit()
    return deleted

def checkpoint_stats(db_uri: str) -> dict:
    """Usage statistics for the checkpoints table."""
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT COUNT(*) as total,
                       COUNT(DISTINCT thread_id) as threads,
                       pg_size_pretty(pg_total_relation_size('checkpoints')) as size
                FROM checkpoints
            """)
            row = cur.fetchone()
            return {"total": row[0], "threads": row[1], "size": row[2]}

Connection to the Project

In this module's project (capsule 08, Research Agent with Persistent Memory):

• PostgresSaver replaces MemorySaver in the Research Agent. If the user closes the browser halfway through an 8-step research run, everything is still there when they come back — the plan, the partial results, the current step.
• The backend selector (create_checkpointer()) allows MemorySaver in tests and PostgresSaver in production.
• Cleanup: checkpoints older than 7 days get deleted automatically.

In later modules, M7 (MCP) persists external tool results, M8 (Multi-Agent) shares a PostgresSaver across sub-agents with different thread_id namespaces, and M10 (Production) uses AsyncPostgresSaver with connection pooling in FastAPI.

Troubleshooting

Problem 1: psycopg.OperationalError: connection refused

Symptom: A connection-refused error when creating the PostgresSaver.

Cause: PostgreSQL isn't running, or the host/port is wrong.

Solution:

docker compose ps          # Is it running?
docker compose up -d       # If not, bring it up
docker compose exec postgres psql -U agents -d agents_db -c "SELECT 1"

For cloud services, check the IP whitelist and the SSL mode.

Problem 2: relation "checkpoints" does not exist

Symptom: The first invocation fails because the table doesn't exist.

Cause: You forgot to call .setup() before compiling the graph.

Solution: Add checkpointer.setup() after creating the PostgresSaver. It's idempotent — always call it at app startup.

Problem 3: The checkpoints aren't shared between server instances

Symptom: You have 3 instances of your API behind a load balancer. A user starts a conversation on instance A, but the next request goes to instance B and the agent remembers nothing.

Cause: Each instance is using its own MemorySaver (in RAM). Or each instance connects to a different database.

Solution: Verify that every instance connects to the same PostgreSQL database with the same DATABASE_URL. With PostgresSaver, the state lives in the DB — any instance can retrieve any thread's checkpoint.

Problem 4: The connection closes unexpectedly (InterfaceError: connection is closed)

Symptom: After hours of operation, invocations fail with closed-connection errors.

Cause: An idle connection timeout on the server or the pool. Connections that go unused for a long time get closed automatically.

Solution: Use the context manager (with PostgresSaver...), which handles reconnections. For FastAPI, make sure you use the lifespan pattern shown in the async configuration section.

Problem 5: The database size grows out of control

Symptom: Millions of rows in the checkpoints table, slow queries.

Cause: No cleanup configured. Every step generates a checkpoint.

Solution: Implement a TTL or per-thread retention (see the "Performance and Scalability" section). Monitor with checkpoint_stats() and set up alerts on the size.

Exercises

Exercise 1: Complete setup with Docker (Easy)

Create a docker-compose.yml for PostgreSQL, configure the connection string as an environment variable, and verify that you can create a PostgresSaver and call .setup() without errors. Print the created tables.

import os
os.environ["DATABASE_URL"] = "postgresql://agents:agents_secret@localhost:5432/agents_db"

from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

DB_URI = os.environ["DATABASE_URL"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    print("Setup complete")

with psycopg.connect(DB_URI) as conn:
    with conn.cursor() as cur:
        cur.execute("""
            SELECT table_name FROM information_schema.tables 
            WHERE table_schema = 'public'
        """)
        print(f"Tables created: {[t[0] for t in cur.fetchall()]}")

Exercise 2: Migrate an agent from MemorySaver to PostgresSaver (Easy)

Take this agent with MemorySaver and migrate it to PostgresSaver. Verify that the thread_id works the same — invoke it twice with the same thread and confirm the agent remembers the first conversation.

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

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

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

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

graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

# MIGRATE: change this line
checkpointer = MemorySaver()
agent = graph.compile(checkpointer=checkpointer)

from dotenv import load_dotenv
load_dotenv()

import os
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated

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

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

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

graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

DB_URI = os.environ["DATABASE_URL"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    agent = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "test-migration-001"}}
    
    result1 = agent.invoke(
        {"messages": [HumanMessage(content="My name is Carlos and I work in fintech")]},
        config
    )
    print("Response 1:", result1["messages"][-1].content[:200])
    
    result2 = agent.invoke(
        {"messages": [HumanMessage(content="What's my name and what do I do?")]},
        config
    )
    print("Response 2:", result2["messages"][-1].content[:200])

Exercise 3: A backend selector by environment (Medium)

Implement a create_checkpointer() function that returns MemorySaver if ENVIRONMENT=development and PostgresSaver if ENVIRONMENT=production. Add a log line indicating which backend is in use. Use the function to compile a graph and verify it works in both modes.

from dotenv import load_dotenv
load_dotenv()

import os
import logging
from langgraph.checkpoint.memory import MemorySaver

logger = logging.getLogger(__name__)

def create_checkpointer():
    env = os.environ.get("ENVIRONMENT", "development")
    if env == "production":
        from langgraph.checkpoint.postgres import PostgresSaver
        checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
        checkpointer.setup()
        logger.info("Checkpointer: PostgresSaver")
        return checkpointer
    else:
        logger.info("Checkpointer: MemorySaver (dev)")
        return MemorySaver()

os.environ["ENVIRONMENT"] = "development"
print(f"Dev: {type(create_checkpointer()).__name__}")  # MemorySaver

os.environ["ENVIRONMENT"] = "production"
print(f"Prod: {type(create_checkpointer()).__name__}")  # PostgresSaver

checkpointer = create_checkpointer()
agent = graph.compile(checkpointer=checkpointer)

Exercise 4: Monitoring checkpoints in production (Medium)

Write functions to: (a) count the total checkpoints and unique threads, (b) find the threads with the most checkpoints (possible leaks), and (c) compute the average checkpoint size. Run them against your local Postgres.

import os
import psycopg
from dotenv import load_dotenv
load_dotenv()

DB_URI = os.environ["DATABASE_URL"]

def checkpoint_overview(db_uri: str) -> dict:
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT COUNT(*) as total,
                       COUNT(DISTINCT thread_id) as threads,
                       pg_size_pretty(pg_total_relation_size('checkpoints')) as size
                FROM checkpoints
            """)
            row = cur.fetchone()
            return {"total": row[0], "threads": row[1], "size": row[2]}

def top_threads(db_uri: str, limit: int = 10) -> list:
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT thread_id, COUNT(*) as cnt
                FROM checkpoints GROUP BY thread_id
                ORDER BY cnt DESC LIMIT %s
            """, (limit,))
            return [{"thread_id": r[0], "checkpoints": r[1]} for r in cur.fetchall()]

def avg_checkpoint_size(db_uri: str) -> dict:
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT AVG(pg_column_size(checkpoint)),
                       MAX(pg_column_size(checkpoint))
                FROM checkpoints
            """)
            row = cur.fetchone()
            return {"avg_bytes": round(row[0] or 0, 1), "max_bytes": row[1] or 0}

overview = checkpoint_overview(DB_URI)
print(f"Total: {overview['total']}, Threads: {overview['threads']}, Size: {overview['size']}")

for t in top_threads(DB_URI, limit=5):
    print(f"  {t['thread_id']}: {t['checkpoints']} checkpoints")

print(f"Avg size: {avg_checkpoint_size(DB_URI)}")

Exercise 5: Automated cleanup with per-thread retention (Hard)

Implement a cleanup system that: (a) keeps the last N checkpoints per thread, (b) deletes threads with no activity for more than M days, and (c) generates a report of how much space was freed. Structure it as a script you could run as a daily cron job.

import os
import psycopg
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()

DB_URI = os.environ["DATABASE_URL"]

def cleanup_checkpoints(db_uri: str, keep_per_thread: int = 5, inactive_days: int = 30) -> dict:
    report = {"timestamp": datetime.now().isoformat(), "deleted_inactive": 0, "deleted_excess": 0}
    
    with psycopg.connect(db_uri) as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT pg_size_pretty(pg_total_relation_size('checkpoints'))"
            )
            report["space_before"] = cur.fetchone()[0]
            
            cur.execute("""
                DELETE FROM checkpoints WHERE thread_id IN (
                    SELECT thread_id FROM checkpoints
                    GROUP BY thread_id
                    HAVING MAX(created_at) < NOW() - INTERVAL '%s days'
                )
            """, (inactive_days,))
            report["deleted_inactive"] = cur.rowcount
            
            cur.execute("""
                DELETE FROM checkpoints WHERE checkpoint_id NOT IN (
                    SELECT checkpoint_id FROM (
                        SELECT checkpoint_id,
                               ROW_NUMBER() OVER (PARTITION BY thread_id ORDER BY created_at DESC) as rn
                        FROM checkpoints
                    ) ranked WHERE rn <= %s
                )
            """, (keep_per_thread,))
            report["deleted_excess"] = cur.rowcount
            conn.commit()
            
            cur.execute("VACUUM ANALYZE checkpoints")
            cur.execute("SELECT pg_size_pretty(pg_total_relation_size('checkpoints'))")
            report["space_after"] = cur.fetchone()[0]
    
    return report

report = cleanup_checkpoints(DB_URI, keep_per_thread=5, inactive_days=30)
print(f"Inactive: {report['deleted_inactive']}, Excess: {report['deleted_excess']}")
print(f"Space: {report['space_before']} → {report['space_after']}")

Summary

In this capsule you learned:

• PostgresSaver is MemorySaver for production. The same API, the same interface, but the checkpoints get written to PostgreSQL instead of RAM. The change is literally one line of code. Your graph, nodes, edges, and invocation logic don't change.

• The setup is minimal. Docker Compose for a local Postgres, pip install langgraph-checkpoint-postgres, a connection string as an environment variable, .setup() to create the tables. In 5 minutes you have durable persistence.

• The async variant exists for FastAPI. AsyncPostgresSaver with await checkpointer.setup() and await agent.ainvoke(). FastAPI's lifespan pattern handles the connection's lifecycle.

• Migrating from MemorySaver is trivial. Change the import, change the instantiation, add .setup(). The existing checkpoints in RAM don't get migrated (because they no longer exist after a restart), but from here on everything persists.

• Each backend has its place. MemorySaver for development and tests (zero setup). PostgresSaver for general production (durable, multi-instance, queryable). Redis for high-throughput and ephemeral sessions. Most apps only need PostgresSaver.

• Performance needs attention in production. Connection pooling for concurrent requests, monitoring the table's size, and periodic cleanup (TTL or per-thread retention) to prevent uncontrolled growth.

Next capsule: Long-term Memory: Cross-Session — memory that persists not just within one conversation, but between different conversations. Your agent will remember who the user is, what they prefer, and what they researched last week.

Additional Resources

1. LangGraph Persistence Docs — Official checkpointing documentation: concepts, backends, configuration
2. langgraph-checkpoint-postgres (PyPI) — The PostgresSaver package: installation, changelog, versions
3. PostgreSQL Docker Image — The official Postgres image for Docker: tags, environment variables, configuration
4. psycopg 3 Documentation — PostgreSQL's driver for Python 3: connection pooling, async, types
5. LangGraph Async Tutorial — How to use LangGraph with async/await: patterns, FastAPI integration