Machine Learning: Implantação de Modelos com FastAPI e Docker

Q: P: Como aplico rate limiting nos requests da API?

R: Use a biblioteca slowapi para rate limiting — limiter = Limiter(key_func=get_remote_address), por exemplo, limitando a 100 requests por minuto. Isso previne abuso e sobrecarga.

Última atualização: 2026-08-26

Um modelo treinado dentro de um Jupyter Notebook não tem valor — seu valor só é desbloqueado quando é implantado em produção.

1. O que você vai aprender

Serviço de modelos com FastAPI: validação Pydantic, endpoints de previsão assíncronos e documentação Swagger gerada automaticamente
Serialização de modelos: salvando modelos sklearn com joblib/pickle e modelos PyTorch com torch.save
Containerização com Docker: escrevendo Dockerfiles, builds multi-stage e otimização de imagens
Orquestração com docker-compose: serviço de modelo + cache Redis + balanceamento de carga Nginx
A API de previsão do Bob: projetando o endpoint POST /predict com latência abaixo de 50ms por previsão

2. A história real de um engenheiro de ML

(1) O problema: um modelo em notebook não atende o negócio

Bob treinou um modelo XGBoost com R²=0,89, mas ele só rodava dentro do seu Jupyter Notebook. Quando o gerente de produto perguntou: "O frontend pode chamar isso?", Bob não tinha ideia de como transformar o modelo em uma API. A lacuna entre o treinamento e a implantação é o maior problema de "última milha" em projetos de ML.

(2) A solução com FastAPI + Docker

FastAPI encapsula o modelo em uma API REST, e Docker o containeriza — qualquer serviço pode então chamar previsões via HTTP.

PYTHON

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])}

(3) O resultado: milhões de requests por dia após o lançamento

Depois que Bob implantou o modelo com FastAPI + Docker, a latência da API ficou abaixo de 50ms enquanto processava 1 milhão+ de requests por dia, podendo ser chamado a partir do frontend, CRM e sistemas de recomendação.

3. Serialização de Modelos

(1) Salvando e carregando modelos

▶ Exemplo: Serialização de modelo sklearn

PYTHON

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

# Treinar e salvar o modelo
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)

# Salvar com joblib (recomendado para sklearn)
joblib.dump(pipe, "salespredict_model.joblib", compress=3)

# Salvar com pickle (alternativa)
with open("salespredict_model.pkl", "wb") as f:
    pickle.dump(pipe, f)

# Carregar e prever
loaded_model = joblib.load("salespredict_model.joblib")
sample = np.array([[50, 30, 20, 10, 5]])
prediction = loaded_model.predict(sample)
print(f"Previsão: {prediction[0]:.2f} mil USD")

Saída:

TEXT 📖 Somente leitura

# Executado com sucesso

Método	Melhor para	Vantagens	Desvantagens
joblib	sklearn/numpy	Eficiente para arrays grandes	Apenas Python
pickle	Qualquer objeto Python	Universal	Riscos de segurança, compatibilidade de versão
torch.save	PyTorch	Flexível (pode salvar state_dict)	Apenas PyTorch
mlflow.sklearn	sklearn	Versionamento + metadados	Requer MLflow
ONNX	Cross-framework	Cross-language/platform	Conversão complexa

▶ Exemplo: Salvando um modelo PyTorch

PYTHON

import torch
import torch.nn as nn

# Salvar state_dict do modelo (recomendado)
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")

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

Saída:

TEXT 📖 Somente leitura

# Função definida com sucesso

4. Serviço de Modelos com FastAPI

(1) Básico do FastAPI

▶ Exemplo: Uma API de previsão completa

PYTHON

# Arquivo: 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")

# Carregar modelo na inicialização
model = joblib.load("salespredict_model.joblib")

class PredictionInput(BaseModel):
    ad_spend_k: float = Field(..., ge=0, description="Gasto com anúncios em mil USD")
    traffic_k: float = Field(..., ge=0, description="Tráfego em milhares")
    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]}

Saída:

TEXT 📖 Somente leitura

# Função definida com sucesso

(2) Executando o serviço FastAPI

BASH

# Instalar dependências
pip install fastapi uvicorn joblib scikit-learn

# Executar o servidor da API
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

# Testar com 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}'

# Acessar a UI do Swagger: http://localhost:8000/docs

Funcionalidade do FastAPI	Descrição
Validação Pydantic	Valida automaticamente tipos e intervalos de entrada
Swagger UI	Gera automaticamente docs interativas (/docs)
Type hints	Gera automaticamente modelos de resposta
Suporte async	async/await para alta concorrência
Tratamento de exceções	Códigos de erro padrão via HTTPException

5. Containerização com Docker

(1) Escrevendo um Dockerfile

▶ Exemplo: A imagem Docker do SalesPredict

DOCKERFILE

# Estágio 1: Construir dependências
FROM python:3.11-slim AS builder

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

# Estágio 2: Runtime (imagem menor)
FROM python:3.11-slim

WORKDIR /app

# Copiar pacotes instalados do builder
COPY --from=builder /install /usr/local

# Copiar código da aplicação e modelo
COPY app.py .
COPY salespredict_model.joblib .

# Usuário não-root para segurança
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"]

TEXT 📖 Somente leitura

# 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

(2) Comandos Docker

BASH

# Construir a imagem
docker build -t salespredict-api:latest .

# Executar o contêiner
docker run -d -p 8000:8000 --name salespredict salespredict-api:latest

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

# Ver logs
docker logs salespredict

# Parar e remover
docker stop salespredict && docker rm salespredict

