PostgreSQL Checkpointing & Memory Setup

You've built LangGraph agents. They loop, call tools, and produce answers. But right now, every conversation evaporates when the process restarts. This guide shows you how to wire PostgreSQL as your checkpointer so agent state survives crashes, conversations persist across sessions, and you get time-travel debugging for free.

By the end of this article, you will have:

• A PostgreSQL database configured for LangGraph checkpointing.
• A production-ready agent with persistent state management.
• Understanding of thread IDs, checkpoint metadata, and memory patterns.

---

Why Checkpointing Matters

Without checkpointing, every graph execution is ephemeral. A process crash, a network interruption, or even a routine deployment means losing all accumulated state. Checkpointing is the foundation for five capabilities:

Capability	What It Solves
Crash recovery	Restart from the last saved step, not from the beginning.
Conversation memory	Users continue where they left off, across sessions and deployments.
Human-in-the-loop	interrupt() requires a checkpointer to pause and resume execution.
Time travel	Inspect or revert to any previous state in the execution history.
Debugging	Replay execution from any checkpoint to reproduce issues.

Why PostgreSQL Over In-Memory

MemorySaver (the default in examples) stores checkpoints in RAM. It is convenient for development but unacceptable in production:

• No durability — restart the process and all checkpoints vanish.
• No sharing — multiple worker processes cannot share state.
• No querying — you cannot search or filter checkpoints.

PostgreSQL gives you durable storage, concurrent access across workers, and the ability to query checkpoint metadata. It is our team's standard for production LangGraph deployments.

Why Not SQLite or Redis?

• SQLite works for single-server deployments but does not handle concurrent writes well. Use it only for local development.
• Redis offers low latency but lacks the query capabilities and ACID guarantees of PostgreSQL. Use it only when checkpoint throughput is the bottleneck.

---

Understanding Checkpointing

Before wiring up PostgreSQL, understand what LangGraph is persisting.

What Is a Checkpoint

A checkpoint is a snapshot of graph state at a super-step boundary. A super-step is one complete "tick" of the graph where all scheduled nodes execute. After each super-step, LangGraph saves the full state to the checkpointer.

For a simple graph START → node_a → node_b → END, four checkpoints are created:

1. Input checkpoint — empty state, next is __start__.
2. After input — user input applied, next is node_a.
3. After node_a — node_a output merged into state, next is node_b.
4. After node_b — final state, next is empty (graph complete).

Each checkpoint is a StateSnapshot containing:

Field	Type	Description
values	dict	State channel values at this checkpoint.
next	tuple[str, ...]	Node names to execute next. Empty means the graph is complete.
config	dict	Contains thread_id, checkpoint_ns, and checkpoint_id.
metadata	dict	Execution metadata: source, writes, step.
created_at	str	ISO 8601 timestamp of checkpoint creation.
parent_config	dict | None	Config of the previous checkpoint. None for the first.

How Checkpoints Are Stored

Under the hood, PostgresSaver writes to three tables:

• checkpoints — the base checkpoint with serialized state and metadata.
• checkpoint_blobs — large values (like message lists) stored separately to keep the main table lean.
• checkpoint_writes — per-node intermediate writes for fault tolerance. If a node fails mid-super-step, successful nodes' writes are already durable and do not need re-execution.

┌─────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│   checkpoints   │     │  checkpoint_blobs│     │checkpoint_writes │
├─────────────────┤     ├──────────────────┤     ├──────────────────┤
│ thread_id       │────▶│ thread_id        │     │ thread_id        │
│ checkpoint_ns   │     │ checkpoint_ns    │     │ checkpoint_ns    │
│ checkpoint_id   │     │ channel          │     │ checkpoint_id    │
│ parent_..._id   │     │ version          │     │ task_id          │
│ checkpoint      │     │ type             │     │ idx              │
│ metadata        │     │ blob             │     │ channel          │
└─────────────────┘     └──────────────────┘     │ blob             │
                                                  └──────────────────┘

Thread IDs and Conversation Threading

A thread is a sequence of checkpoints sharing the same thread_id. It represents a single conversation or workflow session. Every graph invocation must specify a thread_id:

config = {"configurable": {"thread_id": "thread-abc-123"}}

The thread_id is the primary key for storing and retrieving checkpoints. Without it, the checkpointer cannot save state or resume execution.

Thread ID Length Limit

The thread_id column in PostgreSQL has a practical limit. Keep thread_id values under 255 characters. Use a UUID or hash if you need deterministic IDs:

import uuid
thread_id = str(uuid.uuid4())  # Always under 255 chars

