LangGraph AsyncPostgresSaver：修復 FastAPI 中的「psycopg.OperationalError: the connection is closed」錯誤

Disclosure: I work on MemorySync (I'm Mohammed, the founder). I drafted this page with AI assistance and checked it against the source issue and the official docs listed under Sources on Oct 28, 2026. I read the code and docs; I didn't run this FastAPI app for this page.

The question

In langchain-ai/langgraph#1292, "Encountering OperationalError with Postgres Checkpointer in LangGraph" (Aug 9, 2024), the Postgres checkpointer failed with:

psycopg.OperationalError: the connection is closed

The graph was compiled inside a function like this:

async with AsyncPostgresSaver.from_conn_string(os.environ.get("POSTGRES_CONN_STRING")) as checkpointer:
    return workflow.compile(
        checkpointer=checkpointer,
        interrupt_before=[AgentRole.USER_INPUT],
    )

A later commenter asked the follow-up most FastAPI users have: how to compile the graph once in the lifespan and reuse it for every request, instead of compiling per request.

Status (checked Oct 28, 2026 via the GitHub API): closed Aug 9, 2024, after the reply explaining the cause below. The FastAPI lifespan follow-up (Sep 2024, bumped Mar 2025) has no answer in the thread.

Why it happens

from_conn_string is an async context manager. The connection is open only inside the async with block. Returning the compiled graph from inside the block exits it, which closes the connection, but the graph still holds the checkpointer. The first request that reads or writes a checkpoint then hits psycopg's closed connection. As the reply on #1292 put it: when the function returns, "it is already out of the with scope, at which point the database connection has already been closed."

Fix 1: keep the block open for the app's lifetime

Enter the context manager in the lifespan and yield inside it, so it closes only on shutdown:

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

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        await checkpointer.setup()  # first run only: creates the tables
        app.state.graph = builder.compile(checkpointer=checkpointer)
        yield

app = FastAPI(lifespan=lifespan)

Fix 2: use a connection pool

A single connection serves one request at a time. AsyncPostgresSaver also accepts a psycopg AsyncConnectionPool (its connection type in the source is AsyncConnection | AsyncConnectionPool). The package README says manually created connections need autocommit=True and row_factory=dict_row:

from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncConnectionPool(
        DB_URI,
        kwargs={"autocommit": True, "row_factory": dict_row},
    ) as pool:
        checkpointer = AsyncPostgresSaver(pool)
        await checkpointer.setup()
        app.state.graph = builder.compile(checkpointer=checkpointer)
        yield

The source raises an error if you combine a pool with pipeline; pipelines are for a single connection only.

Checklist

1. No return of a compiled graph from inside an async with ... as checkpointer block.
2. Compile once in the lifespan, read app.state.graph in handlers.
3. Call setup() the first time against a new database.
4. Manual connections or pools: autocommit=True and row_factory=dict_row.

Common questions

Why does it work in a script?
In a script you usually run the graph inside the async with block, so the connection is still open.

Does MemorySaver have this problem?
No: the in-memory saver has no connection to close, which is why the app works until you switch to Postgres.

Is the checkpointer my agent's long-term memory?
It's per-thread state (resume, time travel). Memory across threads is a separate store.

If you also need memory across threads

The MemorySync LangGraph guide keeps the two apart: thread state stays on a checkpointer such as langgraph-checkpoint-postgres, and MemorySync supplies long-term memory across threads through a LangGraph BaseStore (MemorySyncStore) passed as store=, plus context injection, a persist node and a search tool. It doesn't replace the checkpointer, so the fix above still applies. Read the guide first. To see save and recall in your browser, try the MemorySync playground.

Sources

• langchain-ai/langgraph#1292 (closed 2024-08-09; read 2026-10-28)
• langgraph libs/checkpoint-postgres/README.md and langgraph/checkpoint/postgres/aio.py, _ainternal.py on main (read 2026-10-28; PyPI latest langgraph-checkpoint-postgres 3.2.0)
• MemorySync docs: docs.memorysync.io/guides/langgraph