Hugging Face's logo

Spaces:
infoclassifier
/
InfoClassifier
Sleeping

App Files Community
InfoClassifier / reasoning.py
UTKARSH MISHRA
Initial commit: InfoClassifier project
e5e36db
Raw History Blame Contribute Delete
36.4 kB
"""
reasoning.py
============
Deterministic reasoning engine for InformationIntegrityEnv.
Architecture
------------
EvidenceAggregator β€” turns List[ScoredSource] into a numeric evidence score
RiskClassifier β€” detects high-risk claim domains (medical, violence, etc.)
SafetyRules β€” override decisions when evidence is insufficient
DeterministicReasoningEngine β€” combines all into a StructuredReasoning result
Evidence score formula
----------------------
For each source:
contribution = stance_weight(stance) Γ— reliability Γ— similarity_boost(similarity)
Stance weights:
supporting β†’ +1.0 (positive contribution)
contradicting β†’ -1.0 (negative contribution)
neutral β†’ 0.0 (ignored)
similarity_boost(sim):
sim β‰₯ 0.60 β†’ 1.20 (high confidence amplifier)
sim β‰₯ 0.40 β†’ 1.00 (normal)
sim < 0.40 β†’ 0.60 (low confidence dampener β€” source is weakly relevant)
evidence_score = Ξ£ contributions (not normalised; bounded by SafetyRules logic)
High-risk multiplier
--------------------
When a claim is classified as high-risk (medical misinformation, harmful
actions, public health emergencies), the evidence score is multiplied by
a configurable factor (default 1.5) to lower the bar for REMOVE.
Decision thresholds (default)
------------------------------
score > 0.30 β†’ ALLOW
score < -0.50 β†’ REMOVE (lowered from -0.70; high-risk claims use -0.35)
otherwise β†’ FLAG
Safety rules (all override the score-based decision)
------------------------------------------------------
1. No sources at all β†’ FLAG
2. All sources are neutral β†’ FLAG
3. REMOVE needs β‰₯ min_remove_sources contradicting sources
4. REMOVE needs contradicting reliability β‰₯ min_remove_reliability
5. Score near boundary (|score| < boundary_margin) β†’ FLAG
6. ALLOW requires β‰₯1 high-reliability supporting source
7. HIGH-RISK + strong_contra + β‰₯1 authoritative source β†’ force REMOVE
Multiple-source confidence boost
---------------------------------
When β‰₯3 contradicting sources agree, confidence is boosted to "high"
regardless of individual reliability scores.
All thresholds are configurable via reasoning_config.yaml or env-vars.
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import yaml
from retrieval import ScoredSource
logger = logging.getLogger("reasoning")
REASONING_CONFIG_PATH = os.getenv("REASONING_CONFIG", "reasoning_config.yaml")
# ─────────────────────────────────────────────────────────────────────────────
# Config
# ─────────────────────────────────────────────────────────────────────────────
_DEFAULTS: Dict[str, float | int] = {
# Decision thresholds
"allow_threshold": 0.30, # score > this β†’ ALLOW
"remove_threshold": -0.50, # score < this β†’ REMOVE
"boundary_margin": 0.15, # |score| < this β†’ force FLAG
# High-risk domain settings
"high_risk_multiplier": 1.50, # amplify score magnitude for risky claims
"high_risk_remove_threshold": -0.35, # lower REMOVE bar for high-risk claims
# Similarity boost knees
"high_sim_cutoff": 0.60,
"low_sim_cutoff": 0.40,
"high_sim_boost": 1.20,
"low_sim_boost": 0.60,
# Safety rule thresholds
"min_remove_sources": 2, # int: min contradicting sources for REMOVE
"min_remove_reliability": 0.35, # avg reliability of contra sources
# Consensus-based confidence scaling
# Confidence level = min(1.0, base + n_contra * per_source_boost)
"consensus_base_confidence": 0.40, # starting confidence with 1 contra source
"consensus_per_source_boost": 0.12, # +0.12 per additional contradicting source
"consensus_high_cutoff": 0.75, # β‰₯ this scaled value β†’ "high"
"consensus_medium_cutoff": 0.50, # β‰₯ this scaled value β†’ "medium"
# No-support confidence boost
"no_support_bonus": 0.15, # added when zero supporting sources found
# Settled-science auto-remove
"settled_science_remove": 1, # 1=enabled, 0=disabled
# Confidence descriptors (fallback when consensus logic doesn't apply)
"high_confidence_threshold": 0.50,
"low_confidence_threshold": 0.20,
}
class ReasoningConfig:
def __init__(self, path: str = REASONING_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"ReasoningConfig has no attribute '{name}'")
_CFG = ReasoningConfig()
# ─────────────────────────────────────────────────────────────────────────────
# Risk Classifier β€” identifies high-stakes claim domains
# ─────────────────────────────────────────────────────────────────────────────
import re as _re
# Each entry: (domain_name, compiled_regex_for_claim_text)
# A claim matching ANY pattern is tagged as high-risk.
_RISK_PATTERNS: List[Tuple[str, "_re.Pattern"]] = [
# Medical / health misinformation (highest harm)
("medical", _re.compile(
r"\b(vaccine|vaccination|vaxx?|covid|coronavirus|sars|flu|cancer|"
r"cure|cures|treat|treatment|drug|medication|dose|overdose|"
r"autism|5g|radiation|chemotherapy|hiv|aids|ebola|"
r"causes?\s+death|kills?|deadly|toxic|poison)\b",
_re.IGNORECASE,
)),
# Harmful actions / violence incitement
("harmful_action", _re.compile(
r"\b(drink\s+bleach|inject\s+disinfectant|eat\s+tide|tide\s+pod|"
r"ivermectin\s+for\s+human|horse\s+dewormer|"
r"suicide\s+method|how\s+to\s+harm|self.harm|bomb\s+making|"
r"attack|assassin|terrorist|shoot|stab|murder)\b",
_re.IGNORECASE,
)),
# Public health emergencies
("public_health", _re.compile(
r"\b(pandemic|epidemic|outbreak|spread|contagious|"
r"herd\s+immunity|lockdown|quarantine|mask|ventilator|"
r"hospital\s+overwhelm|icu|death\s+toll)\b",
_re.IGNORECASE,
)),
# Election / democratic integrity
("election", _re.compile(
r"\b(election|vote|ballot|voter\s+fraud|stolen\s+election|"
r"rigged|electoral|polling|count|recount|certification)\b",
_re.IGNORECASE,
)),
# Climate / environment denial used to block policy
("climate", _re.compile(
r"\b(climate\s+change|global\s+warming|carbon|emissions?|"
r"ipcc|greenhouse|sea\s+level|arctic|glacier)\b",
_re.IGNORECASE,
)),
]
# Authoritative domains that count as "official contradiction"
_AUTHORITATIVE_DOMAINS = {
"who.int", "cdc.gov", "nih.gov", "fda.gov", "nhs.uk",
"ecdc.europa.eu", "nasa.gov", "noaa.gov", "reuters.com",
"apnews.com", "bbc.com", "bbc.co.uk", "snopes.com",
"factcheck.org", "politifact.com", "fullfact.org",
}
class RiskClassifier:
"""
Classifies a claim into a risk domain.
High-risk claims get a score multiplier that lowers the REMOVE threshold,
mirroring how platforms like Meta and YouTube apply stricter rules to
medical misinformation and election integrity content.
"""
def classify(self, claim_text: str) -> Optional[str]:
"""
Returns the risk domain name (e.g. 'medical') or None if not high-risk.
"""
for domain, pattern in _RISK_PATTERNS:
if pattern.search(claim_text):
return domain
return None
def is_authoritative_contra(self, sources: "List[ScoredSource]") -> bool:
"""
Returns True if at least one contradicting source comes from an
authoritative domain (WHO, CDC, Reuters, etc.).
"""
for src in sources:
if src.stance == "contradicting":
domain = self._extract_domain(src.url or "")
if any(auth in domain for auth in _AUTHORITATIVE_DOMAINS):
return True
return False
@staticmethod
def _extract_domain(url: str) -> str:
url = url.lower().split("//")[-1]
return url.split("/")[0]
# ─────────────────────────────────────────────────────────────────────────────
# Settled Science Detector
# ─────────────────────────────────────────────────────────────────────────────
# Claims where scientific consensus is overwhelming (>97% of experts agree).
# When a claim matches AND contradicting sources exist, the system treats it as
# settled β€” confidence is forced to "high" and decision β†’ REMOVE regardless of
# source count or reliability scores.
#
# Each entry: (topic_label, regex_that_matches_the_false_claim)
_SETTLED_SCIENCE_CLAIMS: List[Tuple[str, "_re.Pattern"]] = [
("flat_earth", _re.compile(
r"\b(earth\s+is\s+flat|flat\s+earth|earth\s+isn.t?\s+(round|spherical|a\s+globe))\b",
_re.IGNORECASE,
)),
("vaccines_autism", _re.compile(
r"\bvaccines?\s+(cause[sd]?|linked?\s+to|lead[s]?\s+to)\s+autism\b",
_re.IGNORECASE,
)),
("climate_hoax", _re.compile(
r"\bclimate\s+change\s+is\s+(a\s+)?(hoax|fake|not\s+real|fabricated|made\s+up)\b",
_re.IGNORECASE,
)),
("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,
)),
("moon_landing_hoax", _re.compile(
r"\b(moon\s+landing\s+(was\s+)?(fake|faked|hoax|staged)|never\s+landed\s+on\s+(the\s+)?moon)\b",
_re.IGNORECASE,
)),
("5g_covid", _re.compile(
r"\b5g\s+(causes?|spreads?|transmits?|created?)\s+(covid|coronavirus|virus|cancer)\b",
_re.IGNORECASE,
)),
("bleach_cure", _re.compile(
r"\b(drink(ing)?\s+bleach|bleach\s+cures?|mms\s+cures?|chlorine\s+dioxide\s+cure)\b",
_re.IGNORECASE,
)),
("hiv_not_real", _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,
)),
]
class SettledScienceDetector:
"""
Detects claims that contradict overwhelming scientific consensus.
When matched + contradicting evidence exists, confidence is forced to
"high" and the decision is escalated directly to REMOVE.
"""
def is_settled(self, claim_text: str) -> Optional[str]:
"""
Returns the topic label (e.g. 'flat_earth') if the claim contradicts
settled science, otherwise None.
"""
for label, pattern in _SETTLED_SCIENCE_CLAIMS:
if pattern.search(claim_text):
return label
return None
# ─────────────────────────────────────────────────────────────────────────────
# Data classes
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class SourceEvidence:
"""Per-source evidence contribution (for explainability)."""
source_id: str
title: str
stance: str
reliability: float
similarity: float
contribution: float # signed score contribution
@dataclass
class StructuredReasoning:
"""
Complete reasoning artifact produced by DeterministicReasoningEngine.
Consumed by ExplainabilityEngine and the environment's decide handler.
"""
# Core decision output
decision: str # "allow" | "flag" | "remove"
evidence_score: float # raw aggregated score
confidence: str # "high" | "medium" | "low"
# Per-source breakdown
source_evidence: List[SourceEvidence] = field(default_factory=list)
# Counts
n_supporting: int = 0
n_contradicting: int = 0
n_neutral: int = 0
n_total: int = 0
# Safety rule audit trail
safety_rules_fired: List[str] = field(default_factory=list)
# Risk / settled-science classification
risk_domain: Optional[str] = None # e.g. "medical", "election"
settled_science_topic: Optional[str] = None # e.g. "flat_earth", None
# Human-readable summary
rationale: str = ""
def to_dict(self) -> dict:
return {
"decision": self.decision,
"evidence_score": round(self.evidence_score, 4),
"confidence": self.confidence,
"risk_domain": self.risk_domain,
"settled_science_topic": self.settled_science_topic,
"n_supporting": self.n_supporting,
"n_contradicting": self.n_contradicting,
"n_neutral": self.n_neutral,
"n_total": self.n_total,
"safety_rules_fired": self.safety_rules_fired,
"rationale": self.rationale,
"source_breakdown": [
{
"id": e.source_id,
"title": e.title[:80],
"stance": e.stance,
"reliability": round(e.reliability, 3),
"similarity": round(e.similarity, 3),
"contribution": round(e.contribution, 4),
}
for e in self.source_evidence
],
}
# ─────────────────────────────────────────────────────────────────────────────
# Part 1 β€” Evidence Aggregator
# ─────────────────────────────────────────────────────────────────────────────
class EvidenceAggregator:
"""
Converts a list of scored+filtered sources into a signed numeric score
and a per-source breakdown for transparency.
"""
# Stance β†’ sign of contribution
_STANCE_WEIGHT = {
"supporting": +1.0,
"contradicting": -1.0,
"neutral": 0.0,
}
# Reliability threshold for "high-quality" source classification
_HI_REL = 0.65
def __init__(self, config: Optional[ReasoningConfig] = None) -> None:
self._cfg = config or _CFG
def aggregate(self, sources: List[ScoredSource]) -> Tuple[float, List[SourceEvidence]]:
"""
Returns (evidence_score, per_source_evidence_list).
evidence_score is the sum of signed contributions. It is NOT
normalised β€” callers compare it against the configured thresholds.
"""
evidence: List[SourceEvidence] = []
total_score = 0.0
for src in sources:
stance_w = self._STANCE_WEIGHT.get(src.stance, 0.0)
sim_boost = self._sim_boost(src.similarity)
contribution = stance_w * src.reliability * sim_boost
evidence.append(SourceEvidence(
source_id = src.id,
title = src.title,
stance = src.stance,
reliability = src.reliability,
similarity = src.similarity,
contribution = round(contribution, 4),
))
total_score += contribution
return round(total_score, 4), evidence
def classify_evidence_state(self, sources: List[ScoredSource]) -> str:
"""
Categorises the overall state of evidence. Used by the reasoning
engine for calibrated confidence and safety-rule decisions.
Returns one of:
"no_evidence" β€” no relevant sources at all
"strong_contra" β€” β‰₯2 high-reliability contradicting sources
"weak_contra" β€” contradicting sources but low reliability
"strong_support" β€” β‰₯2 high-reliability supporting sources
"weak_support" β€” some supporting but low reliability
"conflicting" β€” both high-reliability supporting AND contradicting
"neutral_only" β€” sources exist but all neutral
"""
if not sources:
return "no_evidence"
contra = [s for s in sources if s.stance == "contradicting"]
supp = [s for s in sources if s.stance == "supporting"]
hi_contra = [s for s in contra if s.reliability >= self._HI_REL]
hi_supp = [s for s in supp if s.reliability >= self._HI_REL]
if hi_contra and hi_supp:
return "conflicting"
if len(hi_contra) >= 2:
return "strong_contra"
if contra:
return "weak_contra"
if len(hi_supp) >= 2:
return "strong_support"
if supp:
return "weak_support"
return "neutral_only"
def _sim_boost(self, sim: float) -> float:
if sim >= self._cfg.high_sim_cutoff:
return self._cfg.high_sim_boost
if sim >= self._cfg.low_sim_cutoff:
return 1.0
return self._cfg.low_sim_boost
# ─────────────────────────────────────────────────────────────────────────────
# Part 2 β€” Safety Rules
# ─────────────────────────────────────────────────────────────────────────────
class SafetyRules:
"""
Post-processes the score-based decision with hard safety constraints.
Returns (final_decision, list_of_rules_fired).
"""
def __init__(self, config: Optional[ReasoningConfig] = None) -> None:
self._cfg = config or _CFG
def apply(
self,
score_decision: str,
evidence_score: float,
sources: List[ScoredSource],
evidence: List[SourceEvidence],
evidence_state: str = "unknown",
risk_domain: Optional[str] = None,
has_authoritative_contra: bool = False,
) -> Tuple[str, List[str]]:
"""
Returns (final_decision, fired_rules).
Rules are evaluated in priority order; first match wins for decision
overrides, but ALL fired rules are reported for transparency.
"""
fired: List[str] = []
final = score_decision
# Rule 1: No sources at all β†’ must FLAG
if not sources:
fired.append("NO_SOURCES: no evidence retrieved β€” defaulting to FLAG")
return "flag", fired
# Rule 2: All sources neutral β†’ FLAG (can't confirm or deny)
if all(s.stance == "neutral" for s in sources):
fired.append("ALL_NEUTRAL: all sources are neutral β€” defaulting to FLAG")
final = "flag"
# Rule 3: REMOVE requires sufficient contradicting evidence
if final == "remove":
contra = [s for s in sources if s.stance == "contradicting"]
if len(contra) < int(self._cfg.min_remove_sources):
fired.append(
f"INSUFFICIENT_CONTRA: only {len(contra)} contradicting source(s), "
f"need β‰₯{int(self._cfg.min_remove_sources)} for REMOVE β†’ downgrading to FLAG"
)
final = "flag"
# Rule 4: REMOVE requires contradicting sources to be reliable
if final == "remove":
contra = [s for s in sources if s.stance == "contradicting"]
if contra:
avg_rel = sum(s.reliability for s in contra) / len(contra)
if avg_rel < self._cfg.min_remove_reliability:
fired.append(
f"LOW_CONTRA_RELIABILITY: avg reliability of contradicting "
f"sources={avg_rel:.2f} < {self._cfg.min_remove_reliability:.2f} β†’ "
f"downgrading to FLAG"
)
final = "flag"
# Rule 5: Score near boundary β†’ resolve uncertainty with FLAG
boundary = self._cfg.boundary_margin
if abs(evidence_score) < boundary and final != "flag":
fired.append(
f"BOUNDARY_UNCERTAINTY: |score|={abs(evidence_score):.3f} < "
f"margin={boundary:.2f} β€” insufficient evidence β†’ FLAG"
)
final = "flag"
# Rule 6: ALLOW requires at least one credible supporting source.
if final == "allow":
hi_rel_support = [
s for s in sources
if s.stance == "supporting" and s.reliability >= 0.60
]
if not hi_rel_support:
fired.append(
"NO_CREDIBLE_SUPPORT: decision was ALLOW but no high-reliability "
"supporting sources found β€” downgrading to FLAG"
)
final = "flag"
# Rule 7: HIGH-RISK + strong_contra + authoritative source β†’ force REMOVE
# Mirrors Meta/YouTube policy: medical/election misinformation with official
# fact-check contradictions is removed automatically, not just flagged.
if (
risk_domain is not None
and evidence_state in ("strong_contra", "weak_contra")
and has_authoritative_contra
and final in ("flag", "remove")
):
fired.append(
f"HIGH_RISK_AUTO_REMOVE: claim is in high-risk domain '{risk_domain}' "
f"and contradicted by an authoritative source β€” forcing REMOVE"
)
final = "remove"
return final, fired
# ─────────────────────────────────────────────────────────────────────────────
# Part 3 β€” Deterministic Reasoning Engine
# ─────────────────────────────────────────────────────────────────────────────
class DeterministicReasoningEngine:
"""
Main entry point. Consumes a list of already-ranked ScoredSources
(output of RetrievalPipeline) and produces a StructuredReasoning.
Usage
-----
engine = DeterministicReasoningEngine()
result = engine.reason(claim_text="5G causes COVID", sources=scored_sources)
# result.decision β†’ "flag" | "allow" | "remove"
# result.to_dict() β†’ full JSON-serialisable breakdown
"""
def __init__(self, config: Optional[ReasoningConfig] = None) -> None:
self._cfg = config or _CFG
self._aggregator = EvidenceAggregator(config)
self._safety = SafetyRules(config)
self._risk = RiskClassifier()
self._settled = SettledScienceDetector()
def reason(
self,
claim_text: str,
sources: List[ScoredSource],
) -> StructuredReasoning:
"""
Full deterministic reasoning pass for one claim against its sources.
"""
# Step 1: classify evidence state
evidence_state = self._aggregator.classify_evidence_state(sources)
# Step 2: classify claim risk + settled-science
risk_domain = self._risk.classify(claim_text)
has_auth_contra = self._risk.is_authoritative_contra(sources)
settled_topic = self._settled.is_settled(claim_text)
# Step 3: aggregate evidence, apply high-risk multiplier if needed
score, evidence = self._aggregator.aggregate(sources)
if risk_domain is not None and score < 0:
score = round(score * self._cfg.high_risk_multiplier, 4)
logger.debug("High-risk domain '%s': score amplified to %.3f", risk_domain, score)
# Step 4: score β†’ raw decision (tighter threshold for high-risk)
raw_decision = self._score_to_decision(score, risk_domain)
# Step 5: apply safety rules
final_decision, fired = self._safety.apply(
raw_decision, score, sources, evidence,
evidence_state, risk_domain, has_auth_contra,
)
# Step 6: settled-science override β€” any contradicting evidence β†’ REMOVE
n_con = sum(1 for s in sources if s.stance == "contradicting")
if (
settled_topic is not None
and n_con >= 1
and int(self._cfg.settled_science_remove) == 1
):
fired.append(
f"SETTLED_SCIENCE: '{settled_topic}' contradicts scientific consensus "
f"({n_con} source(s) agree) β€” forcing REMOVE with high confidence"
)
final_decision = "remove"
# Step 7: compute calibrated confidence
n_sup = sum(1 for s in sources if s.stance == "supporting")
n_neu = sum(1 for s in sources if s.stance == "neutral")
confidence = self._confidence(
score, sources, evidence_state,
n_con=n_con, n_sup=n_sup,
settled_topic=settled_topic,
)
# Step 8: build rationale
rationale = self._build_rationale(
claim_text, final_decision, score, confidence,
n_sup, n_con, n_neu, fired, evidence_state,
risk_domain, settled_topic,
)
result = StructuredReasoning(
decision = final_decision,
evidence_score = score,
confidence = confidence,
source_evidence = evidence,
n_supporting = n_sup,
n_contradicting = n_con,
n_neutral = n_neu,
n_total = len(sources),
safety_rules_fired = fired,
risk_domain = risk_domain,
settled_science_topic = settled_topic,
rationale = rationale,
)
logger.info(
"Reasoning: claim='%s…' risk=%s settled=%s state=%s score=%.3f "
"raw=%s final=%s conf=%s sources=%d(sup=%d/con=%d/neu=%d) rules=%d",
claim_text[:50], risk_domain or "none", settled_topic or "none",
evidence_state, score, raw_decision, final_decision, confidence,
len(sources), n_sup, n_con, n_neu, len(fired),
)
return result
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _score_to_decision(self, score: float, risk_domain: Optional[str] = None) -> str:
remove_thresh = (
self._cfg.high_risk_remove_threshold
if risk_domain is not None
else self._cfg.remove_threshold
)
if score > self._cfg.allow_threshold:
return "allow"
if score < remove_thresh:
return "remove"
return "flag"
def _confidence(
self,
score: float,
sources: List[ScoredSource],
evidence_state: str,
n_con: int = 0,
n_sup: int = 0,
settled_topic: Optional[str] = None,
) -> str:
"""
Confidence reflects how certain the system is about its decision.
Priority cascade (highest wins):
1. Settled science + any contra evidence β†’ always "high"
2. Consensus scaling: confidence grows with n_contra sources
base=0.40, +0.12 per source, +0.15 bonus when n_sup=0
3. Evidence state shortcuts (strong_contra/strong_support)
4. Reliability-weighted fallback
"""
abs_score = abs(score)
if evidence_state == "no_evidence" or not sources:
return "low"
# Rule 1: settled science is always high certainty when contradicted
if settled_topic is not None and n_con >= 1:
return "high"
# Rule 2: consensus-based scaling
if n_con >= 1:
raw = self._cfg.consensus_base_confidence + (n_con - 1) * self._cfg.consensus_per_source_boost
# Bonus when absolutely no supporting sources exist (no counter-narrative)
if n_sup == 0:
raw += self._cfg.no_support_bonus
raw = min(1.0, raw)
if raw >= self._cfg.consensus_high_cutoff:
return "high"
if raw >= self._cfg.consensus_medium_cutoff:
return "medium"
return "low"
# Rule 3: conflicting evidence
if evidence_state == "conflicting":
return "medium" if abs_score >= 0.3 else "low"
# Rule 4: strong supporting state
if evidence_state == "strong_support" and abs_score >= 0.3:
return "high"
# Rule 5: reliability-weighted fallback
n_hi_rel = sum(1 for s in sources if s.reliability >= 0.70)
if n_hi_rel >= 2 and abs_score >= 0.40:
return "high"
if n_hi_rel >= 1 and abs_score >= 0.20:
return "medium"
if abs_score >= self._cfg.high_confidence_threshold:
return "high"
if abs_score >= self._cfg.low_confidence_threshold:
return "medium"
return "low"
@staticmethod
def _build_rationale(
claim: str,
decision: str,
score: float,
confidence: str,
n_sup: int,
n_con: int,
n_neu: int,
fired: List[str],
evidence_state: str = "",
risk_domain: Optional[str] = None,
settled_topic: Optional[str] = None,
) -> str:
parts = []
# Settled science notice (highest priority label)
if settled_topic:
settled_labels = {
"flat_earth": "flat-Earth (Earth is spherical β€” settled science)",
"vaccines_autism": "vaccine-autism link (disproven by large-scale studies)",
"climate_hoax": "climate change denial (97%+ scientific consensus)",
"evolution_denial": "evolution denial (established biological science)",
"moon_landing_hoax": "moon landing conspiracy (documented historical fact)",
"5g_covid": "5G-COVID link (physically impossible β€” settled science)",
"bleach_cure": "bleach as cure (extremely dangerous pseudoscience)",
"hiv_not_real": "HIV/AIDS denial (settled medical science)",
}
label = settled_labels.get(settled_topic, f"{settled_topic} (settled science)")
parts.append(f"SETTLED SCIENCE: This claim involves {label}.")
elif risk_domain:
domain_labels = {
"medical": "medical/health misinformation",
"harmful_action": "content promoting harmful actions",
"public_health": "public health misinformation",
"election": "election integrity misinformation",
"climate": "climate misinformation",
}
label = domain_labels.get(risk_domain, f"{risk_domain} misinformation")
parts.append(f"HIGH-RISK: This claim involves {label}.")
decision_map = {
"allow": "The claim is supported by available evidence.",
"flag": "The evidence is insufficient or conflicting β€” the claim is flagged for review.",
"remove": "Strong contradicting evidence indicates the claim is false or harmful.",
}
parts.append(decision_map.get(decision, "Decision made."))
state_desc = {
"strong_contra": "Multiple high-credibility sources contradict this claim.",
"weak_contra": "Low-credibility sources contradict this claim.",
"strong_support": "Multiple high-credibility sources support this claim.",
"weak_support": "Some sources support this claim but their credibility is limited.",
"conflicting": "Sources are divided β€” some support and some contradict this claim.",
"neutral_only": "Available sources are neutral and do not confirm or deny the claim.",
"no_evidence": "No relevant sources were found to assess this claim.",
}
if evidence_state in state_desc:
parts.append(state_desc[evidence_state])
# Consensus strength note
if n_con >= 1 and n_sup == 0:
parts.append(f"No supporting sources found β€” all {n_con} source(s) contradict the claim.")
elif n_con >= 3:
parts.append(f"{n_con} independent sources contradict this claim (strong consensus).")
parts.append(
f"Evidence score: {score:+.3f} ({confidence} confidence). "
f"Sources: {n_sup} supporting, {n_con} contradicting, {n_neu} neutral."
)
# Safety rules
if fired:
parts.append("Safety rules applied: " + " | ".join(fired))
return " ".join(parts)
# ─────────────────────────────────────────────────────────────────────────────
# Convenience: reason over multiple claims, return aggregate decision
# ─────────────────────────────────────────────────────────────────────────────
def reason_over_claims(
claim_results: Dict[str, List[ScoredSource]],
engine: Optional[DeterministicReasoningEngine] = None,
) -> Tuple[str, Dict[str, StructuredReasoning]]:
"""
Run reasoning for each claim independently, then derive a single
aggregate content decision.
Parameters
----------
claim_results : dict mapping claim_text β†’ List[ScoredSource]
engine : optional pre-built engine (created if None)
Returns
-------
(aggregate_decision, per_claim_reasoning_dict)
Aggregate logic:
ANY "remove" β†’ aggregate = "remove"
ANY "flag" β†’ aggregate = "flag" (unless already remove)
ALL "allow" β†’ aggregate = "allow"
"""
eng = engine or DeterministicReasoningEngine()
per_claim: Dict[str, StructuredReasoning] = {}
for claim_text, sources in claim_results.items():
per_claim[claim_text] = eng.reason(claim_text, sources)
decisions = [r.decision for r in per_claim.values()]
if "remove" in decisions:
agg = "remove"
elif "flag" in decisions:
agg = "flag"
elif decisions:
agg = "allow"
else:
agg = "flag" # no claims β†’ conservative default
return agg, per_claim