Deploying and Scaling AI Applications: From Prototype to Production

┌─────────────┐
│   Users     │
└──────┬──────┘
       │
┌──────▼───────┐
│ Load Balancer│  (nginx, AWS ALB)
└──────┬───────┘
       │
┌──────▼────────────────────────┐
│  Application Servers (3+)     │
│  - API endpoints              │
│  - Request validation         │
│  - Business logic             │
└──────┬────────────────────────┘
       │
       ├──────────┬────────────┐
       │          │            │
┌──────▼──┐  ┌───▼────┐  ┌───▼───────┐
│  Redis  │  │LLM APIs│  │Vector DB  │
│ (Cache) │  │ (GPT,  │  │(Pinecone, │
│         │  │Claude) │  │ Qdrant)   │
└─────────┘  └────────┘  └───────────┘

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Run with gunicorn (production WSGI server)
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "--timeout", "120", "app:app"]

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379
      - VECTOR_DB_URL=http://qdrant:6333
    depends_on:
      - redis
      - qdrant
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '1'
          memory: 2G

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant_data:/qdrant/storage

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - api

volumes:
  redis_data:
  qdrant_data:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-api
spec:
  replicas: 5  # Auto-scale from 5-20
  selector:
    matchLabels:
      app: ai-api
  template:
    metadata:
      labels:
        app: ai-api
    spec:
      containers:
      - name: api
        image: your-registry/ai-api:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: openai-key
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

---
apiVersion: v1
kind: Service
metadata:
  name: ai-api-service
spec:
  selector:
    app: ai-api
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-api
  minReplicas: 5
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

from functools import lru_cache
import redis
import hashlib

class MultiLayerCache:
    """Three-tier caching: Memory → Redis → Database."""

    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)

    @lru_cache(maxsize=1000)
    def get_from_memory(self, key):
        """Layer 1: In-memory cache (fastest)."""
        return None  # Handled by @lru_cache decorator

    def get_from_redis(self, key):
        """Layer 2: Redis cache (fast)."""
        cached = self.redis_client.get(key)
        if cached:
            return json.loads(cached)
        return None

    def get_from_source(self, key, fetch_func):
        """Layer 3: Source of truth (slowest)."""
        return fetch_func()

    def get(self, key, fetch_func, ttl=3600):
        """Get value with multi-layer caching."""
        # Try memory cache
        result = self.get_from_memory(key)
        if result:
            return result

        # Try Redis
        result = self.get_from_redis(key)
        if result:
            # Warm memory cache
            return result

        # Fetch from source
        result = self.get_from_source(key, fetch_func)

        # Store in Redis
        self.redis_client.setex(
            key,
            ttl,
            json.dumps(result)
        )

        return result

# Usage
cache = MultiLayerCache()

def generate_llm_response(prompt):
    """Generate response with multi-layer caching."""
    cache_key = hashlib.md5(prompt.encode()).hexdigest()

    return cache.get(
        cache_key,
        lambda: call_llm_api(prompt),
        ttl=3600
    )

import asyncio
from collections import defaultdict
from datetime import datetime, timedelta

class RequestBatcher:
    """Batch multiple requests into single API call."""

    def __init__(self, max_batch_size=10, max_wait_ms=100):
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.pending_requests = []
        self.batch_lock = asyncio.Lock()

    async def add_request(self, request):
        """Add request to batch."""
        async with self.batch_lock:
            self.pending_requests.append(request)

            # Trigger batch if full
            if len(self.pending_requests) >= self.max_batch_size:
                return await self._process_batch()

            # Wait for more requests or timeout
            await asyncio.sleep(self.max_wait_ms / 1000)

            if self.pending_requests:
                return await self._process_batch()

    async def _process_batch(self):
        """Process accumulated batch."""
        if not self.pending_requests:
            return

        batch = self.pending_requests.copy()
        self.pending_requests.clear()

        # Single API call for entire batch
        results = await call_embedding_api_batch([r['text'] for r in batch])

        # Distribute results
        for request, result in zip(batch, results):
            request['future'].set_result(result)

# Usage
batcher = RequestBatcher(max_batch_size=100, max_wait_ms=50)

async def get_embedding(text):
    """Get embedding with automatic batching."""
    future = asyncio.Future()

    await batcher.add_request({
        'text': text,
        'future': future
    })

    return await future

from urllib3 import PoolManager
from requests.adapters import HTTPAdapter

