Observabilidade e Avaliações de RAG

Execute a geração aumentada por recuperação (RAG) como um sistema de produção mensurável, não como uma caixa preta.

Pré-requisitos

• Pipeline RAG com estágios de recuperação e geração instrumentados
• Python 3.10+ com bibliotecas de avaliação (ragas, langchain, openai)
• Endpoint Prometheus para exportação de métricas personalizadas
• Conjunto de dados de benchmark com triplas de pergunta/resposta/fonte de padrão ouro
• SDK OpenTelemetry integrado ao serviço RAG

O que Medir

Qualidade da Recuperação

• Recall@k e MRR para os k melhores trechos
• Cobertura de citação e atualidade da fonte
• Deriva de embedding e obsolescência do índice

Qualidade da Geração

• Pontuação de fundamentação (resposta apoiada pelo contexto recuperado)
• Taxa de alucinação por rota/caso de uso
• Aderência à instrução e validade do formato

Confiabilidade e Custo

• Latência p50/p95 dividida por recuperação vs. geração
• Uso de tokens por estágio
• Taxa de acerto do cache e custo por resposta bem-sucedida

Script de Avaliação RAGAS

# rag_eval.py
"""Evaluate RAG pipeline quality using RAGAS metrics."""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_entity_recall,
    answer_similarity,
)
from datasets import Dataset
import json
import sys

def load_eval_dataset(path: str) -> Dataset:
    """Load evaluation dataset with required columns."""
    with open(path) as f:
        data = json.load(f)

    return Dataset.from_dict({
        "question": [d["question"] for d in data],
        "answer": [d["generated_answer"] for d in data],
        "contexts": [d["retrieved_contexts"] for d in data],
        "ground_truth": [d["reference_answer"] for d in data],
    })

def run_evaluation(dataset_path: str, output_path: str):
    """Run full RAGAS evaluation suite."""
    dataset = load_eval_dataset(dataset_path)

    metrics = [
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
        context_entity_recall,
        answer_similarity,
    ]

    results = evaluate(dataset, metrics=metrics)

    # Print summary
    print("=== RAG Evaluation Results ===")
    for metric_name, score in results.items():
        print(f"  {metric_name}: {score:.4f}")

    # Save detailed results
    with open(output_path, "w") as f:
        json.dump({
            "summary": {k: float(v) for k, v in results.items()},
            "dataset_size": len(dataset),
        }, f, indent=2)

    return results

if __name__ == "__main__":
    run_evaluation(sys.argv[1], sys.argv[2])

Pontuação de Fundamentação

# groundedness.py
"""Pontua se as respostas geradas são fundamentadas no contexto recuperado."""
from openai import OpenAI
import json
from typing import List

client = OpenAI()

GROUNDEDNESS_PROMPT = """Você está avaliando se uma resposta de IA é totalmente fundamentada
nos documentos de contexto fornecidos. Pontue cada afirmação na resposta.

Documentos de contexto:
{contexts}

Resposta a ser avaliada:
{answer}

Para cada afirmação distinta na resposta, determine:
1. APOIADO - a afirmação é diretamente apoiada pelo contexto
2. PARCIALMENTE_APOIADO - a afirmação é parcialmente apoiada
3. NÃO_APOIADO - a afirmação não tem suporte no contexto

Retorne JSON:
{{
  "claims": [
    {{"claim": "...", "verdict": "APOIADO|PARCIALMENTE_APOIADO|NÃO_APOIADO", "evidence": "..."}}
  ],
  "groundedness_score": <float 0-1>,
  "unsupported_claims": ["..."]
}}
"""

def score_groundedness(answer: str, contexts: List[str]) -> dict:
    """Pontua a fundamentação de uma única resposta em relação aos seus contextos."""
    context_text = "\n---\n".join(
        f"[Documento {i+1}]: {c}" for i, c in enumerate(contexts)
    )

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": GROUNDEDNESS_PROMPT.format(
                contexts=context_text, answer=answer
            ),
        }],
        response_format={"type": "json_object"},
        temperature=0,
    )

    return json.loads(response.choices[0].message.content)

def batch_groundedness(eval_data: list) -> dict:
    """Pontua a fundamentação para um lote de pares QA."""
    scores = []
    unsupported_count = 0
    total_claims = 0

    for item in eval_data:
        result = score_groundedness(
            item["generated_answer"],
            item["retrieved_contexts"],
        )
        scores.append(result["groundedness_score"])
        unsupported_count += len(result["unsupported_claims"])

…