Production AI: Deploying, Monitoring, and Scaling LLM Features
The gap between a demo and a live system is bigger than you think.

Every developer who has shipped an LLM feature has had the same experience: it works perfectly on localhost, then something goes wrong the moment real users touch it. The production gap is the distance between a happy-path demo and a system that handles real load, real failures, and real costs.

Here are the five most common ways that gap bites teams:

This checkpoint gives you the toolbox to close every one of these gaps before they hit production.

Hardcoded API keys get rotated after a breach. Environment variables scattered across the codebase become unmaintainable. The right pattern is a single config class that loads everything from the environment at startup and fails loudly if anything is missing.

# config.py — single source of truth for all LLM settings
import os
from dataclasses import dataclass, field

@dataclass
class LLMConfig:
    # Provider selection
    provider: str = field(default_factory=lambda: os.getenv("LLM_PROVIDER", "openai"))

    # API keys (never hardcoded)
    openai_api_key: str = field(default_factory=lambda: os.getenv("OPENAI_API_KEY", ""))
    anthropic_api_key: str = field(default_factory=lambda: os.getenv("ANTHROPIC_API_KEY", ""))
    gemini_api_key: str = field(default_factory=lambda: os.getenv("GOOGLE_API_KEY", ""))

    # Model settings
    model: str = field(default_factory=lambda: os.getenv("LLM_MODEL", "gpt-4o-mini"))
    max_tokens: int = field(default_factory=lambda: int(os.getenv("LLM_MAX_TOKENS", "2048")))
    temperature: float = field(default_factory=lambda: float(os.getenv("LLM_TEMPERATURE", "0.7")))
    timeout: int = field(default_factory=lambda: int(os.getenv("LLM_TIMEOUT_SECONDS", "30")))

    # Cost controls
    max_cost_per_request_usd: float = field(
        default_factory=lambda: float(os.getenv("LLM_MAX_COST_PER_REQUEST", "0.10"))
    )

    def validate(self):
        """Fail fast at startup — never mid-request."""
        key_map = {
            "openai": ("openai_api_key", "OPENAI_API_KEY"),
            "anthropic": ("anthropic_api_key", "ANTHROPIC_API_KEY"),
            "gemini": ("gemini_api_key", "GOOGLE_API_KEY"),
        }
        attr, env_var = key_map.get(self.provider, (None, None))
        if attr and not getattr(self, attr):
            raise ValueError(
                f"[LLMConfig] Missing required env var: {env_var}\n"
                f"Provider '{self.provider}' is selected but no key found.\n"
                f"Fix: export {env_var}=your-key-here"
            )
        return self

# Instantiate at module import time — crash at startup, not mid-request
config = LLMConfig().validate()

LLM API costs scale with tokens, not requests. A single user who pastes a 50-page document can spend $5 in one call. Without controls, a burst of traffic can run up thousands of dollars before your credit card alert fires.

Three layers of protection:

# cost_guard.py — estimate cost before sending, reject if over budget
COST_PER_1M_INPUT = {
    "gpt-4o-mini": 0.15,
    "gpt-4o": 2.50,
    "claude-3-5-haiku-20241022": 0.80,
    "gemini-1.5-flash": 0.075,
}
COST_PER_1M_OUTPUT = {
    "gpt-4o-mini": 0.60,
    "gpt-4o": 10.00,
    "claude-3-5-haiku-20241022": 4.00,
    "gemini-1.5-flash": 0.30,
}

def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float:
    input_cost = (input_tokens / 1_000_000) * COST_PER_1M_INPUT.get(model, 1.0)
    output_cost = (output_tokens / 1_000_000) * COST_PER_1M_OUTPUT.get(model, 4.0)
    return input_cost + output_cost

