Evaluation-Driven Quality Gates

# Define QualityGate; implement __init__, evaluate, check_gate
# Key operations: evaluation logic, threshold comparison
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class GateResult:
 """Result of a quality gate evaluation."""
 gate_name: str
 passed: bool
 overall_score: float
 threshold: float
 category_scores: dict[str, float] = field(default_factory=dict)
 regressions: list[str] = field(default_factory=list)
 details: str = ""

class QualityGate:
 """Automated quality gate for LLM deployment pipelines."""

 def __init__(
 self,
 gate_name: str,
 overall_threshold: float = 0.85,
 category_thresholds: Optional[dict[str, float]] = None,
 max_regression: float = 0.03,
 ):
 self.gate_name = gate_name
 self.overall_threshold = overall_threshold
 self.category_thresholds = category_thresholds or {}
 self.max_regression = max_regression

 def check_gate(
 self,
 scores: dict[str, list[float]],
 baseline_scores: Optional[dict[str, list[float]]] = None,
 ) -> GateResult:
 """Evaluate whether the candidate passes the quality gate.

 Args:
 scores: category_name -> list of scores for that category
 baseline_scores: previous production scores for regression check
 """
 # Compute category-level means
 category_means = {
 cat: sum(vals) / len(vals)
 for cat, vals in scores.items() if vals
 }

 # Compute overall mean across all scores
 all_scores = [s for vals in scores.values() for s in vals]
 overall = sum(all_scores) / len(all_scores) if all_scores else 0.0

 # Check for regressions against baseline
 regressions = []
 if baseline_scores:
 for cat, vals in baseline_scores.items():
 baseline_mean = sum(vals) / len(vals) if vals else 0.0
 current_mean = category_means.get(cat, 0.0)
 drop = baseline_mean - current_mean
 if drop > self.max_regression:
 regressions.append(
 f"{cat}: dropped {drop:.3f} "
 f"(baseline {baseline_mean:.3f} "
 f"-> current {current_mean:.3f})"
 )

 # Check category-level thresholds
 category_failures = []
 for cat, threshold in self.category_thresholds.items():
 if cat in category_means and category_means[cat] < threshold:
 category_failures.append(
 f"{cat}: {category_means[cat]:.3f} < {threshold}"
 )

 passed = (
 overall >= self.overall_threshold
 and len(regressions) == 0
 and len(category_failures) == 0
 )

 details_parts = []
 if regressions:
 details_parts.append(
 "Regressions: " + "; ".join(regressions)
 )
 if category_failures:
 details_parts.append(
 "Category failures: " + "; ".join(category_failures)
 )

 return GateResult(
 gate_name=self.gate_name,
 passed=passed,
 overall_score=round(overall, 4),
 threshold=self.overall_threshold,
 category_scores={k: round(v, 4) for k, v in category_means.items()},
 regressions=regressions,
 details=" | ".join(details_parts) if details_parts else "All checks passed",
 )

# Example: pre-deployment quality gate
gate = QualityGate(
 gate_name="pre-deploy",
 overall_threshold=0.85,
 category_thresholds={"safety": 0.95, "factuality": 0.80},
 max_regression=0.03,
)

candidate_scores = {
 "safety": [1.0, 1.0, 0.95, 1.0, 0.90],
 "factuality": [0.85, 0.90, 0.80, 0.88, 0.82],
 "helpfulness": [0.78, 0.82, 0.85, 0.80, 0.76],
}

baseline_scores = {
 "safety": [1.0, 1.0, 1.0, 0.95, 1.0],
 "factuality": [0.82, 0.85, 0.80, 0.84, 0.81],
 "helpfulness": [0.80, 0.83, 0.81, 0.79, 0.82],
}

result = gate.check_gate(candidate_scores, baseline_scores)
print(f"Gate: {result.gate_name}")
print(f"Passed: {result.passed}")
print(f"Overall score: {result.overall_score}")
print(f"Category scores: {result.category_scores}")
print(f"Details: {result.details}")

# Define PromptRegressionTest; implement __init__, run_test, compare
# Key operations: prompt construction, evaluation logic
from dataclasses import dataclass

@dataclass
class TestCase:
 """A single golden test case for prompt regression testing."""
 input_text: str
 expected_criteria: dict[str, str] # criterion_name -> description
 category: str = "general"

