Production Patterns & Best Practices

Building a LangGraph prototype is one thing. Running it reliably at scale with thousands of concurrent users is another. This lesson covers the patterns that separate toy projects from production systems.

Production Readiness Checklist

Error Handling Patterns

1. Retry with Exponential Backoff

import asyncio
from functools import wraps
from typing import TypeVar, Callable
import random

T = TypeVar('T')

def with_retry(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    retryable_exceptions: tuple = (Exception,)
):
    """Decorator for retry with exponential backoff and jitter"""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except retryable_exceptions as e:
                    last_exception = e

                    if attempt == max_retries:
                        raise

                    # Calculate delay with exponential backoff + jitter
                    delay = min(
                        base_delay * (exponential_base ** attempt),
                        max_delay
                    )
                    # Add jitter (0.5 to 1.5x delay)
                    delay = delay * (0.5 + random.random())

                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s")
                    await asyncio.sleep(delay)

            raise last_exception

        return wrapper
    return decorator

# Usage in a node
@with_retry(
    max_retries=3,
    retryable_exceptions=(TimeoutError, ConnectionError)
)
async def call_payment_api(order_id: str, amount: float) -> dict:
    """Call payment API with automatic retry"""
    # Simulated API call
    response = await payment_gateway.charge(order_id, amount)
    return response

2. Circuit Breaker Pattern

import time
from enum import Enum
from dataclasses import dataclass
from typing import Callable

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

@dataclass
class CircuitBreaker:
    """Circuit breaker for external service calls"""
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    half_open_max_calls: int = 3

    def __post_init__(self):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    def can_execute(self) -> bool:
        """Check if request should be allowed"""
        if self.state == CircuitState.CLOSED:
            return True

        if self.state == CircuitState.OPEN:
            # Check if recovery timeout has passed
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False

        if self.state == CircuitState.HALF_OPEN:
            return self.half_open_calls < self.half_open_max_calls

        return False

    def record_success(self):
        """Record successful call"""
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                # Recovered!
                self.state = CircuitState.CLOSED
                self.failure_count = 0
        else:
            self.failure_count = 0

    def record_failure(self):
        """Record failed call"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.state == CircuitState.HALF_OPEN:
            # Failed during recovery, go back to open
            self.state = CircuitState.OPEN
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage
payment_circuit = CircuitBreaker(failure_threshold=5, recovery_timeout=30)

def process_payment_with_circuit_breaker(state: dict) -> dict:
    if not payment_circuit.can_execute():
        return {
            "status": "failed",
            "error": "Payment service temporarily unavailable. Please try again."
        }

    try:
        result = call_payment_service(state)
        payment_circuit.record_success()
        return result
    except Exception as e:
        payment_circuit.record_failure()
        raise

3. Fallback Strategies

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# Primary and fallback models
primary_llm = ChatOpenAI(model="gpt-4o", timeout=30)
fallback_llm = ChatOpenAI(model="gpt-4o-mini", timeout=30)
emergency_llm = ChatAnthropic(model="claude-3-haiku-20240307", timeout=30)

async def call_llm_with_fallback(messages: list, state: dict) -> str:
    """Call LLM with automatic fallback to cheaper/faster models"""

    models = [
        (primary_llm, "gpt-4o"),
        (fallback_llm, "gpt-4o-mini"),
        (emergency_llm, "claude-3-haiku")
    ]

    last_error = None

    for llm, model_name in models:
        try:
            response = await llm.ainvoke(messages)
            # Log which model was used
            state["model_used"] = model_name
            return response.content
        except Exception as e:
            last_error = e
            print(f"Model {model_name} failed: {e}, trying fallback...")
            continue

    # All models failed
    raise Exception(f"All LLM models failed. Last error: {last_error}")

Observability

LangSmith Integration

import os
from langsmith import traceable
from langgraph.graph import StateGraph

# Enable LangSmith tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "ecommerce-agents"

# Nodes are automatically traced, but you can add custom spans
@traceable(name="validate_order", tags=["order", "validation"])
def validate_order(state: dict) -> dict:
    """This function will be traced in LangSmith"""
    # Validation logic...
    return state

# Add custom metadata to traces
@traceable(
    name="process_payment",
    metadata={"service": "stripe", "version": "2024-01"}
)
def process_payment(state: dict) -> dict:
    """Payment processing with rich metadata"""
    # Payment logic...
    return state

Custom Metrics & Logging

import logging
import time
from dataclasses import dataclass
from typing import Dict, Any
from prometheus_client import Counter, Histogram, Gauge

# Configure structured logging
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger("langgraph_ecommerce")

# Prometheus metrics
GRAPH_INVOCATIONS = Counter(
    'langgraph_invocations_total',
    'Total graph invocations',
    ['graph_name', 'status']
)

GRAPH_DURATION = Histogram(
    'langgraph_duration_seconds',
    'Graph execution duration',
    ['graph_name'],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)

NODE_DURATION = Histogram(
    'langgraph_node_duration_seconds',
    'Individual node execution duration',
    ['graph_name', 'node_name']
)

ACTIVE_THREADS = Gauge(
    'langgraph_active_threads',
    'Currently active conversation threads',
    ['graph_name']
)

@dataclass
class GraphMetrics:
    """Wrapper to add metrics to any graph"""

    def __init__(self, graph, graph_name: str):
        self.graph = graph
        self.graph_name = graph_name

    async def invoke_with_metrics(self, state: dict, config: dict) -> dict:
        """Invoke graph with automatic metrics collection"""
        thread_id = config.get("configurable", {}).get("thread_id", "unknown")

        ACTIVE_THREADS.labels(graph_name=self.graph_name).inc()

        start_time = time.time()
        status = "success"

        try:
            logger.info(
                "Graph invocation started",
                extra={
                    "graph": self.graph_name,
                    "thread_id": thread_id,
                    "input_keys": list(state.keys())
                }
            )

            result = await self.graph.ainvoke(state, config)

            logger.info(
                "Graph invocation completed",
                extra={
                    "graph": self.graph_name,
                    "thread_id": thread_id,
                    "duration_ms": (time.time() - start_time) * 1000
                }
            )

            return result

        except Exception as e:
            status = "error"
            logger.error(
                "Graph invocation failed",
                extra={
                    "graph": self.graph_name,
                    "thread_id": thread_id,
                    "error": str(e)
                },
                exc_info=True
            )
            raise

        finally:
            duration = time.time() - start_time
            GRAPH_INVOCATIONS.labels(
                graph_name=self.graph_name,
                status=status
            ).inc()
            GRAPH_DURATION.labels(
                graph_name=self.graph_name
            ).observe(duration)
            ACTIVE_THREADS.labels(graph_name=self.graph_name).dec()

# Usage
metrics_wrapper = GraphMetrics(order_processor, "order_processing")
result = await metrics_wrapper.invoke_with_metrics(order_state, config)

Performance Optimization

1. Streaming Responses

from langgraph.graph import StateGraph

async def stream_customer_service(query: str, config: dict):
    """Stream response tokens as they're generated"""

    async for event in customer_service_app.astream_events(
        {"messages": [{"role": "user", "content": query}]},
        config,
        version="v2"
    ):
        kind = event["event"]

        if kind == "on_chat_model_stream":
            # Stream LLM tokens to the client
            content = event["data"]["chunk"].content
            if content:
                yield content

        elif kind == "on_chain_end":
            # Node completed
            node_name = event.get("name", "unknown")
            print(f"Completed node: {node_name}")