Checkpoint Namespaces and Subgraphs

Each checkpoint has a checkpoint_ns (namespace) field that identifies which graph or subgraph it belongs to:

• "" (empty string) — the checkpoint belongs to the root (parent) graph.
• "node_name:uuid" — the checkpoint belongs to a subgraph invoked as the given node.
• "outer:uuid|inner:uuid" — nested subgraphs use | as a separator.

You can access the checkpoint namespace from within a node via the config:

from langchain_core.runnables import RunnableConfig

def my_node(state: AgentState, config: RunnableConfig) -> dict:
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    # "" for root graph, "node_name:uuid" for a subgraph

Checkpoint Lifecycle Diagram

User invokes graph with thread_id
         │
         ▼
  ┌───────────────┐
  │  Checkpoint 0 │ ◀── Empty state, source="input"
  └───────┬───────┘
          │ node_a executes
          ▼
  ┌───────────────┐
  │  Checkpoint 1 │ ◀── node_a output merged, source="loop"
  └───────┬───────┘
          │ node_b executes
          ▼
  ┌───────────────┐
  │  Checkpoint 2 │ ◀── node_b output merged, source="loop"
  └───────┬───────┘
          │ graph complete (next=())
          │
          │ ─── Process restarts ───
          │
          ▼
  Resume from Checkpoint 2
  (full state restored from PostgreSQL)

---

Setting Up PostgreSQL

Prerequisites

• A running PostgreSQL instance (13 or later recommended).
• A connection string in the format postgresql://user:password@host:port/database.
• uv for Python package management.

Installing Dependencies

# Install the PostgreSQL checkpointer
uv add langgraph-checkpoint-postgres

# The package includes psycopg (Psycopg 3) by default.
# For production, consider psycopg[binary] for better performance:
# uv add "psycopg[binary]"

The package langgraph-checkpoint-postgres (v3.1.0+) provides both synchronous and asynchronous checkpointers.

Creating the Database Schema

LangGraph provides a setup() method that creates all required tables and runs migrations. Call it once when deploying:

from langgraph.checkpoint.postgres import PostgresSaver

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

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables if they don't exist

The setup() method:

1. Creates checkpoint_migrations, checkpoints, checkpoint_blobs, and checkpoint_writes tables.
2. Runs any pending migrations automatically.
3. Creates indexes on thread_id for efficient lookups.

Migration Safety

The migrations table tracks schema version. Calling setup() multiple times is safe — it only applies migrations that haven't run yet.

Environment Configuration

Store your connection string in environment variables. Never hardcode database credentials:

.env File

# .env
LANGGRAPH_DB_URI=postgresql://langgraph_user:secure_password@postgres.internal:5432/langgraph

Python Loading

import os
from dotenv import load_dotenv

load_dotenv()

DB_URI = os.environ["LANGGRAPH_DB_URI"]

Azure Database for PostgreSQL Notes

Since our infrastructure runs on Azure, here are specific considerations:

• SSL is required — Azure Database for PostgreSQL requires SSL connections. Your connection string should include sslmode=require (or verify-full for production):

DB_URI = "postgresql://user:pass@server.postgres.database.azure.com:5432/db?sslmode=require"

• Connection pooling — Azure recommends using a connection pool for production workloads. PostgresSaver accepts a psycopg_pool.ConnectionPool:

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

pool = ConnectionPool(
    conninfo=DB_URI,
    min_size=3,
    max_size=10,
    timeout=5,
    open=True,
)
checkpointer = PostgresSaver(pool)

• Firewall rules — ensure your application's IP is whitelisted in the Azure firewall rules.
• Private endpoints — for VNet-deployed agents, use private endpoints instead of public IP access.

---

Configuring the Checkpointer

Creating a PostgresSaver Instance

There are two ways to create a checkpointer: using the convenience factory or passing a connection directly.

Factory Method (Recommended)

The from_conn_string() class method handles connection lifecycle management:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = os.environ["LANGGRAPH_DB_URI"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)
    # Checkpointer is automatically closed when the context exits

Direct Connection

For more control (e.g., connection pooling, pipeline mode):

from psycopg import Connection
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

# Option A: Single connection
conn = Connection.connect(
    DB_URI,
    autocommit=True,       # Required for setup() to persist tables
    prepare_threshold=0,
    row_factory=dict_row,  # Required: checkpointer accesses rows by name
)
checkpointer = PostgresSaver(conn)
checkpointer.setup()