Otimização do Dockerfile	Efeito
Build multi-stage	Encolhe a imagem de 1,5GB para 200MB
Imagem base slim	Remove pacotes do sistema desnecessários
.dockerignore	Exclui arquivos grandes como .git/data
Usuário não-root	Reforço de segurança
HEALTHCHECK	Verificações de saúde do contêiner

6. Orquestração com docker-compose

Uma arquitetura de implantação completa conecta todos os componentes — o serviço de API, cache, balanceador de carga e monitoramento formam um pipeline completo de ponta a ponta:

graph TB
    CLIENT[Cliente / Frontend] --> NGINX[Nginx<br/>Limite de Taxa + LB]
    NGINX --> API1[Worker FastAPI 1]
    NGINX --> API2[Worker FastAPI 2]
    API1 --> REDIS[(Cache Redis<br/>LRU 256MB)]
    API2 --> REDIS
    API1 --> MODEL[Arquivo de Modelo<br/>.joblib]
    API2 --> MODEL
    PROM[Prometheus<br/>Métricas] --> API1
    PROM --> API2
    GRAF[Grafana<br/>Dashboard] --> PROM

▶ Exemplo: Uma arquitetura de implantação em produção completa

YAML

# 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:

TEXT 📖 Somente leitura

# nginx.conf (balanceador de carga simplificado)
upstream api_servers {
    server api:8000;
}

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

▶ Exemplo: API + Cache Redis

PYTHON

# app.py aprimorado com cache Redis
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")

# Cache Redis (conceitual)
# 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)

    # Verificar cache primeiro
    # 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 por 5 minutos
    # redis_client.setex(cache_key, 300, json.dumps(result))

    return result

Saída:

TEXT 📖 Somente leitura

# Função definida com sucesso

Componente	Papel	Escolha de tecnologia
Serviço de API	Inferência do modelo	FastAPI + Uvicorn
Cache	Cache de previsões quentes	Redis (TTL 5min)
Balanceador de carga	Distribuição de requests	Nginx
Orquestração de contêineres	Gerenciamento de serviços	docker-compose
Verificações de saúde	Detecção de falhas	/health + HEALTHCHECK

❓ Perguntas Frequentes

P: Qual é melhor, pickle ou joblib? R: Use joblib para modelos sklearn (comprime arrays numpy grandes com mais eficiência); use pickle para objetos Python gerais. Ambos carregam riscos de segurança (um arquivo pickle não confiável pode executar código malicioso), então MLflow ou ONNX é mais seguro para produção.

P: Devo usar FastAPI ou Flask? R: Use FastAPI para novos projetos — docs automáticas (Swagger), validação de tipos (Pydantic), suporte async e melhor desempenho. Flask é mais maduro, mas sua experiência de desenvolvimento de API não se compara ao FastAPI.

P: E se minha imagem Docker for muito grande? R: Três truques — 1) builds multi-stage (o estágio de build não vai para a imagem final); 2) use imagens base slim/alpine; 3) use .dockerignore para excluir .git/data e similares.

P: Como atualizo o modelo sem tempo de inatividade? R: Duas abordagens — 1) implantação blue-green (alternando entre versões antiga e nova); 2) atualizações rolling (rolling update do docker-compose). Combine isso com o MLflow Model Registry para gerenciar versões.

P: Como otimizo a latência da API? R: Quatro camadas de otimização — 1) cache de requests quentes com Redis; 2) previsões em lote para reduzir overhead; 3) execute múltiplos workers em paralelo (Uvicorn workers); 4) quantize o modelo (reduza seu tamanho).

P: Como aplico rate limiting nos requests da API? R: Use a biblioteca slowapi para rate limiting — limiter = Limiter(key_func=get_remote_address), por exemplo, limitando a 100 requests por minuto. Isso previne abuso e sobrecarga.

📖 Resumo

Serialização de modelos: use joblib para sklearn, torch.save (state_dict) para PyTorch, e MLflow é recomendado para produção
FastAPI fornece a API REST: Pydantic valida entradas, Swagger gera docs automaticamente, e async lida com alta concorrência
Containerização com Docker: builds multi-stage encolhem a imagem, usuário não-root reforça a segurança, e HEALTHCHECK monitora a saúde
Orquestração com docker-compose: uma implantação de nível de produção com API + cache Redis + balanceamento de carga Nginx
Estratégia de cache: Redis faz cache de resultados de previsão quentes com TTL de 5 minutos, alcançando uma taxa de acerto de 30-50%
Meta de latência da API: <50ms (incluindo inferência do modelo), com previsões em lote reduzindo ainda mais a latência amortizada

📝 Exercícios

Básico (Dificuldade ⭐): Treine um modelo sklearn, salve-o com joblib, depois carregue-o em um script Python separado e faça uma previsão. Dica: joblib.dump/load.
Intermediário (Dificuldade ⭐⭐): Crie um endpoint /predict com FastAPI, incluindo validação de entrada Pydantic e um check /health, depois execute-o com uvicorn e teste-o com curl. Dica: consulte o código completo da API na Seção 4.
Desafio (Dificuldade ⭐⭐⭐): Escreva um Dockerfile (build multi-stage) + docker-compose.yml (API + Redis + Nginx), construa a imagem e execute a stack completa de serviços, depois verifique o balanceamento de carga e o cache. Dica: consulte os arquivos de configuração nas Seções 5-6.

← Anterior: Introdução ao MLOps | Próximo: Teste A/B →

Equipe Técnica Web-Tutorial

Uma plataforma de tutoriais mantida por diversos desenvolvedores. Cada tutorial é escrito e revisado por profissionais da área correspondente. Trabalhamos para manter nosso conteúdo preciso e confiável — se encontrar algum problema, avise-nos.

Sobre Nós