LangGraph for Production: State Machines, Checkpointing, and Subgraphs

LangGraph in a notebook is a prototype. LangGraph in production is a different beast entirely. You need state machines that handle failures gracefully, checkpointing that survives crashes, and subgraphs that keep complex workflows manageable. This guide covers the patterns that separate demo-quality LangGraph apps from production systems.

Production Requirements

Before deploying LangGraph, you need:

• Deterministic state transitions — No ambiguous routing
• Persistent state — Resume after crashes
• Error recovery — Graceful handling of failures
• Observability — Trace every decision
• Scalability — Handle concurrent workflows

State Machines with TypedDict

Define strict state schemas:

from typing import TypedDict, Literal, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END

class OrderState(TypedState):
    order_id: str
    customer_id: str
    items: list[dict]
    total: float
    status: Literal[
        "validating",
        "checking_inventory",
        "processing_payment",
        "fulfilling",
        "completed",
        "failed"
    ]
    errors: Annotated[list[str], add]
    retry_count: int

def validate_order(state: OrderState) -> OrderState:
    """Validate the order before processing."""
    errors = []
    
    if not state["items"]:
        errors.append("Order has no items")
    
    if state["total"] <= 0:
        errors.append("Invalid order total")
    
    if errors:
        return {"status": "failed", "errors": errors}
    
    return {"status": "checking_inventory"}

def check_inventory(state: OrderState) -> OrderState:
    """Check if all items are in stock."""
    out_of_stock = []
    
    for item in state["items"]:
        if not is_in_stock(item["sku"], item["quantity"]):
            out_of_stock.append(item["name"])
    
    if out_of_stock:
        return {
            "status": "failed",
            "errors": [f"Out of stock: {', '.join(out_of_stock)}"]
        }
    
    return {"status": "processing_payment"}

Durable Execution

LangGraph’s checkpointing provides durable execution:

from langgraph.checkpoint.postgres import PostgresSaver

# Production checkpointing with PostgreSQL
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agent_checkpoints"
)

graph = StateGraph(OrderState)
# ... add nodes and edges ...
app = graph.compile(checkpointer=checkpointer)

# Every execution is automatically checkpointed
config = {"configurable": {"thread_id": "order-123"}}

# If this crashes, resume with:
# app.invoke(None, config)  # Continues from last checkpoint

Subgraph Patterns

Break complex workflows into composable subgraphs:

# Payment subgraph
payment_graph = StateGraph(PaymentState)
payment_graph.add_node("validate_card", validate_card)
payment_graph.add_node("charge", charge_card)
payment_graph.add_node("handle_failure", handle_payment_failure)
payment_graph.add_edge(START, "validate_card")
payment_graph.add_conditional_edges(
    "validate_card",
    lambda s: "charge" if s.get("card_valid") else "handle_failure"
)
payment_graph.add_edge("charge", END)
payment_graph.add_edge("handle_failure", END)
payment_subgraph = payment_graph.compile(checkpointer=checkpointer)

# Fulfillment subgraph
fulfillment_graph = StateGraph(FulfillmentState)
fulfillment_graph.add_node("reserve_inventory", reserve_inventory)
fulfillment_graph.add_node("create_shipment", create_shipment)
fulfillment_graph.add_node("track", track_shipment)
fulfillment_graph.add_edge(START, "reserve_inventory")
fulfillment_graph.add_edge("reserve_inventory", "create_shipment")
fulfillment_graph.add_edge("create_shipment", "track")
fulfillment_graph.add_edge("track", END)
fulfillment_subgraph = fulfillment_graph.compile(checkpointer=checkpointer)

# Main graph uses subgraphs
main_graph = StateGraph(OrderState)
main_graph.add_node("validate", validate_order)
main_graph.add_node("inventory", check_inventory)
main_graph.add_node("payment", payment_subgraph)
main_graph.add_node("fulfillment", fulfillment_subgraph)
main_graph.add_node("complete", complete_order)