# Option B: Connection pool (production)
pool = ConnectionPool(
    conninfo=DB_URI,
    min_size=3,
    max_size=10,
    open=True,
)
checkpointer = PostgresSaver(pool)
checkpointer.setup()

Connection Parameters

When creating connections manually, you must include:

• autocommit=True — required for setup() to properly commit checkpoint tables.
• row_factory=dict_row — the checkpointer accesses database rows using dictionary-style syntax (row["column_name"]). Without this, you get TypeError exceptions.

Passing to compile(checkpointer=...)

Once you have a PostgresSaver instance, pass it to compile():

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

# ... define your graph ...

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    app = graph_builder.compile(checkpointer=checkpointer)

Async vs Sync Checkpointer

LangGraph provides both synchronous and asynchronous checkpointers. Choose based on your application's concurrency model:

Sync (PostgresSaver)

from langgraph.checkpoint.postgres import PostgresSaver

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

    result = app.invoke(
        {"messages": [HumanMessage(content="Hello")]},
        config={"configurable": {"thread_id": "thread-1"}},
    )

Async (AsyncPostgresSaver)

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

    result = await app.ainvoke(
        {"messages": [HumanMessage(content="Hello")]},
        config={"configurable": {"thread_id": "thread-1"}},
    )

Connection Pooling Best Practices

For production deployments with multiple concurrent requests:

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

# Create the pool once at application startup
pool = ConnectionPool(
    conninfo=DB_URI,
    min_size=3,         # Minimum idle connections
    max_size=10,        # Maximum connections
    timeout=5,          # Seconds to wait for a connection
    max_idle=300,       # Close idle connections after 5 minutes
    open=True,          # Open the pool immediately
)

# Reuse the same checkpointer instance across requests
checkpointer = PostgresSaver(pool)
checkpointer.setup()

# The pool handles connection lifecycle automatically
app = graph.compile(checkpointer=checkpointer)

Pool Sizing

• min_size — set to the number of expected concurrent graph executions.
• max_size — set to 2-3x min_size to handle traffic spikes.
• Monitor connection usage with Azure Database metrics and adjust accordingly.

Complete Code Example: Agent With PostgreSQL Checkpointing

Here is a production-ready agent with PostgreSQL checkpointing, showing both the graph definition and the checkpointer wiring:

import os
import uuid
from typing_extensions import TypedDict, Annotated
import operator

from langchain.messages import AnyMessage, HumanMessage, SystemMessage, ToolMessage
from langchain.tools import tool
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver


# --- State definition ---
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    query: str
    step_count: int


# --- Tools ---
@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    # Replace with real search API in production
    return f"Search results for '{query}': [simulated results]"


tools = [web_search]
tools_by_name = {tool.name: tool for tool in tools}


# --- Model setup ---
model = init_chat_model("gpt-4o")
model_with_tools = model.bind_tools(tools)


# --- Nodes ---
def call_llm(state: AgentState) -> dict:
    """Call the LLM with current conversation history."""
    system = SystemMessage(
        content="You are a helpful research assistant. Use web_search when needed."
    )
    response = model_with_tools.invoke([system] + state["messages"])
    return {
        "messages": [response],
        "step_count": state.get("step_count", 0) + 1,
    }


def execute_tools(state: AgentState) -> dict:
    """Execute tool calls from the LLM's last response."""
    results = []
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        tool_fn = tools_by_name.get(tool_call["name"])
        if not tool_fn:
            results.append(
                ToolMessage(
                    content=f"Unknown tool: {tool_call['name']}",
                    tool_call_id=tool_call["id"],
                )
            )
            continue

        try:
            observation = tool_fn.invoke(tool_call["args"])
            results.append(
                ToolMessage(content=str(observation), tool_call_id=tool_call["id"])
            )
        except Exception as exc:
            results.append(
                ToolMessage(
                    content=f"Tool failed: {type(exc).__name__}: {exc}",
                    tool_call_id=tool_call["id"],
                )
            )

    return {"messages": results}


def route_after_llm(state: AgentState) -> str:
    """Route based on whether the LLM requested tool calls."""
    if state["messages"][-1].tool_calls:
        return "execute_tools"
    return END


# --- Build the graph ---
graph_builder = StateGraph(AgentState)
graph_builder.add_node("call_llm", call_llm)
graph_builder.add_node("execute_tools", execute_tools)
graph_builder.add_edge(START, "call_llm")
graph_builder.add_conditional_edges("call_llm", route_after_llm, ["execute_tools", END])
graph_builder.add_edge("execute_tools", "call_llm")


