LangGraph

Low-level orchestration for stateful, multi-step LLM workflows: graphs, checkpoints, interrupts, and durable agent loops.

What LangGraph Solves

LangGraph models LLM applications as state machines (graphs) instead of opaque agent loops. Each step is a node function that reads shared state and returns partial updates. Edges define control flow: fixed transitions, conditional routing, or dynamic jumps via the Command API.

This matters because production agents need persistence across restarts, human approval on risky actions, replay for debugging, and explicit caps on loops. LangChain LCEL chains are acyclic pipelines; LangGraph adds cycles, branching, and checkpointed state by design.

Refer: LangGraph low-level concepts · GitHub

Core Concepts: State, Nodes, Edges

State schema

State is a TypedDict (or Pydantic model) defining keys the graph reads and writes. Use Annotated[list, add_messages] reducers so new messages append instead of overwrite:

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_id: str
    step_count: int

Reducers define how concurrent or sequential updates merge. Without reducers, last write wins. Custom reducers implement counters, set unions, or last-value semantics per key.

Nodes

Nodes are Python callables (state, config) -> partial_state | Command. They should be small: call the model, execute one tool batch, run a validator. Heavy side effects belong in tool functions with idempotency keys.

Edges

Building a Graph

from langgraph.graph import StateGraph, START, END

def call_model(state: AgentState):
    response = model.invoke(state["messages"])
    return {"messages": [response], "step_count": state["step_count"] + 1}

def should_continue(state: AgentState):
    last = state["messages"][-1]
    if state["step_count"] > 10:
        return END
    if last.tool_calls:
        return "tools"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tools", END])
builder.add_edge("tools", "agent")

graph = builder.compile()

compile() validates the graph, wires checkpointers, and returns an invokable application. The compiled graph is itself a Runnable with invoke, stream, batch, and async variants.

Prebuilt ReAct Agent

create_react_agent(model, tools, prompt=...) ships a production-ready loop: model decides tool calls, tool node executes, returns to model until done. Customize via state schema extensions, pre/post hooks, or by building from scratch when you need non-standard routing.

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model, tools)
result = agent.invoke(
    {"messages": [("user", "Search docs for rate limits")]},
    config={"configurable": {"thread_id": "user-42"}},
)

Prebuilt agents are a starting point. Complex products (multi-tenant ACL, sub-workflows, approval gates) usually outgrow them and require custom graphs.

Persistence and Checkpointers

Checkpointers save graph state after each super-step so runs survive process crashes and support conversation threads.

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    graph.invoke(input, config={"configurable": {"thread_id": "abc-123"}})

thread_id scopes checkpoint history per conversation or workflow instance. Use separate thread IDs per user session. checkpoint_id enables time travel: resume from an earlier snapshot for debugging or forked execution.

What gets persisted

Full state dict after each node (or super-step in parallel graphs). Large blobs (raw PDFs) should live in object storage with references in state, not megabyte strings in Postgres JSON.

Human-in-the-Loop: Interrupts

interrupt() pauses graph execution before or after a node, surfacing state to an operator UI. Resume with Command(resume=...) after approval or edited input.

from langgraph.types import interrupt, Command

def sensitive_action(state):
  approval = interrupt({
      "action": "delete_records",
      "count": state["pending_deletes"],
  })
  if not approval.get("approved"):
      return Command(goto=END)
  return run_delete(state)

Patterns: approve tool calls, edit model drafts before send, collect missing form fields. Interrupts require a checkpointer; the graph serializes pending interrupt metadata in the checkpoint.

The Command API

Nodes can return Command(update={...}, goto="node_name") for dynamic routing plus state updates in one step. Replaces brittle conditional edge functions when routing depends on rich logic inside the node.

Command(graph=Command.PARENT) supports jumping between parent and child graphs in multi-agent setups.

Subgraphs and Multi-Agent

Compile a subgraph and add it as a node in a parent graph. Child graphs can have isolated state schemas with input/output mappers. Supervisor patterns:

Watch cost and latency: each subgraph invocation may trigger full model calls. Cap delegations and cache specialist outputs.

Streaming Modes

graph.stream(input, stream_mode=...) supports multiple modes (combinable):

UI chat products typically stream messages to the client and log updates server-side for observability.

Configuration and Runtime Context

config["configurable"] passes per-run values without polluting state: thread_id, user_id, feature flags, model name. Access in nodes via config = get_config() or the config argument.

store (LangGraph Store) provides long-term memory across threads: namespace + key-value documents retrievable inside nodes, backed by Postgres or in-memory for dev.

LangGraph Platform and Deployment

LangGraph Platform (managed or self-hosted) runs compiled graphs as APIs with built-in persistence, cron triggers, and assistants UI. You package langgraph.json defining graph entrypoints, dependencies, and env vars.

Self-hosting options: Docker containers behind your API gateway, same auth and rate limits as the rest of your stack. Horizontal scale requires shared Postgres checkpointer, not MemorySaver.

For Kubernetes deployments, run stateless API replicas; all coordination lives in the checkpointer DB and optional message queue for long jobs.

Error Handling and Retries

Node failures can retry with policies on the node decorator. Distinguish transient (rate limit, timeout) from permanent (validation) errors. Surface tool errors as ToolMessages so the model can recover instead of crashing the graph.

Use RetryPolicy with max attempts and backoff. For paid side effects, combine retries with idempotency keys on the tool layer.

Testing LangGraph Applications

LangGraph vs LangChain Agents

ConcernLangChain AgentExecutorLangGraph
Control flow visibilityOpaque loopExplicit graph
PersistenceLimited / customFirst-class checkpointers
Human approvalCustom hacksNative interrupts
Parallel branchesAwkwardSupported
Time travel / replayNoCheckpoint history

New agent features in the ecosystem land in LangGraph first. Treat LangChain as the integration layer (models, tools, prompts) and LangGraph as the runtime.

Production Checklist