def check_cost_budget(model: str, prompt_tokens: int, max_cost_usd: float = 0.10):
    """Reject requests likely to exceed budget. Estimate output as 2x input."""
    estimated = estimate_cost_usd(model, prompt_tokens, prompt_tokens * 2)
    if estimated > max_cost_usd:
        raise ValueError(
            f"Estimated cost ${estimated:.4f} exceeds budget ${max_cost_usd:.2f}. "
            f"Reduce prompt length or increase budget."
        )

# rate_limiter.py — per-user token bucket using Redis
import time
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def check_rate_limit(user_id: str, tokens_requested: int,
                     limit_tokens: int = 100_000,  # 100k tokens per hour per user
                     window_seconds: int = 3600) -> bool:
    key = f"rate:{user_id}:{int(time.time() // window_seconds)}"
    pipe = r.pipeline()
    pipe.incr(key, tokens_requested)
    pipe.expire(key, window_seconds)
    current, _ = pipe.execute()
    if current > limit_tokens:
        raise ValueError(
            f"Rate limit exceeded. User {user_id} has used {current:,} tokens "
            f"this hour (limit: {limit_tokens:,})."
        )
    return True

When a user reports "the AI gave a weird answer last Tuesday," you need to be able to replay exactly what was sent and received. Unstructured logs like print("LLM response:", response) are useless for debugging production issues.

Log structured JSON for every LLM call:

# llm_logger.py — structured logging for every LLM call
import json, time, uuid, logging
from datetime import datetime, timezone

logger = logging.getLogger("llm")

def log_llm_call(
    request_id: str,
    user_id: str,
    model: str,
    messages: list,
    response_text: str,
    input_tokens: int,
    output_tokens: int,
    latency_ms: float,
    finish_reason: str,
    error: str = None,
):
    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "request_id": request_id,
        "user_id": user_id,
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cost_usd": estimate_cost_usd(model, input_tokens, output_tokens),
        "latency_ms": round(latency_ms, 1),
        "finish_reason": finish_reason,
        "error": error,
        # Store prompt hash, not raw text, unless you have PII consent
        "prompt_hash": hash(json.dumps(messages)),
    }
    if finish_reason == "length":
        record["alert"] = "TRUNCATED_RESPONSE — increase max_tokens or shorten prompt"
    logger.info(json.dumps(record))
    return record

# Usage in your LLM wrapper
def complete(messages: list, user_id: str) -> str:
    request_id = str(uuid.uuid4())
    start = time.time()
    try:
        response = client.chat.completions.create(model=config.model, messages=messages)
        text = response.choices[0].message.content
        finish_reason = response.choices[0].finish_reason
        log_llm_call(
            request_id, user_id, config.model, messages, text,
            response.usage.prompt_tokens,
            response.usage.completion_tokens,
            (time.time() - start) * 1000,
            finish_reason,
        )
        if finish_reason == "length":
            raise ValueError("LLM response was truncated — increase max_tokens")
        return text
    except Exception as e:
        log_llm_call(request_id, user_id, config.model, messages, "",
                     0, 0, (time.time() - start) * 1000, "error", str(e))
        raise

OpenAI had 8 incidents in 2024. Anthropic had 4. Every major LLM provider has outages. If your service is 100% dependent on one provider, their downtime is your downtime. A provider fallback chain costs you almost nothing to implement and buys you near-100% uptime.

# fallback_client.py — try primary, then fallback providers
import openai
import anthropic
from typing import Optional

