How to properly use the Langgraph checkpointer in a web framework?

[gbaian10]

Hello everyone, I am a beginner in databases.

If I use Postgres as a checkpointer in a production environment, how should I correctly use Langgraph in my API when using a web framework? (I am using FastAPI)

1. Create a single global graph object and use only one checkpointer.
2. Create a new graph object for each API request but use the same checkpointer.
3. Create a new graph object for each API request, and each graph object uses a different checkpointer.

If I have multiple different graphs for different API, should they use the same checkpointer?

[nfcampos]

Hi neither the graph nor the checkpointer keep any internal state so using one global instance or a new one per request makes no difference. What id definitely recommend doing is ensuring you create a global connection pool for postgres and use that to get a connection to pass to the checkpointer on each request. That will ensure connections are reused, which saves latency

[gmyrianthous]

@nfcampos Can you please share a quick example with PG Bouncer? A reference in the docs would also be very helpful!

[gbaian10]

@nfcampos Thank you for your reply.

It seems we only need one instance and don't need to create a new graph and obtain a new connection for each request. Right?

I read this official document

I saw that the document used an AsyncConnectionPool and took a connection from the pool to give to AsyncPostgresSaver. So it seems that one AsyncPostgresSaver only uses one connection from the pool.

If all my graphs share the same AsyncPostgresSaver, it looks like the connection pool's features are not being utilized. Right?

This brings me back to the initial question: do I need multiple graphs and multiple AsyncPostgresSavers ? (create each AsyncPostgresSaver obtaining a connection from the pool)

If we only need one AsyncPostgresSaver (even if we have many different graph classes and each only class only need to create one instance, but share the same AsyncPostgresSaver), do I not need to use AsyncConnectionPool?

Is the purpose of using AsyncConnectionPool to allow other packages to also use the database? Is one connection enough for the entire langgraph application?

Sorry for asking so many questions; I'm not very familiar with database applications.

[sudo-god]

Did you get a resolution? Would love to know the answer too.

[gbaian10]

Did you get a resolution? Would love to know the answer too.

import secrets
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from functools import lru_cache
from typing import Annotated, Literal, cast

from fastapi import Depends, FastAPI, Path, Request
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import MessagesState, StateGraph
from langgraph.graph.state import CompiledStateGraph
from psycopg_pool import AsyncConnectionPool
from pydantic import BaseModel

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"


@asynccontextmanager
async def lifespan(
    _: FastAPI,
) -> AsyncGenerator[dict[str, AsyncPostgresSaver], None]:
    async with AsyncConnectionPool(conninfo=DB_URI) as pool:
        checkpointer = AsyncPostgresSaver(pool)
        await checkpointer.setup()
        yield {"checkpointer": checkpointer}


api = FastAPI(lifespan=lifespan)


@lru_cache(maxsize=1)
def get_graph(request: Request) -> CompiledStateGraph:
    def append_random_int(_: MessagesState) -> MessagesState:
        return {"messages": [AIMessage(content=f"{secrets.randbelow(100)}")]}

    checkpointer = request.state.checkpointer
    workflow = (
        StateGraph(MessagesState)
        .add_node(append_random_int)
        .set_entry_point(append_random_int.__name__)
    )
    return workflow.compile(checkpointer=checkpointer)


class Message(BaseModel):
    role: Literal["human", "ai"]
    content: str


class MessagesResponse(BaseModel):
    messages: list[Message]


@api.post("/graph/{thread_id}")
async def run_graph(
    thread_id: Annotated[str, Path()],
    *,
    graph: Annotated[CompiledStateGraph, Depends(get_graph)],
) -> MessagesResponse:
    config = RunnableConfig(configurable={"thread_id": thread_id})
    input_ = {"messages": [HumanMessage(content="Please give me a random number")]}
    output = await graph.ainvoke(input_, config=config)
    output = cast("MessagesState", output)
    return MessagesResponse(
        messages=[
            Message(role=message.type, content=message.content)
            for message in output["messages"]
        ]
    )


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(api)

[telecomshy]

According to the official documentation, the way to create and use a PostgresSaver is as follows:

DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # call .setup() the first time you're using the checkpointer
    checkpointer.setup()
    ...

However, I cannot do this within a route function because creating a new checkpointer instance on every request prevents retrieving the stored conversation history via the config.

My question is: Can I create the checkpointer within a lifespan function instead? For example:

@contextmanager
def lifespan(_: FastAPI):
    with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
        checkpointer.setup()
        yield {"checkpointer": checkpointer}

In this case, would the entire FastAPI app use only a single connection, and would this connection remain unreleased until the app shuts down?

I understand that using a connection pool is a good approach, but besides that, are there any other alternatives?