How to deploy a LangGraph app to production

Shipping a prototype LangGraph agent is easy; the hard part is how to deploy LangGraph to production without drowning in state bugs, orphaned tasks, and silent LLM failures. This guide walks through a concrete architecture: a FastAPI service hosting the graph, a Redis-backed checkpointer, and horizontally scalable workers, all containerized and observable.

Step 1: Pin your graph and add a real checkpointer

A LangGraph app in dev often uses an in-memory checkpointer. In production, you need durability. Use RedisSaver so graph state survives process restarts and can be shared across replicas.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.redis import RedisSaver
from typing import TypedDict

class AgentState(TypedDict):
    input: str
    output: str

def node_a(state: AgentState) -> AgentState:
    return {"output": state["input"].upper()}

def build_graph(redis_url: str):
    saver = RedisSaver.from_conn_string(redis_url)
    saver.setup()  # create index/stream keys
    g = StateGraph(AgentState)
    g.add_node("a", node_a)
    g.add_edge(START, "a")
    g.add_edge("a", END)
    return g.compile(checkpointer=saver)

The setup() call is idempotent but required once per Redis instance. Without it, you’ll get missing-key errors under load.

Step 2: Build a stateless API server

HTTP handlers should never hold graph state. They pass a thread_id to the compiled graph and let the checkpointer handle persistence. FastAPI fits well.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import os

app = FastAPI()
graph = build_graph(os.environ["REDIS_URL"])

class InvokeReq(BaseModel):
    thread_id: str
    input: str

@app.post("/invoke")
async def invoke(req: InvokeReq):
    config = {"configurable": {"thread_id": req.thread_id}}
    try:
        result = await graph.ainvoke({"input": req.input}, config)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return result

This gives you a single endpoint to trigger or resume any conversation. The same thread_id can be hit from multiple pods; Redis locks prevent concurrent writes to the same thread.

Step 3: Containerize with a single artifact

Multi-stage builds keep the image small and reproducible. Pin LangGraph and its checkpointer dependencies explicitly.

FROM python:3.11-slim AS base
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt should contain at least:

langgraph==0.2.34
langchain-openai==0.1.25
fastapi==0.111.0
uvicorn==0.30.1
redis==5.0.4

Run locally with docker compose to verify the full stack:

services:
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
  api:
    build: .
    environment:
      REDIS_URL: redis://redis:6379/0
      OPENAI_API_KEY: ${OPENAI_API_KEY}
    ports: ["8000:8000"]
    depends_on: [redis]

Step 4: Decouple long runs from HTTP requests

If your agent calls multiple LLMs or tools, a synchronous HTTP request will time out. Push execution to a worker and poll state. Use a lightweight queue like ARQ or Celery; here’s the pattern with asyncio and a simple background task for clarity.

from fastapi import BackgroundTasks

@app.post("/invoke-async")
async def invoke_async(req: InvokeReq, bg: BackgroundTasks):
    bg.add_task(run_graph, req.thread_id, req.input)
    return {"status": "queued", "thread_id": req.thread_id}

async def run_graph(thread_id: str, user_input: str):
    config = {"configurable": {"thread_id": thread_id}}
    await graph.ainvoke({"input": user_input}, config)

For real scale, replace BackgroundTasks with a separate worker process consuming from Redis so the API pod can scale independently of compute-heavy nodes.

Step 5: Route LLM calls through a resilient gateway

Nodes that call ChatOpenAI should not hardcode a single provider. Use an OpenAI-compatible base URL and let the gateway handle model routing and fallback. Pointing ChatOpenAI at a gateway like n4n.ai gives you one OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is degraded, which simplifies model routing in your LangGraph nodes.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.environ["GATEWAY_KEY"],
    base_url=os.environ["GATEWAY_BASE_URL"],  # e.g. https://api.n4n.ai/v1
    default_headers={"X-Route": "auto"}  # honor client routing directives
)

Set GATEWAY_BASE_URL and GATEWAY_KEY in your deployment secrets. This removes the need to bake provider-specific retry logic into every node.

Step 6: Instrument every transition

You cannot debug a distributed agent without traces. LangGraph supports callbacks; forward them to structured logs or a tracing backend.

from langchain_core.callbacks import StdOutCallbackHandler

async def run_graph_traced(thread_id: str, user_input: str):
    config = {
        "configurable": {"thread_id": thread_id},
        "callbacks": [StdOutCallbackHandler()]
    }
    await graph.ainvoke({"input": user_input}, config)

In production, swap StdOutCallbackHandler for a handler that emits JSON lines with thread_id, node, and latency. Alert on checkpointer write failures—those are early signs of Redis saturation.

Step 7: Ship with health checks and rollback

Kubernetes users should define liveness and readiness probes that hit a /health route checking Redis connectivity.

@app.get("/health")
async def health():
    try:
        await graph.aget_state({"configurable": {"thread_id": "probe"}})
    except Exception:
        return {"status": "unhealthy"}, 500
    return {"status": "ok"}

Use immutable image tags and keep the previous revision available. LangGraph state schemas change; if you rename a state key, old threads in Redis will fail to deserialize. Write a migration that reads and rewrites affected threads before rolling forward.

Verify your deployment

After docker compose up, run a round-trip test:

curl -X POST localhost:8000/invoke \
  -H 'content-type: application/json' \
  -d '{"thread_id":"t1","input":"hello"}'
# expect {"input":"hello","output":"HELLO"}

curl -X POST localhost:8000/invoke \
  -H 'content-type: application/json' \
  -d '{"thread_id":"t1","input":"world"}'
# state persists; new run resumes same thread

Check Redis for the checkpoint key checkpoint:t1. If it exists and the API responds with low p99 latency under locust load, you have a working production deploy.

To validate fallback, temporarily set GATEWAY_BASE_URL to a bad host and confirm the gateway (or your retry wrapper) returns a clean 503 rather than hanging the graph. That behavior is what keeps a LangGraph agent alive when a single model provider is down.

Deploying LangGraph to production is less about the framework and more about the surrounding plumbing: durable state, separated concerns, and observable transitions. Get those three right and the agent scales.