Avaliando RAG como engenheiro: crie um conjunto de testes de recuperação em Python

 golden/v1.jsonl ──> load + validate ──> for each question:
   (question,                               retriever.search(q, k) ──> recall@k, hit rate, MRR
    expected ids,                           generate(q, chunks)    ──> faithfulness (claims vs chunks)
    source)                                                            optional LLM judge
                                                         │
                                                         v
                                    pytest thresholds ──> CI pass / fail + JSON report

python -m venv .venv
source .venv/bin/activate   # no Windows: .venv\Scripts\activate
pip install pytest

import math
import re
from collections import Counter
 
TOKEN = re.compile(r"[a-z0-9]+")
STOPWORDS = frozenset(
    "a an and are as at be by can do does for from how i in is it my of on or the to what when which with you your".split()
)
 
 
def tokenize(text: str) -> list[str]:
    tokens = [t for t in TOKEN.findall(text.lower()) if t not in STOPWORDS]
    # Normalização tosca de plural para "plans" casar com "plan"; um sistema real usa o analyzer do seu motor de busca
    return [t[:-1] if t.endswith("s") and len(t) > 3 else t for t in tokens]
 
 
CHUNKS: dict[str, str] = {
    "billing/refunds#0": "Annual plans can be refunded in full within 30 days of purchase. After 30 days, refunds are prorated by the unused months.",
    "billing/refunds#1": "Monthly plans are not refundable, but you can cancel at any time and keep access until the end of the billing period.",
    "billing/invoices#0": "Invoices are emailed to the billing contact on the first business day of each month and can be downloaded as PDF from the Billing page.",
    "api/rate-limits#0": "The public API allows 600 requests per minute per API key. Requests above the limit receive HTTP 429 with a Retry-After header.",
    "api/webhooks#0": "Webhook deliveries are retried up to 5 times with exponential backoff. Each delivery is signed with an HMAC-SHA256 signature in the X-Signature header.",
    "account/sso#0": "Single sign-on supports SAML 2.0 and OpenID Connect. SSO is available on the Enterprise plan and is configured by an organization admin.",
    "account/password-reset#0": "Password reset links expire after 60 minutes. Users with SSO enabled must reset their password through their identity provider.",
    "data/retention#0": "Deleted projects are kept for 14 days and can be restored by an admin. After 14 days they are permanently erased, including from backups.",
}
 
 
class TfidfRetriever:
    def __init__(self, chunks: dict[str, str]) -> None:
        self.ids = list(chunks)
        counts = [Counter(tokenize(text)) for text in chunks.values()]
        df = Counter(term for doc in counts for term in doc)
        n = len(counts)
        self.idf = {term: math.log((1 + n) / (1 + freq)) + 1 for term, freq in df.items()}
        self.vectors = [self._weigh(doc) for doc in counts]
 
    def _weigh(self, counts: Counter) -> dict[str, float]:
        vector = {term: count * self.idf.get(term, 0.0) for term, count in counts.items()}
        norm = math.sqrt(sum(value * value for value in vector.values())) or 1.0
        return {term: value / norm for term, value in vector.items()}
 
    def search(self, query: str, k: int = 5) -> list[str]:
        query_vector = self._weigh(Counter(tokenize(query)))
        scored = []
        for position, (chunk_id, vector) in enumerate(zip(self.ids, self.vectors)):
            score = sum(weight * vector.get(term, 0.0) for term, weight in query_vector.items())
            if score > 0:
                scored.append((-score, position, chunk_id))
        return [chunk_id for _, _, chunk_id in sorted(scored)[:k]]

{"id": "q001", "question": "Can I get a refund on an annual plan?", "expected_chunk_ids": ["billing/refunds#0"], "source": "logs"}
{"id": "q002", "question": "Is a monthly subscription refundable if I cancel?", "expected_chunk_ids": ["billing/refunds#1"], "source": "logs"}
{"id": "q003", "question": "Which plans have refunds?", "expected_chunk_ids": ["billing/refunds#0", "billing/refunds#1"], "source": "sme"}
{"id": "q004", "question": "How many API requests can I send per minute?", "expected_chunk_ids": ["api/rate-limits#0"], "source": "sme"}
{"id": "q005", "question": "Why am I getting HTTP 429 errors?", "expected_chunk_ids": ["api/rate-limits#0"], "source": "logs"}
{"id": "q006", "question": "How do I check the webhook signature?", "expected_chunk_ids": ["api/webhooks#0"], "source": "synthetic_reviewed"}
{"id": "q007", "question": "Does single sign-on work with SAML?", "expected_chunk_ids": ["account/sso#0"], "source": "synthetic_reviewed"}
{"id": "q008", "question": "My password reset link stopped working", "expected_chunk_ids": ["account/password-reset#0"], "source": "logs"}
{"id": "q009", "question": "Can an admin restore a deleted project?", "expected_chunk_ids": ["data/retention#0"], "source": "sme"}
{"id": "q010", "question": "Where can I download my invoices?", "expected_chunk_ids": ["billing/invoices#0"], "source": "synthetic_reviewed"}
{"id": "q011", "question": "How do I get my money back?", "expected_chunk_ids": ["billing/refunds#0"], "source": "logs"}