# FastAPI endpoint with streaming
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat_endpoint(query: str, session_id: str):
    config = {"configurable": {"thread_id": session_id}}

    return StreamingResponse(
        stream_customer_service(query, config),
        media_type="text/event-stream"
    )

2. Caching

from functools import lru_cache
import hashlib
import json
import redis

# Redis cache for expensive operations
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_key(func_name: str, **kwargs) -> str:
    """Generate cache key from function name and arguments"""
    sorted_kwargs = json.dumps(kwargs, sort_keys=True)
    hash_input = f"{func_name}:{sorted_kwargs}"
    return hashlib.sha256(hash_input.encode()).hexdigest()

def cached_product_search(query: str, category: str = None) -> list:
    """Cache product search results"""
    key = cache_key("product_search", query=query, category=category)

    # Check cache
    cached = redis_client.get(key)
    if cached:
        return json.loads(cached)

    # Perform search
    results = perform_product_search(query, category)

    # Cache for 5 minutes
    redis_client.setex(key, 300, json.dumps(results))

    return results

# LRU cache for LLM prompts (in-memory)
@lru_cache(maxsize=1000)
def get_prompt_template(template_name: str, customer_tier: str) -> str:
    """Cache prompt templates"""
    return load_prompt_from_database(template_name, customer_tier)

3. Rate Limiting

import asyncio
from collections import deque
import time

class TokenBucketRateLimiter:
    """Rate limiter using token bucket algorithm"""

    def __init__(self, tokens_per_second: float, max_tokens: int):
        self.tokens_per_second = tokens_per_second
        self.max_tokens = max_tokens
        self.tokens = max_tokens
        self.last_refill = time.time()
        self._lock = asyncio.Lock()

    async def acquire(self, tokens: int = 1) -> bool:
        """Acquire tokens, returns True if successful"""
        async with self._lock:
            self._refill()

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    async def wait_for_token(self, tokens: int = 1):
        """Wait until tokens are available"""
        while not await self.acquire(tokens):
            await asyncio.sleep(0.1)

    def _refill(self):
        """Refill tokens based on elapsed time"""
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.tokens_per_second
        self.tokens = min(self.max_tokens, self.tokens + new_tokens)
        self.last_refill = now

# Create rate limiters for different services
openai_limiter = TokenBucketRateLimiter(tokens_per_second=50, max_tokens=100)
stripe_limiter = TokenBucketRateLimiter(tokens_per_second=25, max_tokens=50)

