""" RAGAS Regression Validation Script Runs the RAG pipeline against a lightweight golden set and checks that all metrics remain above defined thresholds. Exit code 0 = pass, 1 = fail. Usage: python scripts/validate_ragas_regression.py python scripts/validate_ragas_regression.py --golden-set data/ground_truth/golden_set_ci.json """ import json import os import sys import time from typing import Any, cast sys.path.insert(0, os.getcwd()) from src.reasoning.pipeline import ReasoningPipeline # Stricter thresholds for full 68-pair evaluation on your machine. # CI uses lower thresholds below because the 5-pair subset has high variance. THRESHOLDS = { "faithfulness": 0.80, "answer_relevancy": 0.55, "context_precision": 0.75, "context_recall": 0.45, "answer_completeness": 0.55, } CI_THRESHOLDS = { "faithfulness": 0.30, "answer_relevancy": 0.30, "context_precision": 0.50, "context_recall": 0.30, "answer_completeness": 0.30, } def load_golden_set(path: str) -> list[dict[str, Any]]: with open(path, encoding="utf-8") as f: return cast(list[dict[str, Any]], json.load(f)) def compute_ragas_scores(pipeline: ReasoningPipeline, question: str, ground_truth_answer: str) -> dict[str, float]: """Run a single query and compute proxy RAGAS scores.""" result = pipeline.run(question) answer = str(result.get("generated_answer", "")) contexts_raw = result.get("retrieved_context", []) contexts: list[str] = [] for ctx in contexts_raw: if isinstance(ctx, dict): text = ctx.get("text", ctx.get("content", "")) if text: contexts.append(str(text)) elif isinstance(ctx, str) and ctx: contexts.append(ctx) stopwords = { "the", "a", "an", "is", "are", "was", "were", "be", "been", "being", "have", "has", "had", "do", "does", "did", "will", "would", "could", "should", "may", "might", "must", "shall", "to", "of", "in", "for", "on", "with", "at", "by", "from", "as", "into", "through", "during", "this", "that", "these", "those", "it", "its", "and", "or", "but", "not", "no", "nor", "just", "so", "than", "too", "very", "s", "t", } scores: dict[str, float] = {} # Context precision q_words = set(question.lower().split()) - stopwords if q_words and contexts: relevant = sum(1 for ctx in contexts if len(q_words & set(ctx.lower().split())) >= 2) scores["context_precision"] = relevant / len(contexts) else: scores["context_precision"] = 0.0 # Answer relevancy: keyword overlap ratio between question and answer. # No length multiplier — answer length is already captured by answer_completeness. # This avoids penalizing concise-but-relevant answers and reduces CI flakiness # from LLM output length variance on the 5-pair subset. a_words = set(answer.lower().split()) - stopwords if q_words and a_words: overlap = len(q_words & a_words) scores["answer_relevancy"] = overlap / len(q_words) else: scores["answer_relevancy"] = 0.0 # Answer completeness length = len(answer.split()) if length < 20: scores["answer_completeness"] = 0.3 elif length < 50: scores["answer_completeness"] = 0.6 elif length < 100: scores["answer_completeness"] = 0.8 else: scores["answer_completeness"] = 1.0 # Faithfulness if answer and contexts: import re combined = " ".join(contexts).lower() sentences = [s.strip() for s in re.split(r"[.!?]+", answer) if len(s.strip()) >= 10] if sentences: grounded = 0 for sentence in sentences: words = set(sentence.lower().split()) - stopwords if words: overlap = len(words & set(combined.split())) if overlap / len(words) > 0.3: grounded += 1 scores["faithfulness"] = grounded / len(sentences) else: scores["faithfulness"] = 0.5 else: scores["faithfulness"] = 0.0 # Context recall if ground_truth_answer and contexts: combined = " ".join(contexts).lower() gt_words = set(ground_truth_answer.lower().split()) - stopwords if gt_words: matched = sum(1 for w in gt_words if len(w) >= 4 and w in combined) scores["context_recall"] = matched / len(gt_words) else: scores["context_recall"] = 0.0 else: scores["context_recall"] = 0.0 return scores def main() -> int: golden_set_path = ( sys.argv[1] if len(sys.argv) > 1 and sys.argv[1] != "--golden-set" else ( sys.argv[2] if "--golden-set" in sys.argv and len(sys.argv) > sys.argv.index("--golden-set") + 1 else "data/ground_truth/golden_set_ci.json" ) ) if not os.path.exists(golden_set_path): print(f"Golden set not found: {golden_set_path}") print("Skipping RAGAS regression check (no golden set).") return 0 golden_set = load_golden_set(golden_set_path) print(f"Loaded {len(golden_set)} golden QA pairs from {golden_set_path}") is_ci = "_ci" in golden_set_path thresholds = CI_THRESHOLDS if is_ci else THRESHOLDS if is_ci: print("Using CI thresholds (adjusted for small sample variance)") pipeline = ReasoningPipeline() all_scores: list[dict[str, float]] = [] failures = 0 for item in golden_set: qid = item["question_id"] question = item["question"] gt_answer = item.get("ground_truth_answer", "") print(f" [{qid}] {question[:60]}...", end=" ", flush=True) try: scores = compute_ragas_scores(pipeline, question, gt_answer) all_scores.append(scores) print("OK") time.sleep(1) except Exception as e: print(f"FAIL: {e}") failures += 1 if not all_scores: print("\nNo scores computed. Check pipeline availability.") return 1 if failures > 0 else 0 aggregated: dict[str, float] = {} for metric in thresholds: values = [s.get(metric, 0) for s in all_scores] aggregated[metric] = sum(values) / len(values) if values else 0 print("\n--- RAGAS Regression Report ---") passed = True for metric, threshold in thresholds.items(): actual = aggregated.get(metric, 0) status = "PASS" if actual >= threshold else "FAIL" if status == "FAIL": passed = False print(f" {metric:25s}: {actual:.4f} (threshold: {threshold}) [{status}]") if failures > 0: print(f"\nQuery failures: {failures}") passed = False print(f"\nOverall: {'PASS' if passed else 'FAIL'}") return 0 if passed else 1 if __name__ == "__main__": sys.exit(main())