Deploy LangGraph to Production: A Step-by-Step Tutorial (2026)

1. How long can a single graph run?
   → Drives timeout strategy and checkpoint frequency.

2. Does a run need to survive a worker crash?
   → If yes, you need Postgres (or another durable backend), not MemorySaver.

3. How many concurrent runs at peak?
   → Drives worker count, DB connection pool, and rate-limit budget.

4. What does "done" look like?
   → Fixed number of nodes? LLM decides? Error budget?

5. Who pays when the agent loops?
   → Token cost per run, hard ceiling, and who gets paged when it blows past.

from typing import Annotated, TypedDict, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

class AgentState(TypedDict):
    """Durable state for our production agent.

    Every field here is serialized to Postgres after every node.
    Keep it SMALL — no binary blobs, no giant payloads.
    """
    # Conversation history — add_messages is a reducer that appends
    messages: Annotated[list[AnyMessage], add_messages]

    # User context (stable per run)
    user_id: str
    tenant_id: str

    # Control flow
    step_count: int
    status: Literal["running", "awaiting_tool", "done", "error"]

    # Tool results — bounded by policy (keep last 5 only)
    recent_tool_calls: list[dict]

    # Token budget enforcement
    tokens_used: int
    tokens_budget: int

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
import os

from agent.state import AgentState
from agent.nodes import call_model, execute_tools, should_continue

# Build the graph once at import time — it's stateless
def build_graph():
    workflow = StateGraph(AgentState)
    workflow.add_node("model", call_model)
    workflow.add_node("tools", execute_tools)
    workflow.add_edge(START, "model")
    workflow.add_conditional_edges(
        "model",
        should_continue,
        {"tools": "tools", "done": END},
    )
    workflow.add_edge("tools", "model")
    return workflow

# Connection pool — tune min/max to your worker count
pool = ConnectionPool(
    conninfo=os.environ["DATABASE_URL"],
    min_size=5,
    max_size=20,
    kwargs={"autocommit": True, "prepare_threshold": 0},
)

# Bind the checkpointer — survives restarts, shared across workers
checkpointer = PostgresSaver(pool)
checkpointer.setup()  # Idempotent — safe to call on every boot

graph = build_graph().compile(checkpointer=checkpointer)

from agent.graph import graph

def run_turn(thread_id: str, user_message: str, user_id: str):
    config = {
        "configurable": {"thread_id": thread_id},
        "recursion_limit": 25,   # HARD CAP on node visits
    }
    result = graph.invoke(
        {
            "messages": [{"role": "user", "content": user_message}],
            "user_id": user_id,
            "step_count": 0,
            "tokens_used": 0,
            "tokens_budget": 100_000,
            "status": "running",
            "recent_tool_calls": [],
        },
        config=config,
    )
    return result

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent.run import run_turn
import uuid

app = FastAPI(title="langgraph-agent", version="1.0")

class TurnRequest(BaseModel):
    thread_id: str | None = None
    user_id: str
    message: str

@app.post("/v1/chat")
def chat(req: TurnRequest):
    thread_id = req.thread_id or str(uuid.uuid4())
    try:
        result = run_turn(thread_id, req.message, req.user_id)
    except Exception as e:
        # Log with thread_id — that's your replay handle
        raise HTTPException(status_code=500, detail={
            "thread_id": thread_id,
            "error": str(e),
        })
    return {"thread_id": thread_id, "result": result}

@app.get("/healthz")
def health():
    return {"status": "ok"}

FROM python:3.12-slim AS base

# Non-root user — never run the graph as root
RUN useradd -m -u 10001 agent
WORKDIR /app

# Install deps first (Docker layer cache)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy code
COPY agent/ ./agent/
COPY api/ ./api/

USER agent
EXPOSE 8080

# gunicorn for multi-worker; tune worker count to CPU + DB pool size
CMD ["gunicorn", "api.main:app", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--workers", "4", \
     "--bind", "0.0.0.0:8080", \
     "--timeout", "120", \
     "--graceful-timeout", "30"]

                    ┌────────────────┐
   Client ────►      │  Load Balancer │
                     └────────┬───────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
      ┌────────────┐  ┌────────────┐  ┌────────────┐
      │  Worker 1  │  │  Worker 2  │  │  Worker N  │
      │ (gunicorn) │  │ (gunicorn) │  │ (gunicorn) │
      └──────┬─────┘  └──────┬─────┘  └──────┬─────┘
             │               │               │
             └───────┬───────┴───────┬───────┘
                     ▼               ▼
             ┌───────────────┐ ┌───────────────┐
             │   Postgres    │ │     Redis     │
             │ (checkpoints) │ │ (rate limits) │
             └───────────────┘ └───────────────┘

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: langgraph-agent
  namespace: ai-agents
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: langgraph-agent
  minReplicas: 3
  maxReplicas: 30
  metrics:
    # Scale on custom metric: in-flight graph runs per pod
    - type: Pods
      pods:
        metric:
          name: langgraph_runs_in_flight
        target:
          type: AverageValue
          averageValue: "10"
  behavior:
    scaleDown:
      # Don't scale down aggressively — runs can take minutes
      stabilizationWindowSeconds: 300
      policies:
        - type: Percent
          value: 20
          periodSeconds: 60

import redis
from contextlib import contextmanager
import time

r = redis.Redis.from_url(os.environ["REDIS_URL"])

@contextmanager
def thread_lock(thread_id: str, ttl_sec: int = 300):
    """Serialize graph runs per thread to prevent checkpoint races."""
    key = f"graph:lock:{thread_id}"
    token = f"{time.time()}-{os.getpid()}"
    acquired = r.set(key, token, nx=True, ex=ttl_sec)
    if not acquired:
        raise RuntimeError(f"thread {thread_id} is already running")
    try:
        yield
    finally:
        # Only release if we still own the lock
        if r.get(key) == token.encode():
            r.delete(key)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from functools import wraps

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint=os.environ["OTEL_ENDPOINT"]))
)
tracer = trace.get_tracer("langgraph-agent")

def traced_node(node_name: str):
    """Wrap a LangGraph node so every execution emits a span."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(state):
            with tracer.start_as_current_span(f"node.{node_name}") as span:
                span.set_attribute("thread_id", state.get("thread_id", ""))
                span.set_attribute("step_count", state.get("step_count", 0))
                span.set_attribute("tokens_used", state.get("tokens_used", 0))
                try:
                    result = fn(state)
                    span.set_attribute("status", "ok")
                    return result
                except Exception as e:
                    span.set_attribute("status", "error")
                    span.record_exception(e)
                    raise
        return wrapper
    return decorator

# Apply to each node
@traced_node("call_model")
def call_model(state):
    # ... your model call
    ...

{
  "graphs": {
    "agent": "./agent/graph.py:graph"
  },
  "env": ".env",
  "python_version": "3.12",
  "dependencies": ["."],
  "rapidclaw": {
    "checkpointer": "postgres",
    "min_replicas": 3,
    "max_replicas": 30,
    "scaling_metric": "runs_in_flight",
    "target_value": 10,
    "observability": {
      "otel": true,
      "log_level": "info"
    },
    "security": {
      "egress": "allowlist",
      "allowed_hosts": [
        "api.anthropic.com",
        "api.openai.com"
      ]
    }
  }
}

# install the CLI
npm i -g @rapidclaw/cli

# authenticate
rapidclaw login

# deploy the graph
rapidclaw deploy --manifest langgraph.json

# check status
rapidclaw status
# → agent  v12  running  3 replicas  p95=1.4s  $0.00142/turn