LangGraph Persistence with Postgres

LangGraph persistence with Postgres means wiring a PostgresSaver (or its async twin, AsyncPostgresSaver) into your graph as the checkpointer, so every step of the graph's state gets written to a real database instead of living only in process memory. Once that's in place, a graph run can crash, the server can restart, or a user can walk away for three days, and when they come back you call invoke with the same thread_id and the agent picks up exactly where it left off. This matters the moment you move a LangGraph agent past a demo notebook: in-memory checkpointing (MemorySaver) disappears the instant the process dies, and SQLite checkpointing doesn't survive a multi-instance deployment. Postgres persistence is what makes LangGraph state durable, shareable across workers, and inspectable with plain SQL.

This guide walks through installing the Postgres checkpointer package, standing up the tables, wiring it into a StateGraph, running threaded conversations, doing human-in-the-loop interrupts, time-travel debugging, and cleaning up old checkpoints in production.

Why Postgres over the default checkpointers

LangGraph ships three checkpointer backends out of the box: MemorySaver (dict in RAM), a SQLite-backed saver, and Postgres via the separate langgraph-checkpoint-postgres package. All three implement the same BaseCheckpointSaver interface, so swapping between them is a one-line change. The difference is what happens under real traffic.

MemorySaver is fine for a single script or a test. It has zero setup cost and that's its only advantage. The moment you deploy behind more than one worker process, or restart the app, every in-flight conversation is gone.

SQLite persistence survives a restart on a single machine, but it doesn't survive horizontal scaling. Two API replicas writing to two different SQLite files can't share thread state, and SQLite's file locking gets ugly under concurrent writes from multiple processes.

Postgres persistence with LangGraph solves both problems. Any number of application instances can point at the same Postgres database (or the same Neon / RDS / Supabase Postgres cluster), so a request that lands on worker A can resume a thread that was last touched by worker B. You also get everything a real database gives you for free: connection pooling, backups, point-in-time recovery, and the ability to run SELECT against checkpoint tables when you're debugging why an agent got stuck.

Installing the Postgres checkpointer

The Postgres checkpointer lives in its own package, separate from core langgraph. Install both, plus psycopg for the actual database driver.

pip install langgraph langgraph-checkpoint-postgres "psycopg[binary,pool]"

You'll need a running Postgres instance. For local development, Docker is the fastest path:

docker run --name langgraph-pg -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=langgraph -p 5432:5432 -d postgres:16

For production, point at a managed Postgres provider (Neon, Supabase, RDS, Cloud SQL). LangGraph doesn't care which one, as long as you have a standard connection string.

Setting up PostgresSaver

The core object is PostgresSaver, which wraps a Postgres connection and implements put, get, and list for checkpoints. The simplest way to construct one is from_conn_string, which manages the connection lifecycle for you as a context manager.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5432/langgraph?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

checkpointer.setup() is the piece people forget. It creates the tables LangGraph needs (checkpoints, checkpoint_blobs, checkpoint_writes, and a migrations table) if they don't already exist. Run it once when you first stand up the database, and again after upgrading langgraph-checkpoint-postgres to a version that adds new migrations. It's idempotent, so calling it on every app boot is safe and cheap.

Wiring the checkpointer into a StateGraph

With the tables in place, pass the checkpointer into compile(). Here's a minimal agent graph with two nodes to make the persistence behavior visible.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

class State(TypedDict):
    messages: list
    step_count: int

def greet(state: State) -> State:
    return {
        "messages": state["messages"] + ["hello from node one"],
        "step_count": state["step_count"] + 1,
    }

def respond(state: State) -> State:
    return {
        "messages": state["messages"] + ["response from node two"],
        "step_count": state["step_count"] + 1,
    }

builder = StateGraph(State)
builder.add_node("greet", greet)
builder.add_node("respond", respond)
builder.add_edge(START, "greet")
builder.add_edge("greet", "respond")
builder.add_edge("respond", END)

DB_URI = "postgresql://postgres:postgres@localhost:5432/langgraph?sslmode=disable"

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

    config = {"configurable": {"thread_id": "conversation-1"}}
    result = graph.invoke({"messages": [], "step_count": 0}, config=config)
    print(result)

Every node execution writes a checkpoint tied to thread_id. Run this script again with the same thread_id, and LangGraph loads the last checkpoint instead of starting from a fresh State. That thread_id is the whole mental model for LangGraph persistence with Postgres: one thread equals one durable, resumable execution history, stored as a chain of checkpoints in the database.

Threads: how LangGraph organizes persisted state

A thread is not a special object you create explicitly. It's just a string key you pass in config["configurable"]["thread_id"]. LangGraph uses it to partition rows in the checkpoints table. Give a support bot one thread_id per customer conversation, give a coding agent one thread_id per repository task, and each gets its own independent, resumable state.

To resume a thread later, you don't need to pass the full input state again. Calling invoke(None, config=config) with the same thread_id tells LangGraph "continue from wherever this thread last stopped."

config = {"configurable": {"thread_id": "conversation-1"}}

# later, possibly in a different process
result = graph.invoke(None, config=config)

To inspect what's stored for a thread without running the graph, use get_state:

