Langgraph Graph API

LangGraph Graph API Skill

Expert assistance for the LangGraph Graph API: explicit, declarative agent graphs using StateGraph, typed state, nodes, and edges. Use this API when you need visible, inspectable control flow and want to define your agent as a directed graph.

When to choose Graph API over Functional API:

• You want an explicit graph topology you can visualize
• Your workflow has clear, named stages (nodes)
• You need conditional branching between well-defined steps
• You're building a new agent from scratch (not wrapping an existing one)

Full corpus: ../langgraph/references/llms-txt.md (5.4 MB) and ../../langchain-deepagents/references/llms-full.md (10 MB).

When to Use This Skill

Activate when:

• Defining a StateGraph — creating nodes, edges, and graph topology
• Typing agent state — designing TypedDict state schemas (input, output, overall, private)
• Routing between nodes — using Command for single routing or Send for parallel fan-out
• Adding persistence — wiring InMemorySaver or PostgresSaver checkpointers
• Implementing HITL — calling interrupt() inside a node to pause execution
• Streaming graph output — choosing between values, updates, messages, or events modes
• Using Pydantic state — validating state fields with BaseModel instead of TypedDict
• Building multi-agent systems — routing between specialized subgraph agents

Quick Reference

Minimal StateGraph

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

class State(TypedDict):
    input: str
    result: str

def process(state: State) -> State:
    return {"result": state["input"].upper()}

graph = StateGraph(State)
graph.add_node("process", process)
graph.add_edge(START, "process")
graph.add_edge("process", END)

app = graph.compile(checkpointer=InMemorySaver())
result = app.invoke({"input": "hello"}, config={"configurable": {"thread_id": "1"}})

Separate input / output / private state types

from typing import TypedDict
from langgraph.graph import StateGraph

class InputState(TypedDict):
    user_input: str

class OutputState(TypedDict):
    graph_output: str

class OverallState(TypedDict):  # Internal full state
    foo: str
    user_input: str
    graph_output: str

class PrivateState(TypedDict):  # Only visible within a single node
    bar: str

def node_1(state: InputState) -> OverallState:
    return {"foo": state["user_input"] + " processed"}

def node_2(state: OverallState) -> PrivateState:
    return {"bar": state["foo"] + " private"}

def node_3(state: PrivateState) -> OutputState:
    return {"graph_output": state["bar"]}

graph = StateGraph(OverallState, input=InputState, output=OutputState)
graph.add_node("node_1", node_1)
graph.add_node("node_2", node_2)
graph.add_node("node_3", node_3)

Conditional routing with Command

from langgraph.types import Command
from langgraph.graph import StateGraph, START

def router(state: State) -> Command:
    if state["type"] == "math":
        return Command(goto="math_agent")
    return Command(goto="general_agent")

# Fan-out to multiple agents in parallel with Send
from langgraph.types import Send

def parallel_router(state: State):
    return [
        Send("agent_a", {"task": state["task_a"]}),
        Send("agent_b", {"task": state["task_b"]}),
    ]

LLM-based router with Pydantic state

from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class RouteDecision(BaseModel):
    target: str = Field(description="Which agent to route to")

router_llm = init_chat_model("anthropic:claude-sonnet-4-6").with_structured_output(RouteDecision)

class State(BaseModel):  # Pydantic validates all state writes
    query: str
    route: str = ""
    result: str = ""

Human-in-the-loop interrupt

from langgraph.types import interrupt, Command

def approval_node(state: State) -> State:
    decision = interrupt({
        "prompt": "Approve this action?",
        "action": state["pending_action"],
    })
    return {"approved": decision == "yes"}

# Resume after interrupt
app.invoke(
    Command(resume="yes"),
    config={"configurable": {"thread_id": "1"}},
)

Streaming modes

config = {"configurable": {"thread_id": "1"}}

# Full state after every node
for chunk in app.stream(inputs, config, stream_mode="values"):
    print(chunk)

# Only the delta from each node
for chunk in app.stream(inputs, config, stream_mode="updates"):
    print(chunk)

# LLM token-by-token (for nodes with chat models)
for chunk in app.stream(inputs, config, stream_mode="messages"):
    print(chunk)

Per-node retries and fault tolerance

from langgraph.types import RetryPolicy

graph.add_node(
    "api_call",
    call_external_api,
    retry=RetryPolicy(max_attempts=3, backoff_factor=2.0),
)

Time travel — inspect and replay

# Get full thread history
history = list(app.get_state_history(config={"configurable": {"thread_id": "1"}}))

# Replay from a prior checkpoint
past_config = history[2].config
app.invoke(None, config=past_config)

# Patch state at a checkpoint, then replay
app.update_state(config=past_config, values={"result": "corrected"})
app.invoke(None, config=past_config)

Graph API Concepts

Reference Files

Key doc pages: docs.langchain.com/oss/python/langgraph/graph-api, /use-graph-api, /choosing-apis