#!/usr/bin/env python3 """Regression comparison gate script for prompt and model releases. Compares candidate prompt evaluation scores against baseline outputs for groundedness and relevance using the Azure AI Evaluation SDK or a deterministic local heuristic fallback. Exits with 0 if regression is within tolerance, or 1 if regression exceeds tolerance. """ from __future__ import annotations import argparse import json import os import sys from typing import Any def load_jsonl(path: str) -> list[dict[str, Any]]: """Load JSONL records from disk.""" if not os.path.exists(path): print(f"Error: input file '{path}' not found", file=sys.stderr) sys.exit(2) rows: list[dict[str, Any]] = [] with open(path, "r", encoding="utf-8") as f: for line_no, line in enumerate(f, start=1): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError as err: print(f"Error: invalid JSON on line {line_no} in '{path}': {err}", file=sys.stderr) sys.exit(2) if not rows: print(f"Error: input file '{path}' contains no records", file=sys.stderr) sys.exit(2) return rows def evaluate_triplet( groundedness_evaluator: Any, relevance_evaluator: Any, query: str, context: str, response: str, ) -> tuple[float, float]: """Calculate groundedness and relevance scores (1.0 to 5.0 scale).""" # 1. SDK Evaluator path when cloud credentials are provided if groundedness_evaluator is not None and relevance_evaluator is not None: try: g_res = groundedness_evaluator(query=query, context=context, response=response) r_res = relevance_evaluator(query=query, context=context, response=response) g_score = float(g_res.get("groundedness", 3.0)) r_score = float(r_res.get("relevance", 3.0)) return g_score, r_score except Exception as err: print(f"Warning: Evaluator call failed ({err}); using fallback heuristic", file=sys.stderr) # 2. Local deterministic fallback for CI/offline runs norm_resp = response.lower().strip() norm_ctx = context.lower().strip() norm_qry = query.lower().strip() # Refusal handling: correctly refused queries are 100% grounded if "could not find" in norm_resp or "cannot answer" in norm_resp or not norm_resp: g_score = 5.0 r_score = 4.0 return g_score, r_score # Groundedness: token overlap with retrieved context resp_words = [w for w in norm_resp.split() if len(w) >= 3] g_overlap = sum(1 for w in resp_words if w in norm_ctx) / max(1, len(resp_words)) g_score = 5.0 if g_overlap >= 0.3 else (1.0 + 4.0 * (g_overlap / 0.3)) # Relevance: query intent coverage in response qry_words = [w for w in norm_qry.split() if len(w) >= 3] r_overlap = sum(1 for w in qry_words if w in norm_resp) / max(1, len(qry_words)) r_score = 5.0 if r_overlap >= 0.4 else (1.0 + 4.0 * (r_overlap / 0.4)) return round(g_score, 2), round(r_score, 2) def score_dataset( records: list[dict[str, Any]], groundedness_evaluator: Any, relevance_evaluator: Any, ) -> tuple[float, float]: """Score a dataset and return mean groundedness and relevance.""" g_scores: list[float] = [] r_scores: list[float] = [] for row in records: # Check if pre-scored values exist in dataset if "groundedness" in row and "relevance" in row: g_scores.append(float(row["groundedness"])) r_scores.append(float(row["relevance"])) continue query = str(row.get("query", "")) context = str(row.get("context", "")) response = str(row.get("response", "")) g, r = evaluate_triplet( groundedness_evaluator=groundedness_evaluator, relevance_evaluator=relevance_evaluator, query=query, context=context, response=response, ) g_scores.append(g) r_scores.append(r) mean_g = sum(g_scores) / max(1, len(g_scores)) mean_r = sum(r_scores) / max(1, len(r_scores)) return round(mean_g, 3), round(mean_r, 3) def main() -> int: parser = argparse.ArgumentParser( description="Regression comparison gate for RAG prompt releases", ) parser.add_argument("--baseline", required=True, help="Path to baseline answers JSONL file") parser.add_argument("--candidate", required=True, help="Path to candidate answers JSONL file") parser.add_argument( "--tolerance", type=float, default=0.2, help="Allowed drop in mean evaluator score (team default: 0.2 on 1-5 scale)", ) args = parser.parse_args() baseline_rows = load_jsonl(args.baseline) candidate_rows = load_jsonl(args.candidate) # Initialise cloud evaluators if credentials are provided g_eval = None r_eval = None endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT") deployment = os.environ.get("JUDGE_DEPLOYMENT") if endpoint and deployment: try: from azure.ai.evaluation import ( # type: ignore AzureOpenAIModelConfiguration, GroundednessEvaluator, RelevanceEvaluator, ) from azure.identity import DefaultAzureCredential # type: ignore model_config = AzureOpenAIModelConfiguration( azure_endpoint=endpoint, azure_deployment=deployment, credential=DefaultAzureCredential(), ) g_eval = GroundednessEvaluator(model_config) r_eval = RelevanceEvaluator(model_config) except ImportError: print("Notice: azure.ai.evaluation not installed; using local heuristic fallback", file=sys.stderr) base_g, base_r = score_dataset(baseline_rows, g_eval, r_eval) cand_g, cand_r = score_dataset(candidate_rows, g_eval, r_eval) delta_g = round(cand_g - base_g, 3) delta_r = round(cand_r - base_r, 3) g_passed = delta_g >= -args.tolerance r_passed = delta_r >= -args.tolerance print("\n" + "=" * 62) print("PROMPT EVALUATION REGRESSION GATE REPORT") print("=" * 62) print(f"{'Metric':<16} {'Baseline':<12} {'Candidate':<12} {'Delta':<10} {'Status'}") print("-" * 62) print( f"{'groundedness':<16} {base_g:<12.3f} {cand_g:<12.3f} {delta_g:<+10.3f} " f"{'PASS' if g_passed else 'FAIL'}" ) print( f"{'relevance':<16} {base_r:<12.3f} {cand_r:<12.3f} {delta_r:<+10.3f} " f"{'PASS' if r_passed else 'FAIL'}" ) print("-" * 62) print(f"Regression tolerance: maximum drop of {args.tolerance:.2f} on 1-5 scale") if g_passed and r_passed: print("Gate result: PASS - Candidate prompt is within regression tolerance.\n") return 0 print("Gate result: FAIL - Candidate prompt causes unacceptable regression.\n", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())