""" retrieval.py ============ RAG-style retrieval pipeline for InformationIntegrityEnv. Pipeline: 1. QueryRewriter — rewrites claims into effective search queries 2. RelevanceFilter — scores claim↔source semantic similarity 3. SourceRanker — combines similarity + reliability into a final score 4. RetrievalPipeline — orchestrates all stages, returns ranked sources only Public helper API (importable standalone): rewrite_query(claim) → str generate_queries(claim) → List[str] filter_relevant_sources(claim, sources) → List[dict] Zero runtime API calls. Fully deterministic given the same inputs. Embedding backend (auto-selected at startup): Priority 1 → sentence-transformers (all-MiniLM-L6-v2, fast + accurate) Priority 2 → scikit-learn TF-IDF (good for keyword-rich claims) Priority 3 → BM25-style BoW cosine (pure stdlib, always available) Configuration via retrieval_config.yaml or environment variables. """ from __future__ import annotations import logging import math import os import re import string from dataclasses import dataclass, field from typing import List, Optional, Tuple import yaml logger = logging.getLogger("retrieval") RETRIEVAL_CONFIG_PATH = os.getenv("RETRIEVAL_CONFIG", "retrieval_config.yaml") # ───────────────────────────────────────────────────────────────────────────── # Config # ───────────────────────────────────────────────────────────────────────────── _DEFAULTS = { "relevance_threshold": 0.08, # min TF-IDF similarity (bigrams improve precision enough) "reliability_threshold": 0.30, # min reliability to keep a source "top_k": 6, # max sources returned per claim "similarity_weight": 0.55, # weight in combined score "reliability_weight": 0.45, # weight in combined score "embedding_model": "all-MiniLM-L6-v2", "keyword_boost_min_overlap": 2, # keyword overlap to grant a sim boost to high-rel sources "keyword_boost_amount": 0.08, # added to sim when high-rel source has sufficient overlap "keyword_boost_min_rel": 0.65, # min reliability to receive keyword boost } class RetrievalConfig: def __init__(self, path: str = RETRIEVAL_CONFIG_PATH) -> None: self._cfg = dict(_DEFAULTS) if os.path.exists(path): with open(path, "r", encoding="utf-8") as fh: loaded = yaml.safe_load(fh) or {} self._cfg.update(loaded) def __getattr__(self, name: str): if name == "_cfg": raise AttributeError(name) try: return self._cfg[name] except KeyError: raise AttributeError(f"RetrievalConfig has no attribute '{name}'") _CFG = RetrievalConfig() # ───────────────────────────────────────────────────────────────────────────── # Data class for a source that has been scored # ───────────────────────────────────────────────────────────────────────────── @dataclass class ScoredSource: id: str title: str text: str url: str stance: str reliability: float source_type: str similarity: float = 0.0 # claim↔source semantic similarity [0,1] combined: float = 0.0 # final ranking score [0,1] def to_dict(self) -> dict: return { "id": self.id, "title": self.title, "text": self.text, "url": self.url, "stance": self.stance, "reliability": round(self.reliability, 4), "source_type": self.source_type, "similarity": round(self.similarity, 4), "combined": round(self.combined, 4), } # ───────────────────────────────────────────────────────────────────────────── # Part 1 — Query Rewriter # ───────────────────────────────────────────────────────────────────────────── # ── Claim type detection ───────────────────────────────────────────────────── _CLAIM_TYPE_PATTERNS = [ ("scientific_fact", re.compile( r"\b\d+\s*°[cCfF]\b" r"|\b\d+\s*(degrees?|celsius|fahrenheit|kelvin)\b" r"|\b(speed|velocity)\s+of\s+(light|sound)\b" r"|\b(boil|melt|freez)\w+\s+(point|at)\b" r"|\b(atomic|molecular|chemical|formula|element|compound|periodic)\b" r"|\bph\s*(of|level|value)\b" r"|\b\d+\s*(km|m|cm|mm|kg|g|mg|l|ml|hz|mhz|ghz|nm|km/h|m/s)\b", re.IGNORECASE, )), ("myth", re.compile( r"\b(only\s+use|only\s+uses?)\s+\d+\s*(%|percent)" r"|\b\d+\s*(%|percent)\s+of\s+(your|our|the|their|human)\s+brain\b" r"|\b(flat\s+earth|earth\s+is\s+flat)\b" r"|\b(vaccines?\s+(cause|causes?|link|linked)\s+autism)\b" r"|\b(5g\s+(cause|spread|create)\w*\s+(covid|virus|cancer))\b" r"|\b(bleach\s+cure|drink\s+bleach|inject\s+disinfectant)\b" r"|\b(moon\s+landing\s+(fake|hoax|staged))\b" r"|\b(evolution\s+is\s+(false|fake|wrong|a\s+lie))\b", re.IGNORECASE, )), ("health", re.compile( r"\b(vaccine|vaccination|drug|medication|dose|treatment|cure|cancer|" r"covid|virus|bacteria|disease|symptom|clinical|fda|who|cdc|nhs|" r"study|trial|efficacy|side\s+effect|immune)\b", re.IGNORECASE, )), ("news_event", re.compile( r"\b(government|ministry|president|prime\s+minister|parliament|congress|" r"announce[sd]?|pass(?:ed)?|sign(?:ed)?|reform|policy|election|vote|" r"january|february|march|april|may|june|july|august|september|" r"october|november|december|\b20\d{2}\b)\b", re.IGNORECASE, )), ] # ── Symbol / phrase normalisation ──────────────────────────────────────────── _SYMBOL_NORMALISE = [ (re.compile(r"(\d+)\s*°\s*[Cc]\b"), r"\1 degrees celsius"), (re.compile(r"(\d+)\s*°\s*[Ff]\b"), r"\1 degrees fahrenheit"), (re.compile(r"(\d+)\s*°\s*[Kk]\b"), r"\1 kelvin"), (re.compile(r"(\d+)\s*°\b"), r"\1 degrees"), (re.compile(r"(\d+)\s*%\b"), r"\1 percent"), (re.compile(r"\bkm/h\b"), "kilometres per hour"), (re.compile(r"\bm/s\b"), "metres per second"), (re.compile(r"\bpH\b"), "pH level"), (re.compile(r"\bCO2\b"), "carbon dioxide"), (re.compile(r"\bH2O\b"), "water"), (re.compile(r"\bO2\b"), "oxygen"), (re.compile(r"\bNaCl\b"), "sodium chloride salt"), ] # ── Type-specific rewrite rules ─────────────────────────────────────────────── _REWRITE_RULES = [ ("scientific_fact", re.compile(r"^(water|mercury|lead|iron|alcohol|ethanol|nitrogen|oxygen)\s+boils?\s+at\b", re.IGNORECASE), lambda m: f"boiling point of {m.group(1).lower()}"), ("scientific_fact", re.compile(r"^(.*?)\s+melts?\s+at\b", re.IGNORECASE), lambda m: f"melting point of {m.group(1).lower()}"), ("myth", re.compile(r"humans?\s+only\s+use[sd]?\s+(\d+\s*(?:percent|%))\s+of\s+(?:their|the|our|human)?\s*brain", re.IGNORECASE), lambda m: f"{m.group(1).replace('%','percent')} brain myth scientific fact"), ("myth", re.compile(r"(earth|world)\s+is\s+flat", re.IGNORECASE), lambda m: f"{m.group(1)} spherical globe scientific proof"), ] # ── Negative context keywords (for off-topic filtering) ────────────────────── _NEGATIVE_CONTEXT = { "scientific_fact": ["reactor", "nuclear", "fission", "steam turbine", "power plant", "cooking tip", "recipe for"], "myth": [], "health": ["fictional", "novel", "movie", "song", "game"], "news_event": ["historical", "ancient", "fiction", "fantasy"], "default": [], } # Fact-check site anchors _FACTCHECK_SITES = ( "site:snopes.com OR site:factcheck.org OR site:politifact.com OR " "site:fullfact.org OR site:boomlive.in OR site:altnews.in OR " "site:factchecker.in OR site:vishvasnews.com" ) # Countries for lightweight NER _COUNTRIES = [ "india", "china", "usa", "united states", "uk", "united kingdom", "russia", "pakistan", "france", "germany", "japan", "australia", "canada", "brazil", "nigeria", "south africa", "italy", "spain", "mexico", "indonesia", "turkey", "saudi arabia", "iran", "ukraine", "israel", "egypt", "argentina", ] _TOPIC_TEMPLATES = { "health": ["{claim}", "fact check {claim}", "is it true {claim} evidence study"], "politics": ["{claim}", "fact check {claim}", "{claim} evidence official statement"], "technology": ["{claim}", "{claim} scientific evidence", "fact check {claim}"], "climate": ["{claim}", "{claim} scientific consensus", "fact check {claim}"], "default": ["{claim}", "fact check {claim}", "{claim} true false evidence"], } _TOPIC_SIGNALS = { "health": r"vaccine|virus|cancer|disease|drug|health|medical|cure|symptom|covid|pandemic", "politics": r"government|president|election|congress|policy|law|vote|senator|parliament", "technology": r"5g|ai|chip|algorithm|robot|software|data|internet|cyber|phone", "climate": r"climate|carbon|emission|warming|fossil|renewable|greenhouse|temperature", } def _classify_claim_type(claim: str) -> str: """Return the claim type: scientific_fact | myth | health | news_event | default.""" for label, pattern in _CLAIM_TYPE_PATTERNS: if pattern.search(claim): return label return "default" def _normalise_symbols(text: str) -> str: """Replace symbols with word forms for better search-engine matching.""" for pattern, replacement in _SYMBOL_NORMALISE: text = pattern.sub(replacement, text) return text class QueryRewriter: """ Rewrites a raw claim string into multiple targeted search queries. Key improvements: - Claim type detection (scientific_fact / myth / health / news_event) - Symbol normalisation (°C → degrees celsius, % → percent) - Type-specific rewrite rules (boiling point phrasing, myth framing) - 4-variant expansion: factual | verbatim | question-style | fact-check """ def rewrite(self, claim: str) -> str: """ Return the best single search query for a claim. Steps: strip prefixes → normalise symbols → type-specific rewrite. """ claim = claim.strip().rstrip("?!.") claim = re.sub( r"^(BREAKING|URGENT|CONFIRMED|SHOCKING|EXCLUSIVE)[:\s!]+", "", claim, flags=re.IGNORECASE, ).strip() claim = re.sub( r"^(i heard|they say|apparently|reportedly|rumour has it)[,\s]+", "", claim, flags=re.IGNORECASE, ).strip() normalised = _normalise_symbols(claim) claim_type = _classify_claim_type(claim) for rule_type, pattern, replacement in _REWRITE_RULES: if rule_type != claim_type: continue m = pattern.search(normalised) if m: prefix = replacement(m) tail = normalised[m.end():].strip() result = f"{prefix} {tail}".strip() logger.debug("QueryRewriter [%s]: '%s' → '%s'", claim_type, claim[:60], result[:60]) return result return normalised def expand(self, claim: str) -> List[str]: """ Generate up to 4 orthogonal search query variants. Q1 — rewritten/factual query (precision for scientific facts & myths) Q2 — verbatim claim (recall for news articles) Q3 — question-style (FAQ / encyclopedia pages) Q4 — fact-check site anchor (explicit debunking/verification) """ cleaned = claim.strip().rstrip("?!.") rewritten = self.rewrite(cleaned) claim_type = _classify_claim_type(cleaned) entities = self._extract_entities(cleaned) queries: List[str] = [] # Q1: type-specific primary query if claim_type == "scientific_fact": queries.append(rewritten) elif claim_type == "myth": base = rewritten if rewritten != cleaned else cleaned queries.append(f"{base} myth scientific evidence explanation") elif claim_type == "health": queries.append(f"{rewritten} clinical evidence study research") elif claim_type == "news_event": anchors = [e for e in [entities.get("country"), entities.get("date")] if e] kw = " ".join(entities.get("keywords", [])[:3]) queries.append(f'{" ".join(anchors)} official {kw}'.strip() if anchors else rewritten) else: queries.append(rewritten) # Q2: verbatim (finds news articles that quote the claim directly) if cleaned.lower() != queries[0].lower(): queries.append(cleaned) # Q3: question-style question = self._to_question(rewritten, claim_type) if question not in queries: queries.append(question) # Q4: fact-check site anchor fc_query = f'{_FACTCHECK_SITES} {rewritten}' if fc_query not in queries: queries.append(fc_query) seen: set = set() result = [] for q in queries: if q not in seen: seen.add(q) result.append(q) return result[:4] def to_semantic_text(self, claim: str) -> str: """ Produce the text embedded when comparing a claim against a source. Repeats key entities so TF-IDF gives them higher weight. """ rewritten = self.rewrite(claim) entities = self._extract_entities(claim) parts = [rewritten] if entities.get("country"): parts.append(entities["country"]) if entities.get("date"): parts.append(entities["date"]) parts.extend(entities.get("keywords", [])[:3]) return " ".join(parts) def extract_entities(self, claim: str) -> dict: """Public wrapper around _extract_entities.""" return self._extract_entities(self.rewrite(claim)) def _extract_entities(self, text: str) -> dict: low = text.lower() country = None for c in sorted(_COUNTRIES, key=len, reverse=True): if c in low: country = c.title() break date_match = re.search( r'\b(?:jan(?:uary)?|feb(?:ruary)?|mar(?:ch)?|apr(?:il)?|may|jun(?:e)?|' r'jul(?:y)?|aug(?:ust)?|sep(?:t(?:ember)?)?|oct(?:ober)?|nov(?:ember)?|' r'dec(?:ember)?)[\s.]*\d{1,2}(?:st|nd|rd|th)?\b' r'|\b\d{1,2}[\s/-]\d{1,2}[\s/-]\d{2,4}\b' r'|\b\d{4}\b', text, re.IGNORECASE, ) date = date_match.group(0).strip() if date_match else None qty_match = re.search( r'\b\d+(?:\.\d+)?(?:\s*percent|\s*million|\s*billion|\s*crore|\s*lakh)?\b', text, ) _stop = { "the","a","an","is","are","was","were","in","on","at","to","from", "of","and","or","but","that","this","it","be","will","has","have", "its","by","for","not","with","also","into","than","more","about", } keywords = [ w for w in re.findall(r'\b[a-zA-Z]{4,}\b', text.lower()) if w not in _stop ][:5] return { "country": country, "date": date, "quantity": qty_match.group(0) if qty_match else None, "keywords": keywords, } @staticmethod def _to_question(rewritten: str, claim_type: str) -> str: low = rewritten.lower() if claim_type == "scientific_fact": if "boiling point" in low or "melting point" in low: return f"what is the {rewritten}" return f"is it true that {rewritten}" if claim_type == "myth": return f"is the {rewritten} true or false" if claim_type == "health": return f"does {rewritten} according to research" return f"is it true that {rewritten}" @staticmethod def _detect_topic(text: str) -> str: low = text.lower() for topic, pattern in _TOPIC_SIGNALS.items(): if re.search(pattern, low): return topic return "default" # ───────────────────────────────────────────────────────────────────────────── # Part 2 — Embedding / Similarity backend (auto-selected) # ───────────────────────────────────────────────────────────────────────────── class _EmbeddingBackend: """Abstract interface for all similarity backends.""" name: str = "base" def similarity(self, text_a: str, text_b: str) -> float: raise NotImplementedError def batch_similarity(self, query: str, texts: List[str]) -> List[float]: return [self.similarity(query, t) for t in texts] class _SentenceTransformerBackend(_EmbeddingBackend): name = "sentence-transformers" def __init__(self, model_name: str) -> None: from sentence_transformers import SentenceTransformer import numpy as np self._model = SentenceTransformer(model_name) self._np = np logger.info("Embedding backend: sentence-transformers (%s)", model_name) def batch_similarity(self, query: str, texts: List[str]) -> List[float]: np = self._np all_t = [query] + texts embs = self._model.encode(all_t, normalize_embeddings=True, show_progress_bar=False) q_emb = embs[0] sims = [float(np.dot(q_emb, e)) for e in embs[1:]] return [max(0.0, min(1.0, s)) for s in sims] def similarity(self, text_a: str, text_b: str) -> float: return self.batch_similarity(text_a, [text_b])[0] class _TfidfBackend(_EmbeddingBackend): name = "tfidf" def __init__(self) -> None: from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity as cos_sim import numpy as np self._Tfidf = TfidfVectorizer self._cos = cos_sim self._np = np logger.info("Embedding backend: scikit-learn TF-IDF") def batch_similarity(self, query: str, texts: List[str]) -> List[float]: if not texts: return [] corpus = [query] + texts try: mat = self._Tfidf( stop_words="english", sublinear_tf=True, # log(1+tf) — reduces noise from repeated words ngram_range=(1, 2), # bigrams catch "sunday holiday", "april 5" min_df=1, # don't drop rare claim-specific terms max_features=8000, ).fit_transform(corpus) sims = self._cos(mat[0:1], mat[1:]).flatten().tolist() return [max(0.0, min(1.0, float(s))) for s in sims] except Exception: return [0.0] * len(texts) def similarity(self, text_a: str, text_b: str) -> float: return self.batch_similarity(text_a, [text_b])[0] class _BowBackend(_EmbeddingBackend): """ Bag-of-words cosine similarity. Pure stdlib — zero extra dependencies. Sufficient for keyword-based filtering when better backends are absent. """ name = "bow" def __init__(self) -> None: logger.info("Embedding backend: BoW cosine (stdlib fallback)") self._stop = { "a","an","the","is","it","in","on","at","to","for","of","and", "or","but","not","be","are","was","were","has","have","had", "this","that","with","from","by","as","about","its","their", } def _tokenize(self, text: str) -> dict[str, int]: tokens = re.sub(r"[^a-z0-9\s]", " ", text.lower()).split() freq: dict[str, int] = {} for t in tokens: if t not in self._stop and len(t) > 2: freq[t] = freq.get(t, 0) + 1 return freq def similarity(self, text_a: str, text_b: str) -> float: fa, fb = self._tokenize(text_a), self._tokenize(text_b) if not fa or not fb: return 0.0 vocab = set(fa) | set(fb) dot = sum(fa.get(t, 0) * fb.get(t, 0) for t in vocab) norm_a = math.sqrt(sum(v * v for v in fa.values())) norm_b = math.sqrt(sum(v * v for v in fb.values())) if norm_a == 0 or norm_b == 0: return 0.0 return max(0.0, min(1.0, dot / (norm_a * norm_b))) def batch_similarity(self, query: str, texts: List[str]) -> List[float]: return [self.similarity(query, t) for t in texts] def _build_backend(model_name: str) -> _EmbeddingBackend: """Auto-select the best available embedding backend.""" try: return _SentenceTransformerBackend(model_name) except Exception: pass try: return _TfidfBackend() except Exception: pass return _BowBackend() # Singleton — loaded once at import time _BACKEND: Optional[_EmbeddingBackend] = None def get_backend() -> _EmbeddingBackend: global _BACKEND if _BACKEND is None: _BACKEND = _build_backend(_CFG.embedding_model) return _BACKEND # ───────────────────────────────────────────────────────────────────────────── # Part 3 — Relevance Filter # ───────────────────────────────────────────────────────────────────────────── class RelevanceFilter: """ Scores each source against a claim using semantic similarity. Filters out sources below the relevance threshold. Also enforces a minimum reliability gate. """ def __init__( self, relevance_threshold: float = _CFG.relevance_threshold, reliability_threshold: float = _CFG.reliability_threshold, ) -> None: self._rel_thr = relevance_threshold self._crd_thr = reliability_threshold def filter( self, claim_text: str, sources: List[dict], rewriter: Optional[QueryRewriter] = None, ) -> List[ScoredSource]: """ Returns sources that pass both the relevance and reliability gates, each annotated with its similarity score. High-reliability sources receive a keyword-overlap boost. Off-topic sources (e.g. "boiling water reactor" for a temperature claim) receive a -0.20 sim penalty before the gate is applied. """ if not sources: return [] rw = rewriter or QueryRewriter() query_txt = rw.to_semantic_text(claim_text) claim_type = _classify_claim_type(claim_text) backend = get_backend() source_texts = [ s.get("title", "") + " " + s.get("text", "") for s in sources ] similarities = backend.batch_similarity(query_txt, source_texts) # Precompute keyword overlap for the boost _stop = {"the","a","an","is","are","was","were","in","on","at","to","from", "of","and","or","but","that","this","it","be","will","has","have"} claim_kws = { w for w in re.findall(r'\b[a-z]{3,}\b', claim_text.lower()) if w not in _stop } boost_min_overlap = int(_CFG.keyword_boost_min_overlap) boost_amount = float(_CFG.keyword_boost_amount) boost_min_rel = float(_CFG.keyword_boost_min_rel) scored = [] for src, sim in zip(sources, similarities): rel = float(src.get("reliability", 0.0)) # Keyword boost: high-reliability sources with ≥N overlapping claim # keywords get a sim boost so they aren't penalised for paraphrasing if rel >= boost_min_rel and claim_kws: src_kws = { w for w in re.findall(r'\b[a-z]{3,}\b', (src.get("title","") + " " + src.get("text","")).lower()) if w not in _stop } if len(claim_kws & src_kws) >= boost_min_overlap: sim = min(1.0, sim + boost_amount) # Off-topic noise penalty: penalise sources whose text contains # known negative-context words for this claim type but lacks # any claim keyword. This kills "boiling water reactor" matches # when the claim is about temperature, "recipe for pasta" etc. neg_words = _NEGATIVE_CONTEXT.get(claim_type, []) if neg_words: src_low = (src.get("title", "") + " " + src.get("text", "")).lower() has_neg = any(neg in src_low for neg in neg_words) has_claim_kw = bool(claim_kws & { w for w in re.findall(r'\b[a-z]{3,}\b', src_low) if w not in _stop }) if has_neg and not has_claim_kw: logger.debug( "Off-topic penalty applied to '%s'", src.get("title", "?")[:50], ) sim = max(0.0, sim - 0.20) # Hard gates if sim < self._rel_thr: logger.debug( "Filtered out '%s' (sim=%.3f < threshold=%.3f)", src.get("title", "?")[:50], sim, self._rel_thr, ) continue if rel < self._crd_thr: logger.debug( "Filtered out '%s' (reliability=%.3f < threshold=%.3f)", src.get("title", "?")[:50], rel, self._crd_thr, ) continue scored.append(ScoredSource( id = src.get("id", ""), title = src.get("title", ""), text = src.get("text", ""), url = src.get("url", ""), stance = src.get("stance", "neutral"), reliability = rel, source_type = src.get("source_type", "unknown"), similarity = round(sim, 4), )) return scored # ───────────────────────────────────────────────────────────────────────────── # Part 4 — Source Ranker # ───────────────────────────────────────────────────────────────────────────── class SourceRanker: """ Ranks filtered sources by a combined score: combined = w_sim × similarity + w_rel × reliability Sources that are contradicting/supporting receive a stance boost because they carry more signal than neutral sources. """ STANCE_BOOST = {"supporting": 1.10, "contradicting": 1.10, "neutral": 0.80} def __init__( self, similarity_weight: float = _CFG.similarity_weight, reliability_weight: float = _CFG.reliability_weight, ) -> None: self._w_sim = similarity_weight self._w_rel = reliability_weight def rank( self, sources: List[ScoredSource], top_k: int = _CFG.top_k, ) -> List[ScoredSource]: """Return at most top_k sources sorted by combined score descending.""" for src in sources: boost = self.STANCE_BOOST.get(src.stance, 1.0) src.combined = round( (self._w_sim * src.similarity + self._w_rel * src.reliability) * boost, 4, ) ranked = sorted(sources, key=lambda s: -s.combined) return ranked[:top_k] # ───────────────────────────────────────────────────────────────────────────── # Part 5 — Full RetrievalPipeline # ───────────────────────────────────────────────────────────────────────────── class RetrievalPipeline: """ Orchestrates: rewrite → filter → rank → top-k sources per claim. Usage ----- pipeline = RetrievalPipeline() ranked = pipeline.retrieve(claim_text="5G causes COVID", sources=scenario_sources) # ranked is a List[ScoredSource] — relevant only, sorted by quality """ def __init__( self, rewriter: Optional[QueryRewriter] = None, filt: Optional[RelevanceFilter] = None, ranker: Optional[SourceRanker] = None, top_k: int = _CFG.top_k, ) -> None: self._rewriter = rewriter or QueryRewriter() self._filter = filt or RelevanceFilter() self._ranker = ranker or SourceRanker() self._top_k = top_k def retrieve( self, claim_text: str, sources: List[dict], top_k: Optional[int] = None, ) -> List[ScoredSource]: """ Full pipeline for a single claim against a pool of sources. Returns ------- List[ScoredSource] — relevant, ranked, top-k. May be empty if nothing passes the relevance/reliability gates. """ k = top_k or self._top_k filtered = self._filter.filter(claim_text, sources, self._rewriter) ranked = self._ranker.rank(filtered, top_k=k) logger.info( "Retrieval: claim='%s…' → %d/%d sources pass gates, top-%d returned", claim_text[:50], len(filtered), len(sources), len(ranked), ) return ranked def retrieve_for_all_claims( self, claims: List[str], sources: List[dict], top_k: Optional[int] = None, ) -> dict[str, List[ScoredSource]]: """ Run retrieval for every claim, return a dict keyed by claim text. Sources may appear in multiple claim buckets. """ return { claim: self.retrieve(claim, sources, top_k) for claim in claims } # ───────────────────────────────────────────────────────────────────────────── # Public standalone API # Importable directly: from retrieval import rewrite_query, generate_queries, # filter_relevant_sources # ───────────────────────────────────────────────────────────────────────────── _REWRITER_SINGLETON = QueryRewriter() def rewrite_query(claim: str) -> str: """ Convert a raw claim into an optimised search query. Handles: - Scientific facts → noun-phrase form ("boiling point of water 100 degrees celsius") - Myths → debunking frame ("10 percent brain myth scientific fact") - News events → entity-anchored (keeps country/date context) - Symbols → word form (°C → degrees celsius, % → percent) Examples -------- >>> rewrite_query("Water boils at 100°C at sea level") 'boiling point of water 100 degrees celsius at sea level' >>> rewrite_query("Humans only use 10% of their brain") '10 percent brain myth scientific fact' >>> rewrite_query("India removed Sunday holidays in April") 'India removed Sunday holidays in April' # news event — kept as-is with normalisation """ return _REWRITER_SINGLETON.rewrite(claim) def generate_queries(claim: str) -> List[str]: """ Generate 2–4 orthogonal search query variants for a claim. Variant 1 — factual / rewritten (best precision for scientific & myth claims) Variant 2 — verbatim claim (best recall for news articles) Variant 3 — question-style ("what is the boiling point of water?") Variant 4 — fact-check anchored ("site:snopes.com OR site:factcheck.org …") Examples -------- >>> generate_queries("Water boils at 100°C at sea level") [ 'boiling point of water 100 degrees celsius at sea level', 'Water boils at 100°C at sea level', 'what is the boiling point of water 100 degrees celsius at sea level', 'site:snopes.com OR ... boiling point of water ...', ] """ return _REWRITER_SINGLETON.expand(claim) def filter_relevant_sources( claim: str, sources: List[dict], top_k: int = 5, relevance_threshold: float = 0.08, ) -> List[dict]: """ Filter and rank a list of source dicts by relevance to the claim. Each source dict must contain at least: {"title": str, "snippet": str, "url": str} Optional keys used if present: "reliability" (float), "stance" (str). Scoring ------- relevance_score = keyword_overlap_ratio (40%) + semantic_similarity (60%, via TF-IDF or sentence-transformers) - off_topic_penalty (-0.20 if negative-context words found but no claim keywords present) Off-topic filter example ------------------------ Claim: "Water boils at 100°C at sea level" Rejected: "Boiling Water Reactor — nuclear power plant design" (contains "reactor", "nuclear", no temperature keywords) Accepted: "Boiling point of water — Wikipedia" (contains "boiling", "point", "water", "celsius") Parameters ---------- claim : the raw claim string sources : list of source dicts top_k : max number of results to return (default 5) relevance_threshold: minimum relevance score to keep a source (default 0.08) Returns ------- List of source dicts, each augmented with: "relevance_score" : float (0–1, higher = more relevant) "keyword_overlap" : int (number of claim keywords found in source) Sorted by relevance_score descending, top_k results only. """ if not sources: return [] claim_type = _classify_claim_type(claim) rewritten = _REWRITER_SINGLETON.rewrite(claim) backend = get_backend() _stop = { "the","a","an","is","are","was","were","in","on","at","to","from", "of","and","or","but","that","this","it","be","will","has","have", "its","by","for","not","with","also","into","than","more","about", } # Claim keywords (from both original and rewritten form) claim_kws = { w for w in re.findall(r'\b[a-z]{3,}\b', (claim + " " + rewritten).lower()) if w not in _stop } # Normalise sources so both "snippet" and "text" keys work normalised = [] for s in sources: text = s.get("snippet") or s.get("text") or "" normalised.append({**s, "_body": s.get("title", "") + " " + text}) # Semantic similarity scores bodies = [s["_body"] for s in normalised] sem_scores = backend.batch_similarity( _REWRITER_SINGLETON.to_semantic_text(claim), bodies ) neg_words = _NEGATIVE_CONTEXT.get(claim_type, []) results = [] for src, sem in zip(normalised, sem_scores): body_low = src["_body"].lower() body_kws = {w for w in re.findall(r'\b[a-z]{3,}\b', body_low) if w not in _stop} # Keyword overlap ratio overlap = len(claim_kws & body_kws) max_possible = max(len(claim_kws), 1) kw_score = min(1.0, overlap / max_possible) # Combined relevance score relevance = 0.40 * kw_score + 0.60 * sem # Off-topic penalty if neg_words: has_neg = any(neg in body_low for neg in neg_words) has_claim_kw = bool(claim_kws & body_kws) if has_neg and not has_claim_kw: relevance = max(0.0, relevance - 0.20) if relevance < relevance_threshold: continue out = {k: v for k, v in src.items() if k != "_body"} out["relevance_score"] = round(relevance, 4) out["keyword_overlap"] = overlap results.append(out) results.sort(key=lambda x: -x["relevance_score"]) return results[:top_k]