Building Production AI Applications with LangGraph: State, Graphs & Deployment

Master LangGraph for building production AI applications — stateful graphs, persistence, streaming, human-in-the-loop, and deployment patterns.

Introduction

LangGraph transforms LangChain chains into stateful, cyclic graphs — enabling agents, multi-step workflows, and complex state management. This guide covers production patterns.

Why LangGraph?

LangChain (Chains)	LangGraph (Graphs)
Linear, acyclic	Cyclic, branching
Stateless	Full state management
Single-threaded	Concurrent execution
Hard to debug	Visualizable, inspectable
Limited human-in-the-loop	Built-in interruption

Part 1: Core Concepts

State & Graph Definition

# langgraph/core.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    user_id: str
    session_id: str
    context: dict
    current_step: str
    retry_count: int
    metadata: dict

def create_basic_graph():
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("process_input", process_input_node)
    workflow.add_node("reason", reason_node)
    workflow.add_node("act", act_node)
    workflow.add_node("respond", respond_node)

    # Define edges
    workflow.add_edge(START, "process_input")
    workflow.add_edge("process_input", "reason")
    workflow.add_conditional_edges("reason", route_after_reason, {
        "act": "act",
        "respond": "respond"
    })
    workflow.add_edge("act", "reason")
    workflow.add_edge("respond", END)

    return workflow.compile()

async def process_input_node(state: AgentState):
    # Preprocess, validate, enrich
    return {"current_step": "input_processed"}

async def reason_node(state: AgentState):
    # LLM reasoning
    return {"current_step": "reasoned"}

def route_after_reason(state: AgentState) -> str:
    # Decide next action
    return "act" if needs_action(state) else "respond"

async def act_node(state: AgentState):
    # Tool execution
    return {"current_step": "acted"}

async def respond_node(state: AgentState):
    # Generate final response
    return {"current_step": "responded"}

Checkpointers (Persistence)

# langgraph/persistence.py
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.redis import RedisSaver
import sqlite3

# SQLite (development)
sqlite_conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
sqlite_checkpointer = SqliteSaver(sqlite_conn)

# PostgreSQL (production)
postgres_checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langgraph"
)

# Redis (high-throughput)
redis_checkpointer = RedisSaver.from_conn_string(
    "redis://localhost:6379"
)

# Usage with checkpointer
app = workflow.compile(checkpointer=sqlite_checkpointer)

# Run with thread_id for persistence
config = {"configurable": {"thread_id": "user-123-session-456"}}
result = await app.ainvoke({"messages": [HumanMessage(content="Hello")]}, config)

# Resume later with same thread_id
result2 = await app.ainvoke({"messages": [HumanMessage(content="Continue")]}, config)

Part 2: Advanced Graph Patterns

Conditional Routing

# langgraph/routing.py
from typing import Literal

def route_by_intent(state: AgentState) -> Literal["research", "code", "write", "error"]:
    last_msg = state["messages"][-1].content.lower()

    if any(kw in last_msg for kw in ["research", "find", "search", "lookup"]):
        return "research"
    elif any(kw in last_msg for kw in ["code", "implement", "build", "function"]):
        return "code"
    elif any(kw in last_msg for kw in ["write", "document", "explain", "summarize"]):
        return "write"
    return "error"

def route_by_confidence(state: AgentState) -> Literal["proceed", "clarify", "escalate"]:
    confidence = state.get("confidence", 1.0)

    if confidence > 0.8:
        return "proceed"
    elif confidence > 0.5:
        return "clarify"
    return "escalate"

# Multi-condition routing
def complex_router(state: AgentState) -> dict:
    """Return multiple routing decisions."""
    return {
        "next_node": "process",
        "parallel_nodes": ["validate", "enrich"],
        "skip_nodes": [] if state["retry_count"] < 3 else ["optional_step"]
    }

Parallel Execution

# langgraph/parallel.py
from langgraph.graph import StateGraph
from typing: Annotated, List
import operator

class ParallelState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    research_results: Annotated[List[dict], operator.add]  # Accumulate
    code_results: Annotated[List[dict], operator.add]
    write_results: Annotated[List[dict], operator.add]

async def research_node(state: ParallelState):
    # Run multiple searches in parallel
    import asyncio

    async def search(query):
        return {"query": query, "results": await web_search(query)}

    queries = extract_queries(state["messages"][-1].content)
    results = await asyncio.gather(*[search(q) for q in queries])

    return {"research_results": results}