async def call_llm_rate_limited(messages: list) -> str:
    """Call LLM with rate limiting"""
    await openai_limiter.wait_for_token()
    return await llm.ainvoke(messages)

Security Best Practices

Input Validation

from pydantic import BaseModel, Field, validator
from typing import Optional
import re

class CustomerQuery(BaseModel):
    """Validated customer query input"""
    message: str = Field(..., min_length=1, max_length=5000)
    customer_id: str = Field(..., pattern=r'^CUST-[A-Z0-9]{6,12}

PII Handling

import re
from typing import Tuple

class PIIHandler:
    """Handle PII in logs and storage"""

    PII_PATTERNS = {
        'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
        'ssn': r'\b\d{3}-\d{2}-\d{4}\b',
        'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b'
    }

    @classmethod
    def redact(cls, text: str) -> str:
        """Redact PII from text for logging"""
        redacted = text
        for pii_type, pattern in cls.PII_PATTERNS.items():
            redacted = re.sub(pattern, f'[REDACTED_{pii_type.upper()}]', redacted)
        return redacted

    @classmethod
    def extract_and_redact(cls, text: str) -> Tuple[str, dict]:
        """Extract PII for secure storage, return redacted version"""
        pii_found = {}
        redacted = text

        for pii_type, pattern in cls.PII_PATTERNS.items():
            matches = re.findall(pattern, text)
            if matches:
                pii_found[pii_type] = matches
                redacted = re.sub(pattern, f'[{pii_type.upper()}]', redacted)

        return redacted, pii_found

# Usage in logging
def log_customer_interaction(state: dict):
    safe_messages = [
        PIIHandler.redact(msg.content)
        for msg in state.get("messages", [])
    ]
    logger.info(f"Customer interaction: {safe_messages}")

Deployment Architecture

                    ┌─────────────────────────┐
                    │      Load Balancer      │
                    │    (AWS ALB / nginx)    │
                    └───────────┬─────────────┘
                                │
            ┌───────────────────┼───────────────────┐
            ▼                   ▼                   ▼
    ┌───────────────┐   ┌───────────────┐   ┌───────────────┐
    │   API Server  │   │   API Server  │   │   API Server  │
    │   (FastAPI)   │   │   (FastAPI)   │   │   (FastAPI)   │
    └───────┬───────┘   └───────┬───────┘   └───────┬───────┘
            │                   │                   │
            └───────────────────┼───────────────────┘
                                │
            ┌───────────────────┼───────────────────┐
            ▼                   ▼                   ▼
    ┌───────────────┐   ┌───────────────┐   ┌───────────────┐
    │   LangGraph   │   │    Redis      │   │   PostgreSQL  │
    │   Workers     │   │  (Cache/Queue)│   │ (Checkpoints) │
    └───────────────┘   └───────────────┘   └───────────────┘
                                │
                    ┌───────────┴───────────┐
                    ▼                       ▼
            ┌───────────────┐       ┌───────────────┐
            │   LangSmith   │       │  Prometheus   │
            │  (Tracing)    │       │   + Grafana   │
            └───────────────┘       └───────────────┘

Key Takeaways


  

    
Retry with backoff - Don't hammer failing services; use exponential backoff with jitter
    
Circuit breakers - Fail fast when dependencies are down
    
Observability first - You can't fix what you can't see; use tracing + metrics
    
Security in depth - Validate inputs, redact PII, handle secrets properly
  


Next up: Final Assessment - Test your knowledge with a comprehensive quiz on LangGraph for e-commerce!
)
    session_id: str = Field(..., pattern=r'^[a-f0-9-]{36}

PII Handling

___CODEBLOCK_9___

Deployment Architecture

___CODEBLOCK_10___

Key Takeaways


  

    
Retry with backoff - Don't hammer failing services; use exponential backoff with jitter
    
Circuit breakers - Fail fast when dependencies are down
    
Observability first - You can't fix what you can't see; use tracing + metrics
    
Security in depth - Validate inputs, redact PII, handle secrets properly
  


Next up: Final Assessment - Test your knowledge with a comprehensive quiz on LangGraph for e-commerce!
)

    @validator('message')
    def sanitize_message(cls, v):
        # Remove potential injection attempts
        # (This is defense in depth - LLMs should also be prompted safely)
        dangerous_patterns = [
            r'ignore previous instructions',
            r'system prompt',
            r'<\|.*?\|>'
        ]
        for pattern in dangerous_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError("Invalid message content")
        return v

# Usage in API endpoint
def handle_customer_query(raw_input: dict):
    try:
        validated = CustomerQuery(**raw_input)
        return process_query(validated)
    except ValueError as e:
        return {"error": "Invalid input", "details": str(e)}

PII Handling

Deployment Architecture

Key Takeaways

Next up: Final Assessment - Test your knowledge with a comprehensive quiz on LangGraph for e-commerce!