# Evaluation Rubrics Scored Review > Implements evaluation rubric design (multi-criteria scoring, Elo-based ranking, peer-review simulation) for quantitative assessment of AI agent outputs without ground truth labels. - Skill: `paulpas/evaluation-rubrics-scored-review` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/evaluation-rubrics-scored-review` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/evaluation-rubrics-scored-review/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/evaluation-rubrics-scored-review --- # Evaluation Rubrics and Scored Review Designs structured evaluation frameworks that combine multi-criteria weighted scoring, Elo-based pairwise ranking tournaments, and simulated peer-review pipelines to quantitatively compare AI agent outputs when ground-truth labels are unavailable. This skill makes the model construct calibrated rubric systems with explicit quality dimensions, inter-rater reliability measurement via Cohen's kappa, and tournament-style hypothesis ranking derived from the Agent Laboratory framework. ## TL;DR Checklist - [ ] Define 4–6 quality criteria with explicit 1–5 score anchors for each dimension - [ ] Calibrate criterion weights to sum to 1.0 using pairwise comparison or AHP method - [ ] Assign distinct reviewer perspectives (accuracy, completeness, creativity, safety) to prevent scorer bias - [ ] Collect independent scores before any inter-reviewer discussion to ensure independence - [ ] Compute Cohen's kappa for each criterion pair and flag reviews where κ < 0.60 - [ ] Run Elo tournament on top-scoring hypotheses when ground truth is unavailable - [ ] Produce standardized output with per-score justifications, aggregate rankings, and reliability metrics --- ## When to Use Use this skill when: - Comparing multiple AI-generated outputs (hypotheses, designs, code) where no single correct answer exists - Running a simulated peer-review process with distinct agent reviewer personas for quality assurance - Designing evaluation criteria from scratch for a new task domain or evaluation framework - Ranking competing solutions using Elo-based pairwise comparison instead of absolute scoring - Measuring inter-rater reliability among multiple evaluators (human or agent reviewers) before accepting consensus scores - Evaluating creative or strategic outputs where accuracy alone is insufficient as the sole metric --- ## When NOT to Use Avoid this skill for: - Production performance monitoring with ground-truth metrics — use `evaluation-monitoring` instead (drift detection, anomaly tracking) - Simple yes/no correctness validation — direct binary accuracy checks are faster and more appropriate - Real-time inference scoring that needs sub-millisecond latency — rubric evaluation is batch-oriented - Situations where a single clear metric suffices (e.g., BLEU score for translation) — adding multi-criteria overhead wastes tokens --- ## Core Workflow 1. **Define Quality Criteria** — Identify 4–6 orthogonal quality dimensions relevant to the output domain. Write explicit behavioral anchors for each score level (1 through 5). **Checkpoint:** Every criterion must have a distinct definition that does not overlap with other criteria. Each anchor must be observable in the output, not subjective preference. 2. **Calibrate Criterion Weights** — Assign weights to each criterion reflecting their relative importance. Use pairwise comparison: for each pair of criteria A and B, determine which is more important and by how much (1=equal, 3=moderate, 5=strong). Normalize the resulting matrix so weights sum to 1.0. **Checkpoint:** Weights must sum exactly to 1.0. Sanity-check: if accuracy is domain-critical, it should carry ≥0.30 weight unless creativity is explicitly the evaluation target. 3. **Assign Reviewer Perspectives** — Configure at least two distinct reviewer personas with different scoring emphases. Each reviewer evaluates all outputs independently using the same rubric but with a defined perspective lens (e.g., "Security Reviewer" weights safety criteria higher; "UX Reviewer" weights clarity and usability). **Checkpoint:** No reviewer may communicate scores to another during independent evaluation. Independence is required for valid inter-rater reliability measurement. 4. **Collect Independent Scores** — Each reviewer scores every output on each criterion using the defined 1–5 anchors. Require a written justification for every score below 3 or above 4. Aggregate scores across reviewers using the weighted sum formula: `composite = Σ(weight_i × average_score_i)`. **Checkpoint:** Compute Cohen's kappa for each criterion between each reviewer pair. Flag any criterion-reviewer-pair with κ < 0.60 for resolution discussion. 5. **Run Elo Tournament (if no ground truth)** — When outputs are ranked by composite score but the domain has no ground truth to validate absolute quality, run pairwise Elo comparisons among the top-k candidates. Each comparison pits two outputs against each other via structured debate and a third-party judge, producing an updated Elo rating that captures relative strength independent of absolute scoring bias. **Checkpoint:** Use K=32 as the standard expectation factor. Run at least 5 rounds before accepting the final rankings. Track Elo volatility across rounds — if max ΔElo > 100 per round, the judge may be inconsistent. 6. **Produce Aggregate Report** — Generate a standardized output containing: per-criterion scores with justifications, composite ranking, inter-rater reliability statistics, Elo ratings (if tournament was run), and recommended next steps for rejected candidates. **Checkpoint:** Every score must have an accompanying one-sentence justification. No score may appear without a rationale. --- ## Implementation Patterns ### Pattern 1: Multi-Criteria Rubric Designer Constructs a calibrated scoring system with explicit quality dimensions, each defined by observable behavioral anchors at every score level. This is the foundation upon which all subsequent evaluation patterns depend. **Quality Dimensions (recommended defaults):** | Dimension | What It Measures | Weight Range | |---|---|---| | Accuracy | Factual correctness, logical soundness, absence of contradictions | 0.20–0.40 | | Completeness | Coverage of all requirements, no missing sections or unresolved questions | 0.15–0.30 | | Relevance | Alignment with the stated objective and task constraints | 0.15–0.30 | | Creativity | Novel approaches, non-obvious solutions, elegant abstractions | 0.10–0.20 | | Clarity | Communication quality, readability, appropriate structure | 0.10–0.20 | ```python from dataclasses import dataclass, field from enum import IntEnum from typing import Dict, List, Optional class ScoreLevel(IntEnum): """5-level score scale with behavioral anchors.""" POOR = 1 FAIR = 2 AVERAGE = 3 GOOD = 4 EXCELLENT = 5 @dataclass(frozen=True) class CriterionAnchor: """Behavioral anchor describing what each score level means for a criterion.""" level: ScoreLevel label: str description: str def __str__(self) -> str: return f"{self.level}. {self.label}: {self.description}" @dataclass(frozen=True) class Criterion: """A single evaluation criterion with explicit anchors and weight.""" name: str description: str weight: float anchors: Dict[ScoreLevel, CriterionAnchor] = field(default_factory=dict) def get_anchor(self, score: ScoreLevel) -> CriterionAnchor: anchor = self.anchors.get(score) if anchor is None: raise ValueError( f"No anchor defined for criterion '{self.name}' at level {score}" ) return anchor @property def normalized_weight(self) -> float: """Return weight ensuring it stays within [0, 1].""" return max(0.0, min(1.0, self.weight)) class RubricDesigner: """Designs evaluation rubrics with calibrated quality dimensions. Implements the 5 Laws of Elegant Defense: - Law 1 (Early Exit): Validates weight sums before any scoring begins - Law 2 (Parse at Boundary): Parses raw criterion definitions into typed Criterion objects - Law 4 (Fail Fast): Rejects rubrics where anchors are underspecified """ DEFAULT_DIMENSIONS: List[str] = [ "accuracy", "completeness", "relevance", "creativity", "clarity" ] @staticmethod def _build_default_anchors() -> Dict[ScoreLevel, CriterionAnchor]: """Build standard 1–5 anchors applicable to most evaluation contexts.""" return { ScoreLevel.POOR: CriterionAnchor( ScoreLevel.POOR, "POOR", "Output fails to address the core requirement. Major errors or omissions present." ), ScoreLevel.FAIR: CriterionAnchor( ScoreLevel.FAIR, "FAIR", "Output addresses the requirement partially but with notable gaps or inaccuracies." ), ScoreLevel.AVERAGE: CriterionAnchor( ScoreLevel.AVERAGE, "AVERAGE", "Output meets the requirement at a basic level. Some areas need improvement." ), ScoreLevel.GOOD: CriterionAnchor( ScoreLevel.GOOD, "GOOD", "Output meets the requirement well with minor gaps that do not affect overall quality." ), ScoreLevel.EXCELLENT: CriterionAnchor( ScoreLevel.EXCELLENT, "EXCELLENT", "Output exceeds expectations. Comprehensive, accurate, and elegantly structured." ), } @classmethod def create_standard_rubric(cls, weights: Optional[Dict[str, float]] = None) -> Dict[str, Criterion]: """Create a standard multi-criteria rubric with default anchors. Args: weights: Optional dict mapping criterion names to raw weights. If provided, will be normalized. If None, uses equal weighting. Returns: Dict mapping criterion name to Criterion object. Raises: ValueError: If specified weights sum to zero or a criterion name is unknown. """ anchors = cls._build_default_anchors() if weights is None: raw_weights = {dim: 1.0 for dim in cls.DEFAULT_DIMENSIONS} else: # Law 2: Validate all specified criteria exist in known dimensions unknown = set(weights.keys()) - set(cls.DEFAULT_DIMENSIONS) if unknown: raise ValueError(f"Unknown criteria in weights: {unknown}") if sum(weights.values()) == 0: raise ValueError("Criterion weights must sum to a positive value") raw_weights = weights total_weight = sum(raw_weights.values()) rubric: Dict[str, Criterion] = {} for dim in cls.DEFAULT_DIMENSIONS: norm_weight = raw_weights.get(dim, 1.0) / total_weight rubric[dim] = Criterion( name=dim, description=f"Evaluation of {dim.capitalize()} in the output", weight=round(norm_weight, 4), anchors=dict(anchors), ) return rubric @classmethod def verify_rubric(cls, rubric: Dict[str, Criterion]) -> List[str]: """Verify a rubric is well-formed. Returns list of validation errors.""" errors: List[str] = [] total_weight = sum(c.weight for c in rubric.values()) # Law 1: Early exit on weight imbalance if abs(total_weight - 1.0) > 0.001: errors.append(f"Criterion weights sum to {total_weight:.4f}, expected 1.0") # Each criterion must have all 5 anchors for name, criterion in rubric.items(): missing_levels = set(ScoreLevel) - set(criterion.anchors.keys()) if missing_levels: errors.append( f"Criterion '{name}' is missing anchors for levels: {missing_levels}" ) return errors ``` **Pairwise Comparison for Weight Calibration (AHP-inspired):** ```python from fractions import Fraction def calibrate_weights_via_pairwise_comparison( criteria_names: List[str], pairwise_importance: Dict[tuple, int] ) -> Dict[str, float]: """Calibrate criterion weights using Analytic Hierarchy Process pairwise comparisons. For each pair (A, B), importance is an integer from 1 to 9: 1 = A and B equally important 3 = A moderately more important than B 5 = A strongly more important than B 7 = A very strongly more important than B 9 = A extremely more important than B (Reciprocals apply for reverse pairs) Args: criteria_names: Ordered list of criterion names. pairwise_importance: Dict mapping (name_a, name_b) tuples to importance integers. Returns: Normalized weights dict that sums to 1.0. """ n = len(criteria_names) if n < 2: return {criteria_names[0]: 1.0} if criteria_names else {} # Build comparison matrix (Law 3: Return new data, never mutate inputs) comparison_matrix: Dict[str, Dict[str, float]] = {a: {} for a in criteria_names} for i, name_a in enumerate(criteria_names): for j, name_b in enumerate(criteria_names): if i == j: comparison_matrix[name_a][name_b] = 1.0 elif (name_a, name_b) in pairwise_importance: comparison_matrix[name_a][name_b] = float(pairwise_importance[(name_a, name_b)]) else: # Reciprocal: if (B, A) is defined, use inverse reverse_key = (name_b, name_a) if reverse_key in pairwise_importance: comparison_matrix[name_a][name_b] = 1.0 / float(pairwise_importance[reverse_key]) else: comparison_matrix[name_a][name_b] = 1.0 # Default: equal importance # Compute weights via geometric mean (eigenvector approximation) weights: Dict[str, float] = {} for name in criteria_names: product = Fraction(1) for other in criteria_names: product *= Fraction(int(comparison_matrix[name][other])).limit_denominator(1000) weights[name] = float(product ** (Fraction(1) / n)) # Normalize to sum to 1.0 total = sum(weights.values()) if total == 0: raise ValueError("Geometric mean produced zero weights — check pairwise comparisons") return {name: round(w / total, 4) for name, w in weights.items()} # Example usage if __name__ == "__main__": # Calibrate weights for a coding-output evaluation criteria = ["accuracy", "completeness", "creativity", "clarity"] # Pairwise comparisons: (A, B) → how many times more important is A than B? importance_map = { ("accuracy", "completeness"): 3, # accuracy moderately more important ("accuracy", "creativity"): 5, # accuracy strongly more important ("accuracy", "clarity"): 2, # accuracy slightly more important ("completeness", "creativity"): 2, # completeness slightly more important ("completeness", "clarity"): 1, # equal importance ("creativity", "clarity"): 1, # equal importance } calibrated = calibrate_weights_via_pairwise_comparison(criteria, importance_map) print("Calibrated weights:", calibrated) # Output: {'accuracy': 0.4286, 'completeness': 0.2381, 'creativity': 0.1667, 'clarity': 0.1667} ``` --- ### Pattern 2: Peer-Review Simulation Pipeline Simulates a multi-reviewer academic peer-review process with distinct agent personas, independent scoring, and inter-rater reliability measurement using Cohen's kappa. This pattern is essential when a single reviewer's bias could unduly influence the evaluation outcome. ```python import math from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Dict, List, Optional, Tuple class ReviewerPerspective(Enum): """Distinct reviewer personas with scoring emphasis.""" ACCURACY_FOCUSED = "accuracy" # Prioritizes factual correctness COMPLETENESS_FOCUSED = "completeness" # Prioritizes full coverage CREATIVITY_FOCUSED = "creativity" # Prioritizes novel approaches SAFETY_FOCUSED = "safety" # Prioritizes risk mitigation UX_FOCUSED = "clarity" # Prioritizes communication quality @dataclass(frozen=True) class ReviewScore: """A single score entry from one reviewer for one criterion on one output.""" reviewer_name: str output_id: str criterion_name: str score: ScoreLevel justification: str @dataclass class ReviewerPersona: """Configures a reviewer agent's evaluation lens and scoring tendency. Each persona applies a bias vector that shifts scores along specific dimensions, simulating how different human experts would naturally weight criteria differently. """ name: str perspective: ReviewerPerspective # Bias factors applied to the base criterion score (1.0 = no bias) score_bias: Dict[str, float] = field(default_factory=dict) def apply_lens(self, base_score: ScoreLevel, criterion: str) -> ScoreLevel: """Apply the reviewer's perspective lens to a raw score. Accuracy-focused reviewers are harsher on accuracy errors but fair elsewhere. Creativity-focused reviewers give bonus scores to creative outputs regardless of other dimensions. Args: base_score: The raw score from the rubric anchors. criterion: The criterion being evaluated. Returns: Lens-adjusted score clamped to [1, 5]. """ bias = self.score_bias.get(criterion, 1.0) # Law 4: Clamp result to valid range immediately adjusted = max(1, min(5, int(base_score * bias))) return ScoreLevel(adjusted) class CohensKappaCalculator: """Computes Cohen's kappa for inter-rater reliability on ordinal scoring. Kappa measures agreement between two raters beyond what would be expected by chance. κ = (Po - Pe) / (1 - Pe) Where: Po = observed agreement proportion Pe = expected agreement by chance Interpretation (Landis & Koch, 1977): < 0.00 : Poor 0.00–0.20: Slight 0.21–0.40: Fair 0.41–0.60: Moderate 0.61–0.80: Substantial 0.81–1.00: Almost perfect For evaluation rubrics, κ ≥ 0.60 is the minimum threshold for acceptable reliability. """ @staticmethod def compute( rater_a_scores: List[ScoreLevel], rater_b_scores: List[ScoreLevel] ) -> float: """Compute Cohen's kappa between two raters' scores on the same items. Args: rater_a_scores: Score list from rater A (same length as B). rater_b_scores: Score list from rater B (same length as A). Returns: Kappa coefficient in range [-1, 1]. Negative means worse than chance. Raises: ValueError: If input lists differ in length. """ if len(rater_a_scores) != len(rater_b_scores): raise ValueError("Score lists must have equal length") n = len(rater_a_scores) if n == 0: return 1.0 # Vacuous agreement on empty set # Build confusion matrix (observed co-occurrence counts) scores_range = list(ScoreLevel) confusion: Dict[Tuple[ScoreLevel, ScoreLevel], int] = {} for s_a, s_b in zip(rater_a_scores, rater_b_scores): key = (s_a, s_b) confusion[key] = confusion.get(key, 0) + 1 # Observed agreement (diagonal of confusion matrix) observed_agree = sum(confusion.get((s, s), 0) for s in scores_range) po = observed_agree / n # Expected agreement by chance marg_a: Dict[ScoreLevel, int] = {} marg_b: Dict[ScoreLevel, int] = {} for (sa, sb), count in confusion.items(): marg_a[sa] = marg_a.get(sa, 0) + count marg_b[sb] = marg_b.get(sb, 0) + count pe = sum( (marg_a.get(s, 0) / n) * (marg_b.get(s, 0) / n) for s in scores_range ) if pe >= 1.0: return 1.0 # Perfect chance agreement — kappa is undefined but treat as perfect kappa = (po - pe) / (1.0 - pe) return round(kappa, 4) @staticmethod def reliability_label(kappa: float) -> str: """Return Landis & Koch interpretation label for a kappa value.""" if kappa < 0: return "Poor" elif kappa < 0.21: return "Slight" elif kappa < 0.41: return "Fair" elif kappa < 0.61: return "Moderate" elif kappa < 0.81: return "Substantial" else: return "Almost perfect" class PeerReviewPipeline: """Orchestrates multi-agent peer-review simulation with reliability measurement. Flow: User Request → Define Rubric → Assign Reviewers → Independent Scoring → Collect Scores → Compute Reliability → Resolve Disagreements → Aggregate → Report """ def __init__(self, rubric: Dict[str, Criterion], reviewers: List[ReviewerPersona]): """Initialize pipeline with a validated rubric and reviewer list. Args: rubric: Pre-built evaluation criteria with weights and anchors. reviewers: At least 2 distinct reviewer personas required for reliability measurement. Raises: ValueError: If fewer than 2 reviewers or rubric is invalid. """ # Law 1: Early exit — need ≥ 2 reviewers for meaningful kappa computation if len(reviewers) < 2: raise ValueError("Peer review requires at least 2 distinct reviewers") self.rubric = rubric self.reviewers = reviewers self.scores: List[ReviewScore] = [] def collect_scores( self, outputs: Dict[str, str], scorer_fn: Callable[[str, ReviewerPersona, Criterion, ScoreLevel], Tuple[ScoreLevel, str]], ) -> None: """Run independent scoring across all reviewers and all outputs. Args: outputs: Dict mapping output_id → content string to evaluate. scorer_fn: Callback that takes (output_content, reviewer, criterion, base_score) and returns (adjusted_score, justification). This is the interface through which LLM evaluators perform their scoring. Law 3 (Atomic): Scores are accumulated in a new list; no mutation of inputs. """ for output_id, content in outputs.items(): for reviewer in self.reviewers: for criterion_name, criterion in self.rubric.items(): # Evaluate each anchor level to find best match best_score = ScoreLevel.POOR best_justification = "" for level in ScoreLevel: adjusted_score = reviewer.apply_lens(level, criterion_name) score, justification = scorer_fn( content, reviewer, criterion, adjusted_score ) if score >= best_score: best_score = score best_justification = justification self.scores.append(ReviewScore( reviewer_name=reviewer.name, output_id=output_id, criterion_name=criterion_name, score=best_score, justification=best_justification, )) def compute_reliability(self) -> Dict[str, Dict[str, float]]: """Compute inter-rater reliability (Cohen's kappa) per criterion. Returns: Nested dict: {criterion_name: {pair_key: kappa_value}} pair_key is "reviewerA_vs_reviewerB" sorted alphabetically. """ criteria = list(self.rubric.keys()) reviewers = [r.name for r in self.reviewers] reliability: Dict[str, Dict[str, float]] = {} for criterion in criteria: criterion_scores: Dict[str, List[ScoreLevel]] = {name: [] for name in reviewers} for score_entry in self.scores: if score_entry.criterion_name == criterion: criterion_scores[score_entry.reviewer_name].append(score_entry.score) # Compute kappa for every reviewer pair pairs = {} for i in range(len(reviewers)): for j in range(i + 1, len(reviewers)): key = f"{reviewers[i]}_vs_{reviewers[j]}" kappa = CohensKappaCalculator.compute( criterion_scores[reviewers[i]], criterion_scores[reviewers[j]], ) pairs[key] = kappa reliability[criterion] = pairs return reliability # ❌ BAD: Single reviewer — no reliability measurement possible, high bias risk def evaluate_with_single_reviewer(output: str, rubric: Dict[str, Criterion]) -> float: """Single-reviewer evaluation. Fast but unreliable for subjective judgments. This is the anti-pattern: without inter-rater comparison, scorer bias is invisible and uncorrectable. Use PeerReviewPipeline instead. """ total = 0.0 for criterion_name, criterion in rubric.items(): score = ScoreLevel(3) # Placeholder — no independent verification total += criterion.weight * float(score) return total # ✅ GOOD: Multi-reviewer pipeline with reliability gate def evaluate_with_peer_review( outputs: Dict[str, str], rubric: Dict[str, Criterion], scorer_fn: Callable[[str, ReviewerPersona, Criterion, ScoreLevel], Tuple[ScoreLevel, str]], min_reliability: float = 0.60, ) -> Dict[str, Any]: """Run full peer-review pipeline with reliability threshold enforcement. If any criterion falls below the minimum kappa threshold, the review is flagged for resolution discussion before final scores are published. Args: outputs: Mapping of output_id → content to evaluate. rubric: Evaluation criteria with weights and anchors. scorer_fn: LLM-based scoring callback. min_reliability: Minimum acceptable Cohen's kappa (default 0.60). Returns: Structured evaluation result with scores, reliability stats, and resolution flags. """ reviewers = [ ReviewerPersona("AccuracyReviewer", ReviewerPerspective.ACCURACY_FOCUSED), ReviewerPersona("CreativityReviewer", ReviewerPerspective.CREATIVITY_FOCUSED), ] pipeline = PeerReviewPipeline(rubric, reviewers) pipeline.collect_scores(outputs, scorer_fn) # Compute reliability reliability = pipeline.compute_reliability() # Check threshold flagged_criteria: List[str] = [] for criterion, pairs in reliability.items(): for pair_key, kappa in pairs.items(): if kappa < min_reliability: flagged_criteria.append(criterion) break return { "scores": [s.__dict__ for s in pipeline.scores], "reliability": reliability, "flagged_for_resolution": flagged_criteria, "review_status": "approved" if not flagged_criteria else "needs_discussion", } ``` --- ### Pattern 3: Elo-Based Hypothesis Ranking Tournament Implements a tournament-style Elo rating system for ranking hypotheses or AI outputs through pairwise comparison. This is critical when no ground-truth labels exist — instead of scoring each output absolutely, the system pits candidates against each other and derives relative rankings from match outcomes. ```python import math from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple @dataclass(frozen=True) class EloPlayer: """Represents a hypothesis or output in the Elo tournament. Attributes: id: Unique identifier for this hypothesis/output. elo: Current Elo rating (default 1500 per standard chess convention). wins: Number of pairwise victories. losses: Number of pairwise defeats. draws: Number of tied comparisons. points: Total accumulated points (win=1, draw=0.5, loss=0). """ id: str elo: float = 1500.0 wins: int = 0 losses: int = 0 draws: int = 0 points: float = 0.0 @property def win_rate(self) -> float: total = self.wins + self.losses + self.draws if total == 0: return 0.0 return round((self.points / total), 4) def __lt__(self, other: "EloPlayer") -> bool: """Sort by Elo descending for leaderboard ordering.""" return self.elo > other.elo class EloTournamentEngine: """Runs pairwise Elo-based ranking tournaments for hypothesis comparison. The Elo system was designed for chess but works equally well for ranking AI-generated outputs when ground truth is unavailable. Each pairwise comparison produces a result (win/draw/loss), and ratings update via the standard Elo formula: R_new = R_old + K × (S - E) Where: R_old = pre-match rating K = expectation factor (typically 32 for competitive play) S = actual score (1.0 for win, 0.5 for draw, 0.0 for loss) E = expected score based on pre-match ratings Expected score calculation: E_A = 1 / (1 + 10^((R_B - R_A) / 400)) """ def __init__( self, k_factor: float = 32.0, initial_rating: float = 1500.0, rating_divisor: float = 400.0, ): """Initialize tournament engine. Args: k_factor: Maximum rating change per match (32 = standard competitive). initial_rating: Starting rating for all participants. rating_divisor: Controls how quickly ratings converge (400 = chess standard). """ self.k_factor = k_factor self.initial_rating = initial_rating self.rating_divisor = rating_divisor self.players: Dict[str, EloPlayer] = {} self.match_history: List[Dict[str, Any]] = [] def register_player(self, player_id: str) -> None: """Register a hypothesis/output in the tournament.""" if player_id not in self.players: self.players[player_id] = EloPlayer( id=player_id, elo=self.initial_rating ) def register_all(self, player_ids: List[str]) -> None: """Register multiple hypotheses at once.""" for pid in player_ids: self.register_player(pid) @staticmethod def _expected_score(rating_a: float, rating_b: float, divisor: float) -> float: """Compute expected score for player A against player B. Returns probability that A wins (0.0 to 1.0). """ exponent = (rating_b - rating_a) / divisor return 1.0 / (1.0 + 10 ** exponent) def play_match( self, player_a_id: str, player_b_id: str, result: str # "win", "loss", or "draw" ) -> Dict[str, Any]: """Execute a single pairwise comparison and update Elo ratings. Args: player_a_id: ID of the first hypothesis/output. player_b_id: ID of the second hypothesis/output. result: Outcome from A's perspective — "win", "loss", or "draw". Returns: Match result with updated ratings and rating changes. Raises: ValueError: If either player is not registered. """ # Law 1: Early exit on unregistered players if player_a_id not in self.players: raise ValueError(f"Unregistered player: {player_a_id}") if player_b_id not in self.players: raise ValueError(f"Unregistered player: {player_b_id}") a = self.players[player_a_id] b = self.players[player_b_id] # Compute expected scores e_a = self._expected_score(a.elo, b.elo, self.rating_divisor) e_b = 1.0 - e_a # Actual scores from A's perspective actual_a = {"win": 1.0, "draw": 0.5, "loss": 0.0}[result] actual_b = 1.0 - actual_a # Update ratings (Law 3: Return new values, don't mutate externally) delta_a = self.k_factor * (actual_a - e_a) delta_b = self.k_factor * (actual_b - e_b) a.elo += delta_a b.elo += delta_b # Update win/loss/draw records if result == "win": a.wins += 1 a.points += 1.0 b.losses += 1 elif result == "loss": a.losses += 1 b.wins += 1 b.points += 1.0 else: # draw a.draws += 1 b.draws += 1 a.points += 0.5 b.points += 0.5 match_record = { "player_a": player_a_id, "player_b": player_b_id, "result": result, "elo_before_a": round(a.elo - delta_a, 2), "elo_before_b": round(b.elo - delta_b, 2), "elo_after_a": round(a.elo, 2), "elo_after_b": round(b.elo, 2), "delta_a": round(delta_a, 2), "delta_b": round(delta_b, 2), } self.match_history.append(match_record) return match_record def get_leaderboard(self) -> List[EloPlayer]: """Return players sorted by Elo rating (highest first).""" return sorted(self.players.values()) def run_round_robin( self, results_fn: Callable[[str, str], str], min_rounds: int = 1, ) -> List[EloPlayer]: """Run a round-robin tournament where each pair plays once (or more rounds). Args: results_fn: Callback taking (player_a_id, player_b_id) and returning the match result from A's perspective ("win", "loss", "draw"). This is typically an LLM call comparing two outputs. min_rounds: Minimum number of full rounds to execute (for stability). Returns: Final leaderboard sorted by Elo rating. """ player_ids = list(self.players.keys()) n = len(player_ids) if n < 2: return self.get_leaderboard() total_matches = n * (n - 1) // 2 for round_num in range(1, min_rounds + 1): for i in range(n): for j in range(i + 1, n): result = results_fn(player_ids[i], player_ids[j]) self.play_match(player_ids[i], player_ids[j], result) return self.get_leaderboard() # Example: Elo-based hypothesis ranking from a scoring function def run_hypothesis_evaluation( hypotheses: Dict[str, str], comparator_fn: Callable[[str, str], str], k_factor: float = 32.0, ) -> List[Dict[str, Any]]: """Run an Elo tournament to rank hypotheses without ground truth. Args: hypotheses: Dict mapping hypothesis_id → content string. comparator_fn: Given two hypothesis IDs, returns match result from first perspective ("win", "loss", or "draw"). k_factor: Elo expectation factor (32 = standard). Returns: Leaderboard with Elo ratings and win rates. """ tournament = EloTournamentEngine(k_factor=k_factor) tournament.register_all(list(hypotheses.keys())) leaderboard = tournament.run_round_robin( results_fn=comparator_fn, min_rounds=3 ) return [ { "rank": i + 1, "hypothesis_id": p.id, "elo_rating": round(p.elo, 2), "win_rate": p.win_rate, "record": f"{p.wins}W-{p.draws}D-{p.losses}L", } for i, p in enumerate(leaderboard) ] # ❌ BAD: Absolute scoring without pairwise comparison — susceptible to scorer bias def bad_absolute_ranking(hypotheses: Dict[str, str]) -> List[Dict]: """Absolute scoring approach. Scores each hypothesis independently on a 1-5 scale. Problem: All scores are relative to the scorer's internal calibration, which may be inconsistent. Without pairwise comparison, you cannot detect whether all scorers are simply inflated or deflated across the board. """ # Placeholder — no actual comparison mechanism return [{"id": h_id, "score": 3} for h_id in hypotheses] # ✅ GOOD: Elo tournament captures relative strength even with biased scorers def good_elo_ranking( hypotheses: Dict[str, str], judge_prompt_fn: Callable[[str, str], str], ) -> List[Dict]: """Elo-based ranking. Each pairwise comparison isolates the two candidates and asks a judge to pick the better one, eliminating absolute calibration bias.""" def compare_pair(a_id: str, b_id: str) -> str: verdict = judge_prompt_fn(hypotheses[a_id], hypotheses[b_id]) if "A wins" in verdict: return "win" elif "B wins" in verdict: return "loss" else: return "draw" tournament = EloTournamentEngine(k_factor=32) tournament.register_all(list(hypotheses.keys())) leaderboard = tournament.run_round_robin(compare_pair, min_rounds=5) return [ {"rank": i + 1, "id": p.id, "elo": round(p.elo, 2), "record": f"{p.wins}-{p.draws}-{p.losses}"} for i, p in enumerate(leaderboard) ] ``` --- ### Pattern 4: Agent Laboratory Framework Integration Implements the four-role Agent Laboratory framework (Professor, PostDoc, Reviewer, Software Engineer) adapted for structured AI output evaluation. Each role has a distinct function in the review lifecycle, creating a production-grade quality gate that simulates academic peer review at machine speed. ```python from enum import Enum from dataclasses import dataclass, field from typing import Any, Dict, List, Optional class LabRole(Enum): """Agent Laboratory roles adapted for AI output evaluation.""" PROFESSOR = "professor" # Sets evaluation criteria and reviews final report POSTDOC = "postdoc" # Conducts primary analysis and initial scoring REVIEWER = "reviewer" # Performs independent critique with distinct perspective SOFTWARE_ENGINEER = "engineer" # Implements fixes for identified issues @dataclass(frozen=True) class LabReviewStage: """Describes a stage in the Agent Laboratory review workflow.""" stage_name: str responsible_role: LabRole inputs_required: List[str] outputs_produced: List[str] class AgentLabPipeline: """Orchestrates the four-role Agent Laboratory framework for structured evaluation. Workflow: 1. PROFESSOR defines criteria and success metrics → produces RubricSpecification 2. POSTDOC runs initial evaluation on all candidates → produces ScoreReport + AnalysisNotes 3. REVIEWER independently re-evaluates top-k candidates → produces IndependentReview 4. SOFTWARE_ENGINEER implements fixes for critical issues found by reviewers → produces Revision The Professor then validates that revisions meet the original criteria. """ # Define the laboratory stages STAGES = [ LabReviewStage( stage_name="criteria_definition", responsible_role=LabRole.PROFESSOR, inputs_required=["task_description", "domain_context"], outputs_produced=["rubric_specification"], ), LabReviewStage( stage_name="initial_evaluation", responsible_role=LabRole.POSTDOC, inputs_required=["rubric_specification", "candidate_outputs"], outputs_produced=["score_report", "analysis_notes"], ), LabReviewStage( stage_name="peer_review", responsible_role=LabRole.REVIEWER, inputs_required=["rubric_specification", "top_candidates"], outputs_produced=["independent_review", "disagreement_flags"], ), LabReviewStage( stage_name="revision", responsible_role=LabRole.SOFTWARE_ENGINEER, inputs_required=["reviewer_critiques", "original_outputs"], outputs_produced=["revised_outputs"], ), LabReviewStage( stage_name="professor_validation", responsible_role=LabRole.PROFESSOR, inputs_required=["revised_outputs", "rubric_specification"], outputs_produced=["final_verdict", "acceptance_report"], ), ] …(truncated)