async def code_node(state: ParallelState):
    # Parallel code generation for different components
    tasks = extract_code_tasks(state["messages"][-1].content)
    results = await asyncio.gather(*[generate_code(t) for t in tasks])
    return {"code_results": results}

# Fan-out/fan-in pattern
workflow = StateGraph(ParallelState)
workflow.add_node("research", research_node)
workflow.add_node("code", code_node)
workflow.add_node("synthesize", synthesize_node)

workflow.add_edge(START, "research")
workflow.add_edge(START, "code")  # Both start from START
workflow.add_edge("research", "synthesize")
workflow.add_edge("code", "synthesize")
workflow.add_edge("synthesize", END)

Human-in-the-Loop

# langgraph/human_in_loop.py
from langgraph.types import interrupt, Command
from dataclasses import dataclass

@dataclass
class HumanReview:
    approved: bool
    feedback: str
    modifications: dict

async def human_review_node(state: AgentState):
    # Present to human for review
    review_request = {
        "task": state["messages"][-1].content,
        "proposed_action": state.get("proposed_action"),
        "context": state.get("context")
    }

    # This pauses execution and waits for human input
    human_response = interrupt(review_request)

    # Resume with human decision
    review = HumanReview(**human_response)

    if review.approved:
        return {"human_approved": True, "modifications": review.modifications}
    else:
        return {
            "human_approved": False,
            "feedback": review.feedback,
            "retry_count": state["retry_count"] + 1
        }

def route_after_review(state: AgentState) -> str:
    if state.get("human_approved"):
        return "execute"
    elif state["retry_count"] >= 3:
        return "escalate"
    return "revise"

# Usage
app = workflow.compile(checkpointer=checkpointer, interrupt_before=["human_review"])

# First invocation - pauses at human_review
result = await app.ainvoke(input_data, config)

# Check if interrupted
if "__interrupt__" in result:
    interrupt_info = result["__interrupt__"]
    # Present to human, get response
    human_decision = await get_human_decision(interrupt_info)

    # Resume with human input
    result = await app.ainvoke(Command(resume=human_decision), config)

Subgraphs (Modularity)

# langgraph/subgraphs.py
from langgraph.graph import StateGraph

# Subgraph for research
research_graph = StateGraph(AgentState)
research_graph.add_node("search", search_node)
research_graph.add_node("summarize", summarize_node)
research_graph.add_edge(START, "search")
research_graph.add_edge("search", "summarize")
research_graph.add_edge("summarize", END)

research_app = research_graph.compile()

# Subgraph for code generation
code_graph = StateGraph(AgentState)
code_graph.add_node("plan", plan_code_node)
code_graph.add_node("implement", implement_node)
code_graph.add_node("test", test_node)
code_graph.add_edge(START, "plan")
code_graph.add_edge("plan", "implement")
code_graph.add_edge("implement", "test")
code_graph.add_edge("test", END)

code_app = code_graph.compile()

# Main graph using subgraphs
main_graph = StateGraph(AgentState)
main_graph.add_node("research", research_app)
main_graph.add_node("code", code_app)
main_graph.add_node("integrate", integrate_node)

main_graph.add_edge(START, "research")
main_graph.add_edge("research", "code")
main_graph.add_edge("code", "integrate")
main_graph.add_edge("integrate", END)

# Compile with shared checkpointer
app = main_graph.compile(checkpointer=checkpointer)

Part 3: Streaming & Real-Time

Token Streaming

# langgraph/streaming.py
from langgraph.graph import StateGraph
from typing: AsyncGenerator

async def stream_graph(app, input_data: dict, config: dict) -> AsyncGenerator[dict, None]:
    """Stream graph execution events."""

    async for event in app.astream_events(input_data, config, version="v2"):
        event_type = event["event"]

        if event_type == "on_chain_start":
            yield {"type": "node_start", "node": event["name"]}
        elif event_type == "on_chain_end":
            yield {"type": "node_end", "node": event["name"], "output": event["data"]["output"]}
        elif event_type == "on_chat_model_stream":
            yield {"type": "token", "content": event["data"]["chunk"].content}
        elif event_type == "on_tool_start":
            yield {"type": "tool_start", "tool": event["name"], "input": event["data"]["input"]}
        elif event_type == "on_tool_end":
            yield {"type": "tool_end", "tool": event["name"], "output": event["data"]["output"]}
        elif event_type == "on_error":
            yield {"type": "error", "error": str(event["data"]["error"])}

