4.16 Monitoramento Específico de Machine Learning

Até este ponto, nosso foco esteve concentrado em métricas mais “clássicas”: tempo de resposta, uso de CPU, consumo de memória e afins. Essas métricas são fundamentais — sem elas, operamos no escuro do ponto de vista de infraestrutura. No entanto, em sistemas de Machine Learning, elas são apenas a camada mais superficial do monitoramento.

Isso acontece porque, diferentemente de aplicações tradicionais, o comportamento de um sistema de ML não é determinado apenas pelo código, mas também pelos dados. E dados são dinâmicos por natureza. Eles mudam com o tempo, com o comportamento dos usuários, com sazonalidade, com mudanças de mercado… e, às vezes, mudam de formas sutis e perigosas.

É perfeitamente possível ter um sistema com excelente saúde de infraestrutura — baixa latência, consumo estável de memória, nenhum erro — e ainda assim entregar resultados ruins. Do ponto de vista do Prometheus, está tudo verde. Do ponto de vista do negócio… nem tanto.

Por isso, precisamos começar a observar métricas específicas de Machine Learning. Alguns exemplos importantes:

• Distribuição das entradas (features): acompanhar estatísticas como média, desvio padrão ou até histogramas dos dados de entrada. Isso ajuda a identificar mudanças no perfil dos dados ao longo do tempo.

• Distribuição das predições: se um modelo que antes classificava bem distribuído passa a prever quase sempre a mesma classe, algo está errado.

• Confiança do modelo: em modelos probabilísticos, podemos monitorar o nível médio de confiança. Quedas podem indicar que o modelo está “menos seguro” sobre o que está vendo.

• Taxa de acerto (quando há ground truth disponível): essencial para detectar degradação real de performance.

• Tempo de inferência por tipo de entrada: certos inputs podem ser mais custosos que outros, revelando gargalos escondidos.

Um dos conceitos mais importantes nesse contexto é o data drift. Ele ocorre quando a distribuição dos dados de entrada em produção começa a se afastar da distribuição dos dados usados no treinamento. Em termos simples: o modelo foi treinado para um “mundo”, mas está operando em outro.

Imagine, por exemplo, um classificador de produtos treinado com descrições curtas e objetivas. Se, ao longo do tempo, os usuários passarem a inserir descrições muito mais longas e detalhadas, o modelo pode começar a ter dificuldades — não porque está “quebrado”, mas porque o contexto mudou. Esse tipo de mudança pode não impactar imediatamente CPU ou memória, mas pode afetar diretamente a qualidade das predições.

Existe também o concept drift, que é ainda mais traiçoeiro: quando a relação entre entrada e saída muda. Por exemplo, o significado de certas palavras pode evoluir, ou padrões de comportamento podem mudar com o tempo. Nesse caso, mesmo que os dados “pareçam” semelhantes, o modelo pode estar aprendendo uma regra que já não é mais válida.

Em resumo, monitorar apenas infraestrutura é como medir os sinais vitais de um paciente sem nunca perguntar se ele está sentindo dor. Para sistemas de Machine Learning, precisamos ir além: observar os dados, o comportamento do modelo e, principalmente, como tudo isso evolui ao longo do tempo.

Detectando Data Drift na Prática

Até agora, nosso monitoramento estava focado em como o sistema se comporta (latência, CPU, etc.). Agora vamos dar um passo além: entender se os dados que chegam ao modelo continuam fazendo sentido.

Iremos trabalhar com o primeiro modelo que construímos neste curso, ou seja, o classificador de produtos. Esse cenário é simples e didático:

• Entrada: descricao (texto)

• Saída: categoria

• Dataset de treino: nome;descricao;categoria

Queremos responder à seguinte pergunta:

Os textos que estão chegando em produção são parecidos com os textos usados no treinamento?

Para responder a essa pergunta, podemos extrair algumas estatísticas do dataset de treino:

• Tamanho médio da descrição

• Número médio de palavras

• Distribuição das categorias

• Vocabulário

Essas estatísticas representam o perfil dos dados originais. Posteriormente, iremos monitorar o sistema em produção para ver se essas estatísticas mudam. Caso haja mudanças de maneira significativa e perene, temos um possível data drift, o que pode significar a necessidade de atualizar o treinamento com base em um novo dataset.

Durante o treinamento (ou offline)

O processo começa, idealmente, durante o treinamento, pois é o momento em que o modelo que faz a predição é produzido. Caso ainda tenha acesso, ative o ambiente jupyter-dev criado lá no capítulo 2. Caso contrário, é fácil recriá-lo.

1. Crie uma pasta chamada jupyter

2. Crie o ambiente virtual jupyter-dev e instale notebook

pyenv virtualenv 3.10.2 jupyter-dev
pyenv activate jupyter-dev
pip install notebook

Ou se estiver no conda:

conda create --name jupyter-dev python=3.13.5
conda activate jupyter-dev
pip install notebook

Baixe o dataset e notebook de treinamento, e salve nessa pasta:

