Machine Learning: FastAPIとDockerによるモデルデプロイ — MLモデルの本番サービングガイド

from fastapi import FastAPI
import joblib

app = FastAPI()
model = joblib.load("model.pkl")

@app.post("/predict")
def predict(features: PredictionInput):
    result = model.predict([features.dict()])
    return {"prediction": float(result[0])}

import joblib
import pickle
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
import numpy as np

# Train and save model
rng = np.random.default_rng(42)
X = rng.uniform(0, 100, (1000, 5))
y = 50 + 0.8 * X[:, 0] + 1.2 * X[:, 1] + rng.normal(0, 5, 1000)

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", RandomForestRegressor(n_estimators=100, random_state=42)),
])
pipe.fit(X, y)

# Save with joblib (recommended for sklearn)
joblib.dump(pipe, "salespredict_model.joblib", compress=3)

# Save with pickle (alternative)
with open("salespredict_model.pkl", "wb") as f:
    pickle.dump(pipe, f)

# Load and predict
loaded_model = joblib.load("salespredict_model.joblib")
sample = np.array([[50, 30, 20, 10, 5]])
prediction = loaded_model.predict(sample)
print(f"Prediction: {prediction[0]:.2f} thousand USD")

# Executed successfully

import torch
import torch.nn as nn

# Save model state_dict (recommended)
class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(5, 32), nn.ReLU(), nn.Linear(32, 1))

    def forward(self, x):
        return self.net(x)

model = SimpleModel()
torch.save(model.state_dict(), "pytorch_model.pt")

# Load
loaded = SimpleModel()
loaded.load_state_dict(torch.load("pytorch_model.pt", weights_only=True))
loaded.eval()

# Function defined successfully

# File: app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import joblib
import numpy as np
import time

app = FastAPI(title="SalesPredict API", version="1.0.0")

# Load model at startup
model = joblib.load("salespredict_model.joblib")

class PredictionInput(BaseModel):
    ad_spend_k: float = Field(..., ge=0, description="Ad spend in thousand USD")
    traffic_k: float = Field(..., ge=0, description="Traffic in thousands")
    category_electronics: float = Field(0, ge=0, le=1)
    category_clothing: float = Field(0, ge=0, le=1)
    is_promotion: float = Field(0, ge=0, le=1)

    model_config = {"json_schema_extra": {
        "example": {"ad_spend_k": 50, "traffic_k": 300,
                     "category_electronics": 1, "category_clothing": 0, "is_promotion": 1}
    }}

class PredictionOutput(BaseModel):
    predicted_revenue_k: float
    latency_ms: float

@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": model is not None}

@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
    start = time.time()
    try:
        features = np.array([[input_data.ad_spend_k, input_data.traffic_k,
                               input_data.category_electronics,
                               input_data.category_clothing, input_data.is_promotion]])
        prediction = model.predict(features)[0]
        latency = (time.time() - start) * 1000
        return PredictionOutput(predicted_revenue_k=round(float(prediction), 2),
                                 latency_ms=round(latency, 2))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/predict_batch")
def predict_batch(inputs: list[PredictionInput]):
    features = np.array([[d.ad_spend_k, d.traffic_k, d.category_electronics,
                           d.category_clothing, d.is_promotion] for d in inputs])
    predictions = model.predict(features)
    return {"predictions": [round(float(p), 2) for p in predictions]}

# Function defined successfully

# Install dependencies
pip install fastapi uvicorn joblib scikit-learn

# Run the API server
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

# Test with curl
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"ad_spend_k": 50, "traffic_k": 300, "category_electronics": 1, "category_clothing": 0, "is_promotion": 1}'

# Access Swagger UI: http://localhost:8000/docs

# Stage 1: Build dependencies
FROM python:3.11-slim AS builder

WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: Runtime (smaller image)
FROM python:3.11-slim

WORKDIR /app

# Copy installed packages from builder
COPY --from=builder /install /usr/local

# Copy application code and model
COPY app.py .
COPY salespredict_model.joblib .

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

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s \
  CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

# requirements.txt
fastapi==0.109.0
uvicorn==0.27.0
joblib==1.3.2
scikit-learn==1.4.0
numpy==1.26.4
pydantic==2.5.0

# Build image
docker build -t salespredict-api:latest .

# Run container
docker run -d -p 8000:8000 --name salespredict salespredict-api:latest

# Test
curl http://localhost:8000/health

# View logs
docker logs salespredict

# Stop and remove
docker stop salespredict && docker rm salespredict

# docker-compose.yml
version: "3.8"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    deploy:
      replicas: 2
    restart: unless-stopped

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

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - api
    restart: unless-stopped

volumes:
  redis_data:

# nginx.conf (simplified load balancer)
upstream api_servers {
    server api:8000;
}

server {
    listen 80;
    location / {
        proxy_pass http://api_servers;
        proxy_set_header Host $host;
    }
}

# Enhanced app.py with Redis caching
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np
import hashlib
import json

app = FastAPI(title="SalesPredict API with Cache")
model = joblib.load("salespredict_model.joblib")

# Redis cache (conceptual)
# import redis
# redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))

class PredictionInput(BaseModel):
    ad_spend_k: float
    traffic_k: float
    category_electronics: float = 0
    category_clothing: float = 0
    is_promotion: float = 0

def get_cache_key(input_data: PredictionInput) -> str:
    data_str = json.dumps(input_data.model_dump(), sort_keys=True)
    return f"pred:{hashlib.md5(data_str.encode()).hexdigest()}"

@app.post("/predict")
def predict(input_data: PredictionInput):
    cache_key = get_cache_key(input_data)

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

    features = np.array([[input_data.ad_spend_k, input_data.traffic_k,
                           input_data.category_electronics,
                           input_data.category_clothing, input_data.is_promotion]])
    prediction = float(model.predict(features)[0])
    result = {"predicted_revenue_k": round(prediction, 2)}

    # Cache for 5 minutes
    # redis_client.setex(cache_key, 300, json.dumps(result))

    return result

# Function defined successfully