# --- Wire up PostgreSQL checkpointing ---
DB_URI = os.environ["LANGGRAPH_DB_URI"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    app = graph_builder.compile(checkpointer=checkpointer)

    # First conversation turn
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    result = app.invoke(
        {
            "messages": [HumanMessage(content="What is LangGraph checkpointing?")],
            "step_count": 0,
        },
        config=config,
    )

    # The agent state is now persisted in PostgreSQL.
    # You can resume this conversation later using the same thread_id.
    print(f"Thread ID: {thread_id}")
    print(f"Step count: {result['step_count']}")
    print(f"Messages: {len(result['messages'])}")

---

Using Thread IDs

Generating Thread IDs

Use UUIDs for thread IDs. They are unique, short enough for the database column, and deterministic:

import uuid

# Generate a new thread ID for each conversation
thread_id = str(uuid.uuid4())
# Example: "550e8400-e29b-41d4-a716-446655440000"

config = {"configurable": {"thread_id": thread_id}}

For user-facing applications, you may want a more meaningful ID scheme:

# Combine user ID with a session identifier
thread_id = f"user-{user_id}-session-{uuid.uuid4().hex[:8]}"

Keep thread_id under 255 characters

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

Passing Thread Config to invoke() and stream()

Every graph invocation that should persist state needs the config with thread_id:

# invoke — returns final state
result = app.invoke(
    {"messages": [HumanMessage(content="First message")]},
    config=config,
)

# stream — observe state changes in real time
for event in app.stream(
    {"messages": [HumanMessage(content="Second message")]},
    config=config,
    stream_mode="values",
):
    print(event)

# ainvoke / astream — async variants
result = await app.ainvoke(
    {"messages": [HumanMessage(content="Third message")]},
    config=config,
)

Resuming a Conversation From a Checkpoint

To continue a conversation, invoke the graph again with the same thread_id. The checkpointer loads the last checkpoint and continues from there:

# Turn 1
config = {"configurable": {"thread_id": "thread-abc"}}
app.invoke(
    {"messages": [HumanMessage(content="My name is Alice")]},
    config=config,
)

# Turn 2 — same thread_id, state is restored automatically
app.invoke(
    {"messages": [HumanMessage(content="What did I just tell you?")]},
    config=config,
)
# The agent remembers the name "Alice" from the previous turn

Listing and Managing Threads

You can inspect checkpoint history for any thread:

# Get the latest state snapshot
current_state = app.get_state(config)
print(f"Current state: {current_state.values}")
print(f"Next nodes: {current_state.next}")
print(f"Checkpoint ID: {current_state.config['configurable']['checkpoint_id']}")

# Get full execution history
for snapshot in app.get_state_history(config):
    print(f"Step {snapshot.metadata['step']}: next={snapshot.next}")

# Delete a thread (all checkpoints and writes)
checkpointer.delete_thread("thread-abc")

---

Checkpoint Metadata

Adding Custom Metadata

Every checkpoint automatically includes metadata about its source ("input", "loop", or "update"), the step number, and the writes that produced it. You can add custom metadata by passing it in the config:

config = {
    "configurable": {"thread_id": thread_id},
    "metadata": {
        "user_id": "user-123",
        "agent_version": "2.1.0",
        "environment": "production",
        "session_type": "research",
    },
}

result = app.invoke(
    {"messages": [HumanMessage(content="Research quantum computing")]},
    config=config,
)

Querying Checkpoints by Metadata

The list() method on the checkpointer supports metadata filtering:

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # List all checkpoints for a thread
    all_checkpoints = list(checkpointer.list({"configurable": {"thread_id": thread_id}}))

    # Filter by metadata
    production_checkpoints = list(
        checkpointer.list(
            {"configurable": {"thread_id": thread_id}},
            filter={"metadata.environment": "production"},
        )
    )

    # Limit results
    recent_checkpoints = list(
        checkpointer.list(
            {"configurable": {"thread_id": thread_id}},
            limit=5,
        )
    )

Common Metadata Patterns

Establish a consistent metadata schema across your agents:

Key	Example	Purpose
user_id	"user-123"	Associate checkpoints with a user.
agent_version	"2.1.0"	Track which agent version produced the checkpoint.
environment	"production"	Filter checkpoints by deployment environment.
session_type	"research"	Categorize conversation types.
cost	0.042	Track LLM token costs per checkpoint.
tags	["urgent", "customer-facing"]	Multi-value categorization.

# Consistent metadata in production
def make_config(thread_id: str, user_id: str) -> dict:
    return {
        "configurable": {"thread_id": thread_id},
        "metadata": {
            "user_id": user_id,
            "agent_version": APP_VERSION,
            "environment": os.environ.get("ENVIRONMENT", "development"),
        },
    }

---

Basic Memory Patterns

How Message History Persists

When you use Annotated[list[AnyMessage], operator.add] for the messages field in your state, each checkpoint captures the full conversation history. PostgreSQL stores this as a serialized blob, and it is restored automatically when you resume a thread:

class AgentState(TypedDict):
    # This field accumulates messages across checkpoints
    messages: Annotated[list[AnyMessage], operator.add]
    query: str

When the graph resumes from a checkpoint, the messages list is fully restored — including all HumanMessage, AIMessage, ToolMessage, and SystemMessage objects.

Loading Previous Conversation State

To load and inspect previous conversation state:

# Get the latest checkpoint state
state = app.get_state(config)

# Access message history
messages = state.values.get("messages", [])
for msg in messages:
    print(f"[{msg.type}] {msg.content[:100]}...")

# Get a specific checkpoint by ID
checkpoint_config = {
    "configurable": {
        "thread_id": thread_id,
        "checkpoint_id": "01H9Z7...",  # From a previous get_state() call
    }
}
past_state = app.get_state(checkpoint_config)

Clearing / Resetting Conversation Memory

To start a fresh conversation while keeping the same thread:

# Option 1: Update state to clear messages
app.update_state(
    config,
    {"messages": []},  # Clear message history
)

# Option 2: Delete the entire thread
checkpointer.delete_thread(thread_id)

# Option 3: Use a new thread_id (simplest)
new_config = {"configurable": {"thread_id": str(uuid.uuid4())}}

When to Clear vs. New Thread

• Clear messages — when you want to reset conversation context but keep other state (e.g., user preferences stored in state).
• Delete thread — when the conversation is complete and you want to free database space.
• New thread_id — when starting a genuinely new conversation. This is the most common pattern.

---

Async Pattern for High-Concurrency Applications

For web applications and APIs handling many concurrent requests, use the async checkpointer:

import asyncio
import os
import uuid

from langchain.messages import HumanMessage
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool


# Application startup
async def create_app():
    """Create the application with async checkpointer."""
    DB_URI = os.environ["LANGGRAPH_DB_URI"]

    pool = AsyncConnectionPool(
        conninfo=DB_URI,
        min_size=3,
        max_size=20,
        timeout=5,
        open=True,
    )

    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.setup()

    app = graph_builder.compile(checkpointer=checkpointer)
    return app, pool


# In your API handler
async def handle_chat(request):
    app = request.app["graph"]
    thread_id = request.json.get("thread_id", str(uuid.uuid4()))
    user_message = request.json["message"]

    config = {"configurable": {"thread_id": thread_id}}

    result = await app.ainvoke(
        {"messages": [HumanMessage(content=user_message)]},
        config=config,
    )

    return {
        "thread_id": thread_id,
        "response": result["messages"][-1].content,
    }

---

Security: Strict Serialization

Security Best Practice

Set LANGGRAPH_STRICT_MSGPACK=true in your environment or pass an explicit allowed_msgpack_modules list when creating your checkpointer. This restricts checkpoint deserialization to known-safe types, preventing code execution if the database is compromised:

import os
os.environ["LANGGRAPH_STRICT_MSGPACK"] = "true"

# Or explicitly:
checkpointer = PostgresSaver(
    conn,
    serde=CustomSerde(allowed_types=["langchain.messages", "builtins"]),
)

---

Next Steps

You now have PostgreSQL checkpointing configured for your LangGraph agents. Here is where to go next:

• LangGraph Best Practices — our team's standards for production-grade agent design, including error handling, human-in-the-loop, and testing strategies.
• Building Agentic Workflows — core LangGraph concepts if you need a refresher on graphs, state, and nodes.
• LangGraph Official Docs — Persistence — deep dive into checkpointers, stores, fault tolerance, and time travel.
• LangGraph Official Docs — Memory — short-term and long-term memory patterns, including semantic, episodic, and procedural memory.
• LangSmith Tracing — set up observability to trace, debug, and evaluate checkpointed agent runs.

Production Checklist

Before deploying to production, verify:

• checkpointer.setup() has been called and tables exist.
• Connection string uses SSL (sslmode=require for Azure).
• Connection pool is configured with appropriate sizing.
• LANGGRAPH_STRICT_MSGPACK=true is set.
• Thread IDs are under 255 characters.
• LangSmith tracing is enabled (LANGSMITH_TRACING=true).
• A retention policy exists for old checkpoints (consider a cron job to prune threads older than N days).