class OptimizedAPIClient:
    """API client with connection pooling."""

    def __init__(self, base_url):
        self.session = requests.Session()

        # Configure connection pooling
        adapter = HTTPAdapter(
            pool_connections=20,  # Number of connection pools
            pool_maxsize=100,     # Max connections per pool
            max_retries=3
        )

        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)

        self.base_url = base_url

    def call(self, endpoint, data):
        """Make API call using pooled connection."""
        response = self.session.post(
            f"{self.base_url}/{endpoint}",
            json=data,
            timeout=30
        )
        return response.json()

# Single client instance shared across requests
api_client = OptimizedAPIClient("https://api.openai.com")

# Reuses connections across calls
result1 = api_client.call("chat/completions", data1)
result2 = api_client.call("chat/completions", data2)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def generate_stream(prompt):
    """Stream LLM response to client."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    for chunk in response:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

@app.post("/chat/stream")
async def chat_stream(request: dict):
    """Streaming endpoint for better perceived performance."""
    return StreamingResponse(
        generate_stream(request['message']),
        media_type="text/event-stream"
    )

# Client receives data as it's generated (better UX)

{
  "AutoScalingGroupName": "ai-api-asg",
  "MinSize": 3,
  "MaxSize": 50,
  "DesiredCapacity": 5,
  "HealthCheckType": "ELB",
  "HealthCheckGracePeriod": 300,
  "LaunchTemplate": {
    "LaunchTemplateName": "ai-api-template",
    "Version": "$Latest"
  },
  "TargetGroupARNs": ["arn:aws:elasticloadbalancing:..."],
  "VPCZoneIdentifier": "subnet-1,subnet-2,subnet-3",

  "ScalingPolicies": [
    {
      "PolicyName": "scale-up-cpu",
      "AdjustmentType": "ChangeInCapacity",
      "ScalingAdjustment": 2,
      "Cooldown": 300,
      "MetricAggregationType": "Average",
      "TargetTrackingConfiguration": {
        "PredefinedMetricSpecification": {
          "PredefinedMetricType": "ASGAverageCPUUtilization"
        },
        "TargetValue": 70.0
      }
    },
    {
      "PolicyName": "scale-up-requests",
      "TargetTrackingConfiguration": {
        "CustomizedMetricSpecification": {
          "MetricName": "RequestCount",
          "Namespace": "AI/API",
          "Statistic": "Sum"
        },
        "TargetValue": 1000.0
      }
    }
  ]
}

upstream ai_api {
    # Least connections algorithm
    least_conn;

    # Health checks
    server api-1:8000 max_fails=3 fail_timeout=30s;
    server api-2:8000 max_fails=3 fail_timeout=30s;
    server api-3:8000 max_fails=3 fail_timeout=30s;

    # Keep-alive connections
    keepalive 32;
}

server {
    listen 80;

    # Connection limits
    limit_conn_zone $binary_remote_addr zone=addr:10m;
    limit_conn addr 10;  # Max 10 connections per IP

    # Rate limiting
    limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;
    limit_req zone=api burst=20 nodelay;

    location / {
        proxy_pass http://ai_api;

        # Timeouts
        proxy_connect_timeout 5s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;

        # Headers
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # Keep-alive
        proxy_http_version 1.1;
        proxy_set_header Connection "";
    }

    # Health check endpoint
    location /health {
        access_log off;
        proxy_pass http://ai_api/health;
    }
}

class ShardedVectorDB:
    """Distribute vectors across multiple database shards."""

    def __init__(self, shard_urls):
        self.shards = [
            QdrantClient(url=url) for url in shard_urls
        ]
        self.num_shards = len(self.shards)

    def _get_shard(self, doc_id):
        """Deterministically route document to shard."""
        shard_index = hash(doc_id) % self.num_shards
        return self.shards[shard_index]

    def upsert(self, doc_id, vector, payload):
        """Insert into appropriate shard."""
        shard = self._get_shard(doc_id)
        shard.upsert(
            collection_name="documents",
            points=[{
                "id": doc_id,
                "vector": vector,
                "payload": payload
            }]
        )

    def search(self, query_vector, top_k=10):
        """Search across all shards, merge results."""
        from concurrent.futures import ThreadPoolExecutor

        def search_shard(shard):
            return shard.search(
                collection_name="documents",
                query_vector=query_vector,
                limit=top_k * 2  # Get more from each shard
            )

        # Parallel search across shards
        with ThreadPoolExecutor(max_workers=self.num_shards) as executor:
            shard_results = list(executor.map(search_shard, self.shards))

        # Merge and re-rank
        all_results = []
        for results in shard_results:
            all_results.extend(results)

        # Sort by score and return top_k
        all_results.sort(key=lambda x: x.score, reverse=True)
        return all_results[:top_k]

# Usage: Distribute 10M vectors across 4 shards
sharded_db = ShardedVectorDB([
    "http://shard-1:6333",
    "http://shard-2:6333",
    "http://shard-3:6333",
    "http://shard-4:6333"
])

import celery
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def process_llm_request(self, user_id, prompt):
    """Process LLM request asynchronously."""
    try:
        # Generate response
        response = call_llm_api(prompt)

        # Store result
        store_result(user_id, response)

        # Notify user (webhook, websocket, etc.)
        notify_user(user_id, response)

        return response

    except Exception as e:
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

# API endpoint returns immediately
@app.post("/chat")
async def chat(request: ChatRequest):
    # Queue the task
    task = process_llm_request.delay(request.user_id, request.message)

    return {
        "task_id": task.id,
        "status": "processing",
        "eta": "1-5 seconds"
    }

# Client polls for result
@app.get("/chat/{task_id}")
async def get_result(task_id: str):
    task = process_llm_request.AsyncResult(task_id)

    if task.ready():
        return {"status": "completed", "result": task.result}
    else:
        return {"status": "processing"}

from datadog import initialize, statsd
from functools import wraps
import time

# Initialize DataDog
initialize(
    api_key=os.getenv('DATADOG_API_KEY'),
    app_key=os.getenv('DATADOG_APP_KEY')
)

def monitor_performance(func):
    """Decorator to monitor function performance."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()

        try:
            result = func(*args, **kwargs)

            # Track success
            statsd.increment(
                'ai.request.success',
                tags=[f"function:{func.__name__}"]
            )

            return result

        except Exception as e:
            # Track errors
            statsd.increment(
                'ai.request.error',
                tags=[f"function:{func.__name__}", f"error:{type(e).__name__}"]
            )
            raise

        finally:
            # Track latency
            duration = (time.time() - start_time) * 1000
            statsd.histogram(
                'ai.request.latency',
                duration,
                tags=[f"function:{func.__name__}"]
            )

    return wrapper