• produtos.csv - conjunto de dados anotado contendo produtos e sua classificação. Existem quatro categorias de produtos neste conjunto: game, maquiagem, brinquedo e livro

• produtos.ipynb - notebook com a solução para classificação de produtos

Execute, no terminal (com o ambiente ativo)

jupyter notebook

Execute o notebook e veja como o arquivo model.sav é criado. É o mesmo que está sendo usado para as predições.

Para obter as métricas, adicione uma célula após o pré-processamento:

import json

# --- Reutilizando dados já processados ---
# Aqui usamos a coluna final que alimenta o modelo: "strings"

# Comprimento em caracteres
products_data["desc_len"] = products_data["strings"].str.len()

# Quantidade de palavras
products_data["word_count"] = products_data["strings"].str.split().str.len()

# --- Estatísticas principais ---
baseline = {
    "avg_len": products_data["desc_len"].mean(),
    "std_len": products_data["desc_len"].std(),
    "avg_words": products_data["word_count"].mean(),
    "std_words": products_data["word_count"].std()
}

# Distribuição das categorias (para detectar drift de saída)
category_distribution = (
    products_data["categoria"]
    .value_counts(normalize=True)
    .to_dict()
)

baseline["category_distribution"] = category_distribution

# Ajuda a detectar palavras "desconhecidas" em produção
baseline["vocab_size"] = len(vect.vocabulary_)

# --- Salva baseline ---
with open("baseline.json", "w") as f:
    json.dump(baseline, f, indent=4)

baseline

Depois de executar, você verá uma saída parecida com esta:

{'avg_len': np.float64(971.051440329218),
 'std_len': np.float64(1124.7837726693785),
 'avg_words': np.float64(132.56824417009602),
 'std_words': np.float64(158.27436436420143),
 'category_distribution': {'livro': 0.28737997256515774,
  'maquiagem': 0.27023319615912206,
  'brinquedo': 0.22908093278463648,
  'game': 0.21330589849108367},
 'vocab_size': 35414}

Que também será salva em um arquivo baseline.json.

Copie este arquivo para a pasta http-api-classificacao-produtos-container-unico (é onde iremos implementar as novas medições).

Durante a inferência (produção)

Agora que temos uma base de comparação, basta, para cada nova requisição:

• Medir essas mesmas características

• Comparar com o baseline do treino

• Expor isso como métricas no Prometheus

Dentro do projeto http-api-classificacao-produtos-container-unico, modifique o arquivo app.py:

from flask import Flask, request, jsonify, g, Response
import pickle
import time
+from prometheus_client import Summary, Gauge, generate_latest
+import json
+import nltk
+from nltk.corpus import stopwords
+from nltk.tokenize import RegexpTokenizer

MODEL_LATENCY = Summary(
    'product_classification_model_inference_seconds',
    'Product classification model inference time'
)

+# --- Métricas de drift ---
+INPUT_LENGTH = Gauge(
+    'product_description_length',
+    'Length of processed description (characters)'
+)

+INPUT_WORD_COUNT = Gauge(
+    'product_description_word_count',
+    'Word count of processed description'
+)

+INPUT_LENGTH_DRIFT = Gauge(
+    'product_description_length_drift',
+    'Difference between processed length and training average'
+)

+INPUT_WORD_COUNT_DRIFT = Gauge(
+    'product_description_word_count_drift',
+    'Difference between processed word count and training average'
+)

app = Flask(__name__)

model = pickle.load(open('model.sav', 'rb'))

+# --- Carrega baseline ---
+with open('baseline.json') as f:
+    BASELINE = json.load(f)

+# --- Inicializa pré-processamento (igual ao treino) ---
+nltk.download('stopwords')

+stop_words = set(stopwords.words("portuguese"))
+tokenizer = RegexpTokenizer(r"\w+")

+def preprocess(text):
+    # lowercase + remoção de stopwords
+    no_stop = ' '.join([
+        word for word in text.lower().split()
+        if word not in stop_words
+    ])

+    # tokenização
+    tokens = tokenizer.tokenize(no_stop)

+    # volta para string (igual ao treino)
+    return ' '.join(tokens)

@app.route('/metrics')
def metrics():
    return Response(generate_latest(), mimetype='text/plain')

@app.route('/predizer_categoria', methods=['POST'])
def predizer_categoria():
    request_data = request.get_json()
-    input_message = [request_data['descricao']]
+    descricao = request_data['descricao']

+    # --- Aplica o MESMO pré-processamento do treino ---
+    processed = preprocess(descricao)

+    # --- Métricas de entrada (já processada) ---
+    desc_len = len(processed)
+    word_count = len(processed.split())

+    INPUT_LENGTH.set(desc_len)
+    INPUT_WORD_COUNT.set(word_count)

+    INPUT_LENGTH_DRIFT.set(desc_len - BASELINE["avg_len"])
+    INPUT_WORD_COUNT_DRIFT.set(word_count - BASELINE["avg_words"])

