| """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: |
| |
| 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 |
| ) |
|
|
