Deploying AI Applications

Moving from prototype to production requires careful attention to reliability, performance, cost, and monitoring. This topic covers the essential practices for deploying AI applications.

Production Architecture

Users → Load Balancer → API Gateway → Your App → LLM API
              │                          │
              │                    ┌─────▼──────┐
              │                    │   Cache     │
              │                    │   (Redis)   │
              │                    └─────────────┘
              │
         ┌────▼────┐
         │Monitoring│
         │& Logging │
         └──────────┘

Building an AI API with FastAPI

python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI
import hashlib
import json

app = FastAPI()
client = OpenAI()

class ChatRequest(BaseModel):
    message: str
    conversation_id: str | None = None
    temperature: float = 0.7

class ChatResponse(BaseModel):
    response: str
    tokens_used: int
    cached: bool = False

# Simple in-memory cache (use Redis in production)
cache = {}

def get_cache_key(message: str, temperature: float) -> str:
    content = f"{message}:{temperature}"
    return hashlib.md5(content.encode()).hexdigest()

@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    # Check cache
    cache_key = get_cache_key(request.message, request.temperature)
    if cache_key in cache:
        return ChatResponse(**cache[cache_key], cached=True)

    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": request.message}
            ],
            temperature=request.temperature,
            max_tokens=1000
        )

        result = {
            "response": response.choices[0].message.content,
            "tokens_used": response.usage.total_tokens
        }

        # Cache the result
        cache[cache_key] = result

        return ChatResponse(**result)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Caching Strategies

Response Caching

Cache identical requests to avoid redundant API calls:

python

import redis
import json
import hashlib

redis_client = redis.Redis(host='localhost', port=6379)

def cached_llm_call(prompt, model="gpt-4o-mini", ttl=3600):
    cache_key = f"llm:{hashlib.md5(prompt.encode()).hexdigest()}"

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

    # Call LLM
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0  # Deterministic for caching
    )

    result = response.choices[0].message.content

    # Store in cache
    redis_client.setex(cache_key, ttl, json.dumps(result))

    return result

Semantic Caching

Cache based on meaning, not exact match:

python

def semantic_cache_lookup(query, threshold=0.95):
    query_embedding = get_embedding(query)

    # Search cache for similar queries
    for cached_query, cached_response in cache.items():
        cached_embedding = get_embedding(cached_query)
        similarity = cosine_similarity(query_embedding, cached_embedding)
        if similarity > threshold:
            return cached_response

    return None

Rate Limiting

Protect your API and manage LLM API costs:

python

from fastapi import Request
from datetime import datetime, timedelta

# Simple rate limiter
rate_limits = {}

async def rate_limit(request: Request, limit=10, window=60):
    """Allow 'limit' requests per 'window' seconds."""
    client_ip = request.client.host
    now = datetime.now()

    if client_ip not in rate_limits:
        rate_limits[client_ip] = []

    # Remove old entries
    rate_limits[client_ip] = [
        t for t in rate_limits[client_ip]
        if now - t < timedelta(seconds=window)
    ]

    if len(rate_limits[client_ip]) >= limit:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Try again later."
        )

    rate_limits[client_ip].append(now)

Monitoring and Observability

Logging LLM Calls

python

import logging
import time

logger = logging.getLogger("ai_app")

def monitored_llm_call(messages, model="gpt-4o-mini", **kwargs):
    start_time = time.time()

    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            **kwargs
        )

        duration = time.time() - start_time
        usage = response.usage

        logger.info(
            "LLM call completed",
            extra={
                "model": model,
                "duration_ms": round(duration * 1000),
                "input_tokens": usage.prompt_tokens,
                "output_tokens": usage.completion_tokens,
                "total_tokens": usage.total_tokens,
            }
        )

        return response

    except Exception as e:
        duration = time.time() - start_time
        logger.error(
            f"LLM call failed: {str(e)}",
            extra={
                "model": model,
                "duration_ms": round(duration * 1000),
                "error_type": type(e).__name__
            }
        )
        raise

Key Metrics to Track

Metric	What to Monitor	Alert Threshold
Latency	Time to first token, total time	> 5 seconds
Error rate	Failed API calls	> 5%
Token usage	Input/output tokens per request	Budget limits
Cost	Daily/weekly spend	Budget alerts
Cache hit rate	Percentage of cached responses	< 20% (optimize)
User satisfaction	Feedback scores, thumbs up/down	< 70% positive

Cost Optimization

1. Model Routing

Use cheaper models for simpler tasks:

python

def route_to_model(task_complexity):
    if task_complexity == "simple":
        return "gpt-4o-mini"       # $0.15 / 1M input
    elif task_complexity == "moderate":
        return "gpt-4o-mini"       # Still cost-effective
    else:
        return "gpt-4o"            # $5 / 1M input

2. Prompt Optimization

Shorter prompts = fewer tokens = lower cost:

python

# ❌ Expensive: 500 tokens
system_prompt = "Very long system prompt with lots of examples..."

# ✅ Optimized: 100 tokens
system_prompt = "You are a support agent. Be concise. Format: bullet points."

3. Max Tokens Limits

Always set max_tokens to prevent runaway costs:

python

response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=500  # Cap the response length
)

Deployment Options

Option	Best For	Complexity
Vercel/Netlify	Next.js apps with API routes	Low
AWS Lambda	Serverless, pay-per-use	Medium
Docker + VPS	Full control, predictable costs	Medium
Kubernetes	High scale, enterprise	High

Docker Deployment

dockerfile

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

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

Security Checklist

API keys stored in environment variables (never in code)
Input validation on all user messages
Output filtering for sensitive content
Rate limiting per user/IP
HTTPS only
Authentication for your API endpoints
Prompt injection defenses
PII detection and redaction in logs
CORS properly configured

Production Checklist

Error handling with graceful fallbacks
Response caching (Redis or similar)
Rate limiting
Monitoring and alerting
Cost tracking and budget alerts
Health check endpoint
Logging (structured, searchable)
Load testing completed
Security review passed
Documentation for API consumers

Key Takeaways

Cache aggressively — identical prompts don't need repeated API calls
Monitor everything — latency, errors, costs, quality
Route to cheaper models when possible
Set limits — max tokens, rate limits, budget caps
Security is not optional — protect API keys, validate inputs, filter outputs

Next, we'll cover AI safety and responsible AI practices.