1 Conceitos
Tempo de Inferência (Latency): Em sistemas como o da Playcatch, se a recomendação demorar mais de 200ms, o usuário desiste. Monitorar isso é vital para a experiência do usuário.
Data Drift (Desvio de Dados): O gosto musical muda (ex: uma música nova vira hit). Se o modelo foi treinado com dados de 2 anos atrás, as recomendações ficarão "velhas". O monitoramento detecta quando a distribuição das previsões muda drasticamente.
Anomalias: Se o modelo começar a recomendar a mesma música para 100% dos usuários, o sistema de logs deve disparar um alerta.
2 Implementação de Observabilidade e Testes de Sanidade. Estabilidade: Garantida por testes automáticos que verificam se a API responde em milissegundos.Qualidade: Mantida pelo monitoramento de drift, onde comparamos se as músicas que o modelo recomenda hoje ainda fazem sentido com o comportamento atual dos usuários da Playcatch.
3 Implementação de Monitoramento e Testes (Programação)
Passo 1: API com Logging e Telemetria (main.py)
Adicionamos um middleware para medir o tempo de resposta e registrar logs de cada predição.
import time
import logging
from fastapi import FastAPI, Request
Configuração de Logging para auditoria
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("Playcatch_Ops")
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
# Registra tempo de latência e status da rota
logger.info(f"Path: {request.url.path} | Time: {process_time:.4f}s | Status: {response.status_code}")
return response
@app.get("/recommend/{user_id}")
async def predict(user_id: int):
# Simulação de log de predição para análise de Drift no MLflow
# mlflow.log_text(str(results), f"logs/pred_{user_id}.json")
return {"user_id": user_id, "recommendations": ["Music_1", "Music_2"]}
Passo 2: Teste de Integração Simples (test_api.py)
Um script para garantir que o contêiner Docker está saudável.
import requests
def test_api_health():
response = requests.get("http://localhost:8000/")
assert response.status_code == 200
assert response.json()["status"] == "Online"
def test_recommendation_logic():
response = requests.get("http://localhost:8000/recommend/123")
data = response.json()
assert "recommendations" in data
assert len(data["recommendations"]) > 0
