Como fazer deploy de modelos de ML com FastAPI e Docker

1. Fundamentos do Deploy de Modelos de Machine Learning

1.1. Desafios do deploy: latência, escalabilidade e reprodutibilidade

O deploy de modelos de Machine Learning em produção apresenta três desafios críticos. A latência deve ser minimizada para garantir respostas em tempo real, especialmente em aplicações que exigem inferência instantânea. A escalabilidade é necessária para lidar com picos de requisições sem degradação do serviço. A reprodutibilidade garante que o modelo se comporte consistentemente em diferentes ambientes, eliminando o clássico "funciona na minha máquina".

1.2. Por que FastAPI e Docker? Vantagens para APIs de ML

FastAPI oferece alta performance (comparável a Node.js e Go), validação automática de dados com Pydantic e documentação interativa via Swagger. Docker proporciona isolamento completo do ambiente, garantindo que as dependências do modelo sejam exatamente as mesmas em desenvolvimento e produção. Juntos, formam a combinação ideal para serviços de inferência.

1.3. Arquitetura típica de um serviço de inferência

Cliente → Load Balancer → Container FastAPI → Modelo em memória → Resposta

2. Preparação do Modelo e Ambiente de Desenvolvimento

2.1. Serialização do modelo treinado

Após treinar seu modelo, serialize-o para carregamento eficiente:

import joblib
from sklearn.ensemble import RandomForestClassifier

modelo = RandomForestClassifier()
modelo.fit(X_train, y_train)
joblib.dump(modelo, 'modelo_rf.pkl')

2.2. Estrutura de diretórios recomendada

projeto-ml-deploy/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── schemas.py
│   ├── modelo.py
│   └── preprocess.py
├── modelos/
│   └── modelo_rf.pkl
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── tests/
    └── test_api.py

2.3. Gerenciamento de dependências

fastapi==0.104.1
uvicorn==0.24.0
joblib==1.3.2
scikit-learn==1.3.2
pandas==2.1.3
numpy==1.26.2
pydantic==2.5.2
python-multipart==0.0.6

3. Construção da API com FastAPI

3.1. Criação dos endpoints

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI(title="API de ML - Previsão de Preços")

class DadosEntrada(BaseModel):
    area: float
    quartos: int
    banheiros: int
    idade: int

class PrevisaoSaida(BaseModel):
    preco_previsto: float
    modelo_versao: str

modelo = None

@app.on_event("startup")
async def carregar_modelo():
    global modelo
    modelo = joblib.load("modelos/modelo_rf.pkl")

@app.get("/health")
async def health_check():
    return {"status": "saudável", "modelo_carregado": modelo is not None}

@app.post("/predict", response_model=PrevisaoSaida)
async def predict(dados: DadosEntrada):
    if modelo is None:
        raise HTTPException(status_code=503, detail="Modelo não carregado")

    features = np.array([[dados.area, dados.quartos, dados.banheiros, dados.idade]])
    preco = modelo.predict(features)[0]

    return PrevisaoSaida(preco_previsto=float(preco), modelo_versao="v1.0")

3.2. Tratamento de erros e logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    logger.error(f"Erro na requisição {request.url}: {str(exc)}")
    return {"detail": "Erro interno do servidor"}, 500

4. Integração do Modelo na API

4.1. Pipeline de pré-processamento

class Preprocessador:
    def __init__(self):
        self.scaler = joblib.load("modelos/scaler.pkl")

    def transformar(self, dados: DadosEntrada) -> np.ndarray:
        features = np.array([[dados.area, dados.quartos, dados.banheiros, dados.idade]])
        return self.scaler.transform(features)

4.2. Estratégias para inferência assíncrona

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

@app.post("/predict/batch")
async def predict_batch(lista_dados: list[DadosEntrada]):
    loop = asyncio.get_event_loop()
    tarefas = []

    for dados in lista_dados:
        tarefa = loop.run_in_executor(executor, processar_predicao, dados)
        tarefas.append(tarefa)

    resultados = await asyncio.gather(*tarefas)
    return {"previsoes": resultados}

5. Containerização com Docker

5.1. Dockerfile otimizado com multi-stage build

# Estágio de construção
FROM python:3.11-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dirs -r requirements.txt

# Estágio final
FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY ./app ./app
COPY ./modelos ./modelos

ENV PATH=/root/.local/bin:$PATH
EXPOSE 8000

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

5.2. Configuração do docker-compose.yml

version: '3.8'

services:
  api-ml:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MODELO_PATH=/app/modelos/modelo_rf.pkl
      - LOG_LEVEL=INFO
    volumes:
      - ./modelos:/app/modelos
    restart: unless-stopped

5.3. Boas práticas de segurança

# Dockerfile com usuário não-root
RUN useradd -m -u 1000 mluser
USER mluser

# Variáveis de ambiente para configuração
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

6. Testes e Validação da Aplicação

6.1. Testes unitários com pytest

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_health():
    response = client.get("/health")
    assert response.status_code == 200

def test_predicao_valida():
    dados = {"area": 150, "quartos": 3, "banheiros": 2, "idade": 10}
    response = client.post("/predict", json=dados)
    assert response.status_code == 200
    assert "preco_previsto" in response.json()

6.2. Testes de integração do container

# test_container.py
import docker
import requests

def test_container_funciona():
    client = docker.from_env()
    container = client.containers.run("api-ml:latest", detach=True, ports={"8000": "8000"})

    response = requests.get("http://localhost:8000/health")
    assert response.status_code == 200

    container.stop()
    container.remove()

7. Deploy em Produção

7.1. Publicação da imagem no Docker Hub

docker build -t seuusuario/api-ml:v1.0 .
docker push seuusuario/api-ml:v1.0

7.2. Deploy em cloud

Para Google Cloud Run:

gcloud builds submit --tag gcr.io/seu-projeto/api-ml
gcloud run deploy api-ml --image gcr.io/seu-projeto/api-ml --platform managed

7.3. Monitoramento contínuo

# Adicione métricas Prometheus no FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app)

8. Manutenção e Atualizações do Serviço

8.1. Versionamento de modelos

# app/modelo.py
class GerenciadorModelo:
    def __init__(self):
        self.versoes = {
            "v1.0": "modelos/modelo_rf_v1.pkl",
            "v2.0": "modelos/modelo_rf_v2.pkl"
        }

    def carregar_versao(self, versao: str):
        caminho = self.versoes.get(versao)
        if caminho:
            return joblib.load(caminho)
        raise ValueError(f"Versão {versao} não encontrada")

8.2. Pipeline CI/CD com GitHub Actions

name: Deploy ML Model

on:
  push:
    branches: [main]

jobs:
  test-and-deploy:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3

    - name: Testes
      run: |
        pip install -r requirements.txt
        pytest tests/

    - name: Build e push Docker
      run: |
        docker build -t api-ml:latest .
        docker tag api-ml:latest registry.example.com/api-ml:${{ github.sha }}
        docker push registry.example.com/api-ml:${{ github.sha }}

Referências