# Usage in FastAPI
@app.post("/stream")
async def stream_endpoint(request: QueryRequest):
    async def event_generator():
        async for event in stream_graph(app, request.dict(), config):
            yield f"data: {json.dumps(event)}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

Progress Tracking

# langgraph/progress.py
from dataclasses import dataclass
from typing: Dict
import asyncio

@dataclass
class ProgressState:
    current_step: str
    total_steps: int
    completed_steps: int
    step_details: Dict[str, any]

class ProgressTracker:
    def __init__(self):
        self.subscribers: Dict[str, asyncio.Queue] = {}

    async def track(self, thread_id: str, state: AgentState):
        if thread_id not in self.subscribers:
            self.subscribers[thread_id] = asyncio.Queue()

        progress = ProgressState(
            current_step=state.get("current_step", "unknown"),
            total_steps=state.get("total_steps", 10),
            completed_steps=state.get("completed_steps", 0),
            step_details=state.get("step_details", {})
        )

        await self.subscribers[thread_id].put(progress)

    async def subscribe(self, thread_id: str) -> AsyncGenerator[ProgressState, None]:
        queue = asyncio.Queue()
        self.subscribers[thread_id] = queue

        try:
            while True:
                progress = await queue.get()
                yield progress
        finally:
            del self.subscribers[thread_id]

# Inject into graph nodes
async def tracked_node(state: AgentState, tracker: ProgressTracker):
    result = await actual_node_logic(state)
    await tracker.track(state["configurable"]["thread_id"], result)
    return result

Part 4: Error Handling & Resilience

Retry with Exponential Backoff

# langgraph/resilience.py
from langgraph.graph import StateGraph
from typing: Callable
import asyncio
import random

def with_retry(max_retries: int = 3, base_delay: float = 1.0):
    def decorator(func: Callable):
        async def wrapper(state: AgentState):
            for attempt in range(max_retries + 1):
                try:
                    return await func(state)
                except Exception as e:
                    if attempt == max_retries:
                        raise

                    delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1)
                    await asyncio.sleep(delay)

                    # Update retry count
                    return {
                        **state,
                        "retry_count": state.get("retry_count", 0) + 1,
                        "last_error": str(e)
                    }
        return wrapper
    return decorator

@with_retry(max_retries=3)
async def unreliable_tool_node(state: AgentState):
    # Tool that might fail
    return await call_external_api(state["input"])

# Circuit breaker pattern
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: float = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"

    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.timeout:
                self.state = "half-open"
            else:
                raise Exception("Circuit breaker open")

        try:
            result = await func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "open"
            raise

Fallback Strategies

# langgraph/fallbacks.py
from typing: Optional

async def node_with_fallback(state: AgentState) -> dict:
    """Try primary, fallback to secondary on failure."""

    try:
        return await primary_llm_call(state)
    except Exception as e:
        # Log failure
        logger.warning(f"Primary failed: {e}, trying fallback")

        try:
            return await fallback_llm_call(state)
        except Exception as e2:
            # Final fallback: structured error response
            return {
                "error": "All models failed",
                "fallback_response": generate_cached_response(state["input"]),
                "details": {"primary": str(e), "fallback": str(e2)}
            }

# Model cascade
async def cascade_models(state: AgentState) -> dict:
    models = ["gpt-4o", "gpt-4o-mini", "claude-3-sonnet", "local-llama"]

    for model in models:
        try:
            result = await call_model(model, state["input"])
            return {"model_used": model, "result": result}
        except Exception as e:
            continue

    raise Exception("All models in cascade failed")

Part 5: Production Deployment

FastAPI Server

# langgraph/server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing: Optional, Dict, Any
import uuid

app = FastAPI(title="LangGraph Agent API")

class InvokeRequest(BaseModel):
    input: dict
    thread_id: Optional[str] = None
    config: Optional[Dict[str, Any]] = None

class StreamRequest(BaseModel):
    input: dict
    thread_id: Optional[str] = None

