"""Metric primitives for the GRM evaluation suite (PLAN section 33).""" from __future__ import annotations import math from collections import defaultdict from collections.abc import Iterable, Mapping, Sequence from graph.types import GraphPath, ReasoningResult def answer_exact_match(result: ReasoningResult, expected: Iterable[str]) -> float: """Strictest answer metric: the predicted set must equal the gold set.""" return float({answer.entity_id for answer in result.answers} == set(expected)) def answer_prf(predicted: set[str], expected: set[str]) -> tuple[float, float, float]: """Set precision, recall and F1 for multi-answer entity queries. Gold answer sets are frequently larger than one entity, so exact match alone hides partial credit and makes training progress invisible early on. """ if not expected: empty = float(not predicted) return empty, empty, empty if not predicted: return 0.0, 0.0, 0.0 hits = len(predicted & expected) precision = hits / len(predicted) recall = hits / len(expected) if precision + recall == 0.0: return precision, recall, 0.0 return precision, recall, 2 * precision * recall / (precision + recall) def reciprocal_rank(result: ReasoningResult, expected: Iterable[str]) -> float: """Reciprocal rank of the first gold answer in the ranked prediction list.""" gold = set(expected) if not gold: return 0.0 for position, answer in enumerate(result.answers, start=1): if answer.entity_id in gold: return 1.0 / position return 0.0 def path_exact_match( predicted: Iterable[GraphPath], expected: Iterable[GraphPath] ) -> float: return float( {_key(path) for path in predicted} == {_key(path) for path in expected} ) def path_validity(paths: Iterable[GraphPath]) -> float: """Share of predicted paths that verify; NaN when nothing was predicted. Abstaining predicts no paths, and scoring that as 1.0 would reward it. """ values = list(paths) if not values: return math.nan return sum(path.verified for path in values) / len(values) def path_recall(predicted: Iterable[GraphPath], expected: Iterable[GraphPath]) -> float: actual = {_key(path) for path in predicted} target = {_key(path) for path in expected} if not target: return 1.0 return len(actual & target) / len(target) def path_endpoint_precision( predicted: Iterable[GraphPath], gold_answers: Iterable[str] ) -> float: """Fraction of returned paths that terminate at a gold answer entity. ``path_recall`` is a strict set match on whole gold paths, so it cannot tell two very different failures apart: evidence that reasons to the wrong entity, and evidence that reaches the *right* entity by a route the generator did not happen to record as gold. A graph usually admits several valid routes, so the second is not an error. This metric isolates it -- read together with ``path_recall``, a high endpoint precision beside a low path recall means the model reasons correctly over alternative evidence rather than incorrectly. """ paths = list(predicted) if not paths: # Nothing predicted is not precise, so it is excluded from the mean. return math.nan gold = set(gold_answers) if not gold: return 0.0 return sum(float(path.endpoint_id in gold) for path in paths) / len(paths) def ranking_accuracy( predicted: Sequence[GraphPath], expected: Iterable[GraphPath] ) -> float: """Pairwise accuracy of path scores against the supporting/competing split.""" gold = {_key(path) for path in expected} positives = [path.score for path in predicted if _key(path) in gold] negatives = [path.score for path in predicted if _key(path) not in gold] if not positives or not negatives: return float("nan") wins = sum( float(positive > negative) + 0.5 * float(positive == negative) for positive in positives for negative in negatives ) return wins / (len(positives) * len(negatives)) def ece( confidences: Iterable[float], correctness: Iterable[float], bins: int = 10 ) -> float: """Expected calibration error over equal-width confidence bins.""" values = list(zip(confidences, correctness, strict=True)) total = len(values) if not total: return 0.0 score = 0.0 for index in range(bins): low, high = index / bins, (index + 1) / bins bucket = [ (confidence, correct) for confidence, correct in values if low <= confidence < high or (index == bins - 1 and low <= confidence <= high) ] if bucket: mean_confidence = sum(item[0] for item in bucket) / len(bucket) mean_accuracy = sum(item[1] for item in bucket) / len(bucket) score += len(bucket) / total * abs(mean_confidence - mean_accuracy) return score def brier(confidences: Iterable[float], correctness: Iterable[float]) -> float: values = list(zip(confidences, correctness, strict=True)) if not values: return 0.0 return sum((confidence - correct) ** 2 for confidence, correct in values) / len( values ) def grouped_accuracy(rows: Iterable[tuple[int, bool]]) -> Mapping[str, float]: grouped: dict[str, list[bool]] = defaultdict(list) for hops, correct in rows: grouped[_hop_key(hops)].append(correct) return {key: sum(values) / len(values) for key, values in sorted(grouped.items())} def grouped_mean(rows: Iterable[tuple[int, float]]) -> Mapping[str, float]: grouped: dict[str, list[float]] = defaultdict(list) for hops, value in rows: grouped[_hop_key(hops)].append(value) return {key: sum(values) / len(values) for key, values in sorted(grouped.items())} def format_metric(value: float, digits: int = 3) -> str: """Render a metric, showing an unmeasurable one as ``n/a`` not ``0.000``.""" if value is None or (isinstance(value, float) and math.isnan(value)): return " n/a" return f"{value:.{digits}f}" def _hop_key(hops: int) -> str: return "6+" if hops >= 6 else str(hops) def _key(path: GraphPath) -> tuple[tuple[str, str, str], ...]: return tuple( (step.source_id, step.relation_id, step.target_id) for step in path.steps )