diff --git a/regression.py b/regression.py new file mode 100644 index 0000000..f5b42c8 --- /dev/null +++ b/regression.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import json +import statistics +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parent +EVAL_DIR = ROOT / "data" / "eval" +DEFAULT_DATASET = EVAL_DIR / "islamic_qa_benchmark.jsonl" +DEFAULT_BASELINE = EVAL_DIR / "baseline_results.json" + + +class RegressionRunner: + def __init__(self, dataset_path: Path | str = DEFAULT_DATASET, baseline_path: Path | str = DEFAULT_BASELINE) -> None: + self.dataset_path = Path(dataset_path) + self.baseline_path = Path(baseline_path) + + def load_dataset(self) -> list[dict[str, Any]]: + if not self.dataset_path.exists(): + return [] + records = [] + with open(self.dataset_path, encoding="utf-8") as f: + for line in f: + if line.strip(): + records.append(json.loads(line)) + return records + + def load_baseline(self) -> dict[str, Any] | None: + if not self.baseline_path.exists(): + return None + with open(self.baseline_path, encoding="utf-8") as f: + return json.load(f) + + def save_baseline(self, results: dict[str, Any]) -> None: + self.baseline_path.parent.mkdir(parents=True, exist_ok=True) + with open(self.baseline_path, "w", encoding="utf-8") as f: + json.dump(results, f, ensure_ascii=False, indent=2) + + def evaluate_model_output(self, record: dict[str, Any], answer: str) -> dict[str, Any]: + from scripts.eval_islamic_qa import evaluate_single_response + return evaluate_single_response(record, answer) + + def run_suite(self, model_callable: Any, max_items: int | None = None) -> dict[str, Any]: + records = self.load_dataset() + if max_items is not None: + records = records[:max_items] + + scores = [] + passed_count = 0 + domain_scores: dict[str, list[float]] = {} + + results_detail = [] + for r in records: + question = r["question"] + try: + answer = model_callable(question) + except Exception as e: + answer = f"ERROR: {e}" + + eval_res = self.evaluate_model_output(r, answer) + score = eval_res.get("composite_score", 0.0) + passed = eval_res.get("passed", False) + + scores.append(score) + if passed: + passed_count += 1 + + domain = r.get("domain", "general") + domain_scores.setdefault(domain, []).append(score) + + results_detail.append({ + "id": r["id"], + "domain": domain, + "score": score, + "passed": passed, + }) + + mean_score = statistics.mean(scores) if scores else 0.0 + pass_rate = (passed_count / len(records)) if records else 0.0 + + domain_summary = { + d: statistics.mean(vals) if vals else 0.0 + for d, vals in domain_scores.items() + } + + summary = { + "total_evaluated": len(records), + "mean_composite_score": round(mean_score, 4), + "pass_rate": round(pass_rate, 4), + "domain_scores": domain_summary, + "details": results_detail, + } + return summary + + def compare_with_baseline(self, current_results: dict[str, Any], tolerance: float = 0.02) -> dict[str, Any]: + baseline = self.load_baseline() + if not baseline: + return { + "has_baseline": False, + "degraded": False, + "message": "No baseline found. Current results established as new baseline.", + } + + base_score = baseline.get("mean_composite_score", 0.0) + curr_score = current_results.get("mean_composite_score", 0.0) + + diff = curr_score - base_score + degraded = diff < (-tolerance) + + domain_diffs = {} + base_domains = baseline.get("domain_scores", {}) + curr_domains = current_results.get("domain_scores", {}) + for d, curr_val in curr_domains.items(): + base_val = base_domains.get(d, curr_val) + domain_diffs[d] = round(curr_val - base_val, 4) + + return { + "has_baseline": True, + "baseline_score": base_score, + "current_score": curr_score, + "difference": round(diff, 4), + "degraded": degraded, + "tolerance": tolerance, + "domain_differences": domain_diffs, + "alert": "REGRESSION DETECTED: Performance dropped below baseline tolerance." if degraded else "OK", + } diff --git a/tests/test_regression.py b/tests/test_regression.py new file mode 100644 index 0000000..eea771f --- /dev/null +++ b/tests/test_regression.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from regression import RegressionRunner + + +@pytest.fixture +def sample_dataset(tmp_path: Path) -> Path: + ds_file = tmp_path / "benchmark.jsonl" + records = [ + { + "id": "reg-001", + "domain": "aqeedah", + "question": "What is Tawhid?", + "question_ar": "ما التوحيد؟", + "expected_answer": "Tawhid is the oneness of Allah.", + "expected_answer_ar": "التوحيد هو توحيد الله", + "key_points": ["oneness of Allah"], + "evaluation_criteria": { + "must_include": ["oneness"], + "must_not_include": ["polytheism"], + }, + }, + { + "id": "reg-002", + "domain": "fiqh_ibadat", + "question": "How many daily prayers?", + "question_ar": "كم عدد الصلوات؟", + "expected_answer": "There are five daily prayers.", + "expected_answer_ar": "الصلوات خمس.", + "key_points": ["five daily prayers"], + "evaluation_criteria": { + "must_include": ["five"], + "must_not_include": ["ten"], + }, + }, + ] + with open(ds_file, "w", encoding="utf-8") as f: + for r in records: + f.write(json.dumps(r) + "\n") + return ds_file + + +class TestRegressionRunner: + def test_run_suite_successful(self, sample_dataset: Path, tmp_path: Path): + baseline_file = tmp_path / "baseline.json" + runner = RegressionRunner(dataset_path=sample_dataset, baseline_path=baseline_file) + + def dummy_model(q: str) -> str: + if "Tawhid" in q: + return "Tawhid is the oneness of Allah." + return "There are five daily prayers." + + results = runner.run_suite(dummy_model) + assert results["total_evaluated"] == 2 + assert results["mean_composite_score"] > 0.8 + assert results["pass_rate"] == 1.0 + assert "aqeedah" in results["domain_scores"] + + def test_baseline_comparison_no_regression(self, sample_dataset: Path, tmp_path: Path): + baseline_file = tmp_path / "baseline.json" + runner = RegressionRunner(dataset_path=sample_dataset, baseline_path=baseline_file) + + baseline_data = { + "total_evaluated": 2, + "mean_composite_score": 0.95, + "pass_rate": 1.0, + "domain_scores": {"aqeedah": 0.95, "fiqh_ibadat": 0.95}, + } + runner.save_baseline(baseline_data) + + current_results = { + "total_evaluated": 2, + "mean_composite_score": 0.96, + "pass_rate": 1.0, + "domain_scores": {"aqeedah": 0.96, "fiqh_ibadat": 0.96}, + } + + comp = runner.compare_with_baseline(current_results) + assert comp["has_baseline"] is True + assert comp["degraded"] is False + assert "OK" in comp["alert"] + + def test_baseline_comparison_detects_regression(self, sample_dataset: Path, tmp_path: Path): + baseline_file = tmp_path / "baseline.json" + runner = RegressionRunner(dataset_path=sample_dataset, baseline_path=baseline_file) + + baseline_data = { + "total_evaluated": 2, + "mean_composite_score": 0.95, + "pass_rate": 1.0, + "domain_scores": {"aqeedah": 0.95, "fiqh_ibadat": 0.95}, + } + runner.save_baseline(baseline_data) + + current_results = { + "total_evaluated": 2, + "mean_composite_score": 0.80, + "pass_rate": 0.5, + "domain_scores": {"aqeedah": 0.80, "fiqh_ibadat": 0.80}, + } + + comp = runner.compare_with_baseline(current_results, tolerance=0.02) + assert comp["has_baseline"] is True + assert comp["degraded"] is True + assert "REGRESSION DETECTED" in comp["alert"]