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 With Persistence

From Leeroopedia
Attribute Value
Page Type Implementation (API Doc)
Library langgraph
Workflow Persistence_and_Memory_Setup
Principle Langchain_ai_Langgraph_Graph_Persistence_Compilation
Implementation StateGraph_Compile_With_Persistence
Source libs/langgraph/langgraph/graph/state.py:L1035-1153

Overview

StateGraph.compile() transforms a graph definition into a runnable CompiledStateGraph. When called with checkpointer and/or store parameters, it enables automatic state persistence and cross-thread memory for the compiled graph.

Description

This method validates the graph structure, prepares output and stream channels, and constructs a CompiledStateGraph instance with the specified persistence configuration. The compiled graph implements the Runnable interface and can be invoked, streamed, batched, and run asynchronously.

Usage

Import

from langgraph.graph import StateGraph

Code Reference

Source Location

Method File Lines
StateGraph.compile libs/langgraph/langgraph/graph/state.py L1035-1153

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:
    """Compiles the StateGraph into a CompiledStateGraph object.

    The compiled graph implements the Runnable interface and can be invoked,
    streamed, batched, and run asynchronously.

    Args:
        checkpointer: A checkpoint saver object or flag.
            If provided, serves as a fully versioned "short-term memory" for the graph.
            If None, may inherit the parent graph's checkpointer when used as a subgraph.
            If False, will not use or inherit any checkpointer.
        cache: A cache instance for caching node results.
        store: A BaseStore instance for cross-thread long-term memory.
        interrupt_before: An optional list of node names to interrupt before.
        interrupt_after: An optional list of node names to interrupt after.
        debug: A flag indicating whether to enable debug mode.
        name: The name to use for the compiled graph.

    Returns:
        CompiledStateGraph: The compiled StateGraph.
    """

I/O Contract

Parameter Type Default Description
checkpointer Checkpointer (BaseCheckpointSaver, None, or False) None Checkpoint saver for thread-scoped state persistence. Pass a saver instance to enable, None to allow inheritance, or False to explicitly disable.
cache None None Optional cache for node result caching.
store None None Cross-thread key-value store for long-term memory. Accessible from nodes via dependency injection.
interrupt_before list[str] | None None Node names where execution pauses before the node runs. Requires a checkpointer.
interrupt_after list[str] | None None Node names where execution pauses after the node runs. Requires a checkpointer.
debug bool False Enable debug mode for verbose execution logging.
name None None Name for the compiled graph. Defaults to "LangGraph".
Returns Type Description
CompiledStateGraph A compiled, runnable graph with persistence configured.

Usage Examples

Basic Persistence with InMemorySaver

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot_node)
builder.set_entry_point("chatbot")
builder.set_finish_point("chatbot")

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

# Invoke with thread_id for state persistence
config = {"configurable": {"thread_id": "user-123"}}
result = graph.invoke({"messages": [("user", "Hello!")]}, config)

# Same thread_id resumes conversation
result = graph.invoke({"messages": [("user", "What did I say?")]}, config)

Persistence with Store for Cross-Thread Memory

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph

checkpointer = InMemorySaver()
store = InMemoryStore()

graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
)

# Thread-scoped state via checkpointer
config1 = {"configurable": {"thread_id": "thread-1"}}
graph.invoke(inputs, config1)

# Cross-thread data via store (accessible from any thread)
config2 = {"configurable": {"thread_id": "thread-2"}}
graph.invoke(inputs, config2)

Human-in-the-Loop with Interrupts

from langgraph.checkpoint.memory import InMemorySaver

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

config = {"configurable": {"thread_id": "review-1"}}
# Execution pauses before "human_review" node
result = graph.invoke(inputs, config)

# After human review, resume from checkpoint
result = graph.invoke(None, config)

Disabling Checkpointer Inheritance in Subgraphs

# Subgraph that explicitly opts out of parent's checkpointer
subgraph = sub_builder.compile(checkpointer=False)

Related Pages

Page Connections

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