Deploy AI Model to Production: FastAPI, Docker, and Cloud Deployment Guide

Deploy AI Model to Production: FastAPI, Docker, and Cloud Deployment Guide

The AI model that works in a Jupyter notebook is about 20% of the way to production. The remaining 80% is packaging, serving, monitoring, scaling, and maintaining it.

I've watched three AI prototypes die in staging because nobody planned the deployment. Load spikes crashed single-process servers. Models reloaded on every request. No alerting when the model started returning errors. This guide covers the patterns that prevent these failures.

Part 1: FastAPI Application Structure

# Project structure:
# my_ai_app/
#   app/
#     main.py
#     models.py          # Pydantic models
#     dependencies.py    # Dependency injection
#     routers/
#       predictions.py
#       health.py
#   Dockerfile
#   requirements.txt
#   .env

# app/dependencies.py
from openai import OpenAI
from functools import lru_cache

@lru_cache()
def get_openai_client() -> OpenAI:
    """Singleton OpenAI client — created once, reused."""
    return OpenAI()

# For local models, load on startup:
# from transformers import pipeline
# _model = None
# def get_model():
#     global _model
#     if _model is None:
#         _model = pipeline("sentiment-analysis", ...)
#     return _model

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import logging
import time

logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Load resources on startup, clean up on shutdown."""
    logger.info("Starting up application...")
    # Load models, connect to databases, etc.
    yield
    logger.info("Shutting down application...")
    # Cleanup

app = FastAPI(
    title="AI Model API",
    description="Production AI inference API",
    version="1.0.0",
    lifespan=lifespan
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],  # Restrict in production
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"]
)

# Request logging middleware
@app.middleware("http")
async def log_requests(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = (time.time() - start_time) * 1000
    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} "
        f"duration={process_time:.1f}ms"
    )
    return response

from app.routers import predictions, health
app.include_router(predictions.router, prefix="/api/v1")
app.include_router(health.router)

# app/routers/predictions.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from openai import OpenAI
from app.dependencies import get_openai_client
import time

router = APIRouter()

class PredictionRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=10000)
    model: str = Field(default="gpt-4o-mini")
    max_tokens: int = Field(default=500, ge=1, le=4096)

class PredictionResponse(BaseModel):
    result: str
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float

@router.post("/predict", response_model=PredictionResponse)
async def predict(
    request: PredictionRequest,
    client: OpenAI = Depends(get_openai_client)
):
    start_time = time.time()
    
    try:
        response = client.chat.completions.create(
            model=request.model,
            messages=[{"role": "user", "content": request.text}],
            max_tokens=request.max_tokens
        )
        
        latency_ms = (time.time() - start_time) * 1000
        
        return PredictionResponse(
            result=response.choices[0].message.content,
            model=response.model,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            latency_ms=latency_ms
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/predict/stream")
async def predict_stream(
    request: PredictionRequest,
    client: OpenAI = Depends(get_openai_client)
):
    from fastapi.responses import StreamingResponse
    
    async def generate():
        stream = client.chat.completions.create(
            model=request.model,
            messages=[{"role": "user", "content": request.text}],
            stream=True
        )
        for chunk in stream:
            content = chunk.choices[0].delta.content or ""
            if content:
                yield f"data: {content}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

Part 2: Dockerfile

# Dockerfile
# Multi-stage build: smaller final image

# Build stage
FROM python:3.11-slim as builder

WORKDIR /app

# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies in virtual env
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

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

# Production stage
FROM python:3.11-slim

WORKDIR /app

# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Copy application code
COPY app/ ./app/

# Non-root user for security
RUN useradd -m -u 1000 appuser
USER appuser

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

EXPOSE 8000

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

Part 3: Docker Compose for Local Development

# docker-compose.yml
version: "3.8"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379
    env_file:
      - .env
    depends_on:
      - redis
    restart: unless-stopped
    volumes:
      - ./app:/app/app  # Mount for development hot reload
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload  # Dev mode

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

Part 4: Rate Limiting

# pip install slowapi

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import Request

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

# Apply to routes
@router.post("/predict")
@limiter.limit("30/minute")  # 30 requests per minute per IP
async def predict(request: Request, body: PredictionRequest, ...):
    ...

# API key-based rate limiting
def get_api_key(request: Request) -> str:
    return request.headers.get("X-API-Key", get_remote_address(request))

api_limiter = Limiter(key_func=get_api_key)

@router.post("/predict")
@api_limiter.limit("100/minute")  # Per API key
async def predict(request: Request, ...):
    ...

Part 5: Monitoring with Prometheus

# pip install prometheus-fastapi-instrumentator

from prometheus_fastapi_instrumentator import Instrumentator

# Add to main.py
instrumentator = Instrumentator()
instrumentator.instrument(app).expose(app)

# Custom metrics
from prometheus_client import Counter, Histogram, Gauge

tokens_counter = Counter(
    "ai_tokens_total",
    "Total tokens used",
    ["model", "type"]  # Labels: input/output
)

cost_counter = Counter(
    "ai_api_cost_dollars",
    "Total API cost in dollars",
    ["model"]
)

model_latency = Histogram(
    "ai_model_latency_seconds",
    "Model inference latency",
    ["model"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

# Use in endpoint
@router.post("/predict")
async def predict(...):
    with model_latency.labels(model=request.model).time():
        response = client.chat.completions.create(...)
    
    tokens_counter.labels(model=response.model, type="input").inc(
        response.usage.prompt_tokens
    )
    tokens_counter.labels(model=response.model, type="output").inc(
        response.usage.completion_tokens
    )

Part 6: Cloud Deployment

Railway (Simplest)

# Install Railway CLI
npm install -g @railway/cli

# Login and deploy
railway login
railway init
railway up

# Set environment variables
railway variables set OPENAI_API_KEY=sk-...
# Railway auto-detects Dockerfile and deploys

AWS ECS with Fargate

# Build and push to ECR
aws ecr create-repository --repository-name ai-api
aws ecr get-login-password | docker login --username AWS --password-stdin $ECR_URL
docker build -t ai-api .
docker tag ai-api:latest $ECR_URL/ai-api:latest
docker push $ECR_URL/ai-api:latest

# Create ECS task definition and service
# (Use AWS Console or Terraform/CDK for infrastructure)

Conclusion

Production AI deployment is primarily software engineering — the model is one component among many. FastAPI + Docker + Redis + Prometheus covers the fundamentals for most applications.

The most common production failures: model loaded on every request (fix: startup loading), no rate limiting (fix: slowapi), no monitoring (fix: Prometheus + Grafana), secrets in code (fix: environment variables). Address all four before going live.

For the AI components that power these APIs, see our OpenAI API integration guide. For building the chatbot application that deploys this way, see our AI chatbot guide.

Frequently Asked Questions

What is the best way to serve an AI model in production?

FastAPI for API-based LLM applications. vLLM for self-hosted transformer models (OpenAI-compatible API with GPU optimization). BentoML or Triton for classification/embedding models. Containerize everything with Docker.

How do I handle model loading in a production FastAPI app?

Load once in the startup lifespan handler, inject via Depends(). Never load per-request. For multiple workers: each loads its own copy — ensure sufficient RAM/VRAM.

How do I dockerize an AI application with GPU support?

Use nvidia/cuda base image. Install nvidia-container-toolkit on host. Run with --gpus all. Multi-stage builds reduce image size. Add health checks.

What monitoring should I add to a production AI application?

Latency (p95/p99), throughput, error rates, token usage, API cost, GPU utilization. Prometheus + Grafana for metrics, Sentry for errors, LangSmith for LLM observability. Alert on response time >2s and error rate >1%.

How do I implement rate limiting for an AI API?

Use slowapi library with Redis backend. Apply @limiter.limit('100/minute') decorator per route. Limit by IP address or API key. Return informative 429 responses with Retry-After header.