Observabilidade em Sistemas de IA: Como Monitorar o que Você Não Consegue Prever

Sistemas de IA falham de formas que sistemas tradicionais não falham. Um microsserviço que retorna HTTP 200 está funcionando — por definição. Um LLM que retorna texto coerente pode estar completamente errado, alucinando fatos, violando restrições de negócio ou degradando gradualmente em qualidade sem nenhum erro explícito.

Essa assimetria entre aparência e correção é o problema central de observabilidade em IA. As ferramentas de APM tradicionais medem latência e throughput com precisão, mas são cegas para a dimensão que mais importa: o output faz sentido?

Os Quatro Pilares da Observabilidade em IA

1. Métricas de Runtime

Latência, throughput, taxa de erro, custo por token. As métricas convencionais aplicam-se e são necessárias, mas insuficientes.

from opentelemetry import trace, metrics
meter        = metrics.get_meter("ai.pipeline") llm_latency  = meter.create_histogram("llm.request.duration", unit="ms") token_counter = meter.create_counter("llm.tokens.consumed") quality_gauge = meter.create_gauge("llm.response.quality_score")
def instrumented_llm_call(prompt: str, context: dict) -> str:     with tracer.start_as_current_span("llm.call") as span:         span.set_attribute("llm.model",       context.get("model"))         span.set_attribute("llm.temperature", context.get("temperature", 0.1))         start    = time.perf_counter()         response = llm.invoke(prompt)         llm_latency.record((time.perf_counter() - start) * 1000)         token_counter.add(response.usage.total_tokens)         return response.content

2. Trace Distribuído de Prompts

Em sistemas multi-agent, uma única requisição pode disparar dezenas de chamadas LLM. Sem distributed tracing, é impossível saber qual chamada causou uma resposta ruim.

class TracedLLMChain:
    def invoke(self, inputs: dict) -> dict:
        with self.tracer.start_as_current_span(f"llm_chain.{self.chain_name}") as span:
            prompt = self._build_prompt(inputs)
            span.set_attribute("chain.prompt_hash",
                hashlib.md5(prompt.encode()).hexdigest())
            response = self.llm.invoke(prompt)
            span.set_attribute("chain.finish_reason",
                response.response_metadata.get("finish_reason"))
            return {"output": response.content}

O prompt_hash é fundamental: permite identificar se o mesmo prompt está produzindo respostas diferentes ao longo do tempo — o que indica model drift.

3. Avaliação Contínua de Qualidade

Esta é a dimensão que sistemas tradicionais não têm. A abordagem: um LLM Judge — um segundo modelo que avalia as respostas do modelo principal.

class QualityScore(BaseModel):
    dimension: str
    score:     float
    reasoning: str
    flags:     list[str] = []
async def evaluate_response(question: str, context: str, response: str) -> list[QualityScore]:     judge_response = await judge_llm.ainvoke(         JUDGE_PROMPT.format(question=question, context=context, response=response)     )     scores = parse_quality_scores(judge_response.content)     for score in scores:         quality_gauge.set(score.score, {"dimension": score.dimension})     return scores

4. Drift Detection

O problema mais insidioso em sistemas de IA de longa duração é o drift: a qualidade degrada gradualmente, sem erros explícitos, até que usuários começam a reclamar.

Drift tem duas fontes principais: data drift (a distribuição das queries muda) e model drift (o provider atualiza o modelo silenciosamente).

from scipy import stats
from collections import deque
class DriftDetector:     def __init__(self, window_size: int = 500, significance: float = 0.05):         self.window_size  = window_size         self.significance = significance         self.baseline     = deque(maxlen=window_size)         self.current      = deque(maxlen=window_size)
def add_sample(self, score: float) -> bool:         if len(self.baseline) < self.window_size:             self.baseline.append(score)             return False         self.current.append(score)         if len(self.current) < 50:             return False         _, p_value = stats.ks_2samp(list(self.baseline), list(self.current))         if p_value < self.significance:             self.baseline = deque(list(self.current), maxlen=self.window_size)             self.current.clear()             return True         return False

O teste Kolmogorov-Smirnov detecta diferenças na distribuição sem assumir normalidade — ideal para métricas de qualidade.

O Dashboard que Realmente Importa

Em produção, seis métricas em tempo real:

P95 de qualidade por dimensão — não a média. A média esconde os piores casos.

Taxa de drift alerts por hora — tendência de aumento indica problema sistêmico.

Custo por sessão bem-sucedida — sessões com baixa qualidade estão desperdiçando orçamento.

Distribuição de finish_reason — aumento de max_tokens indica respostas sendo truncadas.

Latência do agente mais lento — o gargalo do pipeline inteiro.

Taxa de escalação para human_review — se aumenta, os agentes estão menos confiantes.

Observabilidade em sistemas de IA não é uma extensão de observabilidade tradicional. É uma disciplina separada que requer instrumentação intencional desde o design. Um sistema de IA sem essas camadas não está em produção — está em beta permanente.