@monitor_performance
def generate_response(prompt):
    """Generate LLM response with monitoring."""
    response = client.chat.completions.create(...)

    # Track token usage
    statsd.gauge('ai.tokens.input', response.usage.prompt_tokens)
    statsd.gauge('ai.tokens.output', response.usage.completion_tokens)

    # Track cost
    cost = calculate_cost(response.usage)
    statsd.gauge('ai.cost.per_request', cost)

    return response

class MetricsDashboard:
    """Collect and expose custom metrics."""

    def __init__(self):
        self.metrics = {
            "requests_total": 0,
            "requests_success": 0,
            "requests_failed": 0,
            "total_tokens": 0,
            "total_cost_usd": 0.0,
            "avg_latency_ms": 0.0,
            "cache_hits": 0,
            "cache_misses": 0
        }
        self.latencies = []

    def record_request(self, success, tokens, cost, latency, cache_hit):
        """Record request metrics."""
        self.metrics["requests_total"] += 1

        if success:
            self.metrics["requests_success"] += 1
        else:
            self.metrics["requests_failed"] += 1

        self.metrics["total_tokens"] += tokens
        self.metrics["total_cost_usd"] += cost

        self.latencies.append(latency)
        self.metrics["avg_latency_ms"] = sum(self.latencies) / len(self.latencies)

        if cache_hit:
            self.metrics["cache_hits"] += 1
        else:
            self.metrics["cache_misses"] += 1

    def get_metrics(self):
        """Get current metrics."""
        return {
            **self.metrics,
            "success_rate": self.metrics["requests_success"] / max(self.metrics["requests_total"], 1),
            "cache_hit_rate": self.metrics["cache_hits"] / max(self.metrics["requests_total"], 1),
            "p95_latency_ms": np.percentile(self.latencies, 95) if self.latencies else 0
        }

# Expose as Prometheus metrics
from prometheus_client import Counter, Histogram, Gauge

requests_total = Counter('ai_requests_total', 'Total AI requests')
request_latency = Histogram('ai_request_latency_seconds', 'Request latency')
cache_hit_rate = Gauge('ai_cache_hit_rate', 'Cache hit rate')

# Or expose as HTTP endpoint
@app.get("/metrics")
def metrics():
    return dashboard.get_metrics()

