Spaces:
Sleeping
Sleeping
| """ | |
| decision_engine.py | |
| ================== | |
| High-level moderation decision engine. | |
| Takes a summarised claim_result dict (produced after stance classification and | |
| source retrieval) and returns a final moderation decision with recalibrated | |
| confidence and a human-readable reason. | |
| This layer sits *above* reasoning.py: | |
| DataCollector → StanceClassifier → reasoning.py → decision_engine.py → UI | |
| Why a separate layer? | |
| reasoning.py works on raw ScoredSource objects (per-source contributions). | |
| This module works on already-aggregated claim summaries, which is the shape | |
| that the UI, API, and tests produce. The two layers complement each other — | |
| reasoning.py computes the evidence score; this module applies the final | |
| decision policy (thresholds, pattern boosts, confidence blending). | |
| Usage | |
| ----- | |
| from decision_engine import make_moderation_decision | |
| result = make_moderation_decision({ | |
| "claim": "Humans only use 10% of their brain", | |
| "status": "FALSE", | |
| "confidence": 0.35, | |
| "supporting_sources": 0, | |
| "contradicting_sources": 3, | |
| "neutral_sources": 1, | |
| "avg_source_reliability": 0.72, | |
| }) | |
| # → {"decision": "AUTO_REMOVE", "final_confidence": 0.88, "reason": "..."} | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass, field | |
| from typing import Optional | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Configurable thresholds | |
| # All values can be overridden per-call via the `config` parameter. | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class DecisionConfig: | |
| # ── AUTO_REMOVE thresholds ──────────────────────────────────────────────── | |
| # Minimum contradicting sources required for AUTO_REMOVE when status=FALSE | |
| auto_remove_min_contra: int = 2 | |
| # OR: single-source override when that one source is very reliable | |
| auto_remove_single_src_rel: float = 0.85 | |
| # Settled-science / known-misinformation pattern always triggers AUTO_REMOVE | |
| # with just 1 contradicting source | |
| auto_remove_pattern_min_contra: int = 1 | |
| # ── ALLOW thresholds ────────────────────────────────────────────────────── | |
| allow_min_support: int = 2 # min supporting sources for ALLOW | |
| allow_single_src_rel: float = 0.90 # single very-reliable source is enough | |
| # ── Confidence blend weights (must sum to 1.0) ──────────────────────────── | |
| weight_stance: float = 0.50 # from status + incoming confidence | |
| weight_evidence: float = 0.30 # from source counts | |
| weight_reliability: float = 0.20 # from avg_source_reliability | |
| # ── Evidence strength formula ───────────────────────────────────────────── | |
| # evidence_strength = contra / (contra + support + 1) for FALSE claims | |
| # evidence_strength = support / (support + contra + 1) for TRUE claims | |
| # The +1 in denominator prevents division-by-zero and dampens single-source certainty. | |
| # ── Pattern boost ───────────────────────────────────────────────────────── | |
| pattern_boost: float = 0.15 # added to final_confidence when a known pattern matches | |
| # ── Confidence floor/ceiling ────────────────────────────────────────────── | |
| min_confidence: float = 0.10 | |
| max_confidence: float = 0.99 | |
| # Default singleton — callers can pass a custom config to override | |
| _DEFAULT_CONFIG = DecisionConfig() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Known misinformation patterns | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Each entry: (label, regex) | |
| # When a claim text matches, the pattern_boost is applied AND the | |
| # auto_remove threshold is lowered to auto_remove_pattern_min_contra. | |
| _KNOWN_MISINFO: list[tuple[str, re.Pattern]] = [ | |
| # Neuroscience myth | |
| ("10pct_brain", re.compile( | |
| r"\b(10\s*%|ten\s+percent).{0,30}brain\b|\bbrain.{0,30}(10\s*%|ten\s+percent)\b", | |
| re.IGNORECASE, | |
| )), | |
| # Flat earth | |
| ("flat_earth", re.compile( | |
| r"\b(earth\s+is\s+flat|flat\s+earth|earth\s+isn.t?\s+(round|spherical|globe))\b", | |
| re.IGNORECASE, | |
| )), | |
| # Bleach / MMS as cure | |
| ("bleach_cure", re.compile( | |
| r"\b(drink(ing)?\s+bleach|bleach\s+cure[sd]?|mms\s+cure[sd]?|" | |
| r"chlorine\s+dioxide\s+cure|inject(ing)?\s+disinfectant)\b", | |
| re.IGNORECASE, | |
| )), | |
| # Vaccine-autism link | |
| ("vaccines_autism", re.compile( | |
| r"\bvaccines?\s+(cause[sd]?|linked?\s+to|lead[s]?\s+to)\s+autism\b", | |
| re.IGNORECASE, | |
| )), | |
| # 5G / COVID | |
| ("5g_covid", re.compile( | |
| r"\b5g\s+(causes?|spreads?|created?|transmits?)\s+(covid|coronavirus|virus)\b", | |
| re.IGNORECASE, | |
| )), | |
| # Climate hoax | |
| ("climate_hoax", re.compile( | |
| r"\bclimate\s+change\s+is\s+(a\s+)?(hoax|fake|not\s+real|fabricated)\b", | |
| re.IGNORECASE, | |
| )), | |
| # Moon landing | |
| ("moon_landing", re.compile( | |
| r"\b(moon\s+landing\s+(was\s+)?(fake|faked|hoax|staged)|never\s+landed\s+on.{0,10}moon)\b", | |
| re.IGNORECASE, | |
| )), | |
| # HIV denial | |
| ("hiv_denial", re.compile( | |
| r"\b(hiv\s+(doesn.t?\s+cause|doesn.t?\s+exist|is\s+fake)|aids\s+is\s+(a\s+)?(hoax|fake))\b", | |
| re.IGNORECASE, | |
| )), | |
| # Evolution denial | |
| ("evolution_denial", re.compile( | |
| r"\b(evolution\s+is\s+(false|fake|a\s+lie|wrong|disproven)|humans?\s+didn.t?\s+evolve)\b", | |
| re.IGNORECASE, | |
| )), | |
| ] | |
| def _match_known_misinfo(claim: str) -> Optional[str]: | |
| """ | |
| Returns the pattern label if the claim matches a known misinformation | |
| pattern, otherwise None. | |
| """ | |
| for label, pattern in _KNOWN_MISINFO: | |
| if pattern.search(claim): | |
| return label | |
| return None | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Confidence calculation | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def _compute_confidence( | |
| status: str, | |
| incoming_confidence: float, | |
| supporting: int, | |
| contradicting: int, | |
| avg_reliability: float, | |
| pattern_label: Optional[str], | |
| cfg: DecisionConfig, | |
| ) -> float: | |
| """ | |
| Blend three independent signals into a single calibrated confidence score. | |
| Component 1 — Stance confidence (weight=0.50) | |
| Directly from the upstream classifier. Clamped to [0, 1]. | |
| Component 2 — Evidence strength (weight=0.30) | |
| For FALSE claims: contra / (contra + support + 1) | |
| For TRUE claims: support / (support + contra + 1) | |
| For UNKNOWN: 0.0 (no directional evidence) | |
| The +1 dampens over-confidence from a single source. | |
| Component 3 — Source reliability (weight=0.20) | |
| avg_source_reliability passed through directly. | |
| Pattern boost (additive, after blend) | |
| +cfg.pattern_boost when a known misinformation pattern is matched. | |
| """ | |
| # Component 1 | |
| stance_conf = max(0.0, min(1.0, incoming_confidence)) | |
| # Component 2 | |
| total_directional = supporting + contradicting | |
| if status == "FALSE" and contradicting > 0: | |
| evidence_strength = contradicting / (total_directional + 1) | |
| elif status == "TRUE" and supporting > 0: | |
| evidence_strength = supporting / (total_directional + 1) | |
| else: | |
| evidence_strength = 0.0 | |
| # Component 3 | |
| reliability = max(0.0, min(1.0, avg_reliability)) | |
| # Weighted blend | |
| blended = ( | |
| cfg.weight_stance * stance_conf + | |
| cfg.weight_evidence * evidence_strength + | |
| cfg.weight_reliability * reliability | |
| ) | |
| # Pattern boost | |
| if pattern_label is not None: | |
| blended += cfg.pattern_boost | |
| return round(max(cfg.min_confidence, min(cfg.max_confidence, blended)), 4) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Core decision logic | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def _decide( | |
| status: str, | |
| contradicting: int, | |
| supporting: int, | |
| avg_reliability: float, | |
| pattern_label: Optional[str], | |
| cfg: DecisionConfig, | |
| ) -> tuple[str, str]: | |
| """ | |
| Returns (decision, reason_fragment). | |
| Decision priority: | |
| 1. AUTO_REMOVE — status is FALSE + sufficient contradicting evidence | |
| 2. ALLOW — status is TRUE + sufficient supporting evidence | |
| 3. FLAG_FOR_REVIEW — everything else | |
| """ | |
| # ── Path 1: AUTO_REMOVE ─────────────────────────────────────────────────── | |
| if status == "FALSE": | |
| # Known misinformation pattern: lower bar to 1 source | |
| if pattern_label is not None and contradicting >= cfg.auto_remove_pattern_min_contra: | |
| return ( | |
| "AUTO_REMOVE", | |
| f"Matches known misinformation pattern '{pattern_label}' " | |
| f"with {contradicting} contradicting source(s).", | |
| ) | |
| # Standard: ≥2 contradicting sources | |
| if contradicting >= cfg.auto_remove_min_contra: | |
| return ( | |
| "AUTO_REMOVE", | |
| f"{contradicting} contradicting source(s) confirm this claim is false.", | |
| ) | |
| # Single very-reliable contradicting source | |
| if contradicting >= 1 and avg_reliability >= cfg.auto_remove_single_src_rel: | |
| return ( | |
| "AUTO_REMOVE", | |
| f"Single highly reliable source (reliability={avg_reliability:.2f}) " | |
| f"contradicts this claim.", | |
| ) | |
| # ── Path 2: ALLOW ───────────────────────────────────────────────────────── | |
| if status == "TRUE": | |
| if supporting >= cfg.allow_min_support: | |
| return ( | |
| "ALLOW", | |
| f"{supporting} supporting source(s) confirm this claim is true.", | |
| ) | |
| if supporting >= 1 and avg_reliability >= cfg.allow_single_src_rel: | |
| return ( | |
| "ALLOW", | |
| f"Single authoritative source (reliability={avg_reliability:.2f}) " | |
| f"confirms this claim.", | |
| ) | |
| # ── Path 3: FLAG_FOR_REVIEW ─────────────────────────────────────────────── | |
| if status == "UNKNOWN": | |
| reason = "Claim status is UNKNOWN — insufficient evidence to decide." | |
| elif status == "FALSE" and contradicting == 0: | |
| reason = "Claim is marked FALSE but no contradicting sources were retrieved." | |
| elif status == "FALSE" and contradicting < cfg.auto_remove_min_contra: | |
| reason = ( | |
| f"Claim is FALSE but only {contradicting} contradicting source(s) found " | |
| f"(need ≥{cfg.auto_remove_min_contra} for auto-removal)." | |
| ) | |
| elif status == "TRUE" and supporting < cfg.allow_min_support: | |
| reason = ( | |
| f"Claim is TRUE but only {supporting} supporting source(s) found " | |
| f"(need ≥{cfg.allow_min_support} for auto-allow)." | |
| ) | |
| else: | |
| reason = "Evidence is insufficient or ambiguous." | |
| return "FLAG_FOR_REVIEW", reason | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Public API | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def make_moderation_decision( | |
| claim_result: dict, | |
| config: Optional[DecisionConfig] = None, | |
| ) -> dict: | |
| """ | |
| Make a final moderation decision for a single claim. | |
| Parameters | |
| ---------- | |
| claim_result : dict | |
| { | |
| "claim": str, | |
| "status": "TRUE" | "FALSE" | "UNKNOWN", | |
| "confidence": float, # from upstream classifier (0–1) | |
| "supporting_sources": int, | |
| "contradicting_sources": int, | |
| "neutral_sources": int, | |
| "avg_source_reliability": float, # 0–1 | |
| } | |
| config : DecisionConfig | None | |
| Override default thresholds. Pass None to use defaults. | |
| Returns | |
| ------- | |
| dict | |
| { | |
| "decision": "AUTO_REMOVE" | "ALLOW" | "FLAG_FOR_REVIEW", | |
| "final_confidence": float, # 0.10–0.99 | |
| "reason": str, | |
| "pattern_matched": str | None, # known misinformation label if triggered | |
| } | |
| """ | |
| cfg = config or _DEFAULT_CONFIG | |
| # ── Extract fields with safe defaults ───────────────────────────────────── | |
| claim = str(claim_result.get("claim", "")) | |
| status = str(claim_result.get("status", "UNKNOWN")).upper() | |
| confidence = float(claim_result.get("confidence", 0.0)) | |
| supporting = int(claim_result.get("supporting_sources", 0)) | |
| contradicting = int(claim_result.get("contradicting_sources", 0)) | |
| avg_rel = float(claim_result.get("avg_source_reliability", 0.5)) | |
| # Normalise status to our three canonical values | |
| if status not in ("TRUE", "FALSE", "UNKNOWN"): | |
| status = "UNKNOWN" | |
| # ── Known misinformation pattern check ──────────────────────────────────── | |
| pattern_label = _match_known_misinfo(claim) | |
| # ── Decision ────────────────────────────────────────────────────────────── | |
| decision, reason_fragment = _decide( | |
| status, contradicting, supporting, avg_rel, pattern_label, cfg | |
| ) | |
| # ── Confidence recalibration ────────────────────────────────────────────── | |
| final_confidence = _compute_confidence( | |
| status, confidence, supporting, contradicting, | |
| avg_rel, pattern_label, cfg, | |
| ) | |
| # ── Build human-readable reason ─────────────────────────────────────────── | |
| reason = _build_reason( | |
| claim, decision, status, final_confidence, | |
| supporting, contradicting, avg_rel, | |
| pattern_label, reason_fragment, | |
| ) | |
| return { | |
| "decision": decision, | |
| "final_confidence": final_confidence, | |
| "reason": reason, | |
| "pattern_matched": pattern_label, | |
| } | |
| def _build_reason( | |
| claim: str, | |
| decision: str, | |
| status: str, | |
| confidence: float, | |
| supporting: int, | |
| contradicting: int, | |
| avg_rel: float, | |
| pattern_label: Optional[str], | |
| core_reason: str, | |
| ) -> str: | |
| """Compose a full human-readable explanation.""" | |
| lines = [] | |
| # Decision headline | |
| headline = { | |
| "AUTO_REMOVE": "Content removed: claim is false and clearly contradicted.", | |
| "ALLOW": "Content allowed: claim is supported by evidence.", | |
| "FLAG_FOR_REVIEW": "Content flagged: insufficient or ambiguous evidence.", | |
| }.get(decision, "Decision made.") | |
| lines.append(headline) | |
| # Core reason from decision logic | |
| lines.append(core_reason) | |
| # Evidence summary | |
| lines.append( | |
| f"Evidence: {supporting} supporting, {contradicting} contradicting " | |
| f"(avg source reliability: {avg_rel:.0%})." | |
| ) | |
| # Confidence | |
| lines.append(f"Final confidence: {confidence:.0%}.") | |
| return " ".join(lines) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Batch helper | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def make_moderation_decisions( | |
| claim_results: list[dict], | |
| config: Optional[DecisionConfig] = None, | |
| ) -> dict: | |
| """ | |
| Run make_moderation_decision over multiple claims and derive a single | |
| aggregate content-level decision. | |
| Aggregate logic (matches real-world platforms): | |
| ANY AUTO_REMOVE → content = AUTO_REMOVE | |
| ANY FLAG_FOR_REVIEW (and no remove) → content = FLAG_FOR_REVIEW | |
| ALL ALLOW → content = ALLOW | |
| Returns | |
| ------- | |
| { | |
| "content_decision": "AUTO_REMOVE" | "ALLOW" | "FLAG_FOR_REVIEW", | |
| "per_claim": [ {claim_result + decision fields}, ... ], | |
| } | |
| """ | |
| per_claim = [] | |
| for cr in claim_results: | |
| result = make_moderation_decision(cr, config) | |
| per_claim.append({**cr, **result}) | |
| decisions = [r["decision"] for r in per_claim] | |
| if "AUTO_REMOVE" in decisions: | |
| content_decision = "AUTO_REMOVE" | |
| elif "FLAG_FOR_REVIEW" in decisions: | |
| content_decision = "FLAG_FOR_REVIEW" | |
| else: | |
| content_decision = "ALLOW" | |
| return { | |
| "content_decision": content_decision, | |
| "per_claim": per_claim, | |
| } | |