snapshot = graph.get_state(config)
print(snapshot.values)       # the current state dict
print(snapshot.next)         # which node would run next
print(snapshot.config)       # the checkpoint id this snapshot represents

Async persistence with AsyncPostgresSaver

Production LangGraph deployments are almost always async, usually behind FastAPI. Use AsyncPostgresSaver with the same API shape, just with await on every call.

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5432/langgraph?sslmode=disable"

async def build_graph():
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    async with checkpointer as cp:
        await cp.setup()
        graph = builder.compile(checkpointer=cp)
        config = {"configurable": {"thread_id": "async-thread-1"}}
        result = await graph.ainvoke({"messages": [], "step_count": 0}, config=config)
        return result

In a long-lived server, don't wrap the checkpointer in a short-lived async with per request. Open the connection pool once at app startup and reuse it across requests, closing it on shutdown. FastAPI's lifespan handler is the natural place for this:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

checkpointer_cm = None
checkpointer = None
graph = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global checkpointer_cm, checkpointer, graph
    checkpointer_cm = AsyncPostgresSaver.from_conn_string(DB_URI)
    checkpointer = await checkpointer_cm.__aenter__()
    await checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    yield
    await checkpointer_cm.__aexit__(None, None, None)

app = FastAPI(lifespan=lifespan)

Using a connection pool instead of a single connection

from_conn_string opens one connection. For any app handling concurrent requests, hand PostgresSaver a psycopg_pool.ConnectionPool (or AsyncConnectionPool) instead, so checkpoint reads and writes from different requests don't queue behind each other.

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

pool = ConnectionPool(
    conninfo=DB_URI,
    max_size=20,
    kwargs={"autocommit": True, "prepare_threshold": 0},
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()
graph = builder.compile(checkpointer=checkpointer)

The async equivalent uses AsyncConnectionPool and AsyncPostgresSaver the same way. Two settings matter here: autocommit=True because the checkpointer manages its own transactions per operation, and prepare_threshold=0 to avoid prepared-statement conflicts when the pool hands out connections to different query shapes across requests.

Human-in-the-loop pauses that survive a restart

This is where Postgres persistence stops being a nice-to-have and becomes the reason you use LangGraph at all. Add interrupt_before (or use the interrupt() function inside a node) to pause a graph before a sensitive step, like sending an email or executing a trade. Because the paused state is a checkpoint row in Postgres, the pause can outlive the process that created it.

graph = builder.compile(checkpointer=checkpointer, interrupt_before=["respond"])

config = {"configurable": {"thread_id": "approval-flow-1"}}
graph.invoke({"messages": [], "step_count": 0}, config=config)

# the graph stopped before "respond" ran; state is durable in Postgres now.
# a human reviews it, hours or days later, possibly from a different service:

state = graph.get_state(config)
print(state.next)  # ('respond',)

# approve and continue
graph.invoke(None, config=config)

Nothing about the approval step needs to happen in the same process, or even the same day, as the original invocation. The thread_id is the only handle a reviewing service needs to fetch the pending state and resume it.

Time travel: replaying and forking from past checkpoints

Because Postgres persistence keeps every checkpoint in a thread's history, not just the latest one, you can list them and re-run the graph from any earlier point. This is invaluable for debugging why an agent took a wrong turn.

config = {"configurable": {"thread_id": "conversation-1"}}

history = list(graph.get_state_history(config))
for snapshot in history:
    print(snapshot.config["configurable"]["checkpoint_id"], snapshot.values.get("step_count"))

# pick an earlier checkpoint and resume from it
earlier_config = history[2].config
graph.invoke(None, config=earlier_config)

Resuming from earlier_config creates a new branch of checkpoints rather than overwriting history, so the original run stays intact in the database alongside the fork. This is the same mechanism you'd use to let a user "edit an earlier message" in a chat UI and continue the conversation down a different path.

Cross-thread memory with the Postgres store

Checkpointers persist state scoped to a single thread. If you need memory that spans threads, like a user's stored preferences that should be visible in every future conversation with that user, LangGraph's Store API is the right tool, and it has its own Postgres-backed implementation, PostgresStore.

from langgraph.store.postgres import PostgresStore

with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()
    store.put(("users", "user-42"), "preferences", {"tone": "concise", "language": "en"})
    item = store.get(("users", "user-42"), "preferences")
    print(item.value)

Pass store=store into compile() alongside checkpointer=checkpointer, and any node can read or write to it through the store argument LangGraph injects. Use the checkpointer for "what happened in this conversation" and the store for "what do we know about this user across all conversations."

Managing checkpoints in production

A Postgres-backed agent that runs for months will accumulate checkpoint rows fast, especially for long threads or high-frequency nodes. A few operational habits keep this manageable.

• Delete threads you no longer need with checkpointer.delete_thread(thread_id), which removes all checkpoints, writes, and blobs for that thread in one call.
• Set a retention job (a cron task or a scheduled Lambda) that queries the checkpoints table for threads older than your retention window and calls delete_thread on each.
• Index on thread_id is created automatically by setup(), but if you query checkpoints by custom metadata for analytics, add your own index rather than scanning the JSONB columns.
• Watch connection pool exhaustion under load. LangGraph checkpointing does a write on every node transition, so a graph with ten nodes per run issues ten round trips to Postgres per invocation. Size your pool to your concurrent-thread count, not your request count.
• Run checkpointer.setup() again after every langgraph-checkpoint-postgres version bump, since new releases occasionally ship schema migrations.

Migrating from MemorySaver without breaking existing threads

If you prototyped with MemorySaver and now need durability, the swap is mechanical: change the import, construct the Postgres-backed checkpointer, call setup() once, and pass it to compile(). Existing in-memory threads can't be migrated because they never persisted anywhere, but no code that calls graph.invoke(..., config=config) needs to change. The thread_id contract is identical across every checkpointer backend, which is the entire point of LangGraph's pluggable persistence layer.

# before
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# after
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(DB_URI)
checkpointer.setup()

Everything downstream, node functions, edges, interrupt_before, get_state_history, keeps working unchanged.

Inspecting LangGraph checkpoints directly with SQL

Because LangGraph persistence with Postgres is just rows in ordinary tables, you don't need the Python SDK to answer operational questions. Connect with psql or any SQL client and query the schema setup() created.

SELECT thread_id, checkpoint_id, created_at
FROM checkpoints
WHERE thread_id = 'conversation-1'
ORDER BY created_at DESC
LIMIT 10;

That query alone answers "is this thread actually writing checkpoints" without spinning up your application. To find threads that haven't been touched in a while, useful for building your own retention job instead of relying on a cron script that walks every thread in Python:

SELECT thread_id, MAX(created_at) AS last_activity
FROM checkpoints
GROUP BY thread_id
HAVING MAX(created_at) < now() - interval '30 days';

The checkpoint_blobs table holds the actual serialized state values, keyed by thread_id and channel, and checkpoint_writes holds the pending writes attached to a checkpoint before they're folded into the next one. You rarely need to touch either table directly, but knowing they exist helps when a support ticket says an agent "forgot" something and you need to prove whether the state was ever written in the first place.

One thing to watch for: checkpoint values are stored as serialized blobs (JSON or pickle, depending on your serializer configuration), not as flat columns. Don't try to write ad hoc SQL that filters on the contents of a user's conversation; do that filtering in your application code after loading state through get_state or get_state_history, and use SQL only for the structural questions like row counts, timestamps, and thread inventories.

Handling connection failures gracefully

Postgres persistence adds a network hop to every node transition, which means it can fail in ways MemorySaver never could: a dropped connection, a pool exhausted under load, a brief failover during a managed-Postgres maintenance window. Two defaults are worth setting explicitly rather than trusting library defaults.

First, configure psycopg_pool.ConnectionPool with reconnect_timeout and a sane max_size so a transient network blip doesn't cascade into every request timing out while waiting for a connection slot. Second, wrap the top-level graph.invoke (or ainvoke) call in your own retry logic for transient psycopg.OperationalError exceptions, since LangGraph itself doesn't retry database writes on your behalf; it assumes the checkpointer's connection is healthy and surfaces the error if it isn't.

import psycopg
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=8))
def run_graph(input_state, config):
    try:
        return graph.invoke(input_state, config=config)
    except psycopg.OperationalError:
        raise

