Spaces:
Sleeping
Sleeping
File size: 5,529 Bytes
dd990bf | 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 | """The *single* fixed graph selector used across all six reported benchmarks.
Only candidate text, its optional title, and a query-specific neural relevance
score are allowed as input. There is no training, gold-label access, or corpus-
specific parameter tuning inside this module.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping, Sequence
import re
import unicodedata
import numpy as np
def normalize_text(text: Any) -> str:
"""Match the exact normalization of the published evaluation scripts."""
return " ".join(
re.sub(r"[^\w\s]", " ", unicodedata.normalize("NFKD", str(text)).casefold()).split()
)
@dataclass(frozen=True)
class Passage:
title: str
text: str
id: str | None = None
@classmethod
def from_value(cls, value: "Passage | Mapping[str, Any] | str") -> "Passage":
if isinstance(value, Passage):
return value
if isinstance(value, str):
return cls(title="", text=value)
if not isinstance(value, Mapping):
raise TypeError("Each passage must be a string, dict, or Passage.")
text = value.get("text", value.get("body", value.get("paragraph_text", "")))
return cls(title=str(value.get("title", "") or ""), text=str(text or ""),
id=None if value.get("id") is None else str(value["id"]))
def graph_features(query: str, passages: Sequence[Passage]) -> tuple[np.ndarray, np.ndarray]:
"""Return directed source->target article-title edges and query/title matches.
This is intentionally the SAME simple string-link rule as the six-dataset
benchmarks. In particular, it does NOT introduce token overlap, semantic
similarity edges, facet decomposition, or learned relation extraction.
"""
q = normalize_text(query)
titles = [normalize_text(p.title) for p in passages]
bodies = [normalize_text(p.text) for p in passages]
q_words = set(q.split())
n = len(passages)
link = np.zeros((n, n), np.float32)
match = np.zeros(n, np.float32)
for j, title in enumerate(titles):
if not title:
continue
title_words = set(title.split())
overlap = len(title_words & q_words) / max(1, len(title_words))
match[j] = 1.0 if " " + title + " " in " " + q + " " else 0.5 * overlap
for i, body in enumerate(bodies):
if i != j and len(title) >= 4 and " " + title + " " in " " + body + " ":
link[i, j] = 1.0
return link, match
def minmax_relevance(scores: Sequence[float]) -> np.ndarray:
scores = np.asarray(scores, dtype=np.float32)
if scores.ndim != 1 or not np.isfinite(scores).all():
raise ValueError("Scores must be a one-dimensional array of finite numbers.")
if not len(scores):
raise ValueError("At least one candidate passage is required.")
lo, hi = float(scores.min()), float(scores.max())
return (scores - lo) / (hi - lo) if hi > lo + 1e-9 else np.zeros_like(scores)
def select_with_features(
scores: Sequence[float], link: np.ndarray, match: np.ndarray, k: int = 3
) -> list[int]:
"""Greedy rank with fixed coefficients .4, .5, .2 and deterministic ties.
At step t the score of candidate j is
0.4 * normalized_relevance_j
+ 0.5 * max_{i previously selected} link[i, j]
+ 0.2 * query_title_match_j.
"""
if isinstance(k, bool) or not isinstance(k, int) or k < 1:
raise ValueError("k must be a positive integer")
norm_scores = minmax_relevance(scores)
n = len(norm_scores)
link = np.asarray(link, dtype=np.float32)
match = np.asarray(match, dtype=np.float32)
if link.shape != (n, n) or match.shape != (n,):
raise ValueError("Link matrix or match vector dimensions do not match scores")
selected: list[int] = []
remaining = set(range(n))
for _ in range(min(k, n)):
nxt = max(remaining, key=lambda j: (
0.4 * norm_scores[j]
+ 0.5 * max((link[i, j] for i in selected), default=0.0)
+ 0.2 * match[j],
-j,
))
selected.append(nxt)
remaining.remove(nxt)
return selected
def rank_with_scores(query: str, passages: Sequence[Passage | Mapping[str, Any] | str],
scores: Sequence[float], k: int = 3) -> list[dict[str, Any]]:
"""Public, network-free inference API if the retriever supplies scores.
Returned `step_score` is the acquisition score at the moment of selection,
not a calibrated relevance probability.
"""
docs = [Passage.from_value(p) for p in passages]
if len(docs) != len(scores):
raise ValueError("Supply exactly one relevance score per candidate passage")
link, match = graph_features(query, docs)
normalized = minmax_relevance(scores)
order = select_with_features(scores, link, match, k)
result = []
selected = []
for idx in order:
bonus = max((float(link[i, idx]) for i in selected), default=0.0)
step = 0.4 * float(normalized[idx]) + 0.5 * bonus + 0.2 * float(match[idx])
item = docs[idx]
result.append({
"index": idx, "id": item.id, "title": item.title, "text": item.text,
"relevance_score": float(scores[idx]),
"normalized_relevance": float(normalized[idx]),
"directed_link_bonus": bonus,
"query_title_match": float(match[idx]),
"step_score": step,
})
selected.append(idx)
return result
|