InfoClassifier / stance_classifier.py
UTKARSH MISHRA
Initial commit: InfoClassifier project
e5e36db
Raw History Blame Contribute Delete
25.9 kB
"""
stance_classifier.py
====================
Three-layer stance classification for source ↔ claim pairs.
Layer 1 β€” Semantic Entailment Heuristics (always runs, zero cost)
Detects indirect contradictions and supports that keyword lists miss.
Example: "Earth is round" contradicts "Earth is flat" via antonym matching.
Layer 2 β€” Expanded Keyword Matching (always runs, zero cost)
Larger keyword lists, threshold=1 for strong signals (false, debunked, etc.)
Falls back on Layer 1 result if ambiguous.
Layer 3 β€” facebook/bart-large-mnli NLI (completely free, no API key needed)
Natural Language Inference: premise=source_text, hypothesis=claim.
ENTAILMENT β†’ SUPPORTS
CONTRADICTION β†’ CONTRADICTS
NEUTRAL β†’ NEUTRAL
Primary: HF Inference API (set HF_TOKEN in .env for higher rate limits;
works without token on public models β€” just slower)
Fallback: local `transformers` pipeline (pip install transformers torch)
Cached in-process β€” same (claim, source) pair never classified twice.
Priority: NLI result > L1 result > L2 result > neutral
"""
from __future__ import annotations
import hashlib
import logging
import os
import re
from typing import Dict, Optional, Tuple
logger = logging.getLogger("stance_classifier")
# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 β€” Semantic Entailment Heuristics
# ─────────────────────────────────────────────────────────────────────────────
# Pairs of (claim_pattern β†’ contradicting_signal) for common flat-earth-style claims
# Each entry: (regex that matches the CLAIM, regex that in the SOURCE indicates contradiction)
_ENTAILMENT_CONTRA_RULES: list[tuple[str, str]] = [
# Shape / science denial
(r"\bflat\s+earth\b|\bearth\s+is\s+flat\b",
r"\bglobe\b|\bspherical\b|\bround\b|\borbits\b|\bnasa\b|\bscientists\b"),
# Vaccine safety
(r"\bvaccine.{0,20}(cause|causes|cause[sd]|causing).{0,20}(autism|dna|death|cancer|infertil)",
r"\bno\s+evidence\b|\bdebunked\b|\bsafe\b|\bclinical\s+trial\b|\bfda\s+approved\b"),
# 5G / radiation
(r"\b5g.{0,30}(spread|cause|transmit|covid|virus|cancer|radiation)",
r"\bno\s+link\b|\bno\s+evidence\b|\bdebunked\b|\bscientifically\b|\bimpossible\b"),
# Climate denial
(r"\bclimate\s+change\s+is\s+(fake|hoax|not\s+real)",
r"\bscientific\s+consensus\b|\bdata\s+shows\b|\bglobal\s+warming\b|\b97\s*%\b"),
# Election / fraud
(r"\belection.{0,20}(stolen|rigged|fraud|fake)",
r"\bno\s+evidence\b|\bcourts?\s+reject\b|\baudit\b.*\bno\s+fraud\b|\bofficial\s+results?\b"),
# General "X never happened"
(r"\b(never\s+happened|didn.t\s+happen|is\s+a\s+hoax|is\s+(fake|fabricated))",
r"\bconfirmed\b|\bverified\b|\bofficial\s+records?\b|\bdocumented\b"),
]
# claim_pattern β†’ source_pattern for SUPPORTING signal
_ENTAILMENT_SUPPORT_RULES: list[tuple[str, str]] = [
# Vaccine efficacy
(r"\bvaccine.{0,20}(effective|efficacy|prevent|protect)",
r"\bclinical\s+trial\b|\befficacy\b|\bphase\s+[23]\b|\bfda\b|\bwho\b"),
# Climate consensus
(r"\bclimate\s+change\s+is\s+real\b|\bglobal\s+warming\b",
r"\bscientific\s+consensus\b|\b97\s*%\b|\bipcc\b|\bnasa\b|\bnoaa\b"),
# Official announcements
(r"\b(government|ministry|official).{0,30}(announce|confirm|pass|enact)",
r"\bofficial\b|\bgovernment\b|\bministry\b|\bgazette\b|\bpib\b|\bofficial\s+statement\b"),
]
# Antonym pairs: if the claim contains word A and source contains word B (or vice versa)
# in the same context window, it's likely a contradiction
_ANTONYM_PAIRS: list[tuple[str, str]] = [
("flat", "spherical"), ("flat", "globe"), ("flat", "round"),
("cure", "no cure"), ("safe", "dangerous"), ("safe", "unsafe"),
("effective", "ineffective"), ("effective", "no evidence"),
("real", "fake"), ("real", "hoax"), ("true", "false"),
("confirmed", "unconfirmed"), ("proven", "disproven"),
("increases", "decreases"), ("causes", "does not cause"),
("linked", "no link"), ("linked", "unrelated"),
("harmful", "harmless"), ("harmful", "safe"),
]
class SemanticHeuristicLayer:
"""
Pattern-matching entailment: catches indirect stances that keyword
lists miss because the source paraphrases rather than explicitly labels.
"""
def classify(self, claim: str, source_text: str) -> Optional[str]:
"""
Returns 'supporting' | 'contradicting' | None.
None means this layer has no signal β€” defer to next layer.
"""
claim_low = claim.lower()
source_low = source_text.lower()
# Check contradiction entailment rules
for claim_pat, source_pat in _ENTAILMENT_CONTRA_RULES:
if re.search(claim_pat, claim_low) and re.search(source_pat, source_low):
logger.debug("SemanticHeuristic CONTRA rule matched: claim_pat=%s", claim_pat[:40])
return "contradicting"
# Check support entailment rules
for claim_pat, source_pat in _ENTAILMENT_SUPPORT_RULES:
if re.search(claim_pat, claim_low) and re.search(source_pat, source_low):
logger.debug("SemanticHeuristic SUPPORT rule matched: claim_pat=%s", claim_pat[:40])
return "supporting"
# Antonym detection: extract key noun/adj from claim, look for antonym in source
for word_a, word_b in _ANTONYM_PAIRS:
if word_a in claim_low and word_b in source_low:
return "contradicting"
if word_b in claim_low and word_a in source_low:
return "contradicting"
return None # no signal
# ─────────────────────────────────────────────────────────────────────────────
# Layer 2 β€” Expanded Keyword Matching
# ─────────────────────────────────────────────────────────────────────────────
# Strong single-hit contradicting signals (threshold=1)
_STRONG_CONTRA = {
"false", "debunked", "disproven", "misinformation", "misleading",
"no evidence", "lacks evidence", "unfounded", "refuted", "myth",
"conspiracy theory", "fact-check", "factcheck", "incorrect",
"not true", "inaccurate", "unproven", "pseudoscience", "hoax",
"wrong", "fake", "fabricated", "baseless", "discredited",
}
# Weak contradicting signals (threshold=2 β€” need two hits)
_WEAK_CONTRA = {
"disputed", "controversial", "unverified", "claims", "alleged",
"questioned", "skeptics", "critics", "challenged", "contradicts",
"refutes", "rejects", "warns against", "cautioned",
}
# Strong single-hit supporting signals (threshold=1)
_STRONG_SUPPORT = {
"confirms", "confirmed", "verified", "validated", "proven",
"evidence shows", "data shows", "research shows", "study finds",
"scientists confirm", "officially announced", "official statement",
"press release", "gazette notification", "government confirms",
"approved", "endorsed", "supported by evidence",
}
# Weak supporting signals (threshold=2)
_WEAK_SUPPORT = {
"suggests", "indicates", "shows", "demonstrates", "according to",
"reports", "states", "announced", "published", "found that",
}
class KeywordLayer:
"""
Expanded keyword matching with differentiated strong/weak thresholds.
Threshold for strong signals = 1 (any single hit is decisive).
Threshold for weak signals = 2 (need corroboration).
"""
def classify(self, text: str) -> Tuple[str, float]:
"""
Returns (stance, confidence).
stance = 'supporting' | 'contradicting' | 'neutral'
confidence = 0.0–1.0 (how certain this layer is)
"""
low = text.lower()
strong_contra = sum(1 for kw in _STRONG_CONTRA if kw in low)
weak_contra = sum(1 for kw in _WEAK_CONTRA if kw in low)
strong_supp = sum(1 for kw in _STRONG_SUPPORT if kw in low)
weak_supp = sum(1 for kw in _WEAK_SUPPORT if kw in low)
contra_score = strong_contra * 2 + weak_contra
supp_score = strong_supp * 2 + weak_supp
# Strong single signal is decisive
if strong_contra >= 1 and strong_contra >= strong_supp:
conf = min(1.0, 0.6 + 0.1 * strong_contra)
return "contradicting", conf
if strong_supp >= 1 and strong_supp >= strong_contra:
conf = min(1.0, 0.6 + 0.1 * strong_supp)
return "supporting", conf
# Weak signals need 2+
if contra_score >= 2 and contra_score > supp_score:
return "contradicting", 0.45
if supp_score >= 2 and supp_score > contra_score:
return "supporting", 0.45
return "neutral", 0.20
# ─────────────────────────────────────────────────────────────────────────────
# Layer 3 β€” facebook/bart-large-mnli (NLI, completely free)
# ─────────────────────────────────────────────────────────────────────────────
_HF_API_URL = "https://router.huggingface.co/hf-inference/models/facebook/bart-large-mnli/v1/zero-shot-classification"
_NLI_ENABLED = os.getenv("STANCE_NLI_ENABLED", "true").lower() != "false"
# Label mapping: HuggingFace returns "ENTAILMENT", "CONTRADICTION", "NEUTRAL"
_NLI_LABEL_MAP = {
"entailment": "supporting",
"contradiction": "contradicting",
"neutral": "neutral",
}
def classify_stance(claim: str, evidence: str) -> dict:
"""
Public NLI helper used by tests / external callers.
premise = evidence (the source text)
hypothesis = claim (what we are verifying)
Returns:
{"stance": "SUPPORTS" | "CONTRADICTS" | "NEUTRAL", "confidence": float}
"""
result = _bart_nli(claim=claim, text=evidence[:512], title="")
if result is None:
return {"stance": "NEUTRAL", "confidence": 0.20}
stance, confidence = result
return {
"stance": {
"supporting": "SUPPORTS",
"contradicting": "CONTRADICTS",
"neutral": "NEUTRAL",
}.get(stance, "NEUTRAL"),
"confidence": confidence,
}
def _bart_nli(claim: str, title: str, text: str) -> Optional[Tuple[str, float]]:
"""
Core NLI call. Returns (stance_internal, confidence) or None on failure.
stance_internal = "supporting" | "contradicting" | "neutral"
Strategy
--------
1. Try HF Inference API (zero-install, rate-limited but free)
2. Try local `transformers` pipeline (no rate limit, needs torch)
3. Return None β†’ caller falls back to keyword result
"""
combined = f"{title} {text}".strip()[:512]
if not combined:
return None
# ── attempt 1: HF Inference API ──────────────────────────────────────────
result = _hf_api_nli(claim=claim, premise=combined)
if result is not None:
return result
# ── attempt 2: local transformers ────────────────────────────────────────
result = _local_nli(claim=claim, premise=combined)
return result # None if both failed
# ── HF Inference API ─────────────────────────────────────────────────────────
def _hf_api_nli(claim: str, premise: str) -> Optional[Tuple[str, float]]:
"""
Calls the HF router zero-shot classification endpoint.
Requires HF_TOKEN β€” set it in .env for free access.
Without a token, skips to local transformers.
Returns (stance, confidence) or None.
"""
try:
import requests as _req
except ImportError:
return None
hf_token = os.getenv("HF_TOKEN", "")
if not hf_token or hf_token.startswith("your_"):
logger.debug("No HF_TOKEN set β€” skipping HF router API, using local transformers")
return None
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {hf_token}",
}
# router.huggingface.co zero-shot-classification format:
# inputs = the text to classify (premise)
# parameters.candidate_labels = the hypothesis options
# multi_label=False β†’ softmax across labels (NLI-style)
nli_payload = {
"inputs": premise,
"parameters": {
"candidate_labels": ["supports the claim", "contradicts the claim", "unrelated to the claim"],
"hypothesis_template": "{}.",
"multi_label": False,
},
}
try:
resp = _req.post(_HF_API_URL, headers=headers, json=nli_payload, timeout=15)
if resp.status_code in (503, 502):
# Model loading β€” skip silently, fall through to local
logger.debug("HF API: model loading (%s), falling back to local", resp.status_code)
return None
if not resp.ok:
logger.debug("HF API error %s: %s", resp.status_code, resp.text[:120])
return None
data = resp.json()
return _parse_hf_router_response(data)
except Exception as exc:
logger.debug("HF API NLI call failed: %s", exc)
return None
def _parse_hf_router_response(data) -> Optional[Tuple[str, float]]:
"""
Parse the HuggingFace router zero-shot-classification response.
Expected shape:
{
"sequence": "...",
"labels": ["supports the claim", "contradicts the claim", "unrelated to the claim"],
"scores": [0.85, 0.10, 0.05]
}
"""
if not data or not isinstance(data, dict):
return None
labels = data.get("labels", [])
scores = data.get("scores", [])
if not labels or not scores or len(labels) != len(scores):
logger.debug("HF router NLI: unexpected response shape: %s", str(data)[:120])
return None
# Map back to internal stance labels
label_to_stance = {
"supports the claim": "supporting",
"contradicts the claim": "contradicting",
"unrelated to the claim": "neutral",
}
# Pick the highest-scoring label
best_idx = scores.index(max(scores))
best_label = labels[best_idx].lower()
best_score = float(scores[best_idx])
stance = label_to_stance.get(best_label, "neutral")
logger.debug("HF router NLI: '%s' β†’ %s (%.2f)", best_label, stance, best_score)
return stance, best_score
def _parse_hf_nli_response(data) -> Optional[Tuple[str, float]]:
"""
Legacy parser kept for reference β€” not called by current code.
Two possible shapes from old api-inference.huggingface.co:
A) List of {"label": "ENTAILMENT", "score": 0.97} dicts
B) Nested list from zero-shot endpoint
"""
if not data:
return None
# Shape A: direct label list
if isinstance(data, list) and isinstance(data[0], dict) and "label" in data[0]:
best = max(data, key=lambda x: x.get("score", 0))
label = best["label"].lower()
score = float(best.get("score", 0.5))
stance = _NLI_LABEL_MAP.get(label, "neutral")
logger.debug("HF NLI API: %s β†’ %s (%.2f)", label, stance, score)
return stance, score
# Shape B: nested list [[{label, score}, ...]]
if isinstance(data, list) and isinstance(data[0], list):
flat = data[0]
best = max(flat, key=lambda x: x.get("score", 0))
label = best["label"].lower()
score = float(best.get("score", 0.5))
stance = _NLI_LABEL_MAP.get(label, "neutral")
logger.debug("HF NLI API (nested): %s β†’ %s (%.2f)", label, stance, score)
return stance, score
logger.debug("HF NLI: unexpected response shape: %s", str(data)[:120])
return None
# ── Local transformers pipeline ───────────────────────────────────────────────
_LOCAL_NLI_PIPE = None # cached pipeline instance
_LOCAL_NLI_TRIED = False # avoid retrying failed imports
def _local_nli(claim: str, premise: str) -> Optional[Tuple[str, float]]:
"""
Local BART-large-MNLI via the `transformers` library.
Returns (stance, confidence) or None.
First call downloads ~1.6 GB model; subsequent calls are fast.
"""
global _LOCAL_NLI_PIPE, _LOCAL_NLI_TRIED
if _LOCAL_NLI_TRIED and _LOCAL_NLI_PIPE is None:
return None # already failed once β€” skip
if _LOCAL_NLI_PIPE is None:
_LOCAL_NLI_TRIED = True
try:
from transformers import pipeline as hf_pipeline
logger.info("Loading facebook/bart-large-mnli locally (first call downloads ~1.6 GB)…")
_LOCAL_NLI_PIPE = hf_pipeline(
"zero-shot-classification",
model="facebook/bart-large-mnli",
device=-1, # CPU; set device=0 for GPU
)
logger.info("BART-MNLI local pipeline ready.")
except Exception as exc:
logger.warning(
"transformers import or model load failed: %s. "
"Install with: pip install transformers torch", exc
)
_LOCAL_NLI_PIPE = None
return None
try:
# zero-shot-classification with a three-way NLI hypothesis template
# We use claim as the candidate_label so BART tests:
# premise ENTAILS claim / CONTRADICTS claim / NEUTRAL
out = _LOCAL_NLI_PIPE(
premise,
candidate_labels=["true", "false", "unrelated"],
hypothesis_template=f"This text {{}} the following: {claim[:200]}",
)
# out = {"labels": [...], "scores": [...]}
label_map = {"true": "supporting", "false": "contradicting", "unrelated": "neutral"}
top_label = out["labels"][0].lower()
top_score = float(out["scores"][0])
stance = label_map.get(top_label, "neutral")
logger.debug("Local BART NLI: '%s' β†’ %s (%.2f)", premise[:40], stance, top_score)
return stance, top_score
except Exception as exc:
logger.warning("Local BART NLI inference failed: %s", exc)
return None
# ── In-process NLI result cache ───────────────────────────────────────────────
_NLI_CACHE: Dict[str, Tuple[str, float]] = {}
class NLIStanceLayer:
"""
Layer 3: facebook/bart-large-mnli NLI stance classifier.
Completely free β€” no API key required for public HF models.
Optionally set HF_TOKEN in .env for higher Inference API rate limits.
Usage is identical to the old LLMStanceLayer interface so the
orchestrator (StanceClassifier) needs no changes.
"""
def __init__(self) -> None:
self._enabled = _NLI_ENABLED
if self._enabled:
logger.info(
"NLI stance layer enabled (facebook/bart-large-mnli). "
"Primary: HF Inference API. Fallback: local transformers."
)
else:
logger.info("NLI stance layer disabled via STANCE_NLI_ENABLED=false")
def available(self) -> bool:
return self._enabled
def classify(self, claim: str, title: str, text: str) -> Optional[str]:
"""
Returns 'supporting' | 'contradicting' | 'neutral' | None.
None = both API and local inference failed; caller uses keyword result.
"""
if not self._enabled:
return None
cache_key = hashlib.md5(
f"{claim[:120]}||{title[:80]}||{text[:300]}".encode()
).hexdigest()
if cache_key in _NLI_CACHE:
cached_stance, _ = _NLI_CACHE[cache_key]
return cached_stance
result = _bart_nli(claim=claim, title=title, text=text)
if result is None:
return None
stance, confidence = result
_NLI_CACHE[cache_key] = (stance, confidence)
logger.debug("NLI layer: '%s…' β†’ %s (%.2f)", claim[:40], stance, confidence)
return stance
# ─────────────────────────────────────────────────────────────────────────────
# Orchestrator β€” combines all three layers
# ─────────────────────────────────────────────────────────────────────────────
class StanceClassifier:
"""
Three-layer stance classifier for source ↔ claim pairs.
Priority cascade:
1. Semantic heuristics (pattern matching, no cost)
2. Keyword layer (expanded lists, no cost)
3. Claude LLM (only when uncertain + high-reliability source)
Usage (drop-in replacement for the old StanceClassifier)
---------------------------------------------------------
classifier = StanceClassifier()
stance = classifier.classify(text=source_text, claim=claim_text)
# returns: "supporting" | "contradicting" | "neutral"
Advanced (with confidence + layer attribution)
----------------------------------------------
stance, confidence, layer = classifier.classify_detailed(
title=source_title, text=source_text, claim=claim_text,
source_reliability=0.85,
)
"""
# Minimum reliability a source needs to trigger LLM call when uncertain
_LLM_MIN_RELIABILITY = float(os.getenv("STANCE_LLM_MIN_REL", "0.40"))
def __init__(self) -> None:
self._heuristic = SemanticHeuristicLayer()
self._keywords = KeywordLayer()
self._llm = NLIStanceLayer()
# ------------------------------------------------------------------
# Drop-in API (matches old StanceClassifier.classify signature)
# ------------------------------------------------------------------
def classify(self, text: str, claim: str) -> str:
"""
Backward-compatible interface: returns stance string only.
title defaults to first 80 chars of text.
"""
title = text[:80]
stance, _, _ = self.classify_detailed(
title=title, text=text, claim=claim, source_reliability=0.50
)
return stance
# ------------------------------------------------------------------
# Full API
# ------------------------------------------------------------------
def classify_detailed(
self,
title: str,
text: str,
claim: str,
source_reliability: float = 0.50,
) -> Tuple[str, float, str]:
"""
Returns (stance, confidence, winning_layer).
stance = "supporting" | "contradicting" | "neutral"
confidence = 0.0–1.0
winning_layer = "heuristic" | "keyword" | "llm" | "default"
"""
combined_text = (title + " " + text).strip()
# ── Layer 1: Semantic heuristics ─────────────────────────────────
h_stance = self._heuristic.classify(claim, combined_text)
if h_stance is not None:
return h_stance, 0.75, "heuristic"
# ── Layer 2: Keywords ─────────────────────────────────────────────
kw_stance, kw_conf = self._keywords.classify(combined_text)
# If keyword layer is confident, use it
if kw_conf >= 0.55:
return kw_stance, kw_conf, "keyword"
# ── Layer 3: NLI (BART-MNLI) β€” only for uncertain + credible sources ─
if (
self._llm.available()
and source_reliability >= self._LLM_MIN_RELIABILITY
):
llm_stance = self._llm.classify(claim, title, text)
if llm_stance is not None:
return llm_stance, 0.90, "nli"
# ── Fallback: use whatever keyword layer found ────────────────────
return kw_stance, kw_conf, "keyword"
# ─────────────────────────────────────────────────────────────────────────────
# Backwards-compat shim: old code did StanceClassifier(scorer)
# New code supports both StanceClassifier() and StanceClassifier(scorer)
# ─────────────────────────────────────────────────────────────────────────────
def build_stance_classifier(scorer=None) -> StanceClassifier:
"""Factory that accepts an optional DomainReliabilityScorer (ignored β€” kept for compat)."""
return StanceClassifier()