Spaces:
Sleeping
Sleeping
| """Model-free live checks for the public demo (no LLM call, no API key). | |
| Before the graph is extracted and the benchmark is frozen, a visitor can still exercise | |
| the two deterministic stages every answer in this project depends on: | |
| * **Text-RAG retrieval**: corpus-wide BM25 over every chunk, packed into the same | |
| context budget the ``text_rag`` method uses, built by ``methods.build_context`` itself. | |
| * **Claim checking**: exact quote anchoring (``quotes.locate``, no fuzzy matching) and the | |
| NLI support score of the same cross-encoder verifier the methods use, with the best | |
| supporting window of each passage. | |
| The predefined questions are real QASPER questions about corpus papers, chosen by the | |
| rule in ``select_demo_questions`` (documented there) from the raw QASPER files, never | |
| written by hand. | |
| """ | |
| from __future__ import annotations | |
| import random | |
| from collections.abc import Callable | |
| from dataclasses import dataclass, field | |
| from .methods import MethodConfig, Resources, build_context | |
| from .quotes import locate | |
| from .retrieval import tokenize | |
| from .schemas import BenchItem, Chunk, Question | |
| from .verification import NLIVerifier, Verdict | |
| DEMO_SEED = 13 | |
| DEMO_N = 12 | |
| MIN_EVIDENCE_WORDS = 8 | |
| MAX_QUESTION_CHARS = 400 | |
| MAX_CLAIM_CHARS = 600 | |
| MIN_CLAIM_CHARS = 8 | |
| MAX_CHECK_CHUNKS = 3 | |
| class DemoInputError(ValueError): | |
| """Bad visitor input (maps to HTTP 422).""" | |
| def select_demo_questions(questions: list[Question], chunks: list[Chunk], titles: dict[str, str], n: int = DEMO_N, seed: int = DEMO_SEED) -> list[dict]: | |
| """Deterministic selection of real QASPER questions for the demo. | |
| Rule: the question is about a corpus paper; no annotator marked it unanswerable; at | |
| least one annotator's highlighted evidence sentence has ``MIN_EVIDENCE_WORDS`` or more | |
| words, at most ``MAX_CLAIM_CHARS`` characters, and anchors verbatim (``quotes.locate``) in one of that paper's corpus chunks. | |
| Eligible questions are sorted by question id, shuffled with ``seed``, and taken in | |
| that order with at most one question per paper, until ``n`` are chosen. The first | |
| anchoring evidence sentence (annotator order, then sentence order) is recorded with | |
| its chunk and character offsets, so the demo can show where the gold evidence is. | |
| """ | |
| by_doc: dict[str, list[Chunk]] = {} | |
| for c in chunks: | |
| by_doc.setdefault(c.doc_id, []).append(c) | |
| eligible: list[tuple[Question, dict]] = [] | |
| for q in questions: | |
| if q.doc_id not in by_doc or q.unanswerable_any or not q.references: | |
| continue | |
| ev = _anchor_evidence(q, by_doc[q.doc_id]) | |
| if ev is not None: | |
| eligible.append((q, ev)) | |
| eligible.sort(key=lambda t: t[0].qid) | |
| random.Random(seed).shuffle(eligible) | |
| out, seen = [], set() | |
| for q, ev in eligible: | |
| if len(out) >= n: | |
| break | |
| if q.doc_id in seen: | |
| continue | |
| seen.add(q.doc_id) | |
| out.append( | |
| { | |
| "qid": q.qid, | |
| "doc_id": q.doc_id, | |
| "title": titles.get(q.doc_id, q.doc_id), | |
| "question": q.question, | |
| "answers": list(dict.fromkeys(r.answer for r in q.references)), | |
| "evidence": ev, | |
| } | |
| ) | |
| return out | |
| def _anchor_evidence(q: Question, doc_chunks: list[Chunk]) -> dict | None: | |
| for r in q.references: | |
| for sent in r.highlighted_evidence: | |
| # Long enough to be a claim, short enough to be checked as one (the demo pre-fills it). | |
| if len(sent.split()) < MIN_EVIDENCE_WORDS or len(sent) > MAX_CLAIM_CHARS: | |
| continue | |
| for c in doc_chunks: | |
| hit = locate(sent, c.text) | |
| if hit is not None: | |
| return {"text": sent, "chunk_id": c.chunk_id, "start": hit[0], "end": hit[1]} | |
| return None | |
| class LiveDemo: | |
| """Holds the BM25 index over the corpus and the (lazily loaded) NLI verifier.""" | |
| chunks: dict[str, Chunk] | |
| titles: dict[str, str] | |
| budget: int = 2048 | |
| nli: NLIVerifier | None = None | |
| _res: Resources | None = field(default=None, repr=False) | |
| def resources(self) -> Resources: | |
| if self._res is None: | |
| # text_rag touches only chunks, titles and BM25; no model client or graph is needed. | |
| self._res = Resources(base=None, kg=None, linker=None, chunks=self.chunks, titles=self.titles) # type: ignore[arg-type] | |
| return self._res | |
| def retrieve(self, question: str, gold_chunk_id: str | None = None) -> dict: | |
| question = (question or "").strip() | |
| if not question: | |
| raise DemoInputError("question is empty") | |
| if len(question) > MAX_QUESTION_CHARS: | |
| raise DemoInputError(f"question is longer than {MAX_QUESTION_CHARS} characters") | |
| if not tokenize(question): | |
| raise DemoInputError("question has no searchable words") | |
| res = self.resources() | |
| item = BenchItem( | |
| item_id="live", | |
| category="single_hop", | |
| difficulty=1, | |
| template_id="live", | |
| question=question, | |
| question_template=question, | |
| gold_decision="answer", | |
| ) | |
| ctx = build_context(MethodConfig(name="text_rag", kind="text_rag", context_budget=self.budget), item, res) | |
| ranking = res.bm25_rank(question) | |
| rank_of = {cid: i + 1 for i, cid in enumerate(ranking)} | |
| scores = res._bm25.get_scores(tokenize(question)) | |
| score_of = dict(zip(res._order, (float(s) for s in scores))) | |
| passages = [] | |
| for handle, cid in ctx.chunk_handles.items(): | |
| c = self.chunks[cid] | |
| passages.append( | |
| { | |
| "handle": handle, | |
| "chunk_id": cid, | |
| "doc_id": c.doc_id, | |
| "title": self.titles.get(c.doc_id, c.doc_id), | |
| "section": c.section_heading, | |
| "rank": rank_of[cid], | |
| "bm25": round(score_of[cid], 4), | |
| "n_tokens": c.n_tokens, | |
| } | |
| ) | |
| gold = None | |
| if gold_chunk_id: | |
| if gold_chunk_id not in self.chunks: | |
| raise DemoInputError(f"unknown chunk {gold_chunk_id!r}") | |
| gold = {"chunk_id": gold_chunk_id, "rank": rank_of[gold_chunk_id], "in_context": gold_chunk_id in ctx.chunk_handles.values()} | |
| return { | |
| "question": question, | |
| "budget": self.budget, | |
| "context_tokens": ctx.tokens, | |
| "corpus_chunks": len(self.chunks), | |
| "passages": passages, | |
| "gold": gold, | |
| "context_text": ctx.text, | |
| } | |
| def check(self, claim: str, chunk_ids: list[str], use_nli: bool = True, verify: Callable[[str, list[str]], Verdict] | None = None) -> dict: | |
| """Anchor ``claim`` in each passage and score it with the NLI verifier. | |
| ``verify`` replaces ``self.nli.verify`` for the scoring call only (the Space routes it | |
| through a ZeroGPU function); thresholds and model metadata still come from ``self.nli``. | |
| """ | |
| claim = (claim or "").strip() | |
| if len(claim) < MIN_CLAIM_CHARS: | |
| raise DemoInputError(f"claim must have at least {MIN_CLAIM_CHARS} characters") | |
| if len(claim) > MAX_CLAIM_CHARS: | |
| raise DemoInputError(f"claim is longer than {MAX_CLAIM_CHARS} characters") | |
| ids = list(dict.fromkeys(chunk_ids or [])) | |
| if not ids: | |
| raise DemoInputError("select at least one passage") | |
| if len(ids) > MAX_CHECK_CHUNKS: | |
| raise DemoInputError(f"select at most {MAX_CHECK_CHUNKS} passages") | |
| missing = [c for c in ids if c not in self.chunks] | |
| if missing: | |
| raise DemoInputError(f"unknown chunk {missing[0]!r}") | |
| texts = [self.chunks[c].text for c in ids] | |
| passages = [] | |
| for cid, text in zip(ids, texts): | |
| hit = locate(claim, text) | |
| passages.append( | |
| { | |
| "chunk_id": cid, | |
| "title": self.titles.get(self.chunks[cid].doc_id, self.chunks[cid].doc_id), | |
| "section": self.chunks[cid].section_heading, | |
| "anchor": {"start": hit[0], "end": hit[1], "elided": hit[2]} if hit else None, | |
| } | |
| ) | |
| out: dict = {"claim": claim, "passages": passages, "nli": None} | |
| if use_nli: | |
| if self.nli is None: | |
| raise RuntimeError("NLI verifier not configured") | |
| v = (verify or self.nli.verify)(claim, texts) | |
| for p, text, ps in zip(passages, texts, v.detail["per_source"]): | |
| win = ps.get("window") | |
| at = locate(win, text) if win else None | |
| p["nli"] = { | |
| "entail": ps["entail"], | |
| "contradict": ps["contradict"], | |
| "window": win, | |
| "window_start": at[0] if at else None, | |
| "window_end": at[1] if at else None, | |
| } | |
| out["nli"] = { | |
| "label": v.label, | |
| "entail": round(v.entail, 4), | |
| "contradict": round(v.contradict, 4), | |
| "joint": v.detail.get("joint"), | |
| "threshold": self.nli.entail_threshold, | |
| "model": self.nli.model_name, | |
| "window_sentences": self.nli.window_sentences, | |
| } | |
| return out | |