@app.post("/invoke")
async def invoke(request: InvokeRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    if request.config:
        config.update(request.config)

    try:
        result = await app.state.graph.ainvoke(request.input, config)
        return {"thread_id": thread_id, "result": result}
    except Exception as e:
        raise HTTPException(500, str(e))

@app.post("/stream")
async def stream(request: StreamRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    async def generate():
        async for event in app.state.graph.astream_events(
            request.input, config, version="v2"
        ):
            yield f"data: {json.dumps(event)}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

@app.get("/state/{thread_id}")
async def get_state(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = await app.state.graph.aget_state(config)
    return state.values if state else {}

@app.post("/resume/{thread_id}")
async def resume(thread_id: str, human_input: dict):
    config = {"configurable": {"thread_id": thread_id}}
    result = await app.state.graph.ainvoke(Command(resume=human_input), config)
    return {"result": result}

@app.get("/health")
async def health():
    return {"status": "healthy", "graph": app.state.graph.name}

# Initialize on startup
@app.on_event("startup")
async def startup():
    app.state.graph = create_production_graph()

Docker Deployment

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# System deps
RUN apt-get update && apt-get install -y \
    gcc postgresql-client \
    && rm -rf /var/lib/apt/lists/*

# Python deps
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# App
COPY . .

# Non-root
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Kubernetes Deployment

# k8s/langgraph-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langgraph-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langgraph-agent
  template:
    metadata:
      labels:
        app: langgraph-agent
    spec:
      containers:
      - name: agent
        image: your-registry/langgraph-agent:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: langgraph-secrets
              key: database-url
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: langgraph-secrets
              key: redis-url
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: langgraph-secrets
              key: openai-key
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: langgraph-agent
spec:
  selector:
    app: langgraph-agent
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: langgraph-agent-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: langgraph-agent
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

Part 6: Testing & Evaluation

Graph Testing

# langgraph/testing.py
import pytest
from langgraph.graph import StateGraph

class GraphTestHarness:
    def __init__(self, graph_fn):
        self.graph_fn = graph_fn

    def create_test_app(self, checkpointer=None):
        return self.graph_fn().compile(checkpointer=checkpointer)

    async def run_test(self, input_data: dict, expected: dict = None):
        app = self.create_test_app()
        config = {"configurable": {"thread_id": "test-" + str(uuid.uuid4())}}

        result = await app.ainvoke(input_data, config)

        if expected:
            assert self._match_result(result, expected), f"Expected {expected}, got {result}"

        return result

    def _match_result(self, actual: dict, expected: dict) -> bool:
        for key, value in expected.items():
            if key not in actual:
                return False
            if isinstance(value, dict):
                if not self._match_result(actual[key], value):
                    return False
            elif actual[key] != value:
                return False
        return True

# Test fixtures
@pytest.fixture
def test_graph():
    return create_test_graph()

@pytest.mark.asyncio
async def test_basic_flow(test_graph):
    harness = GraphTestHarness(lambda: test_graph)
    result = await harness.run_test(
        {"messages": [HumanMessage(content="Hello")]},
        {"current_step": "responded"}
    )
    assert "messages" in result

Property-Based Testing

# langgraph/property_testing.py
from hypothesis import given, strategies as st
import pytest

@given(st.text(min_size=1, max_size=1000))
async def test_input_sanitization(input_text):
    """Graph should handle any input without crashing."""
    app = create_test_graph().compile()
    config = {"configurable": {"thread_id": "property-test"}}

    result = await app.ainvoke({"messages": [HumanMessage(content=input_text)]}, config)

    # Should always produce valid state
    assert "messages" in result
    assert isinstance(result["messages"], list)

@given(st.integers(min_value=0, max_value=10))
async def test_retry_behavior(retry_count):
    """Graph should respect retry limits."""
    state = {"retry_count": retry_count, "messages": []}
    result = await retry_limited_node(state)

    if retry_count >= 3:
        assert result.get("escalated") == True
    else:
        assert result.get("retry_count") == retry_count + 1

Part 7: Observability

LangSmith Integration

# langgraph/observability.py
from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree

@traceable(name="agent_reasoning")
async def reason_node(state: AgentState):
    run_tree = get_current_run_tree()
    if run_tree:
        run_tree.add_metadata({"user_id": state.get("user_id")})

    # Your logic here
    return result

# Custom callbacks
from langgraph.callbacks import BaseCallbackHandler

class ObservabilityCallback(BaseCallbackHandler):
    def on_node_start(self, node_name: str, state: dict):
        # Send to monitoring
        pass

    def on_node_end(self, node_name: str, output: dict):
        # Record metrics
        pass

    def on_error(self, node_name: str, error: Exception):
        # Alert
        pass

app = workflow.compile(
    checkpointer=checkpointer,
    callbacks=[ObservabilityCallback()]
)

Metrics Collection

# langgraph/metrics.py
from dataclasses import dataclass
from datetime import datetime
import time

@dataclass
class NodeMetrics:
    node_name: str
    start_time: float
    end_time: float
    duration_ms: float
    input_tokens: int
    output_tokens: int
    success: bool
    error: str = ""

class MetricsCollector:
    def __init__(self):
        self.metrics: List[NodeMetrics] = []

    def record(self, metric: NodeMetrics):
        self.metrics.append(metric)

    def get_node_stats(self, node_name: str) -> dict:
        node_metrics = [m for m in self.metrics if m.node_name == node_name]
        if not node_metrics:
            return {}

        durations = [m.duration_ms for m in node_metrics]
        return {
            "count": len(node_metrics),
            "avg_duration_ms": sum(durations) / len(durations),
            "p95_duration_ms": sorted(durations)[int(len(durations) * 0.95)],
            "success_rate": sum(1 for m in node_metrics if m.success) / len(node_metrics),
            "total_tokens": sum(m.input_tokens + m.output_tokens for m in node_metrics)
        }

    def get_graph_summary(self) -> dict:
        return {
            "total_nodes": len(set(m.node_name for m in self.metrics)),
            "total_executions": len(self.metrics),
            "overall_success_rate": sum(1 for m in self.metrics if m.success) / len(self.metrics),
            "by_node": {n: self.get_node_stats(n) for n in set(m.node_name for m in self.metrics)}
        }

Decision Framework

LANGGRAPH PATTERN SELECTION:

Simple linear workflow?
├─→ Yes → Basic StateGraph with linear edges
└─→ No
    ├─→ Need cycles/loops?
    │   ├─→ Yes → Conditional edges with retry logic
    │   └─→ No
    │       ├─→ Parallel independent tasks?
    │       │   └─→ Yes → Fan-out/fan-in with Annotated accumulation
    │       │
    │       ├─→ Human review needed?
    │       │   └─→ Yes → interrupt() + Command(resume=...)
    │       │
    │       ├─→ Complex sub-workflows?
    │       │   └─→ Yes → Subgraphs (compile separately, compose)
    │       │
    │       └─→ Production deployment?
    │           └─→ Yes → Checkpointer + Streaming + Metrics + Health checks

Production Checklist

Graph Design

• [ ] State schema defined with TypedDict
• [ ] All nodes are async for concurrency
• [ ] Conditional edges cover all paths
• [ ] No unreachable nodes
• [ ] Retry limits on all external calls
• [ ] Circuit breakers for external APIs

Persistence

• [ ] Checkpointer configured (Postgres/Redis for prod)
• [ ] Thread ID strategy defined (user/session)
• [ ] State migration plan for schema changes
• [ ] Checkpoint TTL/cleanup policy

Reliability

• [ ] Timeout on all node executions
• [ ] Fallback models for LLM calls
• [ ] Graceful degradation on tool failures
• [ ] Idempotent operations where possible
• [ ] Dead letter queue for failed executions

Observability

• [ ] LangSmith or custom tracing
• [ ] Node-level metrics (latency, tokens, errors)
• [ ] Business metrics (task completion, user satisfaction)
• [ ] Alerts on error rates, latency spikes
• [ ] Distributed tracing across services

Deployment

• [ ] Health endpoint returns graph status
• [ ] Graceful shutdown (finish in-flight)
• [ ] Horizontal scaling with sticky sessions (thread_id)
• [ ] Blue-green deployment support
• [ ] Rollback procedure tested

Testing

• [ ] Unit tests for each node
• [ ] Integration tests for full flows
• [ ] Property-based tests for edge cases
• [ ] Chaos testing (node failures, timeouts)
• [ ] Load testing with realistic traffic

Resources

• LangGraph Documentation
• LangGraph Concepts
• LangGraph API Reference
• LangGraph Examples
• Checkpointer Guide
• Streaming Guide
• Human-in-the-Loop

---

LangGraph makes complex AI workflows manageable. Start simple, add persistence early, instrument everything, and test failure paths as thoroughly as success paths.