Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| 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, | |
| } | |
| 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}" | |
| 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] | |