Implementation:Langchain ai Langgraph AsyncPostgresSaver

Knowledge Sources	Langchain_ai_Langgraph
Domains	Checkpointing, Postgres
Last Updated	2026-02-11 16:00 GMT

Overview

`AsyncPostgresSaver` is an asynchronous checkpoint saver that persists LangGraph checkpoints and intermediate writes to a PostgreSQL database using the `psycopg` async driver.

Description

`AsyncPostgresSaver` extends `BasePostgresSaver` to provide a fully asynchronous interface for storing, retrieving, and listing graph checkpoints in Postgres. It supports both single `AsyncConnection` and `AsyncConnectionPool` connection modes, with optional pipeline support for batching database operations.

The class manages checkpoint data across three tables: `checkpoints` for the main checkpoint state, `checkpoint_blobs` for large binary channel values, and `checkpoint_writes` for intermediate task writes. It handles automatic serialization and deserialization of checkpoint data, metadata, and channel values via a configurable `SerializerProtocol`. Database schema creation and migrations are managed through the `setup()` method, which must be called before first use.

`AsyncPostgresSaver` also provides synchronous wrappers (`list`, `get_tuple`, `put`, `put_writes`) that delegate to the async methods via `asyncio.run_coroutine_threadsafe`, enabling use from background threads while the event loop runs in the main thread. Thread safety is ensured through an `asyncio.Lock`.

Usage

Use `AsyncPostgresSaver` when you need durable, production-grade checkpoint persistence for LangGraph agents running in async Python applications. It is the recommended checkpointer for async workflows backed by PostgreSQL. Use it when you need to persist conversation state, enable time-travel debugging, or support multi-turn agent interactions with reliable checkpoint storage.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
• Lines: 1-582

Signature

class AsyncPostgresSaver(BasePostgresSaver):
    def __init__(
        self,
        conn: AsyncConnection | AsyncConnectionPool,
        pipe: AsyncPipeline | None = None,
        serde: SerializerProtocol | None = None,
    ) -> None:

Import

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

I/O Contract

Inputs

Name	Type	Required	Description
conn	AsyncConnectionPool	Yes	An async Postgres connection or connection pool from psycopg.
pipe	None	No	An optional async pipeline for batching operations. Cannot be used with AsyncConnectionPool.
serde	None	No	Custom serializer for encoding/decoding checkpoint data. Defaults to the built-in serializer.

Outputs

Name	Type	Description
AsyncPostgresSaver	AsyncPostgresSaver	An instance of the async Postgres checkpoint saver, ready for use after calling `setup()`.

Key Methods

Method	Description
setup()	Creates checkpoint tables and runs database migrations. Must be called before first use.
aget_tuple(config)	Retrieves a single checkpoint tuple by thread ID and optional checkpoint ID.
alist(config, ...)	Lists checkpoint tuples matching filter criteria, ordered by checkpoint ID descending.
aput(config, checkpoint, metadata, new_versions)	Saves a checkpoint and its metadata to the database.
aput_writes(config, writes, task_id, task_path)	Stores intermediate writes linked to a checkpoint.
from_conn_string(conn_string, ...)	Async classmethod context manager that creates an instance from a Postgres connection string.

Usage Examples

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

# Using from_conn_string (recommended)
async with AsyncPostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/dbname"
) as checkpointer:
    await checkpointer.setup()

    # Use with a LangGraph compiled graph
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "thread-1"}}
    result = await graph.ainvoke({"messages": [("user", "hello")]}, config)

# With pipeline mode for better throughput
async with AsyncPostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/dbname",
    pipeline=True,
) as checkpointer:
    await checkpointer.setup()
    # Operations are batched through the pipeline

Related Pages

• Principle:Langchain_ai_Langgraph_Async_Checkpoint_Persistence