Repository Public
1"""Instance-level reconstruction metrics (with stated limitations) and factual-integrity checks."""2from __future__ import annotations34from collections import Counter56from rapidfuzz.distance import Levenshtein7from rouge_score import rouge_scorer8from sacrebleu.metrics import BLEU, CHRF910from htr.parse.protect import HEDGE_RE, NEGATION_RE, PLACEHOLDER_RE, extract_numbers, placeholder_multiset11from htr.utils.hashing import normalize_for_hash12from htr.utils.text import split_sentences, word_count1314_chrf = CHRF(word_order=2) # chrF++15_bleu = BLEU(effective_order=True)16_rouge = rouge_scorer.RougeScorer(["rougeL"], use_stemmer=False)171819def lexical(hyp: str, ref: str) -> dict[str, float]:20 """Exact match (after whitespace/case normalisation), normalised Levenshtein similarity, chrF++, sentence BLEU,21 ROUGE-L F. Limitations: n-gram overlap rewards copying; it is reported next to the identity baseline, never alone."""22 hn, rn = normalize_for_hash(hyp), normalize_for_hash(ref)23 return {24 "exact_match": float(hn == rn),25 "edit_sim": float(Levenshtein.normalized_similarity(hn, rn)),26 "chrf": float(_chrf.sentence_score(hyp, [ref]).score),27 "bleu": float(_bleu.sentence_score(hyp, [ref]).score),28 "rougeL": float(_rouge.score(ref, hyp)["rougeL"].fmeasure),29 "len_ratio": word_count(hyp) / max(1, word_count(ref)),30 }313233def factual(hyp: str, ref: str) -> dict[str, float]:34 """Preservation of protected spans and numbers relative to the ORIGINAL (reference): placeholder multiset35 equality, number recall/precision, negation and hedge deltas."""36 pm_h, pm_r = placeholder_multiset(hyp), placeholder_multiset(ref)37 n_h, n_r = Counter(extract_numbers(PLACEHOLDER_RE.sub(" ", hyp))), Counter(extract_numbers(PLACEHOLDER_RE.sub(" ", ref)))38 inter = sum((n_h & n_r).values())39 return {40 "placeholders_exact": float(pm_h == pm_r),41 "placeholder_recall": (sum((Counter(pm_h) & Counter(pm_r)).values()) / sum(pm_r.values())) if pm_r else 1.0,42 "number_recall": inter / sum(n_r.values()) if n_r else 1.0,43 "number_precision": inter / sum(n_h.values()) if n_h else 1.0,44 "negation_delta": float(len(NEGATION_RE.findall(hyp)) - len(NEGATION_RE.findall(ref))),45 "hedge_delta": float(len(HEDGE_RE.findall(hyp)) - len(HEDGE_RE.findall(ref))),46 "sentence_count_delta": float(len(split_sentences(hyp)) - len(split_sentences(ref))),47 }484950def aggregate(rows: list[dict]) -> dict[str, float]:51 if not rows:52 return {}53 keys = rows[0].keys()54 return {k: sum(r[k] for r in rows) / len(rows) for k in keys}55