@dataclass
class RegressionResult:
 """Comparison between baseline and candidate prompt performance."""
 total_cases: int
 improved: int
 regressed: int
 unchanged: int
 regression_rate: float
 details: list[dict]

class PromptRegressionTester:
 """Run regression tests when prompts change."""

 def __init__(self, eval_fn, golden_tests: list[TestCase]):
 """
 Args:
 eval_fn: function(prompt, input_text) -> dict of scores
 golden_tests: curated test cases with expected criteria
 """
 self.eval_fn = eval_fn
 self.golden_tests = golden_tests

 def run_comparison(
 self,
 baseline_prompt: str,
 candidate_prompt: str,
 regression_threshold: float = 0.05,
 ) -> RegressionResult:
 """Compare candidate prompt against baseline on golden tests."""
 details = []
 improved = regressed = unchanged = 0

 for test in self.golden_tests:
 baseline_result = self.eval_fn(baseline_prompt, test.input_text)
 candidate_result = self.eval_fn(candidate_prompt, test.input_text)

 baseline_score = sum(baseline_result.values()) / len(baseline_result)
 candidate_score = sum(candidate_result.values()) / len(candidate_result)
 diff = candidate_score - baseline_score

 if diff > regression_threshold:
 improved += 1
 status = "improved"
 elif diff < -regression_threshold:
 regressed += 1
 status = "regressed"
 else:
 unchanged += 1
 status = "unchanged"

 details.append({
 "input": test.input_text[:80],
 "category": test.category,
 "baseline_score": round(baseline_score, 3),
 "candidate_score": round(candidate_score, 3),
 "status": status,
 })

 total = len(self.golden_tests)
 return RegressionResult(
 total_cases=total,
 improved=improved,
 regressed=regressed,
 unchanged=unchanged,
 regression_rate=round(regressed / total, 3) if total > 0 else 0.0,
 details=details,
 )

# Define ContinuousEvalScheduler; implement __init__, run_evaluation_cycle
# Key operations: evaluation logic, scheduling, alerting
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable, Optional

@dataclass
class EvalCycleResult:
 """Result of a single continuous evaluation cycle."""
 timestamp: str
 gate_result: dict
 samples_evaluated: int
 alert_triggered: bool
 alert_message: str = ""

class ContinuousEvalScheduler:
 """Schedule periodic evaluation of production LLM outputs."""

 def __init__(
 self,
 sample_fn: Callable[[], list[dict]],
 eval_fn: Callable[[dict], dict[str, float]],
 gate: "QualityGate",
 baseline_scores: Optional[dict[str, list[float]]] = None,
 ):
 """
 Args:
 sample_fn: returns recent production request/response pairs
 eval_fn: scores a single request/response pair
 gate: QualityGate instance for threshold checking
 baseline_scores: production baseline for regression detection
 """
 self.sample_fn = sample_fn
 self.eval_fn = eval_fn
 self.gate = gate
 self.baseline_scores = baseline_scores
 self.history: list[EvalCycleResult] = []

 def run_evaluation_cycle(self) -> EvalCycleResult:
 """Run one evaluation cycle on sampled production traffic."""
 samples = self.sample_fn()
 if not samples:
 return EvalCycleResult(
 timestamp=datetime.now(timezone.utc).isoformat(),
 gate_result={},
 samples_evaluated=0,
 alert_triggered=False,
 alert_message="No samples available",
 )

 # Evaluate each sample and group scores by category
 category_scores: dict[str, list[float]] = {}
 for sample in samples:
 scores = self.eval_fn(sample)
 for category, score in scores.items():
 category_scores.setdefault(category, []).append(score)

 # Run through the quality gate
 gate_result = self.gate.check_gate(
 category_scores, self.baseline_scores
 )

 alert = not gate_result.passed
 cycle = EvalCycleResult(
 timestamp=datetime.now(timezone.utc).isoformat(),
 gate_result={
 "passed": gate_result.passed,
 "overall_score": gate_result.overall_score,
 "category_scores": gate_result.category_scores,
 "regressions": gate_result.regressions,
 },
 samples_evaluated=len(samples),
 alert_triggered=alert,
 alert_message=gate_result.details if alert else "",
 )
 self.history.append(cycle)
 return cycle

 def get_trend(self, last_n: int = 7) -> list[float]:
 """Return recent overall scores for trend analysis."""
 recent = self.history[-last_n:] if self.history else []
 return [
 h.gate_result.get("overall_score", 0.0)
 for h in recent
 if h.gate_result
 ]