+    # --- Inferência ---
+    input_message = [processed]

    start = time.time()
    input_message = model["vect"].transform(input_message)
    final_prediction = model["clf"].predict(input_message)[0]
    MODEL_LATENCY.observe(time.time() - start)

    response = {
        'categoria': final_prediction
    }

    return jsonify(response)

Além de coletar algumas das métricas da baseline, aproveitamos e trouxemos o pré-processamento para a inferência também.

Nesse sentido, temos um problema aqui. Cada vez que esse container roda, ele faz o download de dados via nltk. Isso não influencia a inferência em si, mas é um custo desnecessário, uma vez que esse conjunto tende a ser fixo, principalmente neste caso, onde estamos tentando reproduzir algo feito durante treinamento. Assim, uma abordagem mais eficiente seria exportar essas stopwords durante o treinamento, salvar em arquivo e incluir no Dockerfile. Assim, quando o container subir, não é necessário esse custo adicional de acesso à rede. Mas aqui, para simplificar, vamos manter assim mesmo.

Outro ponto digno de nota é como essas métricas representam uma diferença absoluta simples. Apesar de funcionar do ponto de vista didático, pode gerar problemas por não ser normalizado, assim é difícil definir padrões que funcionam entre diferentes métricas. Também não considera a variância. Portanto, em cenários mais avançados, poderíamos normalizar esse valor pelo desvio padrão, obtendo uma medida mais robusta de desvio (ex: z-score).

Vamos atualizar o requirements.txt:

Flask==3.1.3
scikit-learn==1.8.0
gunicorn==25.1.0
prometheus-client==0.24.1 # Para monitoramento com Prometheus
+nltk==3.9.3

E também o Dockerfile:

FROM python:3.13.5-slim

# Instalação do nginx

# Vamos rodar apt update sem precisar interagir. Essa flag diz isso.
ENV DEBIAN_FRONTEND=noninteractive

