12-03-04 — Drift Detection e Regressões: Data Drift, Model Drift, Concept Drift

O problema: modelos apodrecem sem avisar

Um modelo treinado em dados de 2024 começa a atender usuários em 2025. O mundo mudou, o vocabulário mudou, as expectativas mudaram. O modelo não sabe disso. Ele continua respondendo como sempre — só que cada vez mais errado.

Os três tipos de drift

1. Data Drift (Covariate Shift)

A distribuição dos dados de entrada muda. Os usuários passaram a fazer perguntas diferentes, usar vocabulário diferente, ou chegam de um canal novo com um perfil diferente.

Exemplos práticos:

• Chatbot de suporte técnico: usuários começam a perguntar sobre uma versão nova do produto que o modelo nunca viu no treino.
• RAG corporativo: a base de conhecimento foi atualizada mas o modelo de embedding não foi retreinado — embeddings antigos e novos estão em espaços diferentes.
• Sistema de classificação: nova categoria de ticket surge organicamente e o modelo não tem label para ela.

2. Model Drift (Performance Drift)

O modelo em si passa a performar pior — não necessariamente porque os dados mudaram, mas porque o próprio modelo mudou (atualização do provider, quantização diferente, mudança de infraestrutura).

3. Concept Drift (Label Drift)

A relação entre input e output "correto" muda. O que era uma boa resposta antes não é mais uma boa resposta hoje — porque o contexto do negócio, as regulações, ou as preferências do usuário mudaram.

Exemplo: Um assistente jurídico que estava respondendo com base na legislação de 2023. Em 2025, nova lei entrou em vigor. As respostas ainda parecem coerentes, ainda têm boa latência — mas estão erradas sob a nova legislação.

Como detectar cada tipo

Detectando Data Drift

Compare a distribuição estatística dos inputs atuais vs. a distribuição de referência (janela de treino ou janela estável).

from scipy.stats import ks_2samp, chi2_contingency
import numpy as np

def detect_data_drift(reference_embeddings, current_embeddings, threshold=0.05):
    """
    Usa KS test para comparar distribuição de embeddings.
    Retorna True se drift for detectado.
    """
    # Compara cada dimensão do embedding (ou componentes PCA)
    p_values = []
    for dim in range(reference_embeddings.shape[1]):
        stat, p_value = ks_2samp(
            reference_embeddings[:, dim],
            current_embeddings[:, dim]
        )
        p_values.append(p_value)
    
    # Drift detectado se mais de X% das dimensões mostram p < threshold
    drift_dims = sum(p < threshold for p in p_values)
    drift_ratio = drift_dims / len(p_values)
    
    return drift_ratio > 0.1  # mais de 10% das dimensões driftaram

def detect_topic_drift(reference_topics, current_topics):
    """Compara distribuição de tópicos usando chi-quadrado"""
    contingency = np.array([reference_topics, current_topics])
    stat, p_value, dof, expected = chi2_contingency(contingency)
    return p_value < 0.05
  

Detectando Model Drift

Mantenha um golden test set: um conjunto fixo de queries com respostas esperadas. Execute periodicamente e monitore a taxa de acerto.

class ModelDriftMonitor:
    def __init__(self, golden_set: list[dict], llm_judge):
        self.golden_set = golden_set  # [{"query": ..., "expected": ..., "rubric": ...}]
        self.judge = llm_judge
        self.baseline_score = None
    
    def calibrate(self):
        """Roda na versão estável para estabelecer baseline"""
        scores = [self._eval(item) for item in self.golden_set]
        self.baseline_score = sum(scores) / len(scores)
        return self.baseline_score
    
    def check_drift(self, alert_threshold=0.05) -> dict:
        """Detecta regressão comparando com baseline"""
        current_scores = [self._eval(item) for item in self.golden_set]
        current_score = sum(current_scores) / len(current_scores)
        
        delta = self.baseline_score - current_score
        has_regression = delta > alert_threshold
        
        return {
            "baseline": self.baseline_score,
            "current": current_score,
            "delta": delta,
            "regression_detected": has_regression,
            "failed_items": [
                self.golden_set[i] for i, s in enumerate(current_scores)
                if s < 0.5  # itens individuais com falha
            ]
        }
    
    def _eval(self, item) -> float:
        response = self.model.call(item["query"])
        return self.judge.score(response, item["expected"], item["rubric"])
  

Detectando Concept Drift

Mais difícil — requer signal humano ou proxy de negócio. Estratégias:

• Annotation pipeline: Amostragem periódica de conversas para avaliação humana. Compara com avaliações históricas.
• Business proxy metrics: Taxa de escalação para humano, taxa de abandono pós-resposta, resolução no primeiro contato.
• Automated fact-checking: Para domínios factuais, use um segundo modelo como juiz e monitore a concordância.

Ferramentas de drift detection

Pipeline de resposta ao drift

1. Detectar: Monitoramento automático dispara alerta.
2. Classificar: É data drift, model drift ou concept drift? A causa determina a solução.
3. Quantificar impacto: Quais usuários? Qual fração do tráfego? Qual métrica foi afetada?
4. Remediar:
   • Data drift → atualizar embedding model, retreinar retrieval, ajustar prompt com mais contexto.
   • Model drift → rollback de versão, troca de provider, ajuste de parâmetros.
   • Concept drift → retreinamento ou fine-tuning com dados novos, atualização da base de conhecimento.
5. Validar: Confirmar resolução com A/B test ou golden set.