Is a LangGraph compiled graph thread-safe / advised for concurrent use?

[JoshuaC215]

I just wanted to validate if it's ok to initialize/compile the graph once and then use it to serve multiple parallel requests in a web application. In other words is the shared state passed from node to node is thread-safe?

E.g. is it advisable to do something like this (kinda pseudo-code)? Or is there an advantage to re-compiling the graph for each new request?

from fastapi import FastAPI
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel

app = FastAPI()

class Question(BaseModel):
    question: str

# Global variable to store the chatbot
chatbot = None

@app.on_event("startup")
async def startup_event():
    global chatbot
    model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    chatbot = create_react_agent(model, [])

@app.post("/ask")
async def ask_question(question: Question):
    response = chatbot.ainvoke({"messages": [HumanMessage(content=question.question)]})
    return {"answer": response}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

(Also asked in langchain-ai/langchain#23630)

[nfcampos]

It is entirely safe to share a graph between executions, whether they happen concurrently or not, whether in same thread or not. No state is ever stored on the graph instance, and the graph instance isn't ever mutated in any way during any execution of the graph

[davidrosenstark]

what about checkpointer -- if i am using a database like postgres i would assume i need to pull a separate connection from the pool for each run of the graph (thread). Is this a correct assumption?

[sayansahaacc]

for the Postgres Check pointer, how we should manage the DB connection. If we put that in start_up event, after sometime the connection is closing and giving issue in graph astream/ainvoke.

[davidrosenstark]

What i have done to solve this is to have a thread running to continuosly get a connection and do "select 1" (took the idea from java connection pools i have worked with that include a test query). The PG connection pool will automatically create new connections when they go bad

[sayansahaacc]

If I use AsyncConnectionPool in the FastAPI Lifespan, will that resolve this connection issue? or I have to hit the memory as you suggested. Below is the code snippet I am using. @nfcampos can you please suggest if this approach is correct ? or I should follow the workaround @davidrosenstark suggested above.

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Use SqlLite or Postgresql DB for persistance of graph state based on configuration
    if CheckpointDatabase.lower() == "sqlite": 
        print("Construct agent with Sqlite checkpointer")
        # Construct agent with Sqlite checkpointer
        async with AsyncSqliteSaver.from_conn_string("checkpointerDB/checkpoints.db") as saver:
            research_assistant = graph.compile(checkpointer=saver)
            app.state.agent = research_assistant
            yield
        # context manager will clean up the AsyncSqliteSaver on exit
    else:
        print("Construct agent with Postgres checkpointer")
        # Construct agent with Postgres checkpointer
        connection_kwargs = { "autocommit": True }
        async with AsyncConnectionPool(conninfo=DB_URI, kwargs=connection_kwargs,) as pool:
            saver = AsyncPostgresSaver(pool)
            await saver.setup()
            research_assistant = graph.compile(checkpointer=saver)
            app.state.agent = research_assistant
            yield