This keeps a brief Postgres blip from surfacing as a user-facing failure, while still failing loudly after real, sustained outages instead of retrying forever.

FAQ

Do I need to call setup() every time my app starts? Yes, and it's safe to do so. setup() checks whether the checkpoint tables and migrations already exist and only applies what's missing. Skipping it on a fresh database means your first invoke call will fail with a missing-table error.

Can multiple LangGraph app instances share the same Postgres checkpointer safely? Yes, that's the main reason to use Postgres over SQLite or in-memory checkpointing. Point every instance at the same connection string or pool, and each thread_id stays consistent no matter which instance handles a given request, since Postgres itself serializes the writes.

What's the difference between the checkpointer and the store? The checkpointer persists the full state history of a single thread, node by node, so a specific conversation or task can be paused and resumed. The store persists arbitrary key-value data that's meant to be shared across threads, like user profiles or long-term memory, and it doesn't track step-by-step history the way checkpoints do.

Does PostgresSaver work with any Postgres provider, or only self-hosted instances? Any standard Postgres instance works, including Neon, Supabase, Amazon RDS, and Cloud SQL. PostgresSaver talks to Postgres over psycopg, using a normal connection string, so there's nothing provider-specific about it beyond making sure sslmode matches what your provider requires.

How do I clear out a stuck or corrupted thread? Call checkpointer.delete_thread(thread_id) to remove every checkpoint, write, and blob associated with that thread. There's no partial-delete API for a single checkpoint within a thread, since checkpoints reference each other in a chain, so deletion works at the thread level.

Is PostgresSaver safe to use with async FastAPI endpoints? Use AsyncPostgresSaver for async code paths rather than mixing it with the sync PostgresSaver. Both implement the same checkpoint schema, but calling sync PostgresSaver methods from inside an async endpoint blocks the event loop, which defeats the purpose of running FastAPI async in the first place.