Deploying LangGraph Agents: LangServe, Docker, LangGraph Platform, and Production Observability

Scenario context

Swap InMemorySaver → PostgresSaver, add LangServe + Docker, trace with LangSmith.

<!--

Your LangGraph agent works perfectly in the demo. You deploy it to a single FastAPI instance. Two users hit it at the same time. They see each other's conversation history. The InMemorySaver you used in development doesn't know about concurrency — and it never will, because it was never designed for it.

This is the most co…

This walkthrough focuses on The Demo-to-Production Gap: Why Notebook Agents Fail at Scale.

Before making a design choice, clarify:

• Your LangGraph agent works perfectly in the demo.
• You deploy it to a single FastAPI instance.
• Two users hit it at the same time.

Concepts in play: llm-agents-and-langgraph, graph-traversal-algorithms, api-security-oauth2, api-gateway

Stage 1 — LangServe: Wrapping a LangGraph Agent as a FastAPI App

• LangServe wraps any LangChain Runnable — including a compiled LangGraph — as a FastAPI application.
• Its core API is a single function: add routes .
• add routes auto-generates six endpoints from the graph's input/output schema: RemoteRunnable client — callers can use a typed Python client instead of raw HTTP: Limitation: LangServe is in maintenance mode as of 2025.
• directs new projects toward LangGraph Platform for managed deployments.

# app.py
import os
from fastapi import FastAPI, Request, HTTPException
from langserve import add_routes
from agent import build_agent  # your compiled LangGraph

app = FastAPI(title="LangGraph Agent API", version="1.0")

# --- API key middleware ---
@app.middleware("http")
async def verify_api_key(request: Request, call_next):
    if request.url.path.startswith("/agent"):
        key = request.headers.get("X-API-Key")
        if key != os.environ["AGENT_API_KEY"]:
            raise HTTPException(status_code=401, detail="Invalid API key")
    return await call_next(request)

# --- Mount the graph ---
agent = build_agent()

add_routes(
    app,
    agent,
    path="/agent",
    enable_feedback_endpoint=True,
    enable_public_trace_link_endpoint=True,
)

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

from langserve import RemoteRunnable

agent = RemoteRunnable("http://localhost:8000/agent")

# Single turn
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Summarize Q4 results"}]},
    config={"configurable": {"thread_id": "user-42-session-7"}},
)

# Streaming
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Now compare to Q3"}]},
    config={"configurable": {"thread_id": "user-42-session-7"}},
):
    print(chunk)

Observable outcome: directs new projects toward LangGraph Platform for managed deployments.

Stage 2 — Deep Dive: Concurrency, Persistence, and Production-Grade Checkpointing

• The Internals When a POST /agent/invoke request arrives at a LangServe worker, here is the exact execution path: 1.
• Deserialize — LangServe uses the graph's InputType schema to parse the JSON body.
• Config extraction — the config.configurable.thread id is extracted and passed to the checkpointer.
• Checkpoint read — AsyncPostgresSaver opens a connection from its asyncpg pool and runs a SELECT for the latest checkpoint row matching thread id .

Worker 1 memory: { "user-42": [msg1, msg2] }
Worker 2 memory: { }           ← empty, knows nothing about msg1 or msg2

User's turn 3 routes to Worker 2 → fresh conversation, no history

# Development (single process only)
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

# Production (multi-worker safe)
import asyncpg
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async def build_production_graph():
    pool = await asyncpg.create_pool(dsn=os.environ["DATABASE_URL"], min_size=2, max_size=10)
    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.setup()   # idempotent: creates tables if absent
    return builder.compile(checkpointer=checkpointer)

Observable outcome: Checkpoint read — AsyncPostgresSaver opens a connection from its asyncpg pool and runs a SELECT for the latest checkpoint row matching thread id .

Stage 3 — Beyond Request-Response: Auth Patterns, Background Runs, and LangGraph Platform

• Authentication and Per-User Thread Isolation LangServe provides no authentication out of the box — everything added in the middleware snippet earlier is your responsibility.
• The three common production patterns are: Pattern 1 — Static API key (internal tools and service-to-service) Simple, effective for a single service identity.
• Does not support per-user tracking or revocation.
• Pattern 2 — JWT with per-user thread isolation The thread id construction here is the security boundary.

# In FastAPI middleware — shown earlier in ⚙️ section
key = request.headers.get("X-API-Key")
if key != os.environ["AGENT_API_KEY"]:
    raise HTTPException(status_code=401)

import jwt
from fastapi import Depends, Request

SECRET = os.environ["JWT_SECRET"]

async def get_current_user(request: Request) -> dict:
    token = request.headers.get("Authorization", "").removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return payload  # contains {"sub": "user-42", "tenant": "acme"}

@app.post("/agent/invoke")
async def invoke(body: dict, user: dict = Depends(get_current_user)):
    thread_id = f"{user['tenant']}-{user['sub']}-{body['session_id']}"
    return await graph.ainvoke(
        body["input"],
        config={"configurable": {"thread_id": thread_id}},
    )

Observable outcome: Pattern 2 — JWT with per-user thread isolation The thread id construction here is the security boundary.

Stage 4 — Production Architecture: From Load Balancer to LangSmith

• The following diagram shows the full request path for a multi-worker LangServe deployment with PostgreSQL persistence and LangSmith observability.
• The dashed arrows are async/non-blocking paths that do not add to user-facing latency.
• Enabling Tracing What is captured per run: Reading a Trace to Debug a Failed Run 1.
• Open the LangSmith dashboard → select your project.

# .env (or container environment)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__...            # from app.langsmith.com
LANGCHAIN_PROJECT=react-agent-prod  # project name in LangSmith dashboard

Example trace tree for a failed ReAct run:

  ✅ graph.invoke (1.8s)
    ✅ node: classify_intent (0.3s)
    ✅ node: search_tool (0.6s)
    ❌ node: format_response (0.2s)
         ↳ KeyError: 'product_name' not found in tool output
         ↳ Input state: {"search_results": [], "messages": [...]}

Observable outcome: Open the LangSmith dashboard → select your project.

End-to-end outcome

After working through the stages, you should be able to explain:

• InMemorySaver is a deployment hazard, not just a development convenience.
• It works in every test, fails silently in every multi-worker deploy.

Alternatives considered

Deployment Options Compared: LangServe vs LangGraph Platform vs Self-Hosted

Before writing a single line of deployment code, choose the right layer.

Failure and degraded modes

The source does not isolate a separate failure mode. Limits that matter for the scenario are called out in the stage decisions above.