Degradação graciosa em produtos com IA: o que acontece quando o modelo some

Existe uma pergunta que separa integrações de IA maduras das improvisadas, e ela cabe em uma linha:

Se o provedor de modelo ficar indisponível por duas horas, o que para de funcionar?

Se a resposta for "a triagem automática fica desligada e os tickets entram sem classificação", você tem um produto resiliente. Se for "o POST /tickets retorna 500", você tem uma dependência crítica que ninguém decidiu conscientemente criar.

A segunda resposta é a mais comum, e quase nunca é intencional. Ela aparece porque o modelo foi adicionado dentro do caminho da requisição, e nada no código diz o que fazer quando ele não responde.

Este artigo é sobre construir a primeira resposta de propósito.

TL;DR

• Classifique cada uso de IA como enriquecimento, assistência ou crítico. A política de falha decorre da classe.
• Enriquecimento falha aberto: o dado principal entra sem a avaliação.
• Crítico falha fechado: sem julgamento confiável, exige humano ou bloqueia.
• Timeout vem do p95 medido, não de um número redondo escolhido na reunião.
• Circuit breaker evita que uma indisponibilidade vire fila de requisições lentas.
• Fallback precisa ser tipado igual ao caminho feliz, senão a degradação vira bug de contrato.
• O teste que importa não é unitário: é desligar o provider em staging e ver o que acontece.

Passo 1: classifique o uso, não o modelo

A mesma API pode ser enriquecimento em um fluxo e crítica em outro. A classificação é por uso.

A regra por trás disso é antiga e continua valendo: falhe aberto quando a ausência do recurso é um inconveniente; falhe fechado quando a ausência é um risco.

Um erro frequente é tratar moderação como enriquecimento. Se o classificador de conteúdo cai e o sistema publica tudo por padrão, você não construiu degradação graciosa — construiu um bypass de segurança com aparência de resiliência.

from enum import Enum

class Criticality(str, Enum):
    ENRICHMENT = "enrichment"     # ausência é inconveniente
    ASSISTANCE = "assistance"     # ausência é invisível
    CRITICAL = "critical"         # ausência é risco

FALLBACK_POLICY = {
    Criticality.ENRICHMENT: "proceed_without",
    Criticality.ASSISTANCE: "proceed_silent",
    Criticality.CRITICAL: "require_human",
}

Coloque isso em configuração, ao lado da feature. Quando alguém adicionar uma chamada nova, a classe é uma decisão explícita, não um esquecimento.

Passo 2: timeout derivado do p95, não do hábito

timeout=30 é o número que as pessoas escrevem quando não mediram nada. Ele é longo demais para o usuário e curto demais para o pior caso legítimo.

O timeout correto sai da distribuição real:

timeout = p95_observado × 1.5,  limitado pelo orçamento de latência do endpoint

Se o p95 da avaliação é 700 ms e o POST /tickets tem orçamento total de 1,5 s, o timeout fica em ~1,05 s — e não 30 s. A chamada que passar disso não vale a pena esperar: o usuário já foi embora.

import httpx

class Timeouts:
    """Timeouts derivados de medição, não de hábito."""
    CONNECT = 1.0          # handshake deve ser rápido ou não vai acontecer
    READ = 1.05            # p95 (700ms) x 1.5
    TOTAL = 2.0            # teto absoluto incluindo retry

    @classmethod
    def httpx(cls) -> httpx.Timeout:
        return httpx.Timeout(connect=cls.CONNECT, read=cls.READ, write=1.0, pool=1.0)

Separe connect de read. Falha de conexão é quase sempre rápida e indica indisponibilidade; leitura lenta indica sobrecarga. Tratar os dois com o mesmo número esconde qual dos dois está acontecendo.

E defina um deadline total que inclua os retries. Sem ele, três tentativas com backoff transformam um timeout de 1 s em 7 s de espera.

import time

class Deadline:
    def __init__(self, budget_s: float):
        self.expires_at = time.monotonic() + budget_s

    @property
    def remaining(self) -> float:
        return max(0.0, self.expires_at - time.monotonic())

    def expired(self) -> bool:
        return self.remaining <= 0.0

Passo 3: circuit breaker, para não gastar timeout à toa

Quando o provedor está fora, cada requisição paga o timeout inteiro antes de falhar. Com tráfego, isso vira fila: threads presas, pool esgotado, e a lentidão contamina endpoints que nem usam IA.

O circuit breaker corta isso. Depois de N falhas consecutivas, ele para de tentar por um período e falha imediatamente.

import time
from enum import Enum

class State(str, Enum):
    CLOSED = "closed"        # operação normal
    OPEN = "open"            # falhando rápido, sem chamar
    HALF_OPEN = "half_open"  # testando uma requisição

class CircuitBreaker:
    def __init__(self, threshold: int = 5, recovery_s: float = 30.0):
        self.threshold = threshold
        self.recovery_s = recovery_s
        self.failures = 0
        self.opened_at = 0.0
        self.state = State.CLOSED

    def allow(self) -> bool:
        if self.state is State.OPEN:
            if time.monotonic() - self.opened_at >= self.recovery_s:
                self.state = State.HALF_OPEN     # deixa uma passar para testar
                return True
            return False
        return True

    def record_success(self) -> None:
        self.failures = 0
        self.state = State.CLOSED

    def record_failure(self) -> None:
        self.failures += 1
        if self.state is State.HALF_OPEN or self.failures >= self.threshold:
            self.state = State.OPEN
            self.opened_at = time.monotonic()
            metrics.increment("llm.circuit_opened")

Dois detalhes que costumam sair errado:

Só conte como falha o que é falha de disponibilidade. Timeout, 5xx, 429 e erro de conexão contam. Uma resposta válida que não passou na validação semântica não conta — ela indica problema de contrato, não de saúde do provedor, e abrir o circuito por isso mascara o bug real.

O estado precisa ser compartilhado em múltiplas instâncias. Um breaker em memória de processo significa que dez réplicas abrem o circuito dez vezes, cada uma pagando suas cinco falhas. Em escala, use Redis com TTL.