main_graph.add_edge(START, "validate")
main_graph.add_conditional_edges(
    "validate",
    lambda s: "inventory" if s["status"] != "failed" else END
)
main_graph.add_conditional_edges(
    "inventory",
    lambda s: "payment" if s["status"] != "failed" else END
)
main_graph.add_edge("payment", "fulfillment")
main_graph.add_edge("fulfillment", "complete")
main_graph.add_edge("complete", END)

production_app = main_graph.compile(checkpointer=checkpointer)

Error Handling and Retries

from langgraph.types import RetryPolicy

def resilient_node(state: OrderState) -> OrderState:
    """Node with automatic retry on failure."""
    try:
        result = call_external_service(state)
        return {"status": result["next_status"]}
    except TemporaryError as e:
        # LangGraph will retry this node
        raise
    
# Add retry policy to nodes
graph.add_node("payment", payment_node, retry=RetryPolicy(max_attempts=3))

# Custom retry logic
def should_retry(state: OrderState) -> Literal["retry", "fail"]:
    if state["retry_count"] < 3:
        return "retry"
    return "fail"

Observability

Structured Logging

import logging
from opentelemetry import trace

tracer = trace.get_tracer("order-processing")

def traced_node(node_name: str):
    def decorator(func):
        def wrapper(state: OrderState) -> OrderState:
            with tracer.start_as_current_span(
                node_name,
                attributes={
                    "order.id": state.get("order_id"),
                    "order.status": state.get("status"),
                }
            ) as span:
                try:
                    result = func(state)
                    span.set_attribute("result.status", result.get("status", "unknown"))
                    return result
                except Exception as e:
                    span.set_status(trace.StatusCode.ERROR, str(e))
                    span.record_exception(e)
                    raise
        return wrapper
    return decorator

@traced_node("validate_order")
def validate_order(state: OrderState) -> OrderState:
    # ... validation logic ...
    pass

Metrics Collection

from prometheus_client import Counter, Histogram

order_counter = Counter("orders_total", "Total orders processed", ["status"])
processing_time = Histogram("order_processing_seconds", "Order processing time")

def metrics_node(state: OrderState) -> OrderState:
    """Collect metrics for monitoring."""
    order_counter.labels(status=state["status"]).inc()
    return state

Testing Production Workflows

import pytest

@pytest.fixture
def production_app():
    checkpointer = MemorySaver()  # Use memory for tests
    graph = build_order_graph(checkpointer)
    return graph.compile()

def test_happy_path(production_app):
    config = {"configurable": {"thread_id": "test-1"}}
    
    result = production_app.invoke({
        "order_id": "test-1",
        "items": [{"sku": "ABC", "quantity": 1}],
        "total": 29.99,
        "status": "validating",
        "errors": [],
        "retry_count": 0,
    }, config)
    
    assert result["status"] == "completed"

def test_crash_recovery(production_app):
    config = {"configurable": {"thread_id": "test-2"}}
    
    # Start processing
    production_app.invoke({"order_id": "test-2", ...}, config)
    
    # Simulate crash and resume
    result = production_app.invoke(None, config)
    
    # Should resume from checkpoint
    assert result["status"] in ["processing_payment", "completed"]

Deployment

Docker

FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt

# Health check
HEALTHCHECK CMD curl -f http://localhost:8080/health || exit 1

EXPOSE 8080
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]

Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: order-agent
  template:
    spec:
      containers:
      - name: order-agent
        image: order-agent:latest
        resources:
          limits:
            memory: "512Mi"
            cpu: "500m"
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: database-url

Conclusion

Production LangGraph requires more than just compiling a graph. You need typed state for predictability, checkpointing for durability, subgraphs for composability, error handling for resilience, and observability for debugging. Start with typed states and checkpointing, add subgraphs as complexity grows, and instrument everything from day one.