Objetivo da Aula
Ao concluir esta aula, você será capaz de:
Implementar reranking com LLM como cross-encoder para melhorar precisão da recuperação
Avaliar a faithfulness de respostas RAG — se o LLM realmente usou os documentos ou inventou
Construir logging JSONL de queries RAG para análise de qualidade e debugging
Comparar estratégias de chunking e seleção de top-k com trade-offs mensuráveis
Debugar pipelines RAG identificando falhas nas etapas de retrieval, reranking e geração
Por que isso importa
RAG básico funciona em demos. RAG em produção falha de formas sutis que só aparecem com dados reais e usuários reais.
O problema mais comum: recall inadequado. O sistema busca os 5 documentos mais próximos por cosine similarity, mas o mais relevante para a pergunta específica está em #8 porque a similaridade vetorial não captura relevância semântica profunda. A resposta parece razoável mas está incompleta ou levemente errada.
O segundo problema: alucinação mascarada. O LLM recebe 3 documentos de contexto, mas se a pergunta não for respondível com esses 3 documentos, em vez de dizer “não encontrei no contexto”, ele inventa uma resposta plausível usando seu conhecimento de treino. O usuário não percebe. A confiança no sistema cai quando alguém finalmente percebe.
O terceiro problema: sem observabilidade. Quando um usuário reclama “o sistema deu uma resposta errada”, você não tem logs suficientes para saber: o retrieval trouxe os documentos errados? O reranker priorizou mal? O LLM ignorou o contexto? Sem logging detalhado, debugging é adivinhação.
Esta unidade te dá as ferramentas para os três problemas: reranking que melhora precisão, avaliação de faithfulness que detecta alucinação, e logging estruturado para observabilidade real.
Conceitos Fundamentais
O Problema do Bi-Encoder Retrieval
O retrieval básico funciona assim: 1. Cada documento é convertido em vetor (embedding) offline 2. A query é convertida em vetor em tempo real 3. Você encontra os k documentos com vetores mais próximos da query (cosine similarity)
Isso é chamado de bi-encoder — documento e query são codificados separadamente.
O problema: a similaridade de vetores captura semelhança semântica geral mas não relevância específica. Um documento sobre “machine learning para análise de sentimentos” tem alta similarity com uma query sobre “análise de sentimentos” — mas se a query pergunta especificamente sobre “análise de sentimentos em reviews curtos de 2-3 palavras”, um outro documento mais específico pode ter similarity menor mas ser muito mais relevante.
Por que isso acontece? O modelo de embedding (ex: text-embedding-ada-002) comprime o documento inteiro em um vetor de 1536 dimensões. Informações específicas são “diluídas” pela representação geral do documento.
Cross-Encoder: O Que É e Por Que Funciona Melhor
Um cross-encoder vê a query E o documento juntos, processando a relação entre eles. É como fazer o LLM perguntar: “dado que a query é X, quanto esse documento Y é relevante? Score de 0-10.”
Bi-encoder: embed(query) → vetor_query
embed(doc) → vetor_doc
similarity = cosine(vetor_query, vetor_doc)
[processa separados, compara vetores]
Cross-encoder: score = model(query + doc)
[processa juntos — captura relação direta]Cross-encoders são mais precisos mas mais lentos (O(k × n) vs O(n) para bi-encoder). Por isso o padrão em produção é:
1. Bi-encoder: busca rápida dos top-K candidatos (K = 20-50)
2. Cross-encoder: reranking dos K candidatos para top-k final (k = 3-5)
3. LLM: geração com os k documentos rerankeadosDecisão de Arquitetura: LLM-as-Judge
Usar um LLM para avaliar a saída de outro LLM é uma técnica de avaliação escalável quando gabarito humano não existe. Três cuidados não-opcionais: (1) Use um juiz diferente do gerador — o mesmo modelo que aluciná tende a avaliar sua própria alucinação como correta; (2) Randomize a ordem de candidatos quando comparar saídas — modelos têm viés posicional (preferem a primeira ou a última opção); (3) Meça taxa de discordância — se o juiz concorda 100% com o gerador, provavelmente está enviesado. Métricas de qualidade para instrumentar em produção: faithfulness (a resposta é suportada pelos documentos recuperados?) e context relevance (os documentos recuperados são relevantes para a pergunta?). Logue essas métricas por request para detectar degradação ao longo do tempo.
Cross-encoder com LLM como juiz: Em vez de um modelo dedicado (Cohere Rerank, MS MARCO), você pode usar o próprio Claude como cross-encoder. É mais lento e caro, mas não requer modelo adicional. Útil para prototipagem e volumes baixos.
Faithfulness: Medindo Alucinação em RAG
Faithfulness mede se cada afirmação na resposta pode ser suportada pelas fontes fornecidas como contexto. É a métrica mais importante em RAG — garante que o sistema não está inventando.
Resposta: "A empresa foi fundada em 1985 e tem 5000 funcionários."
Fontes: [doc sobre a empresa sem mencionar data de fundação]
Faithfulness = 0 (afirmação sobre 1985 não tem suporte nas fontes)O framework RAGAS (Retrieval Augmented Generation Assessment) define faithfulness como:
faithfulness = (número de claims suportados pelas fontes) / (total de claims na resposta)Implementar isso com Claude como juiz:
def avaliar_faithfulness_llm(query: str, resposta: str, fontes: list[str]) -> dict:
contexto = '\n\n'.join(f"[Fonte {i+1}]: {doc}" for i, doc in enumerate(fontes))
prompt = f"""Avalie se a resposta abaixo está completamente suportada pelas fontes.
FONTES:
{contexto}
RESPOSTA PARA AVALIAR:
{resposta}
Para cada afirmação factual na resposta:
1. Identifique a afirmação
2. Determine se está nas fontes (sim/não)
3. Se sim, cite a fonte [N]
Depois, dê um score de 0-5:
- 5: Totalmente fundamentada nas fontes
- 3: Maioria fundamentada, algumas inferências razoáveis
- 1: Pouco fundamentada, muitas afirmações sem suporte
- 0: Inventou conteúdo não presente nas fontes
Retorne JSON:
{{
"score": 0-5,
"tem_alucinacao": true/false,
"afirmacoes_sem_suporte": ["afirmação1", ...],
"justificativa": "explicação breve"
}}"""
response = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=400,
messages=[{'role': 'user', 'content': prompt}]
)
return extrair_json(response.content[0].text)Logging JSONL para Observabilidade de RAG
Cada query num pipeline RAG precisa ser logada com suficiente contexto para debugging. O formato JSONL (JSON Lines — um JSON por linha) é ideal para isso:
{"timestamp": "2025-01-15T10:30:00", "query": "...", "docs_recuperados": 5, "docs_pos_rerank": 3, "resposta": "...", "faithfulness_score": 4, "latencia_ms": 1200}
{"timestamp": "2025-01-15T10:31:00", "query": "...", "docs_recuperados": 5, "docs_pos_rerank": 3, "resposta": "...", "faithfulness_score": 2, "latencia_ms": 890}Vantagens do JSONL: - Uma linha por evento — fácil de parsear com Python - Streaming — pode escrever sem carregar o arquivo inteiro - Compatível com ferramentas de log analysis (jq, Python, Pandas) - Resiliente — um JSON malformado não corrompe o arquivo inteiro
Estratégias de Chunking
O tamanho do chunk afeta diretamente a qualidade do retrieval:
Chunks menores (100-300 tokens): - Precisão maior: vetores mais específicos - Mais chunks para indexar e buscar - Risco: chunk pode não ter contexto suficiente para responder sozinho - Uso: FAQs, documentação técnica com seções curtas
Chunks médios (300-600 tokens): - Balance entre precisão e contexto - Uso: artigos, relatórios, documentação
Chunks grandes (600-1200 tokens): - Mais contexto por chunk - Similarity menos específica - Uso: documentos narrativos, contratos
Parent-Child Chunking (mais avançado): - Busca em chunks pequenos (filhos) para precisão - Mas retorna o chunk pai (maior) para mais contexto - Melhor dos dois mundos: precisão no retrieval, contexto na geração
Aprofundamento Técnico
Implementando Reranker com LLM
O reranking com LLM funciona fazendo o modelo avaliar cada par (query, documento) individualmente:
import anthropic
import json
import re
client = anthropic.Anthropic()
def rerankar_com_llm(query: str, documentos: list[str], top_k: int = 3) -> list[str]:
"""
Usa LLM como cross-encoder para re-ordenar documentos por relevância.
Retorna os top_k mais relevantes para a query.
"""
if len(documentos) <= top_k:
return documentos # sem necessidade de reranking
scores = []
for i, doc in enumerate(documentos):
# Para cada documento, pede ao Claude um score de relevância
prompt = f"""Avalie a relevância deste documento para responder a pergunta.
Score de 0-10 onde:
- 10: Responde diretamente a pergunta
- 7: Altamente relevante, contém informações importantes
- 4: Parcialmente relevante
- 1: Pouco relevante
- 0: Irrelevante
PERGUNTA: {query}
DOCUMENTO: {doc[:500]}...
Retorne APENAS: {{"score": X, "motivo": "uma frase"}}"""
response = client.messages.create(
model='claude-haiku-4-5-20251001', # Haiku: rápido e barato para scoring
max_tokens=100,
messages=[{'role': 'user', 'content': prompt}]
)
try:
resultado = json.loads(response.content[0].text)
score = resultado.get('score', 0)
except:
score = 0 # se falhar, score 0 (vai para o fim da lista)
scores.append((score, doc))
# Ordena por score decrescente e retorna top_k
scores.sort(key=lambda x: x[0], reverse=True)
return [doc for _, doc in scores[:top_k]]Custo do reranking: Para K=20 documentos, isso faz 20 chamadas de API com ~200 tokens cada = 4000 tokens de input. Com Haiku ($0.25/1M), custa $0.001 por query. Viável para volumes moderados.
Para volumes altos (>1000 queries/hora), considere Cohere Rerank API ($1/1000 queries) ou um modelo de reranking dedicado local (cross-encoder de Hugging Face).
O Pipeline Completo com LangChain e ChromaDB
A versão do starter usa LangChain + ChromaDB para a parte de indexação e retrieval. O LangChain abstrai embeddings e vector store:
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Indexação (feito uma vez, offline)
def indexar_documentos(textos: list[str]) -> Chroma:
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # tamanho do chunk em caracteres
chunk_overlap=50, # overlap entre chunks para não perder contexto nas bordas
separators=["\n\n", "\n", " "], # tenta quebrar por parágrafo, depois linha, depois palavra
)
chunks = splitter.create_documents(textos)
embeddings = OpenAIEmbeddings() # ou HuggingFaceEmbeddings para self-hosted
vectorstore = Chroma.from_documents(chunks, embeddings)
return vectorstore
# Retrieval (em tempo real para cada query)
def buscar(vectorstore: Chroma, query: str, k: int = 20) -> list[str]:
# similarity_search retorna os k chunks mais próximos
docs = vectorstore.similarity_search(query, k=k)
return [doc.page_content for doc in docs]Estrutura do Logging JSONL
import datetime
import json
from pathlib import Path
def logar_query(
log_path: str,
entrada: dict,
) -> None:
"""Adiciona uma entrada ao log JSONL de queries RAG."""
# Adiciona timestamp se não presente
if 'timestamp' not in entrada:
entrada['timestamp'] = datetime.datetime.now().isoformat()
# Abre em append mode — não sobrescreve, apenas adiciona
with open(log_path, 'a', encoding='utf-8') as f:
f.write(json.dumps(entrada, ensure_ascii=False) + '\n')
# Uso típico:
logar_query('rag_queries.jsonl', {
'query': 'Como funciona RAG?',
'docs_recuperados': 20,
'docs_pos_rerank': 3,
'resposta': '...',
'faithfulness': {'score': 4, 'tem_alucinacao': False},
'latencia_total_ms': 1450,
'tokens_usados': 2300,
})
# Análise posterior:
def analisar_logs(log_path: str) -> dict:
queries = []
with open(log_path, 'r') as f:
for linha in f:
queries.append(json.loads(linha.strip()))
scores = [q.get('faithfulness', {}).get('score', 0) for q in queries if 'faithfulness' in q]
alucinacoes = [q for q in queries if q.get('faithfulness', {}).get('tem_alucinacao', False)]
return {
'total_queries': len(queries),
'faithfulness_media': sum(scores) / len(scores) if scores else 0,
'taxa_alucinacao': len(alucinacoes) / len(queries) if queries else 0,
}Exemplos Anotados
Exemplo 1: Pipeline RAG Completo com Fallback
import anthropic
import json
import re
import datetime
client = anthropic.Anthropic()
class RAGSimples:
"""RAG in-memory sem dependências externas — para aprendizado."""
def __init__(self, log_path: str = 'rag_log.jsonl'):
self._docs: list[str] = []
self.log_path = log_path
def indexar(self, documentos: list[str]) -> None:
self._docs.extend(documentos)
print(f"Indexados {len(documentos)} documentos. Total: {len(self._docs)}")
def buscar_bm25_simples(self, query: str, top_k: int = 20) -> list[str]:
"""Busca por palavra-chave (proxy barato para embedding similarity)."""
palavras_query = set(query.lower().split())
scored = []
for doc in self._docs:
palavras_doc = set(doc.lower().split())
# Intersection = palavras em comum = score de relevância simples
score = len(palavras_query & palavras_doc)
scored.append((score, doc))
scored.sort(reverse=True)
return [doc for _, doc in scored[:top_k] if _[0] > 0] # só docs com alguma relevância
def rerankar(self, query: str, documentos: list[str], top_k: int = 3) -> list[str]:
"""Cross-encoder simples usando Claude como juiz."""
if len(documentos) <= top_k:
return documentos
scores = []
for doc in documentos[:10]: # limita a 10 para custo controlado
prompt = f"""Score 0-10: relevância deste trecho para a pergunta.
Retorne APENAS: {{"score": N}}
PERGUNTA: {query}
TRECHO: {doc[:300]}"""
try:
r = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=30,
messages=[{'role': 'user', 'content': prompt}]
)
dados = json.loads(r.content[0].text)
score = float(dados.get('score', 0))
except:
score = 0
scores.append((score, doc))
scores.sort(reverse=True)
return [doc for _, doc in scores[:top_k]]
def gerar_resposta(self, query: str, contexto: list[str]) -> str:
"""Geração grounded — instrui o modelo a usar APENAS o contexto."""
if not contexto:
return "Não encontrei documentos relevantes para responder esta pergunta."
ctx_texto = '\n\n'.join(f"[{i+1}] {doc}" for i, doc in enumerate(contexto))
response = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=400,
messages=[{
'role': 'user',
'content': f"""Responda APENAS com base no contexto fornecido.
Se a resposta não estiver no contexto, diga: "Não encontrei essa informação no contexto fornecido."
Cite as fontes pelo número [N].
CONTEXTO:
{ctx_texto}
PERGUNTA: {query}"""
}]
)
return response.content[0].text
def avaliar_faithfulness(self, query: str, resposta: str, fontes: list[str]) -> dict:
"""Avalia se a resposta está ancorada nas fontes."""
contexto = '\n'.join(f"[{i+1}] {f[:200]}" for i, f in enumerate(fontes))
prompt = f"""Avalie se a resposta está fundamentada nas fontes.
Score: 0=inventou, 3=parcial, 5=totalmente fundamentada.
FONTES: {contexto}
RESPOSTA: {resposta}
Retorne JSON: {{"score": 0-5, "tem_alucinacao": bool, "justificativa": "..."}}"""
try:
r = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=200,
messages=[{'role': 'user', 'content': prompt}]
)
return json.loads(r.content[0].text)
except:
return {'score': None, 'tem_alucinacao': None, 'justificativa': 'erro na avaliação'}
def _logar(self, entrada: dict) -> None:
"""Adiciona linha ao log JSONL."""
entrada['timestamp'] = datetime.datetime.now().isoformat()
with open(self.log_path, 'a', encoding='utf-8') as f:
f.write(json.dumps(entrada, ensure_ascii=False) + '\n')
def query(self, pergunta: str) -> str:
"""Pipeline completo."""
# 1. Retrieval
docs_brutos = self.buscar_bm25_simples(pergunta, top_k=20)
# 2. Reranking
docs_rerankeados = self.rerankar(pergunta, docs_brutos, top_k=3)
# 3. Geração
resposta = self.gerar_resposta(pergunta, docs_rerankeados[:3])
# 4. Avaliação
faithfulness = self.avaliar_faithfulness(pergunta, resposta, docs_rerankeados[:3])
# 5. Logging
self._logar({
'pergunta': pergunta,
'docs_recuperados': len(docs_brutos),
'docs_pos_rerank': len(docs_rerankeados),
'resposta': resposta,
'faithfulness_score': faithfulness.get('score'),
'tem_alucinacao': faithfulness.get('tem_alucinacao'),
})
return resposta
# Uso
rag = RAGSimples()
rag.indexar([
"Python é uma linguagem interpretada criada em 1991 por Guido van Rossum.",
"Python usa indentação para definir blocos de código.",
"NumPy é uma biblioteca de computação numérica para Python.",
])
print(rag.query("Quem criou o Python?"))
# Deve responder com [1] e faithfulness alta
print(rag.query("Qual a temperatura em São Paulo hoje?"))
# Deve dizer "não encontrei no contexto" com faithfulness alta (não aluciou)Exemplo 2: Análise de Qualidade do Pipeline via Logs
import json
def analisar_qualidade_rag(log_path: str) -> dict:
"""Lê o log JSONL e gera relatório de qualidade do pipeline RAG."""
queries = []
with open(log_path, 'r', encoding='utf-8') as f:
for linha in f:
linha = linha.strip()
if linha:
try:
queries.append(json.loads(linha))
except json.JSONDecodeError:
continue # ignora linhas malformadas
if not queries:
return {'erro': 'Nenhuma query no log'}
# Faithfulness
scores = [q['faithfulness_score'] for q in queries if q.get('faithfulness_score') is not None]
alucinacoes = [q for q in queries if q.get('tem_alucinacao') is True]
# Retrieval
docs_medios = sum(q.get('docs_recuperados', 0) for q in queries) / len(queries)
docs_pos_rerank = sum(q.get('docs_pos_rerank', 0) for q in queries) / len(queries)
return {
'total_queries': len(queries),
'faithfulness': {
'media': round(sum(scores) / len(scores), 2) if scores else None,
'taxa_alucinacao': round(len(alucinacoes) / len(queries), 2),
'queries_com_alucinacao': len(alucinacoes),
},
'retrieval': {
'docs_recuperados_media': round(docs_medios, 1),
'docs_pos_rerank_media': round(docs_pos_rerank, 1),
},
'queries_problemáticas': [
q['pergunta'] for q in alucinacoes
]
}Padrões e Armadilhas
Padrões
Padrão 1: Pipeline em 3 estágios para balancear custo e qualidade
Retrieval barato (embeddings) → pool grande de candidatos
Reranking (LLM/cross-encoder) → filtragem precisa
Geração (LLM) → resposta com top-k rerankeadosNão inverta a ordem (reranking antes de retrieval não faz sentido) e não pule o reranking em casos de alta precisão necessária.
Padrão 2: Instrução explícita de grounding no prompt de geração Sempre instrua o LLM a: 1. Usar APENAS o contexto fornecido 2. Citar fontes pelo número [N] 3. Dizer explicitamente quando não encontrar a informação
Sem isso, o modelo usa seu conhecimento de treino livremente e faithfulness cai.
Padrão 3: Logar todos os steps do pipeline, não só o resultado final
# Log mínimo por query:
{
'pergunta': str,
'docs_recuperados': int, # quantos vieram do retrieval
'docs_pos_rerank': int, # quantos ficaram após reranking
'docs_usados': list[str], # os documentos reais usados (para debugging)
'resposta': str,
'faithfulness_score': int,
'latencia_retrieval_ms': int,
'latencia_geracao_ms': int,
}Quando algo der errado, você saberá em qual etapa ocorreu.
Armadilhas
⚠️ Armadilha 1: Confiar em faithfulness score alto sem verificar o prompt de avaliação O mesmo LLM que alucina pode avaliar sua própria alucinação como “5/5 fundamentada”. Use sempre um modelo diferente (ou pelo menos diferente da geração) para avaliar faithfulness, ou use múltiplas avaliações.
⚠️ Armadilha 2: k muito pequeno no retrieval inicial
# ERRADO: busca só 3 candidatos — pode perder o documento mais relevante
docs = buscar(query, k=3)
# CORRETO: busca candidatos suficientes para o reranker trabalhar
candidatos = buscar(query, k=20)
docs_finais = rerankar(query, candidatos, top_k=3)⚠️ Armadilha 3: Chunks muito grandes diluem a similarity
# PROBLEMA: chunk de 2000 tokens tem o assunto diluído
# Uma pergunta específica sobre um parágrafo específico pode não achar esse chunk
# SOLUÇÃO: chunk de 300-500 tokens + overlap de 50-100 tokens
splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=60)⚠️ Armadilha 4: Abrir arquivo de log em modo ‘w’ em vez de ‘a’
# ERRADO: apaga o log a cada query
with open(log_path, 'w') as f: # 'w' = write, apaga tudo
# CORRETO: adiciona ao final do arquivo
with open(log_path, 'a') as f: # 'a' = append, preserva histórico⚠️ Armadilha 5: Não tratar erros no JSON do reranker O LLM às vezes não retorna JSON válido no scoring. Se não tratar:
# SEM tratamento de erro: quebra o pipeline se o modelo retornar texto ruim
score = json.loads(r.content[0].text)['score'] # JSONDecodeError!
# COM tratamento: score 0 como fallback, não quebra o pipeline
try:
score = float(json.loads(r.content[0].text).get('score', 0))
except:
score = 0 # documento vai para o fim — não é excluído, só desprioritizadoSe não for realizar o laboratório, pule para o próximo capítulo.
Ponte para o Lab
O starter tem 3 TODOs que completam o pipeline RAG avançado:
TODO 1 — Em rerankar(): use Claude como cross-encoder para avaliar relevância de cada documento. Para cada doc da lista, construa um prompt que pede um score 0-10 com a query como contexto. Parse o JSON retornado e extraia o score. Ordene documentos por score decrescente e retorne os top_k mais relevantes. Trate exceções de parsing com score=0 como fallback. Seção de referência: Aprofundamento Técnico → “Implementando Reranker com LLM”.
TODO 2 — Em avaliar_faithfulness(): construa um prompt que fornece as fontes e a resposta ao Claude, pedindo avaliação de faithfulness com score 0-5 e campo tem_alucinacao. Retorne o JSON extraído. O stub já existe com o shape do retorno esperado: {'score': int, 'tem_alucinacao': bool, 'justificativa': str}. Seção de referência: Conceitos Fundamentais → “Faithfulness: Medindo Alucinação em RAG”.
TODO 3 — Em _logar(): adicione o campo 'timestamp': datetime.datetime.now().isoformat() à entrada antes de salvar. Abra self.log_path em modo 'a' (append), escreva json.dumps(entrada, ensure_ascii=False) + '\n'. Seção de referência: Aprofundamento Técnico → “Estrutura do Logging JSONL”.
Após implementar os três TODOs, rode o starter e observe: 1. O reranker mudando a ordem dos documentos 2. O score de faithfulness variando por pergunta 3. O arquivo rag_log.jsonl crescendo com cada query
Inspecione o log com cat rag_log.jsonl | python -m json.tool para ver as entradas formatadas.
Agora você está pronto para o lab.