What is the return type of StateGraph.compile()?

OSS Product Help
1

Hi everyone,

I’m using LangGraph and I have an async helper to initialize my agent with a checkpointer:

Blockquote
async def get_agent():
“”“Get initialized agent with checkpointer.”“”
checkpointer = await get_checkpointer()
return agent_graph.compile(checkpointer=checkpointer)

My questions are:

  1. What is the exact return type of agent_graph.compile(checkpointer=checkpointer)?
  2. Since this function is async, should I annotate it as → CompiledWorkflow or something else?
  3. Can this compiled workflow be safely stored in FastAPI app.state?

I want to make sure I’m handling type hints and state correctly in an async environment.

Any clarification or examples would be greatly appreciated.

2

hi @meharaz733

1. The Return Type Is CompiledStateGraph

StateGraph.compile() returns a CompiledStateGraph object. You can see this directly in the source code:

# langgraph/graph/state.py

def compile(
    self,
    checkpointer: Checkpointer = None,
    *,
    cache: BaseCache | None = None,
    store: BaseStore | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    debug: bool = False,
    name: str | None = None,
) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:

Source: langgraph/graph/state.py (line 1036)

Import Path

from langgraph.graph.state import CompiledStateGraph

Class Hierarchy

CompiledStateGraph inherits from Pregel, which itself implements Runnable:

CompiledStateGraph
  └── Pregel
        └── PregelProtocol
              └── Runnable (LangChain)

This means your compiled graph supports all standard Runnable methods: invoke(), ainvoke(), stream(), astream(), batch(), etc.

2. Type Annotation for Your Async Function

Since compile() itself is not async (it’s a synchronous method), you annotate the return type directly with CompiledStateGraph:

from langgraph.graph.state import CompiledStateGraph

async def get_agent() -> CompiledStateGraph:
    """Get initialized agent with checkpointer."""
    checkpointer = await get_checkpointer()
    return agent_graph.compile(checkpointer=checkpointer)

Note: CompiledStateGraph is generic (CompiledStateGraph[StateT, ContextT, InputT, OutputT]), but for most practical use cases the simple CompiledStateGraph annotation is sufficient.

3. Storing the Compiled Graph in FastAPI’s app.state

Yes, you can store it in app.state. The compiled graph object is reusable and thread-safe for concurrent invocations. A common pattern is:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from langgraph.graph.state import CompiledStateGraph

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize once at startup
    app.state.agent = await get_agent()
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(message: str):
    result = await app.state.agent.ainvoke({"messages": [("user", message)]})
    return result
  • Use ainvoke()/astream() in async contexts (not invoke()/stream()), as these are the async-native methods.
  • The checkpointer must also be async-compatible. If you’re using AsyncPostgresSaver or AsyncSqliteSaver, they work natively in async FastAPI. Avoid synchronous checkpointers like MemorySaver in async code - they block the event loop.
  • One compiled graph instance can handle multiple concurrent requests - each call to ainvoke()/astream() operates on its own thread/state based on the thread_id in the config.

References

1 Like
3

@pawel-twardziak thank you so much dude.

1 Like
4

This topic was automatically closed 12 hours after the last reply. New replies are no longer allowed.