class FallbackLLMClient:
    """Try providers in order. First success wins."""

    def __init__(self):
        self.providers = [
            self._call_openai,
            self._call_anthropic,
        ]

    def complete(self, messages: list, max_tokens: int = 1024) -> str:
        last_error = None
        for provider_fn in self.providers:
            try:
                return provider_fn(messages, max_tokens)
            except Exception as e:
                last_error = e
                logger.warning(f"Provider failed, trying next: {e}")
                continue
        raise RuntimeError(f"All providers failed. Last error: {last_error}")

    def _call_openai(self, messages: list, max_tokens: int) -> str:
        client = openai.OpenAI(api_key=config.openai_api_key)
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            max_tokens=max_tokens,
            timeout=config.timeout,
        )
        if resp.choices[0].finish_reason == "length":
            raise ValueError("Response truncated — would propagate bad data")
        return resp.choices[0].message.content

    def _call_anthropic(self, messages: list, max_tokens: int) -> str:
        client = anthropic.Anthropic(api_key=config.anthropic_api_key)
        # Anthropic uses a different messages format
        system = next((m["content"] for m in messages if m["role"] == "system"), "")
        user_msgs = [m for m in messages if m["role"] != "system"]
        resp = client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=max_tokens,
            system=system,
            messages=user_msgs,
        )
        return resp.content[0].text

LLM calls are expensive and slow. Many real-world requests are semantically identical — the same FAQ question, the same document summary, the same code review for unchanged code. Caching identical requests is the easiest latency and cost win available to you.

# cache.py — Redis cache for LLM responses
import hashlib, json
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
CACHE_TTL = 3600 * 24  # 24 hours

def cache_key(model: str, messages: list, temperature: float) -> str:
    """Deterministic key from model + messages + temperature."""
    payload = json.dumps({"model": model, "messages": messages, "temperature": temperature},
                         sort_keys=True)
    return "llm:" + hashlib.sha256(payload.encode()).hexdigest()

def get_cached(model: str, messages: list, temperature: float = 0.0) -> Optional[str]:
    """Only cache deterministic calls (temperature=0)."""
    if temperature != 0.0:
        return None  # Non-deterministic — never cache
    key = cache_key(model, messages, temperature)
    return r.get(key)

def set_cached(model: str, messages: list, temperature: float,
               response: str, ttl: int = CACHE_TTL):
    if temperature != 0.0:
        return
    key = cache_key(model, messages, temperature)
    r.setex(key, ttl, response)

# Wrap your complete() function
def complete_with_cache(messages: list, temperature: float = 0.0) -> str:
    cached = get_cached(config.model, messages, temperature)
    if cached:
        return cached  # No API call, no cost, ~1ms latency
    result = complete(messages)
    set_cached(config.model, messages, temperature, result)
    return result

LLM latency is non-deterministic. The same prompt can take 800ms at 10am and 4000ms at 3pm depending on API load. You need to track P50, P95, and P99 latency — not average (averages hide tail latency spikes).

# monitoring.py — latency tracking with Prometheus (or just file logs)
import time, functools
from collections import deque
from threading import Lock
import statistics

class LatencyTracker:
    """Thread-safe rolling latency tracker. No external dependency."""
    def __init__(self, window_size: int = 1000):
        self._samples = deque(maxlen=window_size)
        self._lock = Lock()

    def record(self, latency_ms: float):
        with self._lock:
            self._samples.append(latency_ms)

    def stats(self) -> dict:
        with self._lock:
            if not self._samples:
                return {}
            s = sorted(self._samples)
            n = len(s)
            return {
                "count": n,
                "p50_ms": round(s[n // 2], 1),
                "p95_ms": round(s[int(n * 0.95)], 1),
                "p99_ms": round(s[int(n * 0.99)], 1),
                "max_ms": round(s[-1], 1),
            }

tracker = LatencyTracker()

def monitor_latency(fn):
    """Decorator that records latency for any function."""
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = fn(*args, **kwargs)
            tracker.record((time.time() - start) * 1000)
            return result
        except Exception:
            tracker.record((time.time() - start) * 1000)
            raise
    return wrapper

# FastAPI health endpoint
# GET /health returns current latency stats
@app.get("/health")
def health():
    stats = tracker.stats()
    status = "ok"
    if stats.get("p95_ms", 0) > 5000:
        status = "degraded"  # P95 over 5 seconds — alert!
    return {"status": status, "latency": stats}

These are the failures that have actually taken down real AI features in production. Each one is preventable with the patterns from this checkpoint.