Postgres connection closed with Langgraph astream_events

[david-sfernandes]

Hi,

I'm starting to learn LangGraph and ran into some issues.

I'm building a chatbot using FastAPI, LangChain, and LangGraph.

I've been trying for some time to add a checkpointer to store the chat history, but whenever I try to execute the astream_event method on my CompiledGraph, I get the following error:

psycopg.OperationalError: the connection is closed

I'm trying to use AsyncPostgresSaver as the checkpointer and adding it when building the graph.

Here’s my graph builder function:

async def graph_builder(retrieve, generate, query_or_respond):
  db_uri = os.getenv("DB_URI")
  tools = ToolNode([retrieve])

  builder = StateGraph(MessagesState)

  builder.add_node(query_or_respond)
  builder.add_node(tools)
  builder.add_node(generate)

  builder.set_entry_point("query_or_respond")
  builder.add_conditional_edges(
      "query_or_respond",
      tools_condition,
      {END: END, "tools": "tools"},
  )
  builder.add_edge("tools", "generate")
  builder.add_edge("generate", END)

  async with AsyncPostgresSaver.from_conn_string(db_uri) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    return graph

I execute this function inside a FastAPI lifespan function:

@asynccontextmanager
async def lifespan(app: FastAPI):
  global graph
  graph = await graph_builder(retrieve, generate, query_or_respond)
  yield
  await graph.close()

However, when I call astream_event from an endpoint, I get that error.

Can anyone point out where I might be making a mistake?

Thanks in advance!

OBS: The astream_events works without this checkpointer config
Main libs I'm using:
fastapi==0.115.8
fastapi-cli==0.0.7
langgraph==0.3.5
langgraph-checkpoint==2.0.15
langgraph-checkpoint-postgres==2.0.15
psycopg==3.2.5
psycopg-binary==3.2.5
psycopg-pool==3.2.6

[aryanranderiya]

did you figure this out?

[hinthornw]

In your code, your checkpoint connection would close before you actually use the graph.

Convert graph_builder into a context manager as well.

@asynccontextmanager
async def graph_builder(retrieve, generate, query_or_respond):
  db_uri = os.getenv("DB_URI")
  tools = ToolNode([retrieve])

  builder = StateGraph(MessagesState)

  builder.add_node(query_or_respond)
  builder.add_node(tools)
  builder.add_node(generate)

  builder.set_entry_point("query_or_respond")
  builder.add_conditional_edges(
      "query_or_respond",
      tools_condition,
      {END: END, "tools": "tools"},
  )
  builder.add_edge("tools", "generate")
  builder.add_edge("generate", END)

  async with AsyncPostgresSaver.from_conn_string(db_uri) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    yield graph

@asynccontextmanager
async def lifespan(app: FastAPI):
  global graph
  async with graph_builder(retrieve, generate, query_or_respond) as built:
      graph = built
      yield
  await graph.close()