| """Frozen 31-slot schema over supplied text and verifier observations. |
| |
| This does not reconstruct benchmark packets or run inference. The caller must |
| supply the same segmentation/chunk counts and evidence-unit coverage. |
| """ |
| from __future__ import annotations |
| from typing import Any, Mapping |
| import math, re |
| import numpy as np |
| FT5_THRESHOLD=0.763 |
| FEATURE_NAMES=['ft5_score', 'ft5_abs_distance_to_calibrated_threshold', 'b192_score', 'ft5_b192_delta', 'ft5_b192_disagreement', 'table_or_mixed_indicator', 'ft5_margin_x_table', 'ft5_full_score', 'selected_full_score_delta', 'selected_full_disagreement', 'log_ft5_selected_chunks', 'log_ft5_full_chunks', 'log_claim_word_count', 'log_selected_evidence_words', 'log_full_evidence_words', 'log_selected_source_count', 'selected_source_coverage_ratio', 'ft5_selected_margin_to_050', 'ft5_selected_margin_to_0763', 'ft5_full_minus_selected', 'b192_minus_ft5_selected', 'b192_ft5_decision_disagreement', 'full_only_support_signal', 'b192_disagreement_x_table_or_mixed', 'selected_full_discrepancy_x_evidence_length_ratio', 'claim_negation_proxy_count', 'claim_comparator_quantifier_count', 'claim_numeral_unit_count', 'claim_titlecase_entity_proxy_count', 'claim_selected_number_overlap_ratio', 'claim_entity_selected_minus_full_overlap'] |
| F5_NAMES=('ft5_selected_margin_to_050', 'ft5_selected_margin_to_0763', 'ft5_full_minus_selected', 'b192_minus_ft5_selected', 'b192_ft5_decision_disagreement', 'full_only_support_signal', 'b192_disagreement_x_table_or_mixed', 'selected_full_discrepancy_x_evidence_length_ratio') |
| F2_NAMES=('claim_negation_proxy_count', 'claim_comparator_quantifier_count', 'claim_numeral_unit_count', 'claim_titlecase_entity_proxy_count', 'claim_selected_number_overlap_ratio', 'claim_entity_selected_minus_full_overlap') |
| def _finite_number(value: Any) -> float | None: |
| if isinstance(value, bool) or not isinstance(value, (int, float)): |
| return None |
| value = float(value) |
| return value if math.isfinite(value) else None |
|
|
| def _text(value: Any) -> str | None: |
| return value if isinstance(value, str) else None |
|
|
| def _tokens(value: str) -> list[str]: |
| return [token.casefold() for token in re.findall(r"[A-Za-z0-9]+(?:['’][A-Za-z0-9]+)?", value)] |
|
|
| def _content_terms(value: str) -> set[str]: |
| stop = {"a", "an", "the", "and", "or", "of", "to", "in", "on", "for", "by", "with", "is", "are", "was", "were", "be", "as", "at", "from"} |
| return {token for token in _tokens(value) if len(token) > 1 and token not in stop} |
|
|
| def _numbers(value: str) -> set[str]: |
| return { |
| re.sub(r",(?=\d{3}(?:\D|$))", "", match.casefold()) |
| for match in re.findall(r"(?<![A-Za-z])[-+]?\d[\d,]*(?:\.\d+)?%?(?![A-Za-z])", value) |
| } |
|
|
| def _entities(value: str) -> set[str]: |
| |
| matches = re.findall(r"\b[A-Z][a-z]{2,}(?:\s+[A-Z][a-z]{2,})*\b", value) |
| return {" ".join(_tokens(match)) for match in matches} |
|
|
| def _overlap(left: set[str], right: set[str]) -> float: |
| return len(left & right) / len(left) if left else 0.0 |
|
|
| class RouterFitError(ValueError): |
| pass |
|
|
| def _base_features(input_row: Mapping[str, Any], selected: Mapping[str, Any], full: Mapping[str, Any], b192: Mapping[str, Any], selected_geometry: Mapping[str, Any], full_geometry: Mapping[str, Any]) -> np.ndarray: |
| ft5 = float(selected["score"]) |
| ft5_full = float(full["score"]) |
| b192_score = float(b192["score"]) |
| if not all(math.isfinite(value) and 0.0 <= value <= 1.0 for value in (ft5, ft5_full, b192_score)): |
| raise RouterFitError("CHEAP_MODEL_SCORE_INVALID") |
| table = float(input_row.get("modality") == "TABLE_OR_MIXED") |
| ft5_calibrated = ft5 > FT5_THRESHOLD |
| b192_prediction = b192.get("prediction") |
| selected_prediction = selected.get("prediction") |
| full_prediction = full.get("prediction") |
| if not all(isinstance(value, bool) for value in (b192_prediction, selected_prediction, full_prediction)): |
| raise RouterFitError("CHEAP_MODEL_PREDICTION_INVALID") |
| selected_full_delta = ft5_full - ft5 |
| ft5_full_disagreement = int((ft5 > FT5_THRESHOLD) != (ft5_full > FT5_THRESHOLD)) |
| features = [ |
| ft5, |
| abs(ft5 - FT5_THRESHOLD), |
| b192_score, |
| ft5 - b192_score, |
| float(ft5_calibrated != b192_prediction), |
| table, |
| abs(ft5 - FT5_THRESHOLD) * table, |
| ft5_full, |
| selected_full_delta, |
| float(ft5_full_disagreement), |
| math.log1p(int(selected_geometry["chunk_count"])), |
| math.log1p(int(full_geometry["chunk_count"])), |
| math.log1p(len(str(input_row.get("claim", "")).split())), |
| math.log1p(len(str(input_row.get("selected_context", "")).split())), |
| math.log1p(len(str(input_row.get("full_fixed_evidence_packet", "")).split())), |
| math.log1p(int(input_row.get("selected_source_count", 0))), |
| float(input_row.get("selected_source_coverage_ratio", 0.0)), |
| ] |
| if len(features) != 17 or not np.isfinite(np.asarray(features, dtype=np.float64)).all(): |
| raise RouterFitError("A2_FEATURE_VECTOR_INVALID") |
| return np.asarray(features, dtype=np.float64) |
|
|
| def _extra_features(input_row: Mapping[str, Any], pair_row: Mapping[str, Any]) -> tuple[dict[str, float], dict[str, bool]]: |
| values: dict[str, float] = {} |
| available: dict[str, bool] = {} |
| cheap = pair_row.get("cheap_model_features") |
| cheap = cheap if isinstance(cheap, dict) else {} |
| ft5 = _finite_number(cheap.get("ft5_score")) |
| full = _finite_number(cheap.get("ft5_full_score")) |
| b192 = _finite_number(cheap.get("b192_score")) |
| full_words = _finite_number(cheap.get("full_document_word_count")) |
| selected_words = _finite_number(cheap.get("selected_document_word_count")) |
| table = float(str(input_row.get("modality", pair_row.get("modality", ""))) == "TABLE_OR_MIXED") |
| f5_raw: dict[str, float | None] = { |
| F5_NAMES[0]: abs(ft5 - 0.5) if ft5 is not None else None, |
| F5_NAMES[1]: abs(ft5 - FT5_THRESHOLD) if ft5 is not None else None, |
| F5_NAMES[2]: full - ft5 if full is not None and ft5 is not None else None, |
| F5_NAMES[3]: b192 - ft5 if b192 is not None and ft5 is not None else None, |
| F5_NAMES[4]: float(bool(cheap.get("ft5_b192_disagreement"))) if isinstance(cheap.get("ft5_b192_disagreement"), (bool, int)) else None, |
| F5_NAMES[5]: float(full > FT5_THRESHOLD and ft5 <= FT5_THRESHOLD) if full is not None and ft5 is not None else None, |
| F5_NAMES[6]: float(bool(cheap.get("ft5_b192_disagreement"))) * table if isinstance(cheap.get("ft5_b192_disagreement"), (bool, int)) else None, |
| F5_NAMES[7]: abs(full - ft5) * math.log1p(max(0.0, full_words)) / max(1.0, math.log1p(max(0.0, selected_words))) |
| if full is not None and ft5 is not None and full_words is not None and selected_words is not None else None, |
| } |
| for key, value in f5_raw.items(): |
| available[key] = value is not None |
| values[key] = float(value) if value is not None else float("nan") |
|
|
| claim = _text(input_row.get("claim")) or _text(input_row.get("claim_text")) |
| selected = _text(input_row.get("selected_context")) |
| full_text = _text(input_row.get("full_fixed_evidence_packet")) |
| text_ready = claim is not None and selected is not None and full_text is not None |
| if text_ready: |
| claim_terms = _content_terms(claim) |
| selected_terms = _content_terms(selected) |
| full_terms = _content_terms(full_text) |
| claim_entities = _entities(claim) |
| selected_entities = _entities(selected) |
| full_entities = _entities(full_text) |
| claim_nums = _numbers(claim) |
| selected_nums = _numbers(selected) |
| negation_matches = list(re.finditer( |
| r"\b(?:no|not|never|neither|nor|without|cannot|can't|doesn't|don't|didn't|isn't|aren't|wasn't|weren't)\b", |
| claim, flags=re.IGNORECASE, |
| )) |
| negation_scope_terms = 0 |
| for match in negation_matches: |
| |
| scope = re.split(r"[.!?;:]", claim[match.end():], maxsplit=1)[0] |
| negation_scope_terms += len(_content_terms(" ".join(_tokens(scope)[:5]))) |
| comparators = re.findall(r"\b(?:more|less|greater|fewer|higher|lower|at least|at most|exactly|only|all|every|each|most|least|equal(?:s)?|between|than|over|under)\b", claim, flags=re.IGNORECASE) |
| numeric_units = re.findall(r"(?<![A-Za-z])[-+]?\d[\d,]*(?:\.\d+)?\s*(?:%|percent|km|miles?|meters?|kg|grams?|years?|months?|days?|hours?|million|billion)?", claim, flags=re.IGNORECASE) |
| f2 = { |
| F2_NAMES[0]: float(len(negation_matches) + negation_scope_terms), |
| F2_NAMES[1]: float(len(comparators)), |
| F2_NAMES[2]: float(len(numeric_units)), |
| F2_NAMES[3]: float(min(20, len(claim_entities))), |
| F2_NAMES[4]: _overlap(claim_nums, selected_nums), |
| F2_NAMES[5]: _overlap(claim_entities, selected_entities) - _overlap(claim_entities, full_entities), |
| } |
| else: |
| f2 = {name: None for name in F2_NAMES} |
| for key, value in f2.items(): |
| available[key] = value is not None |
| values[key] = float(value) if value is not None else float("nan") |
|
|
| return values, available |
|
|
| def build_features(*,claim,selected_evidence,full_evidence,selected_ft5_score,full_ft5_score,deberta_score, |
| selected_chunks,full_chunks,selected_source_count=1,source_coverage=1.0, |
| table_or_mixed=False,adapter="external"): |
| if adapter not in ("dev","external"): raise ValueError("UNKNOWN_FEATURE_ADAPTER") |
| if any(not isinstance(x,str) for x in (claim,selected_evidence,full_evidence)): |
| raise ValueError("SUPPLIED_TEXT_REQUIRED") |
| if any(type(x) is not int or x<0 for x in (selected_chunks,full_chunks,selected_source_count)): |
| raise ValueError("NONNEGATIVE_COUNTS_REQUIRED") |
| if not isinstance(source_coverage,(int,float)) or not math.isfinite(source_coverage) or not 0<=source_coverage<=1: |
| raise ValueError("SOURCE_COVERAGE_INVALID") |
| if type(table_or_mixed) is not bool: raise ValueError("MODALITY_FLAG_INVALID") |
| s,f,d=map(float,(selected_ft5_score,full_ft5_score,deberta_score)) |
| inp=dict(claim=claim,selected_context=selected_evidence,full_fixed_evidence_packet=full_evidence, |
| modality="TABLE_OR_MIXED" if table_or_mixed else "TEXT",selected_source_count=selected_source_count, |
| selected_source_coverage_ratio=source_coverage) |
| base=_base_features(inp,dict(score=s,prediction=s>0.5),dict(score=f,prediction=f>0.5), |
| dict(score=d,prediction=d>=0.398),dict(chunk_count=selected_chunks),dict(chunk_count=full_chunks)) |
| delta_a=float((s>(0.763 if adapter=="dev" else 0.5))!=(d>=0.398)) |
| cheap=dict(ft5_score=s,ft5_full_score=f,b192_score=d,ft5_b192_disagreement=int(delta_a), |
| selected_document_word_count=len(selected_evidence.split()),full_document_word_count=len(full_evidence.split())) |
| values,_=_extra_features(inp,dict(cheap_model_features=cheap)) |
| x=np.asarray(list(base)+[values[n] for n in FEATURE_NAMES[17:]],dtype=float) |
| if x.shape!=(31,) or not np.isfinite(x).all(): raise ValueError("FEATURE_VECTOR_INVALID") |
| return x |
|
|
