RAG Evaluation

from dataclasses import dataclass
from statistics import mean
from typing import List

@dataclass
class RAGSample:
    query: str
    retrieved_context: List[str]    # các đoạn retrieved
    answer: str                      # câu trả lời cuối của LLM

class RAGEvaluator:
    """Bộ ba RAG Triad — reference-free, dùng LLM-as-judge."""

    def __init__(self, judge_llm, embedder, n_generated: int = 3):
        self.judge = judge_llm
        self.embed = embedder
        self.n_generated = n_generated

    # ── 1. FAITHFULNESS ──────────────────────────────────────
    def faithfulness(self, s: RAGSample) -> float:
        claims = self.judge.decompose_to_claims(s.answer)
        if not claims:
            return 1.0  # không khẳng định gì = không bịa
        supported = 0
        for c in claims:
            if self.judge.is_supported(claim=c, context=s.retrieved_context):
                supported += 1
        return supported / len(claims)

    # ── 2. CONTEXT RELEVANCE ─────────────────────────────────
    def context_relevance(self, s: RAGSample) -> float:
        sentences = self.judge.split_into_sentences(s.retrieved_context)
        if not sentences:
            return 0.0
        relevant = sum(
            1 for sent in sentences
            if self.judge.is_relevant(sentence=sent, query=s.query)
        )
        return relevant / len(sentences)

    # ── 3. ANSWER RELEVANCE ──────────────────────────────────
    def answer_relevance(self, s: RAGSample) -> float:
        generated_qs = self.judge.generate_questions(
            answer=s.answer, n=self.n_generated
        )
        q_vec = self.embed(s.query)
        sims = [cosine(q_vec, self.embed(q)) for q in generated_qs]
        return mean(sims) if sims else 0.0

    # ── TRIAD ────────────────────────────────────────────────
    def evaluate(self, samples: List[RAGSample]) -> dict:
        rows = []
        for s in samples:
            rows.append({
                "faithfulness": self.faithfulness(s),
                "context_rel":  self.context_relevance(s),
                "answer_rel":   self.answer_relevance(s),
            })
        agg = {
            "faithfulness": mean(r["faithfulness"] for r in rows),
            "context_rel":  mean(r["context_rel"]  for r in rows),
            "answer_rel":   mean(r["answer_rel"]   for r in rows),
        }
        agg["triad_mean"] = mean(agg.values())
        return agg

def cosine(a, b):
    import numpy as np
    a, b = np.array(a), np.array(b)
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

pipeline:
  dataset: s3://ai-edu/evals/rag-golden-v3.jsonl  # 180 samples
  sample_budget: 180

judge:
  model: anthropic/claude-opus-4.7
  temperature: 0.0

embedder:
  model: bge-m3               # đa ngôn ngữ VN + EN tốt
  dim: 1024

metrics:
  - faithfulness
  - context_relevance
  - answer_relevance

slo:
  faithfulness_min: 0.90
  context_relevance_min: 0.80
  answer_relevance_min: 0.85
  regression_tolerance: 0.02  # sụt > 2% so với baseline = fail

reporting:
  format: json+markdown
  sink: bigquery://ai.evals.rag_runs
  on_regression:
    - slack://#rag-quality
    - create_issue: repo/ragapp