Spaces:
Running on Zero
Running on Zero
File size: 5,645 Bytes
465b477 | 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 | """Claim-versus-source verifiers.
Both verifiers judge a claim against the ORIGINAL text of the cited chunks, never against
notes or summaries, so compression errors cannot launder themselves into "support".
* ``NLIVerifier``: a local NLI cross-encoder. A chunk is longer than the model's input,
so the premise is scored as sliding windows of sentences and the best window wins.
When a claim cites several chunks, the best window of each is also concatenated, so
claims that combine facts from two places can still be entailed.
* ``LLMVerifier``: a second model with a strict three-way rubric.
"""
from __future__ import annotations
from dataclasses import dataclass
from functools import lru_cache
from typing import Literal, Protocol
import numpy as np
from .chunking import split_sentences
from .llm import LLMClient, UsageMeter, call_structured
from .prompts import VERIFY_SYSTEM, VerifyOut, verify_user
Label = Literal["entailed", "contradicted", "not_enough_evidence"]
@dataclass
class Verdict:
label: Label
entail: float
contradict: float
detail: dict
class Verifier(Protocol):
name: str
def verify(self, claim: str, sources: list[str], meter: UsageMeter | None = None) -> Verdict: ...
def strip_headings(text: str) -> str:
return "\n".join(line for line in text.split("\n") if not line.startswith("§ ")).strip()
def windows(text: str, size: int = 3) -> list[str]:
sents = split_sentences(strip_headings(text).replace("\n\n", " "))
if len(sents) <= size:
return [" ".join(sents)] if sents else []
return [" ".join(sents[i : i + size]) for i in range(len(sents) - size + 1)]
@lru_cache(maxsize=2)
def _cross_encoder(model_name: str):
from sentence_transformers import CrossEncoder
return CrossEncoder(model_name, device="cpu")
class NLIVerifier:
def __init__(
self,
model_name: str = "cross-encoder/nli-deberta-v3-small",
entail_threshold: float = 0.5,
contradict_threshold: float = 0.5,
window_sentences: int = 3,
encoder=None,
):
"""``encoder`` injects an already loaded ``CrossEncoder`` (e.g. one placed on CUDA by the
Space entry point); otherwise the model is loaded on CPU and cached per name."""
self.encoder = encoder
self.model_name = model_name
self.name = f"nli:{model_name}"
self.entail_threshold = entail_threshold
self.contradict_threshold = contradict_threshold
self.window_sentences = window_sentences
def _encoder(self):
return self.encoder if self.encoder is not None else _cross_encoder(self.model_name)
def _label_index(self) -> dict[str, int]:
cfg = self._encoder().model.config
return {v.lower(): int(k) for k, v in cfg.id2label.items()}
def score_pairs(self, pairs: list[tuple[str, str]]) -> np.ndarray:
"""Probabilities with columns (entailment, contradiction) for (premise, hypothesis) pairs."""
if not pairs:
return np.zeros((0, 2))
logits = np.asarray(self._encoder().predict(pairs, batch_size=16, show_progress_bar=False))
probs = np.exp(logits - logits.max(axis=1, keepdims=True))
probs /= probs.sum(axis=1, keepdims=True)
li = self._label_index()
return np.stack([probs[:, li["entailment"]], probs[:, li["contradiction"]]], axis=1)
def verify(self, claim: str, sources: list[str], meter: UsageMeter | None = None) -> Verdict:
per_source = [windows(s, self.window_sentences) for s in sources]
flat = [(w, claim) for ws in per_source for w in ws]
scores = self.score_pairs(flat)
best_windows: list[str] = []
detail: dict = {"per_source": []}
offset = 0
best_e = best_c = 0.0
for ws in per_source:
sc = scores[offset : offset + len(ws)]
offset += len(ws)
if len(ws) == 0:
detail["per_source"].append({"entail": 0.0, "contradict": 0.0, "window": None})
continue
i = int(sc[:, 0].argmax())
best_windows.append(ws[i])
e, c = float(sc[:, 0].max()), float(sc[:, 1].max())
detail["per_source"].append({"entail": round(e, 4), "contradict": round(c, 4), "window": ws[i]})
best_e, best_c = max(best_e, e), max(best_c, c)
if len(best_windows) > 1:
joint = self.score_pairs([(" ".join(best_windows), claim)])[0]
detail["joint"] = {"entail": round(float(joint[0]), 4), "contradict": round(float(joint[1]), 4)}
best_e = max(best_e, float(joint[0]))
if best_e >= self.entail_threshold:
label: Label = "entailed"
elif best_c >= self.contradict_threshold:
label = "contradicted"
else:
label = "not_enough_evidence"
return Verdict(label=label, entail=best_e, contradict=best_c, detail=detail)
class LLMVerifier:
def __init__(self, client: LLMClient):
self.client = client
self.name = f"llm:{client.spec.model}"
def verify(self, claim: str, sources: list[str], meter: UsageMeter | None = None) -> Verdict:
passages = "\n\n".join(f"Passage {i + 1}:\n{strip_headings(s)}" for i, s in enumerate(sources))
out = call_structured(self.client, meter, "verify", VERIFY_SYSTEM, verify_user(claim, passages), VerifyOut, max_tokens=1024)
return Verdict(
label=out.label,
entail=1.0 if out.label == "entailed" else 0.0,
contradict=1.0 if out.label == "contradicted" else 0.0,
detail={"rationale": out.rationale},
)
|