import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
 
ALLOWED_SOURCES = {"logs", "sme", "synthetic_reviewed"}
 
 
@dataclass(frozen=True)
class GoldenQuestion:
    id: str
    question: str
    expected_chunk_ids: frozenset[str]
    source: str
 
 
@dataclass(frozen=True)
class GoldenSet:
    version: str
    sha256: str
    questions: list[GoldenQuestion]
 
 
def load_golden_set(path: Path, known_chunk_ids: set[str]) -> GoldenSet:
    raw = path.read_bytes()
    questions: list[GoldenQuestion] = []
    seen: set[str] = set()
    for line_no, line in enumerate(raw.decode("utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        row = json.loads(line)
        q = GoldenQuestion(row["id"], row["question"], frozenset(row["expected_chunk_ids"]), row["source"])
        problems = []
        if q.id in seen:
            problems.append("duplicate id")
        if not q.expected_chunk_ids:
            problems.append("no expected chunks")
        # Refazer o chunking renomeia ids; falhe alto em vez de avaliar contra chunks que não existem mais
        unknown = q.expected_chunk_ids - known_chunk_ids
        if unknown:
            problems.append(f"unknown chunk ids {sorted(unknown)}")
        if q.source not in ALLOWED_SOURCES:
            problems.append(f"unknown source {q.source!r}")
        if problems:
            raise ValueError(f"{path.name}:{line_no} ({q.id}): {'; '.join(problems)}")
        seen.add(q.id)
        questions.append(q)
    return GoldenSet(version=path.stem, sha256=hashlib.sha256(raw).hexdigest()[:12], questions=questions)

from collections import defaultdict
from dataclasses import dataclass
from statistics import mean
from typing import Callable
 
from dataset import GoldenSet
 
SearchFn = Callable[[str, int], list[str]]
 
 
def recall_at_k(retrieved: list[str], relevant: frozenset[str], k: int) -> float:
    return len(set(retrieved[:k]) & relevant) / len(relevant)
 
 
def hit_at_k(retrieved: list[str], relevant: frozenset[str], k: int) -> float:
    return 1.0 if set(retrieved[:k]) & relevant else 0.0
 
 
def reciprocal_rank(retrieved: list[str], relevant: frozenset[str]) -> float:
    for rank, chunk_id in enumerate(retrieved, start=1):
        if chunk_id in relevant:
            return 1.0 / rank
    return 0.0
 
 
@dataclass(frozen=True)
class RetrievalReport:
    k: int
    recall: float
    hit_rate: float
    mrr: float
    misses: list[str]
    recall_by_source: dict[str, float]
 
 
def evaluate_retrieval(golden: GoldenSet, search: SearchFn, k: int = 3) -> RetrievalReport:
    recalls, hits, ranks, misses = [], [], [], []
    by_source: dict[str, list[float]] = defaultdict(list)
    for q in golden.questions:
        retrieved = search(q.question, k)
        recall = recall_at_k(retrieved, q.expected_chunk_ids, k)
        recalls.append(recall)
        hits.append(hit_at_k(retrieved, q.expected_chunk_ids, k))
        ranks.append(reciprocal_rank(retrieved[:k], q.expected_chunk_ids))
        by_source[q.source].append(recall)
        if recall < 1.0:
            misses.append(f"{q.id}: expected {sorted(q.expected_chunk_ids)}, got {retrieved}")
    return RetrievalReport(
        k=k,
        recall=round(mean(recalls), 3),
        hit_rate=round(mean(hits), 3),
        mrr=round(mean(ranks), 3),
        misses=misses,
        recall_by_source={source: round(mean(values), 3) for source, values in sorted(by_source.items())},
    )

import json
from dataclasses import asdict
from pathlib import Path
 
from dataset import load_golden_set
from metrics import evaluate_retrieval
from retriever import CHUNKS, TfidfRetriever
 
if __name__ == "__main__":
    golden = load_golden_set(Path(__file__).parent / "golden" / "v1.jsonl", set(CHUNKS))
    report = evaluate_retrieval(golden, TfidfRetriever(CHUNKS).search, k=3)
    print(json.dumps({"dataset": golden.version, "sha256": golden.sha256, **asdict(report)}, indent=2))

import re
from dataclasses import dataclass
 
from retriever import tokenize
 
SENTENCE_END = re.compile(r"(?<=[.!?])\s+")
NUMBER = re.compile(r"\d+(?:\.\d+)?")
 
 
@dataclass(frozen=True)
class ClaimCheck:
    claim: str
    support: float
    missing_numbers: tuple[str, ...]
    supported: bool
 
 
@dataclass(frozen=True)
class FaithfulnessReport:
    score: float
    claims: list[ClaimCheck]
 
    @property
    def unsupported(self) -> list[str]:
        return [c.claim for c in self.claims if not c.supported]
 
 
def split_claims(answer: str) -> list[str]:
    return [s.strip() for s in SENTENCE_END.split(answer.strip()) if s.strip()]
 
 
def check_faithfulness(answer: str, contexts: list[str], min_support: float = 0.6) -> FaithfulnessReport:
    context_text = " ".join(contexts)
    context_tokens = set(tokenize(context_text))
    context_numbers = set(NUMBER.findall(context_text))
    checks = []
    for claim in split_claims(answer):
        tokens = set(tokenize(claim))
        support = len(tokens & context_tokens) / len(tokens) if tokens else 1.0
        # Um número que o contexto nunca menciona é a alucinação mais barata de pegar
        missing = tuple(sorted(set(NUMBER.findall(claim)) - context_numbers))
        checks.append(ClaimCheck(claim, round(support, 3), missing, support >= min_support and not missing))
    # Uma resposta vazia é uma falha de geração, não uma resposta perfeitamente fiel
    score = sum(c.supported for c in checks) / len(checks) if checks else 0.0
    return FaithfulnessReport(score=round(score, 3), claims=checks)

import importlib
import json
import os
from dataclasses import dataclass
from statistics import mean
from typing import Callable
 
Complete = Callable[[str, str], str]  # (model, prompt) -> texto bruto do modelo
 
PROMPT = """You check whether an answer is grounded in the provided context.
For each numbered claim, answer true only if the context directly supports it.
Reply with JSON only, one boolean per claim: {{"verdicts": [true, false]}}
 
Context:
{context}
 
Claims:
{claims}
"""
 
 
@dataclass(frozen=True)
class JudgeResult:
    verdicts: list[bool]
    agreement: float
    valid_samples: int
 
 
class LLMJudge:
    def __init__(self, complete: Complete, model: str, samples: int = 3) -> None:
        if samples < 1:
            raise ValueError("samples must be at least 1")
        self.complete = complete
        self.model = model
        self.samples = samples
 
    def judge(self, claims: list[str], contexts: list[str]) -> JudgeResult:
        prompt = PROMPT.format(
            context="\n\n".join(contexts),
            claims="\n".join(f"{i}. {claim}" for i, claim in enumerate(claims, start=1)),
        )
        runs = []
        for _ in range(self.samples):
            parsed = self._parse(self.complete(self.model, prompt), len(claims))
            if parsed is not None:
                runs.append(parsed)
        if not runs:
            raise RuntimeError("judge returned no parseable verdicts")
        verdicts, agreement = [], []
        for i in range(len(claims)):
            yes = sum(run[i] for run in runs)
            # Empate conta como não sustentado: o judge não chegar a um acordo não é evidência de grounding
            verdicts.append(yes * 2 > len(runs))
            agreement.append(max(yes, len(runs) - yes) / len(runs))
        return JudgeResult(verdicts, round(mean(agreement), 3), len(runs))
 
    @staticmethod
    def _parse(text: str, expected: int) -> list[bool] | None:
        start, end = text.find("{"), text.rfind("}")
        if start == -1 or end <= start:
            return None
        try:
            verdicts = json.loads(text[start : end + 1]).get("verdicts")
        except (json.JSONDecodeError, AttributeError):
            return None
        if not isinstance(verdicts, list) or len(verdicts) != expected:
            return None
        if not all(isinstance(v, bool) for v in verdicts):
            return None
        return verdicts
 
 
def load_judge_from_env(samples: int = 3) -> LLMJudge | None:
    model = os.environ.get("RAG_EVAL_JUDGE_MODEL")
    target = os.environ.get("RAG_EVAL_JUDGE_CLIENT")  # "package.module:function"
    if not model or not target:
        return None
    module_name, _, function_name = target.partition(":")
    complete = getattr(importlib.import_module(module_name), function_name)
    return LLMJudge(complete, model, samples)

from pathlib import Path
from statistics import mean
 
import pytest
 
from dataset import load_golden_set
from faithfulness import check_faithfulness, split_claims
from judge import LLMJudge, load_judge_from_env
from metrics import evaluate_retrieval, recall_at_k, reciprocal_rank
from retriever import CHUNKS, TfidfRetriever
 
GOLDEN_PATH = Path(__file__).parent / "golden" / "v1.jsonl"
K = 3
THRESHOLDS = {"recall": 0.85, "mrr": 0.8, "faithfulness": 0.95}
LIVE_JUDGE = load_judge_from_env()
 
 
@pytest.fixture(scope="module")
def golden():
    return load_golden_set(GOLDEN_PATH, set(CHUNKS))
 
 
@pytest.fixture(scope="module")
def retriever():
    return TfidfRetriever(CHUNKS)
 
 
def fake_generate(question: str, contexts: list[str]) -> str:
    # Substituto da chamada ao LLM para o gate rodar offline: responde com a primeira frase do chunk do topo
    return split_claims(contexts[0])[0]
 
 
def test_metric_math():
    assert recall_at_k(["a", "b", "c"], frozenset({"b", "x"}), k=2) == 0.5
    assert reciprocal_rank(["a", "b"], frozenset({"b"})) == 0.5
    assert reciprocal_rank(["a"], frozenset({"z"})) == 0.0
 
 
def test_retrieval_meets_thresholds(golden, retriever):
    report = evaluate_retrieval(golden, retriever.search, k=K)
    context = f"dataset {golden.version} ({golden.sha256}), misses: {report.misses}"
    assert report.recall >= THRESHOLDS["recall"], f"recall@{K} = {report.recall}; {context}"
    assert report.mrr >= THRESHOLDS["mrr"], f"MRR@{K} = {report.mrr}; {context}"
 
 
def test_answers_are_grounded(golden, retriever):
    scores, failures = [], []
    for q in golden.questions:
        contexts = [CHUNKS[chunk_id] for chunk_id in retriever.search(q.question, K)]
        if not contexts:
            continue  # um miss de retrieval, já contado pelo gate de retrieval
        report = check_faithfulness(fake_generate(q.question, contexts), contexts)
        scores.append(report.score)
        failures += [f"{q.id}: {claim}" for claim in report.unsupported]
    assert mean(scores) >= THRESHOLDS["faithfulness"], f"unsupported claims: {failures}"
 
 
def test_faithfulness_flags_invented_facts():
    context = [CHUNKS["billing/refunds#0"]]
    wrong_number = check_faithfulness("Annual plans can be refunded in full within 90 days of purchase.", context)
    assert wrong_number.claims[0].missing_numbers == ("90",)
    off_topic = check_faithfulness("Refunds are sent as store credit to your crypto wallet.", context)
    assert off_topic.unsupported == ["Refunds are sent as store credit to your crypto wallet."]
 
 
def test_judge_majority_vote_and_agreement():
    replies = iter(['{"verdicts": [true, false]}', 'Sure! {"verdicts": [true, true]}', "not json"])
    judge = LLMJudge(lambda model, prompt: next(replies), model="fake-judge", samples=3)
    result = judge.judge(["claim one", "claim two"], ["some context"])
    assert result.valid_samples == 2
    assert result.verdicts == [True, False]  # o empate de 1 a 1 na segunda afirmação conta como não sustentado
    assert result.agreement == 0.75
 
 
@pytest.mark.skipif(LIVE_JUDGE is None, reason="set RAG_EVAL_JUDGE_MODEL and RAG_EVAL_JUDGE_CLIENT")
def test_live_judge_confirms_grounding(golden, retriever):
    for q in golden.questions[:3]:
        contexts = [CHUNKS[chunk_id] for chunk_id in retriever.search(q.question, K)]
        claims = split_claims(fake_generate(q.question, contexts))
        result = LIVE_JUDGE.judge(claims, contexts)
        assert all(result.verdicts), f"{q.id}: judge rejected a claim (agreement {result.agreement})"

pytest -q -rs
python report.py > eval-report.json

name: rag-eval
 
on:
  pull_request:
  schedule:
    - cron: "0 3 * * *"
 
jobs:
  gate:
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install pytest
      - run: pytest -q -rs
      - run: python report.py > eval-report.json
        if: always()
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: eval-report
          path: eval-report.json
 
  nightly-judge:
    if: github.event_name == 'schedule'
    runs-on: ubuntu-latest
    env:
      RAG_EVAL_JUDGE_MODEL: ${{ vars.RAG_EVAL_JUDGE_MODEL }}
      RAG_EVAL_JUDGE_CLIENT: ${{ vars.RAG_EVAL_JUDGE_CLIENT }}
      LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install pytest  # mais o SDK que a sua função complete() usa
      - run: pytest -q -rs -k live_judge