class AlertManager:
    """Monitor metrics and trigger alerts."""

    def __init__(self, slack_webhook_url):
        self.slack_webhook = slack_webhook_url
        self.thresholds = {
            "error_rate": 0.05,  # 5%
            "p95_latency_ms": 2000,  # 2 seconds
            "cost_per_hour": 50,  # $50/hour
            "cache_hit_rate": 0.3  # 30%
        }

    def check_metrics(self, metrics):
        """Check metrics against thresholds."""
        alerts = []

        # High error rate
        if metrics["error_rate"] > self.thresholds["error_rate"]:
            alerts.append({
                "severity": "critical",
                "metric": "error_rate",
                "value": metrics["error_rate"],
                "threshold": self.thresholds["error_rate"]
            })

        # High latency
        if metrics["p95_latency_ms"] > self.thresholds["p95_latency_ms"]:
            alerts.append({
                "severity": "warning",
                "metric": "p95_latency",
                "value": metrics["p95_latency_ms"],
                "threshold": self.thresholds["p95_latency_ms"]
            })

        # High costs
        hourly_cost = metrics["total_cost_usd"] / (metrics["uptime_hours"] or 1)
        if hourly_cost > self.thresholds["cost_per_hour"]:
            alerts.append({
                "severity": "warning",
                "metric": "cost_per_hour",
                "value": hourly_cost,
                "threshold": self.thresholds["cost_per_hour"]
            })

        # Send alerts
        for alert in alerts:
            self.send_alert(alert)

    def send_alert(self, alert):
        """Send alert to Slack."""
        message = f"""
        🚨 Alert: {alert['metric']}

        Severity: {alert['severity']}
        Current value: {alert['value']:.2f}
        Threshold: {alert['threshold']:.2f}

        Action required: Investigate immediately
        """

        requests.post(self.slack_webhook, json={"text": message})

import boto3
from botocore.exceptions import ClientError

def get_secret(secret_name):
    """Retrieve secret from AWS Secrets Manager."""
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name='us-east-1'
    )

    try:
        response = client.get_secret_value(SecretId=secret_name)
        return json.loads(response['SecretString'])
    except ClientError as e:
        logging.error(f"Failed to retrieve secret: {e}")
        raise

# Get API keys securely
secrets = get_secret("production/ai-api-keys")
openai_key = secrets['OPENAI_API_KEY']
anthropic_key = secrets['ANTHROPIC_API_KEY']

# Never hardcode or log API keys!

from pydantic import BaseModel, validator
import bleach

class ChatRequest(BaseModel):
    message: str
    user_id: str

    @validator('message')
    def validate_message(cls, v):
        # Length limits
        if len(v) < 1:
            raise ValueError("Message too short")
        if len(v) > 10000:
            raise ValueError("Message too long (max 10000 chars)")

        # Sanitize HTML
        v = bleach.clean(v, tags=[], strip=True)

        # Check for prompt injection patterns
        suspicious_patterns = [
            "ignore previous instructions",
            "ignore all previous",
            "system:",
            "<|im_start|>",
            "###"
        ]

        lower_message = v.lower()
        for pattern in suspicious_patterns:
            if pattern in lower_message:
                logging.warning(f"Potential prompt injection detected: {pattern}")
                # Could block or sanitize

        return v

# Usage
@app.post("/chat")
async def chat(request: ChatRequest):
    # Pydantic automatically validates and sanitizes
    response = process_message(request.message, request.user_id)
    return {"response": response}

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from fastapi import FastAPI, Request

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/chat")
@limiter.limit("10/minute")  # 10 requests per minute per IP
async def chat(request: Request, chat_request: ChatRequest):
    # Additional user-based rate limiting
    user_id = chat_request.user_id

    if not check_user_rate_limit(user_id):
        raise HTTPException(
            status_code=429,
            detail="User rate limit exceeded. Please wait before trying again."
        )

    return process_chat(chat_request)

def check_user_rate_limit(user_id):
    """Check Redis-based user rate limit."""
    key = f"rate_limit:user:{user_id}"
    count = redis_client.incr(key)

    if count == 1:
        redis_client.expire(key, 60)  # 1 minute window

    return count <= 20  # 20 requests per minute per user

import logging
import json
from datetime import datetime

class AuditLogger:
    """Log all requests for compliance and debugging."""

    def __init__(self):
        self.logger = logging.getLogger('audit')
        handler = logging.FileHandler('audit.log')
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_request(self, user_id, request_type, input_data, output_data, metadata):
        """Log request with all relevant details."""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "request_type": request_type,
            "input_hash": hashlib.sha256(str(input_data).encode()).hexdigest(),
            "output_hash": hashlib.sha256(str(output_data).encode()).hexdigest(),
            "metadata": metadata
        }

        # Don't log PII or full content (hash instead)
        self.logger.info(json.dumps(log_entry))

# Usage
audit_logger = AuditLogger()

def process_request(user_id, prompt):
    response = call_llm(prompt)

    audit_logger.log_request(
        user_id=user_id,
        request_type="chat",
        input_data=prompt,
        output_data=response,
        metadata={
            "model": "gpt-4o-mini",
            "tokens": 150,
            "latency_ms": 850
        }
    )

    return response