Implementation:Langchain ai Langgraph StateGraph Compile With Interrupts

Property	Value
API	`StateGraph.compile(..., interrupt_before=[...], interrupt_after=[...])` and `create_react_agent(..., interrupt_before=["tools"], ...)`
Type	API Doc
Workflow	Human_in_the_Loop_Agent
Pipeline Stage	Graph Compilation
Repository	Langchain_ai_Langgraph
Source	`libs/langgraph/langgraph/graph/state.py:L1035-1045`, `libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py:L302-303`

Overview

StateGraph.compile() transforms a StateGraph builder into an executable CompiledStateGraph (which extends Pregel). The interrupt_before and interrupt_after parameters specify which nodes should trigger execution pauses for human-in-the-loop workflows. The create_react_agent() function wraps this compilation and passes interrupt parameters through.

Description

The compile() method performs the following steps relevant to interrupts:

Validates the checkpointer via ensure_valid_checkpointer().
Defaults interrupt_before and interrupt_after to empty lists if None.
Validates that all named interrupt nodes exist in the graph via self.validate().
Constructs a CompiledStateGraph instance, passing interrupt_before_nodes and interrupt_after_nodes to the underlying Pregel constructor.

The create_react_agent() function builds a StateGraph with "agent" and "tools" nodes internally, then calls workflow.compile() with the user-provided interrupt parameters. The typical human-in-the-loop pattern uses interrupt_before=["tools"].

Usage

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.prebuilt import create_react_agent

# Approach 1: Direct StateGraph compilation with interrupts
memory = InMemorySaver()
graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"],
    interrupt_after=["agent"],
)

# Approach 2: create_react_agent with interrupts
graph = create_react_agent(
    model,
    tools=[my_tool],
    checkpointer=InMemorySaver(),
    interrupt_before=["tools"],
)

Code Reference

Source Location

Item	Path
`StateGraph.compile()`	`libs/langgraph/langgraph/graph/state.py`, Lines 1035-1153
`create_react_agent()`	`libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py`, Lines 278-1002
Interrupt parameter pass-through	`libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py`, Lines 995-999

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

def create_react_agent(
    model: ...,
    tools: ...,
    *,
    checkpointer: Checkpointer | None = None,
    interrupt_before: list[str] | None = None,
    interrupt_after: list[str] | None = None,
    ...
) -> CompiledStateGraph:

Import

from langgraph.graph import StateGraph
from langgraph.prebuilt import create_react_agent

I/O Contract

`StateGraph.compile()` interrupt-related parameters

Parameter	Type	Default	Description
`checkpointer`	bool \| BaseCheckpointSaver)	`None`	The checkpoint saver. Required for interrupt functionality. `None` inherits from parent graph; `False` disables; `True` is for subgraphs only.
`interrupt_before`	list[str] \| None	`None`	Node names to interrupt before. Use `"*"` for all nodes.
`interrupt_after`	list[str] \| None	`None`	Node names to interrupt after. Use `"*"` for all nodes.

Returns: CompiledStateGraph -- The compiled graph with interrupt behavior configured.

Raises: ValueError -- If an interrupt node name is not found in the graph.

Usage Examples

Human-in-the-loop tool approval

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: list

def agent(state: State):
    # LLM call that may produce tool calls
    return {"messages": [ai_response]}

def tools(state: State):
    # Execute tool calls
    return {"messages": [tool_result]}

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

# Compile with interrupt before tools for human approval
graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["tools"],
)

config = {"configurable": {"thread_id": "review-thread"}}

# Run until interrupt
for event in graph.stream({"messages": [user_msg]}, config):
    print(event)
# Graph pauses before "tools" node

# Inspect pending tool calls via get_state()
state = graph.get_state(config)
# state.next == ("tools",)

# Resume execution
for event in graph.stream(None, config):
    print(event)

Using the wildcard for all nodes

graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before="*",  # Interrupt before every node
)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment