Jump to content

Connect Leeroopedia MCP: Equip your AI agents to search best practices, build plans, verify code, diagnose failures, and look up hyperparameter defaults.

Implementation:Langchain ai Langgraph StateGraph Compile

From Leeroopedia
Metadata Value
Type Implementation (API Doc)
Library langgraph
Source File libs/langgraph/langgraph/graph/state.py
Lines L1035-1153
Workflow Building_a_Stateful_Graph

Overview

StateGraph.compile validates the graph topology and transforms the builder into an executable CompiledStateGraph (which extends Pregel). The compiled graph implements the LangChain Runnable interface, supporting invoke(), stream(), ainvoke(), and astream().

Description

The compile() method performs the following steps:

  1. Validates the checkpointer -- Ensures the checkpointer value is valid using ensure_valid_checkpointer().
  2. Validates the graph topology -- Calls self.validate() to check that all edge sources/targets exist, an entry point is defined, and interrupt nodes are valid.
  3. Prepares output and stream channels -- Determines which channels constitute the graph's output and which channels should be streamed.
  4. Creates the CompiledStateGraph -- Instantiates CompiledStateGraph with all channels, the checkpointer, interrupt configuration, store, cache, and debug settings.
  5. Attaches nodes -- Calls attach_node() for the virtual START node and each user-defined node, wiring up trigger channels, state readers, and output writers.
  6. Attaches edges -- Calls attach_edge() for each static edge, setting up channel writes that trigger downstream nodes.
  7. Attaches branches -- Calls attach_branch() for each conditional edge, setting up routing logic.
  8. Final validation -- Calls validate() on the compiled graph to ensure internal consistency.

Usage

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    x: int

def my_node(state: State) -> dict:
    return {"x": state["x"] + 1}

builder = StateGraph(State)
builder.add_node(my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)

graph = builder.compile()
result = graph.invoke({"x": 0})
# {'x': 1}

Code Reference

Source Location

Item Path Lines
compile libs/langgraph/langgraph/graph/state.py L1035-1153
validate libs/langgraph/langgraph/graph/state.py L986-1033
CompiledStateGraph libs/langgraph/langgraph/graph/state.py L1156-1476

Signature

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]:

Import

from langgraph.graph import StateGraph

I/O Contract

Parameters

Parameter Type Default Description
checkpointer Checkpointer None A checkpoint saver instance, None (inherit from parent), or False (disable). Enables pause/resume/replay when provided.
cache None None Optional cache backend for node result caching.
store None None Optional persistent key-value store accessible by nodes.
interrupt_before list[str] | None None Node names at which to pause execution before the node runs. Use "*" for all nodes.
interrupt_after list[str] | None None Node names at which to pause execution after the node runs. Use "*" for all nodes.
debug bool False Enable debug mode for verbose output during execution.
name None None Optional name for the compiled graph. Defaults to "LangGraph".

Returns: CompiledStateGraph[StateT, ContextT, InputT, OutputT] -- An executable graph that extends Pregel and implements the Runnable interface.

Raises:

  • ValueError -- If no entry point is defined (no edge from START).
  • ValueError -- If an edge references an unknown node.
  • ValueError -- If an interrupt node does not exist.

Usage Examples

Basic Compilation

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    x: int

def inc(state: State) -> dict:
    return {"x": state["x"] + 1}

builder = StateGraph(State)
builder.add_node(inc)
builder.add_edge(START, "inc")
builder.add_edge("inc", END)

graph = builder.compile()
graph.invoke({"x": 0})
# {'x': 1}

Compilation with Checkpointer

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Must provide thread_id when using a checkpointer
config = {"configurable": {"thread_id": "session-1"}}
graph.invoke({"x": 0}, config)

Human-in-the-Loop with Interrupts

graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["inc"],
)

config = {"configurable": {"thread_id": "review-1"}}
# Execution pauses before "inc" runs
result = graph.invoke({"x": 0}, config)
# Resume by invoking with None input
result = graph.invoke(None, config)

Named Graph

graph = builder.compile(name="MyCustomGraph")
graph.get_name()
# 'MyCustomGraph'

Debug Mode

graph = builder.compile(debug=True)
# Execution will print detailed step-by-step output
graph.invoke({"x": 0})

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment