Observar e avaliar RAG: um trace por requisição e um conjunto dourado

Lede. Trocar o modelo de embedding, mexer no corte, ajustar o prompt: três mudanças, e você não sabe qual delas melhorou — sabe apenas que a resposta ficou melhor "naquele dia". Este artigo monta a instrumentação que transforma isso em número: um trace por requisição, um conjunto dourado de perguntas com resposta esperada, e um alerta que avisa antes do usuário. E começa por um aviso: a API do Langfuse mudou, e o código que quase todo tutorial mostra está deprecado. Onde você está na linha. Este é o passo 11 de 13 — observar e avaliar. Antes dele: o 01, que dá o mapa e os contratos, e os artigos 02 a 04. Depois dele: o 12 · Guia de decisão e o 13 · Do protótipo à produção. Se você já tem coverage_at_k rodando (artigo 01, seção 7), pule para a seção 7.

---

1. "Melhorou" é opinião até existir com o que comparar

Duas pessoas mexem no mesmo sistema. Uma troca o modelo de vetor, a outra muda o

tamanho do corte. Na sexta-feira a resposta melhorou. Na segunda seguinte alguém

troca a biblioteca de busca e, três dias depois, a resposta piorou — e ninguém

sabe qual das três mudanças causou o quê.

Isso não é falta de talento. É falta de contra-fato: sem um registro do que o

sistema fez, "melhorou" é uma impressão, e a única evidência é o relato de um

usuário. Ou seja: a sua primeira versão de observabilidade é escrita por quem

reclamou, e chega tarde.

A primeira coisa a construir não é um painel. É o trace: o registro de uma

requisição inteira, com o que entrou, o que saiu e as etapas do meio, cada uma

aninhada dentro da anterior. A analogia que funciona é o envelope de uma transação

bancária — protocolo, valor, horário e a trilha de quem autorizou o quê, sem a

qual ninguém concerta o problema do cartão. Cada etapa dentro do trace é um

span: um pedaço nomeado, com pai, com o que recebeu e com o que devolveu.

Sem o trace por etapa, "a resposta está ruim" tem sete causas e nenhuma pista.

Com ele, você lê a sequência: entrou 1.842 documentos, saíram 3 trechos, o terceiro

foi cortado pelo Budget, o modelo recebeu 3.180 caracteres e respondeu sem citar

nada.

⚠️ Trace sem etapa é um log de texto, e log de texto não localiza defeito
Registrar uma linha por requisição com a pergunta e a resposta serve para

auditoria e não serve para diagnóstico: a linha não diz se o trecho certo foi

recuperado, se o corte tem a informação, ou se o modelo recebeu contexto. O

valor do trace está no aninhamento — é ele que permite perguntar "em qual

etapa o trecho sumiu".

---

2. As sete peças, e o que cada uma precisa deixar no trace

Cada peça do pipeline, do artigo 01, produz um

número. O trace existe para que esse número esteja disponível depois, junto com os

vizinhos. Antes de montar a instrumentação, vale decidir o que é identificador e o que é texto inteiro: a lista abaixo é de números, e texto só entra quando a

peça é a geração, e mesmo assim com filtro.

Peça	O que registrar	A pergunta que esse número responde
Entrada	documentos lidos, ignorados e o motivo	"o sistema parou de ver a base?"
Corte	número de trechos, tamanho médio em token, tamanho máximo	"o corte está esmagando a tabela?"
Vetor	nome e versão do modelo, dimensão	"a busca piorou depois do deploy?"
Índice	campos do filtro, valores do escopo, k pedido	"o filtro entrou na consulta?"
Recuperação	ref_id, score e a origem do score de cada trecho	"o trecho certo estava na lista?"
Montagem	o que o Budget gastou, por tipo de bloco	"o que saiu do contexto por estourar?"
Geração	modelo, tamanho da entrada, tamanho da saída, citações	"a resposta usou o que recebeu?"
Avaliação	os escores de cada caso do conjunto dourado	"a última mudança mexeu no quê?"

Duas colunas se complementam em vez de se repetir. O identificador do trecho

recuperado responde "chegou até a montagem"; a citação na resposta responde

"chegou até a geração". Quando a primeira está certa e a segunda não, o defeito

está na instrução de citar, não na busca. E a avaliação entra como etapa do trace

como qualquer outra, o que permite rodar o conjunto dourado dentro do mesmo

registro. A seção 4 mostra essa árvore com código que roda.

---

3. A v4 do Langfuse: por que o tutorial antigo não roda

A maior parte do material do Langfuse na internet mostra a API v2: um

cliente explícito, com Langfuse().trace(), .span() e .generation() chamados

à mão. Essa interface está deprecada, e o aviso está no

repositório do SDK. A versão atual

do SDK Python é a v4, construída sobre

OpenTelemetry e estável

desde março de 2026. A assinatura usada a partir daqui está no

guia do SDK.

A mudança não é cosmética. Na v4 não existe mais a distinção entre "criar um

trace" e "criar um span": o decorator @observe marca a função como span raiz,

e as etapas aninhadas são filhas dele. Não existe mais update_trace() — está

deprecado — e o contexto da requisição (user_id, session_id, tags) é

propagado por mecanismo próprio, sem você passar o objeto do Langfuse por

parâmetro. O que continua igual é o essencial: a plataforma é a mesma, o painel é

o mesmo, e o modelo mental de "uma requisição, com etapas dentro" não mudou. O que

mudou é o nome das funções.

Três fatos de versão decidem se a sua instalação funciona, e nenhum deles

aparece no tutorial:

• Servidor self-hosted precisa da versão 3.63.0 ou superior para a API de

  observations que o SDK v4 usa, e a exigência está na página de

  instalação self-hosted — diferente da

  exigência do SDK.

• Langfuse Cloud corta o v3 em 2026-11-16, junto com a ingestão legada. Se o

  seu código é v2, essa data é o prazo.

• Escores da API v3 exigem SDK Python 4.8.1 ou superior. É o detalhe que

  morde quando o problema é a avaliação, e não a instrumentação: observar funciona,

  o escore falha com erro de versão, e o erro não diz "atualize".

⚠️ Copiar tutorial antigo é o erro mais provável deste artigo
O sintoma é um ImportError numa linha que "vinha funcionando", ou um aviso de

depreciação que ninguém lê. A defesa é uma linha no requirements.txt com

langfuse>=4.8.1 fixado, e a checagem de versão no CI. Nenhuma linha do código

antigo sobrevive sem edição — a forma dos chamadas é outra.

---

4. Instrumentação: o decorator, a geração e os atributos

Aqui está a v4 de verdade, na forma que vai para o seu código. Ela não roda neste artigo: o pacote não está instalado no ambiente onde o texto é

verificado, e instalar uma biblioteca para mostrar uma assinatura seria falso

teste. O que importa é a forma das chamadas.

def instrumentar_requisicao(pergunta: str, tenant: str, session_id: str) -> str:
    """Como fica a instrumentação de uma requisição, na v4.

    Não é chamada neste artigo: `langfuse` não está instalado no ambiente de
    verificação, e o ponto do bloco é a forma da API, não a execução. Para usar:
    `pip install langfuse` e as variáveis LANGFUSE_PUBLIC_KEY,
    LANGFUSE_SECRET_KEY e LANGFUSE_BASE_URL.
    """
    # `get_client()` devolve o cliente singleton da biblioteca, já configurado
    # pelas variáveis de ambiente. Instanciar o cliente na mão é erro.
    from langfuse import get_client, observe, propagate_attributes

    langfuse = get_client()

    @observe(name="consulta")
    @propagate_attributes(user_id=tenant, session_id=session_id, tags=["rag"])
    def responder(pergunta: str) -> str:
        # `propagate_attributes` carrega o contexto da requisição (quem
        # pergunta, de qual sessão, com que etiquetas) sem passar o cliente
        # por parâmetro.
        #
        # `as_type` escolhe o tipo da observation: "span" para etapa de
        # código, "generation" para a chamada ao modelo, "event" para um ponto
        # sem entrada e saída (uma citação, um erro).
        with langfuse.start_as_current_observation(
            as_type="span", name="retrieve"
        ) as busca:
            busca.update(input={"pergunta": pergunta, "k": 10,
                                "scope": {"tenant": tenant}})
            # `output` recebe o que a busca devolveu, com a origem do número
            # -- nunca o texto inteiro da base.
            busca.update(output=[{"ref_id": "c17", "score": 0.82, "source": "dense"}])

        with langfuse.start_as_current_observation(
            as_type="generation", name="generate", model="modelo-escolhido"
        ) as geracao:
            geracao.update(input={"caracteres": 3180})
            resposta = "..."
            geracao.update(output=resposta)
        return resposta

    return responder(pergunta)

Três detalhes merecem nome, porque são os que não estão em tutorial antigo.

Primeiro: o nó decorado é raiz, então a função que você decora é o topo da lista

de traces, e não uma etapa dentro dele. Segundo: update() recebe input e

output no começo e no fim da etapa — registrar só a saída deixa de fora a

metade importante. Terceiro: a identificação da requisição sai do

propagate_attributes; a deprecada update_trace() é o caminho antigo para isso,

e é por isso que ela não aparece aqui.

Agora a parte que roda. Para ver a estrutura do trace antes de ter servidor

nenhum — para entender a forma, ou para afirmar coisas sobre o registro dentro de

um teste —, o formato é uma lista de etapas com pai. É o que o Langfuse guarda:

from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any, Iterator


@dataclass
class Registro:
    """Uma etapa do trace. `pai` é o id de quem a abriu, ou None na raiz."""

    id: str
    tipo: str  # "span" | "generation" | "event"
    nome: str
    pai: str | None
    entrada: Any = None
    saida: Any = None
    erro: str | None = None


class Tracer:
    """A mesma forma que o Langfuse guarda, sem depender de servidor."""

    def __init__(self) -> None:
        self.registros: list[Registro] = []
        self._abertos: list[str] = []

    @contextmanager
    def span(self, tipo: str, nome: str) -> Iterator[Registro]:
        # O pai é a última etapa ainda aberta: é isso que faz o aninhamento.
        registro = Registro(str(len(self.registros)), tipo, nome,
                            self._abertos[-1] if self._abertos else None)
        self.registros.append(registro)
        self._abertos.append(registro.id)
        try:
            yield registro
        except Exception as erro:  # noqa: BLE001
            # O erro vai para dentro do trace: a etapa que quebrou fica
            # visível ao lado das que funcionaram.
            registro.erro = f"{type(erro).__name__}: {erro}"
            raise
        finally:
            self._abertos.pop()

    def render(self) -> str:
        """Desenha a árvore. É o que você olha quando a resposta saiu errada."""
        por_id = {r.id: r for r in self.registros}
        linhas: list[str] = []
        for r in self.registros:
            # A profundidade é a contagem de ancestrais: mais barata que
            # montar a lista de filhos, e o resultado é o mesmo.
            nivel, pai = 0, r.pai
            while pai is not None:
                nivel, pai = nivel + 1, por_id[pai].pai
            recuo = "  " * nivel + ("- " if nivel else "")
            sufixo = f"   << {r.erro} >>" if r.erro else ""
            linhas.append(f"{recuo}{r.nome} ({r.tipo}){sufixo}")
        return "\n".join(linhas)


# Uma requisição completa, com uma etapa que falha no meio.
tracer = Tracer()
with tracer.span("span", "consulta") as raiz:
    raiz.entrada = {"pergunta": "qual o prazo de garantia", "tenant": "empresa-a"}
    with tracer.span("span", "retrieve") as busca:
        busca.saida = {"c17": 0.82, "c22": 0.61, "c31": 0.44}
    with tracer.span("span", "assemble") as montagem:
        montagem.saida = {"gasto": 3180, "piso": 900}
    try:
        with tracer.span("generation", "generate") as geracao:
            raise TimeoutError("provedor nao respondeu em 30s")
    except TimeoutError:
        # A excecao sobe para quem chamou, como sempre. O trace ja guardou
        # que a montagem funcionou e que a geracao foi onde parou.
        geracao.saida = None

print(tracer.render())

A árvore impressa mostra o que nenhuma linha de log faria: a montagem terminou, a

geração é que quebrou, e a etapa que falhou continua no registro com o motivo.

---

5. O que entra no registro e o que nunca deve entrar

A lista da seção 2 é o que entra. O que não entra tem nome, e o nome é risco

jurídico.

O primeiro é o texto integral do trecho recuperado: ele é grande, custa caro por

requisição, e é o vetor mais direto de vazamento entre tenants. O buscar_errado

do artigo 01, que recupera 50 e filtra depois,

não vaza na resposta — vaza aqui, porque os 50 documentos de outros tenants

foram lidos e estão prestes a ser gravados no registro. O que se grava é o

identificador, o número e a origem do número.

O segundo é o dado pessoal que veio do corpus e da pergunta: um documento com

CPF, e-mail e telefone entra no índice, e o trecho que sai da busca carrega os

três. A função abaixo é a mais barata da lista, e o ponto não é perfectionismo —

é que o registro de observabilidade costuma ser exportado para ferramenta de

terceiro, e a partir desse momento a correção do vazamento deixa de estar com

quem você decidiu que estava.

import re

# Cada padrão tem o motivo declarado: é o que permite acusar um vazamento
# olhando o campo que faltou, e não diffando o log inteiro. O saneamento
# completo do payload, com o mesmo conjunto de padrões, está no artigo 13.
RE_CPF_CNPJ = re.compile(r"\b\d{3}\.?\d{3}\.?\d{3}-?\d{2}\b")
RE_EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]{2,}")


def redigir(texto: str) -> str:
    """Troca dado pessoal por um marcador antes de gravar.

    Aplicado na fronteira: o que vai para o trace passa por aqui, o que fica
    no banco de vetores não.
    """
    saida = RE_CPF_CNPJ.sub("<doc>", texto)
    return RE_EMAIL.sub("<email>", saida)


# Uma pergunta do canal de atendimento, com dado de terceiro junto.
print(redigir("O titular 123.456.789-09 ligou no telefone (11) 98888-1234 e "
              "pediu a nota do e-mail joao.silva@empresa.com.br"))

A saída troca por marcadores o que era CPF e e-mail. Esse é o estado desejado: o

registro continua legível para depurar, e parou de carregar o dado. O telefone

precisa do mesmo cuidado e está no artigo 13, junto com o saneamento do que vai

para o índice. O terceiro item da lista é o segredo: chave de API, senha de banco

e cabeçalho de autorização não entram no input de uma generation — eles não

fazem parte do prompt, e se apareceram no log foi porque alguém imprimiu a

variável inteira no lugar do valor.

⚠️ O trace é o lugar onde a sua base vaza, não a resposta
Vale repetir o ponto do artigo 01 com o outro lado da tela: a resposta pode

estar perfeitamente filtrada enquanto o registro de observabilidade guarda os

50 documentos que a busca leu antes do filtro. Se você usa user_id para

agrupar sessões, cada sessão vira um índice dos trechos que aquele usuário podia

ver — o que é exatamente um índice de quem tem acesso a quê.

---

6. Conjunto dourado: a avaliação que roda sem ninguém

O conjunto dourado (golden set) é um conjunto de perguntas com a resposta

que você considera correta, mantido no repositório ao lado do código. Não é uma

planilha: é um artefato versionado, revisado como código, que roda em integração

contínua. Cada pergunta carrega também o trecho que deveria ter sido

recuperado — não a frase da resposta. A diferença importa: a resposta depende do

modelo, e trocar de modelo deve reprovar o conjunto, não invalidar a anotação.

Offline é a avaliação que roda sem usuário: você monta as perguntas com base no

acervo, executa o pipeline inteiro e compara. Custa uma chamada de modelo, é

repetível, e é o que permite dizer "a mudança X piorou a cobertura de 0,92 para

0,88". Online é a avaliação que vem do uso real: as perguntas que as pessoas

fizeram, com o polegar em cima ou em baixo. Online é mais fiel e tem duas falhas:

chega filtrada pelo que já deu problema, e chega sem rótulo. Use as duas em sentidos

opostos — offline para decidir se uma mudança entra, online para descobrir quais

perguntas escrever no conjunto dourado.

from __future__ import annotations

import re
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any


class ScoreSource(StrEnum):
    """De que tipo de busca veio o número. Guardar a origem é obrigatório."""

    VECTOR = "vector"
    LEXICAL = "bm25"
    GRAPH = "graph"
    COMMUNITY = "community"


@dataclass(frozen=True)
class Scored:
    """Um trecho escolhido para entrar na conversa.

    Mesmos campos da ficha do artigo 01. Aqui é um dataclass sem validação
    porque o exemplo não valida entrada: o que está em exercício é a medição.
    """

    ref_id: str
    score: float
    source: ScoreSource
    payload: dict[str, Any] = field(default_factory=dict)


# Um corpus de quatro trechos. Não é a sua base: é o mínimo para o exemplo
# ter pergunta com resposta, pergunta sem resposta, e pergunta ambígua.
CORPUS: dict[str, str] = {
    "c1": "A garantia padrão é de doze meses contados da entrega.",
    "c2": "O reembolso é liberado em até trinta dias após a solicitação.",
    "c3": "Danos por mau uso não são cobertos pela garantia.",
    "c4": "O suporte técnico atende pelo e-mail em horário comercial.",
}

# Piso de similaridade abaixo do qual o sistema prefere não responder.
# É uma decisão de produto, não um número mágico: abaixo dele a chance de
# inventar é maior do que a de acertar.
PISO = 0.35

# Pontuação vira espaço. Sem isso, a última palavra do trecho ("garantia.")
# nunca casa com a pergunta ("garantia"), e o recuperador age como se o
# documento não falasse do assunto. Um índice lexical de verdade faz isso ao
# indexar, não na consulta.
RE_PONTUACAO = re.compile(r"[^\w\s]", re.UNICODE)

def normalizar(texto: str) -> list[str]:
    """Minúsculas e sem pontuação: o pré-requisito de qualquer busca por palavra."""
    return RE_PONTUACAO.sub(" ", texto.lower()).split()


def recuperar(pergunta: str, k: int = 3) -> list[Scored]:
    """Busca por sobreposição de termos. Sem vetor, sem modelo.

    Nota de honestidade: o número é contagem de termo em comum sobre o tamanho
    da pergunta. Ele **não** representa similaridade de significado, e existe
    para exercitar a medição -- não a qualidade da busca, assunto dos
    artigos 03 e 04.
    """
    termos = set(normalizar(pergunta))
    achados: list[Scored] = []
    for ref_id, texto in CORPUS.items():
        comuns = termos & set(normalizar(texto))
        if comuns:
            achados.append(
                Scored(ref_id=ref_id, score=len(comuns) / len(termos),
                       source=ScoreSource.LEXICAL)
            )
    achados.sort(key=lambda s: s.score, reverse=True)
    return achados[:k]


def responder(pergunta: str, obtidos: list[Scored]) -> str | None:
    """Escreve a resposta com o melhor trecho, ou recusa.

    Placeholder determinístico com o mesmo formato de saída do seu modelo: texto
    quando tem evidência, None quando não tem. A medição abaixo não muda quando
    você trocar esta função pela chamada real.
    """
    if not obtidos or obtidos[0].score < PISO:
        return None
    return CORPUS[obtidos[0].ref_id]


@dataclass(frozen=True)
class Caso:
    """Uma pergunta do conjunto dourado.

    `esperado_refs` guarda o **trecho** que responde, não a frase da resposta.
    `exige_recusa` marca a pergunta que a base não cobre -- serve para medir se
    o sistema sabe quando não sabe.
    """

    pergunta: str
    esperado_refs: tuple[str, ...] = ()
    exige_recusa: bool = False


@dataclass
class Execucao:
    """O que aconteceu com um caso: o que a busca devolveu e o que foi dito."""

    caso: Caso
    obtidos: list[Scored] = field(default_factory=list)
    resposta: str | None = None


# Cinco casos, e eles caem em três situações: três com resposta óbvia, um que a
# base não cobre, e um que a base cobre mas que o recuperador por palavra traz
# com folga insuficiente.
GOLDEN_SET: tuple[Caso, ...] = (
    Caso("quanto tempo dura a garantia", ("c1",)),
    Caso("em quantos dias o reembolso é liberado", ("c2",)),
    Caso("mau uso tem cobertura", ("c3",)),
    Caso("o que a garantia não cobre quando o uso for indevido", ("c3",)),
    Caso("qual o prazo do contrato de supplying anual", exige_recusa=True),
)


def rodar(casos: tuple[Caso, ...]) -> list[Execucao]:
    """Executa o pipeline inteiro para cada caso.

    A mesma função serve para o teste de integração contínua e para a medição de
    um deploy: ela é a unidade de comparação.
    """
    execucoes: list[Execucao] = []
    for caso in casos:
        obtidos = recuperar(caso.pergunta)
        resposta = responder(caso.pergunta, obtidos)
        execucoes.append(Execucao(caso, obtidos, resposta))
    return execucoes

---

7. As métricas que valem a pena olhar todo dia

Quatro números, cada um respondendo a uma pergunta diferente. Misturá-los é o

erro: um sistema com cobertura alta e resposta ruim é um defeito de montagem ou de

geração, e trocar o recuperador não conserta.

A cobertura é a coverage_at_k do artigo 01 com outro nome — o nome que

aparece em ferramenta é recall@k: das perguntas que tinham resposta na base, em

que proporção o trecho que respondia apareceu entre os k primeiros. O ranking médio recíproco (MRR) responde uma pergunta diferente: não só apareceu, apareceu

onde. Cobertura 1,0 com MRR 0,4 significa que a base está toda certa e a

ordenação está errada — e a correção é outra, não a base.

A taxa de recusa é a fração de perguntas em que o sistema respondeu "não

encontrei evidência". Aqui está a armadilha: ela pode ser levada a 100% e o painel

fica lindo. Por isso ela só é lida junto com a recusa indevida — as perguntas

que tinham resposta na base e foram recusadas. Taxa de recusa que sobe sem que a

indevida caia é o sistema ficando mais burro. A ancoragem é a versão

quantitativa do groundedness: a resposta afirma coisas que estão nos trechos que

ela cita. A seção 8 mostra como medi-la sem confiar em um modelo.

def cobertura(execucoes: list[Execucao], k: int = 3) -> float:
    """Proporção de casos, entre os que têm resposta na base, em que algum
    trecho esperado apareceu entre os k primeiros.

    Casos marcados como `exige_recusa` ficam fora: eles não têm trecho certo,
    e incluí-los faria a métrica cair sem que a busca tenha errado.
    """
    com_resposta = [e for e in execucoes if e.caso.esperado_refs]
    if not com_resposta:
        return 0.0
    acertos = sum(
        1 for e in com_resposta
        if set(e.caso.esperado_refs) & {s.ref_id for s in e.obtidos[:k]}
    )
    return acertos / len(com_resposta)


def mrr(execucoes: list[Execucao]) -> float:
    """Ranking médio recíproco: 1/posição do trecho certo, médio dos casos.

    Ignora o que veio abaixo do primeiro acerto. É por isso que ela
    complementa a cobertura: a cobertura não sabe se o acerto foi o primeiro.
    """
    com_resposta = [e for e in execucoes if e.caso.esperado_refs]
    if not com_resposta:
        return 0.0
    soma = 0.0
    for execucao in com_resposta:
        for posicao, achado in enumerate(execucao.obtidos, start=1):
            if achado.ref_id in execucao.caso.esperado_refs:
                soma += 1.0 / posicao
                break
    return soma / len(com_resposta)


def taxa_de_recusa(execucoes: list[Execucao]) -> float:
    """Fração de casos em que o sistema não respondeu."""
    return sum(1 for e in execucoes if e.resposta is None) / len(execucoes)


def recusa_indevida(execucoes: list[Execucao]) -> int:
    """Quantas perguntas COM resposta na base foram recusadas.

    É o par que dá sentido à taxa de recusa. Recusar tudo zera este número.
    """
    return sum(1 for e in execucoes if e.caso.esperado_refs and e.resposta is None)

execucoes = rodar(GOLDEN_SET)
for nome, valor in [
    ("casos", len(execucoes)),
    ("cobertura@3", cobertura(execucoes, 3)),
    ("mrr", mrr(execucoes)),
    ("recusa", taxa_de_recusa(execucoes)),
    ("recusa indevida", recusa_indevida(execucoes)),
]:
    print(f"{nome:<18} {valor if isinstance(valor, int) else f'{valor:.3f}'}")
for execucao in execucoes:
    situacao = "recusou" if execucao.resposta is None else "respondeu"
    obtido = [s.ref_id for s in execucao.obtidos]
    print(f"- {execucao.caso.pergunta[:40]:<40} {situacao:<9} {obtido}")

O caso "o que a garantia não cobre quando o uso for indevido" é o ponto do

artigo. O trecho c3 está em primeiro na lista, a cobertura está no máximo,

e mesmo assim o sistema recusou, porque o número ficou abaixo do piso. As duas

métricas de recuperação estão verdes e a resposta não existe: elas medem se o

trecho chegou, e o defeito está na decisão de responder. O que precisa mudar é o

piso, não o índice.

Cinco casos é pouco, e é o começo. O que faz o conjunto crescer não é a

quantidade: é cada caso novo que entra depois de uma falha real de produção — e

vale cobrir os quatro desvios que quebram de verdade, que são a pergunta literal,

a paráfrase, a pergunta sem resposta na base, e a pergunta que só um trecho

específico responde. Registre também o que o sistema recusou fazer: um caso de

"não encontrei evidência" prova que o limiar de similaridade está no ponto certo,

e sem ele você não tem como saber se a taxa de recusa subiu porque a base encolheu

ou porque o sistema ficou conservador.

---

8. O juiz tem viés, e a checagem determinística que corrige

Para medir ancoragem, a saída óbvia é pedir a outro modelo: "dado este contexto e esta resposta, a resposta é sustentada pelo contexto?". É barato, é rápido, e é o

que o RAGAS popularizou — junto com as métricas

de fidelidade, relevância e precisão do contexto, que são o mesmo par de perguntas

que eu acabei de listar. Funciona, e não é confiável sozinho: o modelo usado como

juiz prefere a resposta que aparece primeiro na comparação, prefere a resposta

mais longa, e tende a avaliar melhor a saída de modelos parecidos com ele

mesmo. Esses três efeitos estão medidos no

MT-Bench, e nenhum deles é o seu defeito de

busca — é o viés do juiz aparecendo como se fosse. Um juiz que prefere respostas

longas aprova um sistema que enche a resposta de ressalvas inúteis, e você otimiza

para isso durante três semanas.

A correção não é trocar de juiz. É complementar o juiz com verificação

determinística: uma função que confere afirmações verificáveis contra o texto, sem

modelo nenhum. A segunda das três abaixo pega alucinação de citação, e é a mais

barata do artigo: não custa chamada de modelo, e roda em cada requisição em vez de

rodar só na amostra.

import re

# Números com separador de milhar ou decimal: "1.234", "1.234,56", "30".
RE_NUMERO = re.compile(r"\d+(?:[.,]\d+)*")


def numeros_de(texto: str) -> set[str]:
    """Números do texto, normalizados: '1.234' e '1234' viram a mesma coisa.

    É uma aproximação, e vale saber onde ela falha: a fonte tem "12" e a
    resposta tem "12,0", e o juiz não reconhece. O uso correto é como
    **sinalizador** -- ausência aqui significa "olhe esse caso", não "essa
    resposta está errada".
    """
    achados: set[str] = set()
    for bruto in RE_NUMERO.findall(texto):
        achados.add(bruto.replace(".", "").replace(",", "."))
    return achados


def checar_ancoragem(
    resposta: str, trechos: dict[str, str], citacoes: list[str]
) -> dict[str, object]:
    """Confere a resposta contra os trechos que ela mesma cita.

    Três verificações, nenhuma delas é modelo: toda citação aponta para um
    trecho que foi de fato recuperado; todo número da resposta aparece em algum
    dos trechos citados; a resposta não está vazia. Devolve o veredito e o motivo.
    """
    if not citacoes:
        return {"ancorada": False, "motivo": "resposta sem citação", "faltando": []}
    inexistentes = [c for c in citacoes if c not in trechos]  # citação inventada
    if inexistentes:
        return {"ancorada": False, "motivo": "citação fora da recuperação",
                "faltando": inexistentes}
    contexto = " ".join(trechos[c] for c in citacoes)
    faltando = sorted(numeros_de(resposta) - numeros_de(contexto))
    if faltando:
        return {"ancorada": True, "motivo": "número sem lastro", "faltando": faltando}
    return {"ancorada": True, "motivo": "ok", "faltando": []}


# Três respostas. A segunda cita um trecho que a busca não devolveu -- que é
# o que o modelo faz quando o contexto está ruim e ele improvisa a fonte.
for texto, citacoes in [
    ("A garantia padrão é de doze meses contados da entrega.", ["c1"]),
    ("A garantia padrão é de doze meses contados da entrega.", ["c9"]),
    ("A garantia padrão é de 24 meses contados da entrega.", ["c1"]),
]:
    v = checar_ancoragem(texto, CORPUS, citacoes)
    print(f"ancorada={v['ancorada']}  motivo={v['motivo']}  faltando={v['faltando']}")

A segunda resposta é a que importa: ela é perfeita e está errada. A citação

c9 não existe na base. Nenhum juiz vai reprovar uma resposta dessas com

convicção, porque ela soa exatamente como a resposta certa. A checagem

determinística reprova em microssegundos, e é por isso que ela roda em toda

requisição enquanto o juiz roda em amostra.

⚠️ Juiz com o mesmo peso na decisão dá falso negativo com frequência
O ponto do uso do juiz não é substituir a regra: é produzir um número que pode

acompanhar uma família inteira de mudanças sem você reescrever o avaliador.

Trate a saída dele como alerta — o que dispara revisão e regressão — e

nunca como veredito de quem bloqueia o deploy sozinho.

---

9. Regressão: baseline, limiar e quem é avisado

A parte que transforma medição em proteção é o último passo: declarar limites. Uma

métrica sem limiar declarado é um número que ninguém reage: a equipe se acostuma a

ver 0,88, depois 0,87, depois 0,81, e só percebe quando alguém reclama. O que

impede isso é o baseline — o valor da métrica na branch principal, gravado

junto com o código — e o limiar — quanto de queda você aceita antes de chamar

de regressão. O limiar não é moralidade, é custo de ruído: perto demais do zero,

qualquer variação de tamanho de amostra vira alarme e a equipe silencia o alerta na

primeira semana. Uma forma de escolher o valor sem número mágico: o limiar é menor

do que a menor diferença que importa para quem usa o sistema.

Repare também que o sentido do alerta não é o mesmo para toda métrica. Uma taxa de

recusa que sobe é sinal de alarme, não de melhora — é por isso que o limiar

carrega o sinal. E é por isso que a recusa indevida precisa estar na mesma

execução: sem o par, o alerta de subida não sabe se está certo.

from dataclasses import dataclass


@dataclass(frozen=True)
class Baseline:
    """O valor de referência de uma métrica e quanto de variação é tolerado.

    `limiar` é diferença absoluta, em ponto. `subir_e_ruim` existe porque nem
    toda métrica tem o mesmo sentido: subir é ruim para recusa, é neutro para MRR.
    """

    metrica: str
    valor: float
    limiar: float
    subir_e_ruim: bool = False


def rebaixou(atual: dict[str, float], base: Baseline) -> bool:
    """True quando a métrica de `atual` saiu da faixa aceita por `base`."""
    valor = atual.get(base.metrica)
    # Métrica ausente do relatório é regressão: sem número não há como provar
    # que nada mudou, e "não mediu" não é "não quebrou".
    if valor is None:
        return True
    if base.subir_e_ruim:
        return valor > base.valor + base.limiar
    return valor < base.valor - base.limiar


# O baseline da branch principal, versionado no repositório.
BASELINE: tuple[Baseline, ...] = (
    Baseline("cobertura@3", 1.000, 0.050),
    Baseline("mrr", 1.000, 0.050),
    Baseline("recusa", 0.200, 0.100, subir_e_ruim=True),
    Baseline("recusa_indevida", 0.000, 0.001, subir_e_ruim=True),
    Baseline("ancorada", 1.000, 0.020),
)

# A execução de hoje: o time levantou o piso de similaridade de 0,30 para 0,35,
# para o sistema "parar de chutar" nas perguntas duvidosas. Cobertura e
# ranking não se mexeram. A recusa subiu, e subiu junto da que ninguém queria.
relatorio_de_hoje = {
    "cobertura@3": cobertura(execucoes, 3),
    "mrr": mrr(execucoes),
    "recusa": taxa_de_recusa(execucoes),
    "recusa_indevida": float(len(execucoes) - recusa_indevida(execucoes)) / len(execucoes),
    "ancorada": 1.0,
}

alertas = [
    f"{b.metrica}: {'ACIMA' if b.subir_e_ruim else 'ABAIXO'} "
    f"({b.valor:.3f} ± {b.limiar:.3f})"
    for b in BASELINE
    if rebaixou(relatorio_de_hoje, b)
]
for metrica, valor in relatorio_de_hoje.items():
    print(f"{metrica:<18} {valor:.3f}")
print()
for linha in alertas or ["nenhuma regressão: deploy liberado"]:
    print(linha)

Repare no que o alerta pegou. As duas métricas de recuperação ficaram idênticas

ao baseline — a mudança não mexeu em nada que elas medem — e mesmo assim o deploy

foi barrado, por causa de uma métrica que é contagem de problema, não média. É

por isso que a recusa indevida tem limiar 0,001 e subir_e_ruim=True: não existe

"um pouco" de recusa indevida. Cada ponto é um usuário que tinha resposta na base

e recebeu "não encontrei evidência". E quando a métrica some do relatório,

rebaixou devolve True: um pipeline de avaliação que falha em silêncio é o

pior tipo de pipeline de avaliação, porque "tudo verde" passa a ser o estado

padrão quando o código quebra.

💡 Grave o baseline por ambiente, e comece por uma versão que já é boa
O número do seu notebook não é o número de produção: o corpus é outro, o

modelo de vetor pode ser outro, e o filtro de permissão muda a lista. Grave

o baseline do ambiente alvo na primeira semana, mesmo que ele seja ruim. Um

baseline ruim medido é infinitamente mais útil do que um baseline perfeito

que nunca existiu.

Se você parou aqui, o seu sistema tem um trace com uma etapa por peça do pipeline,

um conjunto dourado versionado que roda em integração contínua, quatro métricas

nomeadas com o par que impede leitura errada, e um alerta de regressão com

baseline declarado. O artigo 12 usa esses números

para escolher a arquitetura da sua base, e o

artigo 13 usa o trace para descobrir onde o

dinheiro está vazando.

---

TL;DR

• Trace é o registro de uma requisição inteira, com as etapas aninhadas.

  Sem ele, "a resposta ficou ruim" tem sete causas e nenhuma pista; com ele, a

  etapa que quebrou é uma linha.

• A API do Langfuse mudou de vez. A v4 é sobre OpenTelemetry, usa

  get_client() e start_as_current_observation, e o que os tutoriais mostram

  está deprecado — com prazo em 2026-11-16 no Langfuse Cloud.

• No trace entra identificador e número; não entra o texto inteiro nem dado pessoal. O registro de observabilidade é onde a base vaza, não a resposta, e

  é o lugar que costuma ser exportado para terceiro.

• Conjunto dourado é a resposta correta anotada, versionada como código. É o

  único jeito de provar que uma mudança melhorou; online serve para descobrir

  quais perguntas escrever nele.

• Cobertura e ranking medem coisas diferentes, e recusa só se lê junto com recusa indevida. Cobertura e MRR no máximo com resposta ausente é o caso que

  mais confunde quem procura defeito no recuperador — e juiz de modelo tem viés

  medido, então sempre ao lado de uma checagem determinística.

---

Referências

• SDK overview — Langfuse — a referência da v4, com as assinaturas usadas na seção 4
• langfuse-python — GitHub — onde fica o aviso de depreciação da API legada, e a versão mínima
• Self-hosting — Langfuse — a versão mínima do servidor, que é diferente da do SDK
• Convenções semânticas de GenAI — OpenTelemetry — por que a v4 mudou de forma, e não só de nome
• RAGAS: Automated Evaluation of Retrieval Augmented Generation — as métricas de fidelidade e relevância que o artigo reorganiza em quatro
• Judging LLM-as-a-Judge with MT-Bench and Chatbot Arena — as medidas dos três vieses que a seção 8 descreve