File size: 6,419 Bytes
3e363e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | """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
)
|