"""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 @dataclass 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