ML API Endpoint Expert

Expert in designing and deploying machine learning API endpoints.

Core Principles

API Design

• Stateless Design: Each request contains all necessary information
• Consistent Response Format: Standardize success/error structures
• Versioning Strategy: Plan for model updates
• Input Validation: Rigorous validation before inference

FastAPI Implementation

Basic ML Endpoint

python
1from fastapi import FastAPI, HTTPException
2from pydantic import BaseModel, validator
3import joblib
4import numpy as np
5
6app = FastAPI(title="ML Model API", version="1.0.0")
7
8model = None
9
10@app.on_event("startup")
11async def load_model():
12    global model
13    model = joblib.load("model.pkl")
14
15class PredictionInput(BaseModel):
16    features: list[float]
17
18    @validator('features')
19    def validate_features(cls, v):
20        if len(v) != 10:
21            raise ValueError('Expected 10 features')
22        return v
23
24class PredictionResponse(BaseModel):
25    prediction: float
26    confidence: float | None = None
27    model_version: str
28    request_id: str
29
30@app.post("/predict", response_model=PredictionResponse)
31async def predict(input_data: PredictionInput):
32    features = np.array([input_data.features])
33    prediction = model.predict(features)[0]
34
35    return PredictionResponse(
36        prediction=float(prediction),
37        model_version="v1",
38        request_id=generate_request_id()
39    )

Batch Prediction

python
1class BatchInput(BaseModel):
2    instances: list[list[float]]
3
4    @validator('instances')
5    def validate_batch_size(cls, v):
6        if len(v) > 100:
7            raise ValueError('Batch size cannot exceed 100')
8        return v
9
10@app.post("/predict/batch")
11async def batch_predict(input_data: BatchInput):
12    features = np.array(input_data.instances)
13    predictions = model.predict(features)
14
15    return {
16        "predictions": predictions.tolist(),
17        "count": len(predictions)
18    }

Performance Optimization

Model Caching

python
1class ModelCache:
2    def __init__(self, ttl_seconds=300):
3        self.cache = {}
4        self.ttl = ttl_seconds
5
6    def get(self, features):
7        key = hashlib.md5(str(features).encode()).hexdigest()
8        if key in self.cache:
9            result, timestamp = self.cache[key]
10            if time.time() - timestamp < self.ttl:
11                return result
12        return None
13
14    def set(self, features, prediction):
15        key = hashlib.md5(str(features).encode()).hexdigest()
16        self.cache[key] = (prediction, time.time())

Health Checks

python
1@app.get("/health")
2async def health_check():
3    return {
4        "status": "healthy",
5        "model_loaded": model is not None
6    }
7
8@app.get("/metrics")
9async def get_metrics():
10    return {
11        "requests_total": request_counter,
12        "prediction_latency_avg": avg_latency,
13        "error_rate": error_rate
14    }

Docker Deployment

dockerfile
1FROM python:3.9-slim
2
3WORKDIR /app
4COPY requirements.txt .
5RUN pip install --no-cache-dir -r requirements.txt
6
7COPY . .
8EXPOSE 8000
9
10CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Best Practices

• Use async/await for I/O operations
• Validate data types, ranges, and business rules
• Cache predictions for deterministic models
• Handle model failures with fallback responses
• Log predictions, latencies, and errors
• Support multiple model versions
• Set memory and CPU limits