# Agora vamos instalar nginx, passando a opção -y para aceitar os prompts
# Após a instalação, vamos remover os caches criados pelo apt update,
# Assim nossa imagem fica mais limpa
RUN apt update && apt install -y nginx && rm -rf /var/lib/apt/lists/*

# Vamos copiar os arquivos estáticos para www (pasta configurada para o nginx)
COPY webclient/* /www/

# E agora vamos copiar o arquivo de configuração do nginx
COPY default /etc/nginx/sites-available/default

# A porta 80 será a utilizada pelo nginx. Vamos expô-la
EXPOSE 80/tcp

# Agora vamos às configurações do nosso app original
WORKDIR /usr/src/app

# Aqui vamos copiar os arquivos um a um, pois agora temos outros
# arquivos que não fazem parte da aplicação e não faz sentido copiar
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

COPY model.sav ./
COPY app.py ./
+COPY baseline.json ./

# Agora vamos copiar um script que roda as duas coisas
# nginx e gunicorn

COPY entrypoint.sh ./
RUN chmod 755 ./entrypoint.sh

CMD ["./entrypoint.sh"]

Basta reconstruir a imagem, subir o container novamente, e pronto:

docker build -t http-api-classificacao-produtos-container-unico .

docker run -d -p 8080:80 --network minharede --name http-api-classificacao-produtos-container-unico-container http-api-classificacao-produtos-container-unico

Agora vamos configurar o Grafana para visualizar essas métricas. Porém, não vamos ver nada a não ser que tenhamos uma forma de testar isso.

Simulador de requisições com drift

Crie uma pasta chamada drift-test, e um ambiente virtual drift-test.

pyenv virtualenv 3.10.2 drift-test
pyenv activate drift-test

Ou se estiver no conda:

conda create --name drift-test python=3.13.5
conda activate drift-test

Copie para esta pasta o arquivo produtos.csv e baseline.json que criamos no início desta seção.

Crie o arquivo requirements.txt:

pandas==3.0.1
nltk==3.9.3
requests==2.32.5

Instale os pacotes com: pip install -r requirements.txt (não esqueça de verificar se o ambiente está ativo)

Agora crie o arquivo drift-dados.py:

import pandas as pd
import nltk
import random
import time
import json
import requests
from nltk.corpus import stopwords
from nltk.tokenize import RegexpTokenizer

# ==============================
# 🎛️ CONFIGURAÇÃO DE DRIFT
# ==============================

DRIFT_VOCABULARIO = 0   # percentual de palavras FORA do vocabulário (0 a 100)
DRIFT_TAMANHO = 0       # variação média no número de palavras (+/-)

API_URL = "http://localhost:8080/api/predizer_categoria"

# ==============================
# 📚 SETUP
# ==============================

nltk.download('stopwords')
nltk.download('mac_morpho')

df = pd.read_csv("produtos.csv", delimiter=';', encoding='utf-8')

stop_words = set(stopwords.words("portuguese"))
tokenizer = RegexpTokenizer(r"\w+")

def preprocess(text):
    no_stop = ' '.join([
        word for word in text.lower().split()
        if word not in stop_words
    ])
    return tokenizer.tokenize(no_stop)

# ==============================
# 🧠 VOCABULÁRIOS
# ==============================

print("Construindo vocabulário do dataset...")
dataset_vocab = set()

for text in df["nome"].fillna('') + " " + df["descricao"].fillna(''):
    dataset_vocab.update(preprocess(text))

dataset_vocab = list(dataset_vocab)

print("Carregando léxico externo...")
from nltk.corpus import mac_morpho

external_vocab = list(set(word.lower() for word, _ in mac_morpho.tagged_words()))
external_vocab = [w for w in external_vocab if w.isalpha()]

# remove palavras que já estão no dataset
external_vocab = list(set(external_vocab) - set(dataset_vocab))

# ==============================
# 📊 BASELINE
# ==============================

with open("baseline.json") as f:
    BASELINE = json.load(f)

avg_words = int(BASELINE["avg_words"])

print(f"Média baseline: {avg_words} palavras")

# ==============================
# 🧪 GERADOR CONTROLADO
# ==============================

def generate_text():
    # --- tamanho alvo ---
    base_size = avg_words + DRIFT_TAMANHO

    # pequena variação natural (ruído leve)
    natural_noise = random.randint(-2, 2)

    size = max(1, int(base_size + natural_noise))

    words = []

    for _ in range(size):
        # decisão baseada no drift de vocabulário
        if random.random() < (DRIFT_VOCABULARIO / 100):
            words.append(random.choice(external_vocab))
        else:
            words.append(random.choice(dataset_vocab))

    return " ".join(words)

# ==============================
# 🔁 LOOP
# ==============================

print("Iniciando geração de dados...")

while True:
    try:
        descricao = generate_text()

        payload = {
            "descricao": descricao
        }

        response = requests.post(API_URL, json=payload)

        print(f"Status: {response.status_code} | Drift vocab: {DRIFT_VOCABULARIO}% | Drift tamanho: {DRIFT_TAMANHO}")

        time.sleep(1)

    except Exception as e:
        print("Erro:", e)
        time.sleep(2)

Execute com python drift-dados.py

Visualizar no Grafana

Crie um novo dashboard (nome API classificação de produtos), e dois painéis do tipo Time series:

• Query: product_description_word_count_drift

• Title: Drift instantâneo (palavras)

• Color scheme: From thresholds (by value)

• Thresholds:

  • 25 (vermelho)

  • 10 (laranja)

  • -10 (verde)

  • -25 (laranja)

  • Base (vermelho)

• Show thresholds: As filled regions

• Query: avg_over_time(product_description_word_count_drift[5m])

• Title: Drift média (palavras)

• Color scheme: From thresholds (by value)

• Thresholds:

  • 25 (vermelho)

  • 10 (laranja)

  • -10 (verde)

  • -25 (laranja)

  • Base (vermelho)

• Show thresholds: As filled regions

Execute o simulador de drift com diferentes configurações da variável DRIFT_TAMANHO para ver o resultado.

Drift de vocabulário

Agora vamos ajustar nosso serviço para detectar se estão aparecendo muitas palavras novas. Neste caso, a coleta é simples, pois nosso modelo já tem o vocabulário de treino carregado, basta usar as transformações corretas para detectar novas palavras.

Dentro da pasta http-api-classificacao-produtos-container-unico, modifique o arquivo app.py:

from flask import Flask, request, jsonify, g, Response
import pickle
import time
from prometheus_client import Summary, Gauge, generate_latest
import json
import nltk
from nltk.corpus import stopwords
from nltk.tokenize import RegexpTokenizer

MODEL_LATENCY = Summary(
    'product_classification_model_inference_seconds',
    'Product classification model inference time'
)

# --- Métricas de drift ---
INPUT_LENGTH = Gauge(
    'product_description_length',
    'Length of processed description (characters)'
)

INPUT_WORD_COUNT = Gauge(
    'product_description_word_count',
    'Word count of processed description'
)

INPUT_LENGTH_DRIFT = Gauge(
    'product_description_length_drift',
    'Difference between processed length and training average'
)

INPUT_WORD_COUNT_DRIFT = Gauge(
    'product_description_word_count_drift',
    'Difference between processed word count and training average'
)

+VOCAB_DRIFT = Gauge(
+    'product_description_vocab_drift_percentage',
+    'Percentage of words not seen during training'
+)

app = Flask(__name__)

model = pickle.load(open('model.sav', 'rb'))

+# Carrega vocabulário do modelo
+MODEL_VOCAB = set(model["vect"].vocabulary_.keys())

# --- Carrega baseline ---
with open('baseline.json') as f:
    BASELINE = json.load(f)

# --- Inicializa pré-processamento (igual ao treino) ---
nltk.download('stopwords')

stop_words = set(stopwords.words("portuguese"))
tokenizer = RegexpTokenizer(r"\w+")

def preprocess(text):
    # lowercase + remoção de stopwords
    no_stop = ' '.join([
        word for word in text.lower().split()
        if word not in stop_words
    ])

    # tokenização
    tokens = tokenizer.tokenize(no_stop)

    # volta para string (igual ao treino)
    return ' '.join(tokens)

@app.route('/metrics')
def metrics():
    return Response(generate_latest(), mimetype='text/plain')

@app.route('/predizer_categoria', methods=['POST'])
def predizer_categoria():
    request_data = request.get_json()
    descricao = request_data['descricao']

    # --- Aplica o MESMO pré-processamento do treino ---
    processed = preprocess(descricao)

+    # --- Detectar palavras fora do vocabulário
+    tokens = processed.split()

+    unknown_words = [w for w in tokens if w not in MODEL_VOCAB]

+    if len(tokens) > 0:
+        vocab_drift = (len(unknown_words) / len(tokens)) * 100
+    else:
+        vocab_drift = 0

+    VOCAB_DRIFT.set(vocab_drift)

    # --- Métricas de entrada (já processada) ---
    desc_len = len(processed)
    word_count = len(processed.split())

    INPUT_LENGTH.set(desc_len)
    INPUT_WORD_COUNT.set(word_count)

    INPUT_LENGTH_DRIFT.set(desc_len - BASELINE["avg_len"])
    INPUT_WORD_COUNT_DRIFT.set(word_count - BASELINE["avg_words"])

    # --- Inferência ---
    input_message = [processed]

    start = time.time()
    input_message = model["vect"].transform(input_message)
    final_prediction = model["clf"].predict(input_message)[0]
    MODEL_LATENCY.observe(time.time() - start)

    response = {
        'categoria': final_prediction
    }

    return jsonify(response)

Reconstrua a imagem e rode o container novamente.

No Grafana, adicione mais dois painéis do tipo Time series:

• Query: product_description_vocab_drift_percentage

• Title: Drift instantâneo (vocabulário %)

• Unit: Percent

• Color scheme: From thresholds (by value)

• Thresholds:

  • 25 (vermelho)

  • 10 (laranja)

  • Base (verde)

• Show thresholds: As filled regions

• Query: avg_over_time(product_description_vocab_drift_percentage[5m])

• Title: Drift média (vocabulário %)

• Unit: Percent

• Color scheme: From thresholds (by value)

• Thresholds:

  • 25 (vermelho)

  • 10 (laranja)

  • Base (verde)

• Show thresholds: As filled regions

No drift-test, ajuste a variável DRIFT_VOCABULARIO e veja o resultado nos dois painéis.

Como interpretar e gerar alertas para essas métricas?

Com esses quatro painéis, você está observando o comportamento do seu sistema sob duas perspectivas fundamentais:

• instantânea (tempo real)

• agregada (tendência ao longo do tempo)

E para dois tipos de drift:

• estrutural (tamanho / número de palavras)

• semântico básico (vocabulário)

Essa combinação é extremamente poderosa, pois permite identificar tanto anomalias pontuais quanto mudanças consistentes no comportamento dos dados.

Os painéis de drift instantâneos são úteis para detectar:

• picos abruptos

• inputs fora do padrão

• comportamento inesperado imediato

No entanto, são painéis naturalmente ruidosos. Oscilações rápidas são normais e, isoladamente, não indicam problema. A dica é usar este painel como um “monitor cardíaco”: ele mostra o pulso do sistema, mas não conta a história completa.

Os painéis de média ajudam a responder à seguinte pergunta:

“O comportamento dos dados está mudando de forma consistente?”

Se esse painel começa a se afastar do zero de forma contínua, isso é um forte indicativo de data drift real.

Com base nesses painéis, você pode definir alertas simples e eficazes, sempre com base nos painéis de média, que melhor representam situações de alerta.

Drift de saída

Até agora, nossas métricas estavam focadas em drift de entrada — tamanho das descrições, quantidade de palavras, vocabulário utilizado. Esse tipo de monitoramento tem uma característica importante: ele pode ser feito requisição a requisição, de forma independente. Cada chamada à API traz todas as informações necessárias para calcular a métrica, sem depender de histórico acumulado.

Por isso, até este ponto, conseguimos implementar tudo em memória, dentro do próprio processo da aplicação. Mesmo com múltiplos workers (como no caso do Gunicorn), isso não era um problema relevante, já que cada worker conseguia medir o drift de entrada de forma isolada, sem comprometer a utilidade da métrica.

No entanto, ao avançarmos para o drift de saída, o cenário muda completamente.

Aqui, não estamos mais interessados em uma observação pontual, mas sim em algo agregado:

A distribuição das predições ao longo do tempo está diferente daquela observada no treinamento?

Para responder essa pergunta, precisamos acumular informações:

• quantas vezes cada categoria foi predita

• qual o total de predições realizadas

• como essas proporções evoluem ao longo do tempo

Esse tipo de análise exige estado compartilhado e persistente.

Se continuássemos usando memória local, teríamos alguns problemas sérios:

• Cada worker manteria sua própria contagem, gerando métricas inconsistentes

• Ao reiniciar o container, todo o histórico seria perdido

• Em um ambiente com múltiplas instâncias, não haveria nenhuma visão global do sistema

Em outras palavras, a estratégia anterior deixa de funcionar porque agora o problema deixou de ser local e passou a ser distribuído.

É nesse ponto que introduzimos o Redis.

O Redis funciona como um armazenamento externo, rápido e compartilhado, permitindo que todos os workers da aplicação leiam e escrevam no mesmo conjunto de dados. Com isso, conseguimos:

• manter contadores globais de predições

• calcular distribuições de forma consistente

• preservar o estado mesmo com múltiplos processos

• desacoplar o ciclo de vida da aplicação do ciclo de vida dos dados

Essa mudança, embora simples do ponto de vista técnico, representa um avanço importante na arquitetura:

saímos de um modelo puramente stateless para um modelo com estado controlado e compartilhado.

E esse é um passo fundamental quando começamos a lidar com problemas reais de MLOps, onde muitas métricas dependem não apenas do que acontece agora, mas de como o sistema se comporta ao longo do tempo.

Colocando a mão na massa

Para rodar o Redis, obviamente vamos usar o docker:

docker run -d --name redis --network minharede redis

Pronto. Atenção para o uso da rede aqui.

Agora, na pasta http-api-classificacao-produtos-container-unico, modifique o arquivo requirements.txt:

Flask==3.1.3
scikit-learn==1.8.0
gunicorn==25.1.0
prometheus-client==0.24.1 # Para monitoramento com Prometheus
nltk==3.9.3
+redis==7.3.0

Agora vamos alterar o arquivo app.py para coletar a métrica:

from flask import Flask, request, jsonify, g, Response
import pickle
import time
from prometheus_client import Summary, Gauge, generate_latest
import json
import nltk
from nltk.corpus import stopwords
from nltk.tokenize import RegexpTokenizer
+import redis

MODEL_LATENCY = Summary(
    'product_classification_model_inference_seconds',
    'Product classification model inference time'
)

# --- Métricas de drift ---
INPUT_LENGTH = Gauge(
    'product_description_length',
    'Length of processed description (characters)'
)

INPUT_WORD_COUNT = Gauge(
    'product_description_word_count',
    'Word count of processed description'
)

INPUT_LENGTH_DRIFT = Gauge(
    'product_description_length_drift',
    'Difference between processed length and training average'
)

INPUT_WORD_COUNT_DRIFT = Gauge(
    'product_description_word_count_drift',
    'Difference between processed word count and training average'
)

VOCAB_DRIFT = Gauge(
    'product_description_vocab_drift_percentage',
    'Percentage of words not seen during training'
)

+CATEGORY_DRIFT = Gauge(
+    'product_category_distribution_drift',
+    'Difference between predicted category distribution and training distribution',
+    ['categoria']
+)

+CATEGORY_CURRENT_DIST = Gauge(
+    'product_category_current_distribution',
+    'Current predicted category distribution',
+    ['categoria']
+)

app = Flask(__name__)

model = pickle.load(open('model.sav', 'rb'))

+redis_client = redis.Redis(
+    host='redis',
+    port=6379,
+    decode_responses=True
+)

+WINDOW_SIZE = 200
+REDIS_WINDOW_KEY = "prediction_window"

# Carrega vocabulário do modelo
MODEL_VOCAB = set(model["vect"].vocabulary_.keys())

# --- Carrega baseline ---
with open('baseline.json') as f:
    BASELINE = json.load(f)

# --- Inicializa pré-processamento (igual ao treino) ---
nltk.download('stopwords')

stop_words = set(stopwords.words("portuguese"))
tokenizer = RegexpTokenizer(r"\w+")

def preprocess(text):
    # lowercase + remoção de stopwords
    no_stop = ' '.join([
        word for word in text.lower().split()
        if word not in stop_words
    ])

    # tokenização
    tokens = tokenizer.tokenize(no_stop)

    # volta para string (igual ao treino)
    return ' '.join(tokens)

@app.route('/metrics')
def metrics():
    return Response(generate_latest(), mimetype='text/plain')

@app.route('/predizer_categoria', methods=['POST'])
def predizer_categoria():
    request_data = request.get_json()
    descricao = request_data['descricao']

    # --- Aplica o MESMO pré-processamento do treino ---
    processed = preprocess(descricao)

    # --- Detectar palavras fora do vocabulário
    tokens = processed.split()

    unknown_words = [w for w in tokens if w not in MODEL_VOCAB]

    if len(tokens) > 0:
        vocab_drift = (len(unknown_words) / len(tokens)) * 100
    else:
        vocab_drift = 0

    VOCAB_DRIFT.set(vocab_drift)

    # --- Métricas de entrada (já processada) ---
    desc_len = len(processed)
    word_count = len(processed.split())

    INPUT_LENGTH.set(desc_len)
    INPUT_WORD_COUNT.set(word_count)

    INPUT_LENGTH_DRIFT.set(desc_len - BASELINE["avg_len"])
    INPUT_WORD_COUNT_DRIFT.set(word_count - BASELINE["avg_words"])

    # --- Inferência ---
    input_message = [processed]

    start = time.time()
    input_message = model["vect"].transform(input_message)
    final_prediction = model["clf"].predict(input_message)[0]
    MODEL_LATENCY.observe(time.time() - start)

+    # ==============================
+    # Drift de saída com Redis
+    # ==============================

+    # Adiciona nova predição
+    redis_client.lpush(REDIS_WINDOW_KEY, final_prediction)

+    # Mantém só as últimas N
+    redis_client.ltrim(REDIS_WINDOW_KEY, 0, WINDOW_SIZE - 1)

+    # Agora vamos coletar os dados dentro da janela deslizante
+    window_data = redis_client.lrange(REDIS_WINDOW_KEY, 0, -1)

+    total_predictions = len(window_data)

+    if total_predictions > 0:
+        category_counts = {}

+        # Conta manualmente
+        for categoria in window_data:
+            category_counts[categoria] = category_counts.get(categoria, 0) + 1

+        for categoria in BASELINE["category_distribution"].keys():
+            count = category_counts.get(categoria, 0)

+            current = count / total_predictions
+            baseline = BASELINE["category_distribution"].get(categoria, 0)

+            drift = (current - baseline) * 100

+            CATEGORY_CURRENT_DIST.labels(categoria=categoria).set(current * 100)
+            CATEGORY_DRIFT.labels(categoria=categoria).set(drift)

    response = {
        'categoria': final_prediction
    }

    return jsonify(response)

Note os cálculos de classificação e drift, usando o Redis como armazenamento. Note também como a estratégia adotada é a de considerar apenas uma janela deslizante baseada na quantidade de requisições, de forma que as medidas sejam refletidas de acordo com tendências atuais. Por outro lado, essa abordagem não captura tendências de longo prazo. Em sistemas reais, é comum combinar múltiplas janelas (curta e longa) para obter uma visão mais completa.

Outro ponto que vale a pena comentar é que, em cenários de alta escala, esse cálculo pode ser deslocado para um processo assíncrono, evitando custo adicional em cada requisição associado não apenas ao cálculo mas também o acesso ao Redis.

Construa a imagem e suba o container novamente.

Antes de ir ao Grafana, vamos criar um novo simulador de drift, agora de saída. Na pasta drift-test, crie o arquivo drift-saida.py:

import pandas as pd
import random
import time
import requests

CATEGORY_DISTRIBUTION = {
    "livro": 0.25,
    "game": 0.25,
    "maquiagem": 0.25,
    "brinquedo": 0.25
}

API_URL = "http://localhost:8080/api/predizer_categoria"

df = pd.read_csv("produtos.csv", delimiter=';', encoding='utf-8')

# junta nome + descrição (igual ao treino)
df["texto"] = df["nome"].fillna('') + " " + df["descricao"].fillna('')

# agrupa por categoria
grouped = {
    categoria: group["texto"].tolist()
    for categoria, group in df.groupby("categoria")
}

categorias = list(CATEGORY_DISTRIBUTION.keys())
probabilidades = list(CATEGORY_DISTRIBUTION.values())

def generate_text():
    # escolhe categoria com base no drift configurado
    categoria = random.choices(categorias, weights=probabilidades, k=1)[0]

    # escolhe um exemplo real daquela categoria
    texto = random.choice(grouped[categoria])

    return texto, categoria

print("Iniciando simulação de drift de saída...")

while True:
    try:
        descricao, categoria_esperada = generate_text()

        payload = {
            "descricao": descricao
        }

        response = requests.post(API_URL, json=payload)

        try:
            data = response.json()
            categoria_predita = data.get("categoria", "N/A")
        except:
            categoria_predita = "erro"

        print(
            f"Status: {response.status_code} | "
            f"Esperado: {categoria_esperada} | "
            f"Predito: {categoria_predita} | "
            f"Distribuição alvo: {CATEGORY_DISTRIBUTION}"
        )

        time.sleep(0.5)

    except Exception as e:
        print("Erro:", e)
        time.sleep(2)

A variável CATEGORY_DISTRIBUTION é a que iremos utilizar para gerar a distribuição desejada. Veja como esse código usa exemplos do próprio dataset de treinamento para ter esse controle.

Agora podemos ir ao Grafana para criar dois painéis.

• Tipo: Bar chart

• Query: product_category_current_distribution

• Legend: {{categoria}}

• Title: Distribuição da predição

• X-axis labels minimum spacing Small

• Stacking: Stacking

• Gradient mode: Opacity

• Unit: Percent

Esse painel mostra a linha do tempo das distribuições, em percentagem, de maneira empilhada.

• Tipo: Time series

• Query: sum(abs(product_category_distribution_drift)) / 2

• Title: Drift de saída (predição de categoria)

• Unit: Percent

• Color scheme: From thresholds (by value)

• Thresholds:

  • 25 (vermelho)

  • 10 (laranja)

  • Base (verde)

• Show thresholds: As filled regions

Agora vale a pena explicar com um pouco mais de profundidade o que está sendo calculado nessa query, pois ela condensa toda a informação de drift em um único indicador.

A expressão:

sum(abs(product_category_distribution_drift)) / 2

realiza três etapas importantes:

1. Cálculo do desvio por categoria Cada série product_category_distribution_drift{categoria="..."} já representa a diferença entre a distribuição atual e a distribuição de treino para aquela categoria específica. Esse valor pode ser:

• positivo (a categoria está super-representada), ou

• negativo (sub-representada)

1. Conversão para magnitude (valor absoluto) Ao aplicar abs(...), ignoramos o sinal do desvio e consideramos apenas sua intensidade. Isso é necessário porque, em uma distribuição, aumentos e reduções sempre se compensam — a soma simples dos drifts seria sempre zero.

2. Agregação global e normalização A função sum(...) soma os desvios absolutos de todas as categorias, produzindo uma medida do quanto a distribuição total se afastou do baseline.

Entretanto, esse valor ainda está duplicado, pois todo aumento em uma categoria corresponde necessariamente a uma diminuição em outra. Por isso, dividimos o resultado por 2:

desvio real = soma dos desvios absolutos / 2

O valor final pode ser interpretado como:

a fração total da distribuição que precisou ser “redistribuída” para sair do baseline e chegar ao estado atual.

Por exemplo:

• 0% → nenhuma mudança

• 10% → pequena alteração na distribuição

• 25% → mudança significativa

• 50% → comportamento fortemente diferente do treino

Essa métrica é equivalente, em termos estatísticos, à chamada distância de variação total (Total Variation Distance), amplamente utilizada para comparar distribuições de probabilidade.

Ela é particularmente útil em MLOps porque:

• produz um único valor facilmente monitorável

• é sensível a mudanças reais no comportamento do modelo

• pode ser usada diretamente para definição de alertas

Dessa forma, esse painel não mostra apenas valores isolados por categoria, mas sim um indicador global de mudança no comportamento do modelo ao longo do tempo.

Simulando drift de saída e observando os resultados no Grafana

Veja como os dois gráficos mostram que as categorias estão equilibradas, e que o drift em relação aos dados de treino é pequeno (o dataset original estava razoavelmente bem balanceado).

Vamos mudar isso. No drift-test, no arquivo drift-saida.py, altere a distribuição:

CATEGORY_DISTRIBUTION = {
    "livro": 0.5,
    "game": 0.15,
    "maquiagem": 0.15,
    "brinquedo": 0.20
}

Veja como, aos poucos, o drift começa a ser percebido no Grafana. Tente alterar os valores para ver o resultado.

Neste caso, não foi necessário calcular a média no Grafana porque a forma de cálculo já considera uma média dos últimos valores lidos apenas. Mas se quiser atenuar os ruídos e eliminar picos instantâneos, é possível acrescentar a média no próprio Grafana.

Assim que estiver satisfeito com o resultado, exporte o novo dashboard e provisione-o junto com os demais, via Docker.

Considerações finais

Para encerrar, é fundamental destacar uma distinção conceitual que frequentemente gera confusão: drift não é sinônimo de qualidade do modelo.

As métricas de drift, como as que construímos neste capítulo, têm como objetivo detectar mudanças no comportamento dos dados ou das predições ao longo do tempo. No caso do drift de saída, estamos observando como a distribuição das categorias previstas pelo modelo se afasta daquela observada durante o treinamento. Isso nos dá um excelente indicativo de que “algo mudou” — seja no perfil dos dados de entrada, no contexto de uso ou até mesmo no próprio ambiente de negócio.

No entanto, essa detecção de mudança não implica, por si só, que o modelo está acertando mais ou menos. Um modelo pode apresentar baixo drift e ainda assim estar cometendo erros sistemáticos, caso o mundo real tenha evoluído sem que isso se reflita imediatamente na distribuição das predições. Da mesma forma, um aumento no drift não significa necessariamente uma piora: pode indicar, por exemplo, que o modelo está se adaptando a um novo padrão legítimo de dados.

Esse cenário nos leva a um conceito mais amplo: o chamado concept drift, em que a relação entre entrada e saída muda ao longo do tempo. Nesse caso, mesmo que a distribuição das entradas ou das predições não se altere drasticamente, o significado dessas relações pode ter mudado — e apenas métricas de desempenho, como acurácia ou F1-score, seriam capazes de capturar esse efeito.

Por esse motivo, em ambientes de MLOps maduros, as métricas de drift devem ser interpretadas como sinais de alerta, e não como diagnósticos finais. Elas indicam quando vale a pena investigar mais a fundo, reavaliar o modelo ou até iniciar um processo de re-treinamento. Idealmente, essas métricas são utilizadas em conjunto com indicadores de performance monitorados continuamente, formando uma visão mais completa do comportamento do sistema.

Em última análise, monitorar sistemas de Machine Learning não é apenas observar números, mas interpretar sinais. Drift, latência e distribuição são pistas — e cabe ao engenheiro entender quando essas pistas indicam uma simples variação natural ou o início de um problema real.

Nota sobre o desenvolvimento do material

O conteúdo desta seção contou com o apoio do ChatGPT como ferramenta auxiliar na estruturação de explicações, revisão de conceitos e refinamento técnico. Todas as decisões de conteúdo, exemplos e direcionamento didático foram conduzidas pelo autor.