"""
ui.py — Information Integrity Analyzer
One input. Automatic pipeline. Rich verdict output.
"""
# Load .env first so all API keys are available before any module reads os.getenv()
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
import re
import time
from typing import Generator
import gradio as gr
from data_collector import CollectionRequest, DataCollector
from environment import InformationIntegrityEnv
from explainability import ExplainabilityEngine, build_reasoning_from_episode
from models import Action
from tasks.graders import GraderFactory
try:
from reasoning import DeterministicReasoningEngine, reason_over_claims
from retrieval import RetrievalPipeline
from decision_engine import make_moderation_decisions
_REASONING_AVAILABLE = True
except Exception:
_REASONING_AVAILABLE = False
collector = DataCollector()
explainer = ExplainabilityEngine()
# ─────────────────────────────────────────────────────────────────────────────
# Claim extractor
# ─────────────────────────────────────────────────────────────────────────────
_SIGNALS = re.compile(
r"\b(confirm|prove|show|reveal|say|claim|report|find|cause|lead to|linked to|"
r"associated with|increase|decrease|never|always|all|every|none|proven|fact|"
r"truth|confirmed|vaccine|virus|cancer|disease|drug|radiation|5g|covid|"
r"lockdown|percent|%|study|research|scientist|doctor|expert|government|"
r"permanently|alter|spread|hide|secret|conspiracy|banned|cure|prevent)\b",
re.IGNORECASE,
)
_SKIP = re.compile(
r"^(share|follow|click|subscribe|watch|read|download|buy|sign up|please|"
r"thank you|breaking:|update:|alert:|warning:)",
re.IGNORECASE,
)
def extract_claims(text: str) -> list[str]:
sentences = re.split(r"(?<=[.!?])\s+|\s*[|•–—]\s*", text.strip())
claims, seen = [], set()
for s in sentences:
s = s.strip().strip("\"'.,!?")
if len(s) < 12 or len(s) > 280:
continue
if _SKIP.match(s):
continue
if _SIGNALS.search(s) and s.lower() not in seen:
seen.add(s.lower())
claims.append(s)
if not claims:
claims = [s.strip() for s in sentences if len(s.strip()) >= 15][:3]
return claims[:6]
# ─────────────────────────────────────────────────────────────────────────────
# HTML card builders
# ─────────────────────────────────────────────────────────────────────────────
def _progress_html(stage: str, pct: int, msg: str) -> str:
stages = ["Extracting claims", "Searching sources", "Running analysis", "Complete"]
dots = "".join(
f''
for i, s in enumerate(stages)
if (i <= stages.index(stage)) or True
)
return f"""
"""
def _verdict_color(decision: str) -> tuple[str, str, str]:
"""Returns (bg_gradient, text_color, icon)."""
return {
"remove": ("linear-gradient(135deg,#ff4444,#d32f2f)", "#fff", "🚫"),
"flag": ("linear-gradient(135deg,#ff9800,#e65100)", "#fff", "⚠️"),
"allow": ("linear-gradient(135deg,#43a047,#1b5e20)", "#fff", "✅"),
}.get(decision, ("linear-gradient(135deg,#607d8b,#37474f)", "#fff", "❓"))
def _reliability_bar(score: float) -> str:
filled = int(score * 12)
color = "#e53935" if score < 0.4 else "#fb8c00" if score < 0.7 else "#43a047"
bar = f'{"█"*filled}' \
f'{"█"*(12-filled)}'
return f'{bar} ' \
f'{score:.2f}'
def _stance_badge(stance: str) -> str:
cfg = {
"contradicting": ("#fdecea", "#c62828", "✗ Contradicts"),
"supporting": ("#e8f5e9", "#2e7d32", "✓ Supports"),
"neutral": ("#f3f4f6", "#555", "· Neutral"),
}.get(stance, ("#f3f4f6", "#555", stance))
bg, fg, label = cfg
return (f'{label}')
def _claim_badge(status: str) -> str:
cfg = {
"false": ("#fdecea", "#c62828", "FALSE"),
"true": ("#e8f5e9", "#2e7d32", "TRUE"),
"partially_true": ("#fff8e1", "#e65100", "PARTIAL"),
"unknown": ("#f3f4f6", "#555", "UNKNOWN"),
"unverified": ("#f3f4f6", "#9e9e9e", "UNVERIFIED"),
}.get(status.lower() if isinstance(status, str) else "unverified",
("#f3f4f6", "#555", status))
bg, fg, label = cfg
return (f'{label}')
def build_result_html(statement: str, scenario: dict, obs, result, explanation: str,
per_claim_reasoning: dict | None = None) -> str:
decision = result.decision or "unknown"
grad, txt_col, icon = _verdict_color(decision)
correct_badge = (
'✓ Correct'
if result.correct else
'✗ Incorrect'
)
# ── Claims section
claims_html = ""
for c in obs.claims:
status = c.verification_status
if hasattr(status, "value"):
status = status.value
conf_pct = int(c.confidence * 100)
claims_html += f"""
{c.text}
{_claim_badge(status)}
conf {conf_pct}%
"""
# ── Sources section (top 6 by reliability)
sources = sorted(scenario["sources"], key=lambda s: -s["reliability"])[:6]
sources_html = ""
for s in sources:
domain = re.sub(r"https?://(www\.)?", "", s["url"]).split("/")[0]
sources_html += f"""
{s['title'][:65]}
{_stance_badge(s['stance'])}
{domain} · {s['source_type']}
{s['text'][:180]}{'…' if len(s['text'])>180 else ''}
Reliability: {_reliability_bar(s['reliability'])}
"""
if not sources_html:
sources_html = 'No sources fetched — try adding API keys for richer results.
'
# ── Reward breakdown
bd = result.reward_breakdown
reward_rows = ""
for label, val in [
("Claim verification", bd.verify_claim_reward),
("Source checks", bd.check_source_reward),
("Multi-source bonus", bd.multi_source_bonus),
("Decision accuracy", bd.decision_reward),
("Premature penalty", bd.premature_penalty),
("Redundancy penalty", bd.redundant_penalty),
("Inefficiency", bd.inefficiency_penalty),
("Over-aggression", bd.aggression_penalty),
]:
if val == 0:
continue
color = "#2e7d32" if val > 0 else "#c62828"
reward_rows += f"""
{label}
{val:+.2f}
"""
# ── Grader sub-scores
grader = GraderFactory.get(scenario["difficulty"])
gbd = grader.score_breakdown(
decision = result.decision or "none",
ground_truth = result.ground_truth,
sources_checked = list(range(result.sources_checked)),
claims_verified = list(range(result.claims_verified)),
step_count = result.steps_taken,
all_source_ids = [s["id"] for s in scenario["sources"]],
all_claim_ids = [c["id"] for c in scenario["claims"]],
)
grader_bars = ""
for sub_label, sub_val in [
("Decision", gbd["decision_sub_score"]),
("Evidence", gbd["evidence_sub_score"]),
("Reasoning", gbd["reasoning_sub_score"]),
("Efficiency", gbd["efficiency_sub_score"]),
]:
pct = int(sub_val * 100)
colr = "#e53935" if pct < 50 else "#fb8c00" if pct < 75 else "#43a047"
grader_bars += f"""
"""
# ── Deterministic Reasoning breakdown (new)
reasoning_html = ""
if per_claim_reasoning:
reasoning_html = """
🧠 Deterministic Reasoning Engine
"""
for claim_text, r in per_claim_reasoning.items():
score = r["evidence_score"]
conf = r["confidence"]
dec = r["decision"]
n_sup = r["n_supporting"]
n_con = r["n_contradicting"]
n_neu = r["n_neutral"]
dec_col = {"allow": "#2e7d32", "flag": "#e65100", "remove": "#c62828"}.get(dec, "#555")
score_col = "#c62828" if score < -0.3 else "#2e7d32" if score > 0.3 else "#e65100"
reasoning_html += f"""
"{claim_text[:100]}{'…' if len(claim_text)>100 else ''}"
{dec}
Score: {score:+.3f}
Confidence: {conf}
▲ {n_sup} supporting
▼ {n_con} contradicting
· {n_neu} neutral
"""
# Safety rules fired
if r.get("safety_rules_fired"):
for rule in r["safety_rules_fired"]:
rule_name = rule.split(":")[0]
rule_desc = rule[len(rule_name)+1:].strip()
reasoning_html += f"""
⚠ {rule_name}: {rule_desc}
"""
# Source breakdown (top 4)
sb = r.get("source_breakdown", [])[:4]
if sb:
reasoning_html += '
'
for src in sb:
contrib = src["contribution"]
c_col = "#c62828" if contrib < 0 else "#2e7d32" if contrib > 0 else "#888"
reasoning_html += f"""
{src['title'][:70]}
{_stance_badge(src['stance'])}
{contrib:+.3f}
"""
reasoning_html += "
"
reasoning_html += "
"
reasoning_html += "
"
# ── Explanation (pre-formatted)
expl_lines = explanation.replace("<", "<").replace(">", ">")
expl_html = f'{expl_lines}'
return f"""
{icon}
Moderation Verdict
{decision}
Difficulty: {scenario['difficulty']} ·
Steps: {result.steps_taken} ·
Reward: {result.total_reward:+.2f} ·
Grader: {result.grader_score:.0%}
{correct_badge}
Statement Reviewed
"{statement}"
Claims Verified ({len(obs.claims)})
{claims_html}
Grader Score Breakdown
{grader_bars}
Reward Breakdown
{reward_rows}
Total Reward
{result.total_reward:+.2f}
Evidence Sources ({len(scenario['sources'])} found)
{sources_html}
{reasoning_html}
📋 Full Reasoning & Explanation
{expl_html}
"""
# ─────────────────────────────────────────────────────────────────────────────
# Main pipeline (streaming generator)
# ─────────────────────────────────────────────────────────────────────────────
def analyze(statement: str) -> Generator:
"""
Difficulty is NOT a UI input.
The UI is a demo tool; evaluation difficulty is controlled exclusively
by the backend API (/reset?difficulty=...) and inference.py.
DataCollector always uses 'medium' here — it only affects evidence
retrieval depth, not grading.
"""
statement = statement.strip()
if not statement:
yield "⚠️ Please enter a statement to analyze.
"
return
# Stage 1 — Extract claims
yield _progress_html("Extracting claims", 10, "detecting verifiable assertions…")
claims = extract_claims(statement)
if not claims:
claims = [statement]
claim_list = " • ".join(f"{c[:60]}" for c in claims)
yield _progress_html("Searching sources", 30,
f"found {len(claims)} claim(s): {claim_list[:120]}…")
# Stage 2 — Collect evidence
# Difficulty is fixed to "medium" here — the UI does NOT control evaluation logic.
# To run a proper benchmark, use: python inference.py --benchmark
try:
req = CollectionRequest(content=statement, claims=claims,
difficulty="medium")
scenario = collector.collect(req)
except Exception as e:
yield f"❌ Evidence collection failed: {e}
"
return
n_src = len(scenario["sources"])
yield _progress_html("Running analysis", 65,
f"fetched {n_src} source(s) — running moderation episode…")
# Stage 3 — Run episode
env = InformationIntegrityEnv(scenario=scenario, max_steps=20)
obs = env.reset()
per_claim_reasoning = None # populated by reasoning engine at decide step
# Auto-agent: verify all claims → check all sources → decide
for claim in obs.claims:
try:
obs, _, done, _ = env.step(
Action(type="verify_claim", claim_id=claim.id,
reasoning="auto-verify"))
if done:
break
except Exception:
pass
if not env.done:
for src in obs.available_sources:
try:
obs, _, done, _ = env.step(
Action(type="check_source", source_id=src.id,
reasoning="auto-check"))
if done:
break
except Exception:
pass
if not env.done:
# Use DeterministicReasoningEngine if available, else weighted fallback
if _REASONING_AVAILABLE:
try:
pipeline = RetrievalPipeline()
engine = DeterministicReasoningEngine()
raw_sources = [
{
"id": s.id, "title": s.title, "text": s.text,
"url": s.url, "stance": s.stance,
"reliability": s.reliability, "source_type": s.source_type,
}
for s in obs.available_sources
]
claim_texts = [c.text for c in obs.claims]
claim_results_map = pipeline.retrieve_for_all_claims(claim_texts, raw_sources)
agg_decision, per_claim_reasoning = reason_over_claims(claim_results_map, engine)
# Convert StructuredReasoning objects → dicts for the HTML builder
per_claim_reasoning = {ct: r.to_dict() for ct, r in per_claim_reasoning.items()}
# ── decision_engine: recalibrate each claim's confidence & decision ──
de_inputs = []
for ct, r in per_claim_reasoning.items():
src_list = claim_results_map.get(ct, [])
de_inputs.append({
"claim": ct,
"status": "FALSE" if r["decision"] == "remove"
else "TRUE" if r["decision"] == "allow"
else "UNKNOWN",
"confidence": r.get("evidence_score", 0.0),
"supporting_sources": r["n_supporting"],
"contradicting_sources": r["n_contradicting"],
"neutral_sources": r["n_neutral"],
"avg_source_reliability": (
sum(s.reliability for s in src_list) / len(src_list)
if src_list else 0.5
),
})
de_result = make_moderation_decisions(de_inputs)
# Map content decision back to env decision format
_de_map = {"AUTO_REMOVE": "remove", "ALLOW": "allow", "FLAG_FOR_REVIEW": "flag"}
decision = _de_map.get(de_result["content_decision"], agg_decision)
# Inject decision_engine output into per_claim_reasoning for display
for entry in de_result["per_claim"]:
ct = entry["claim"]
if ct in per_claim_reasoning:
per_claim_reasoning[ct]["de_decision"] = entry["decision"]
per_claim_reasoning[ct]["de_confidence"] = entry["final_confidence"]
per_claim_reasoning[ct]["de_reason"] = entry["reason"]
per_claim_reasoning[ct]["de_pattern"] = entry.get("pattern_matched")
except Exception:
per_claim_reasoning = None
# Fallback to weighted ratio
contra = sum(s.reliability for s in obs.available_sources if s.stance == "contradicting")
supp = sum(s.reliability for s in obs.available_sources if s.stance == "supporting")
total = contra + supp or 1.0
ratio = supp / total
decision = "remove" if ratio < 0.25 else "flag" if ratio < 0.60 else "allow"
else:
per_claim_reasoning = None
contra = sum(s.reliability for s in obs.available_sources if s.stance == "contradicting")
supp = sum(s.reliability for s in obs.available_sources if s.stance == "supporting")
total = contra + supp or 1.0
ratio = supp / total
decision = "remove" if ratio < 0.25 else "flag" if ratio < 0.60 else "allow"
try:
obs, _, _, _ = env.step(
Action(type="decide", decision=decision,
reasoning="deterministic engine decision"))
except Exception:
pass
else:
per_claim_reasoning = None
yield _progress_html("Complete", 95, "generating explanation…")
# Stage 4 — Build result
result = env.episode_result()
reasoning = build_reasoning_from_episode(
obs=obs,
decision=result.decision or "unknown",
episode_id=env._episode_id,
difficulty=scenario["difficulty"],
)
explanation = explainer.generate_explanation(reasoning)
# Use engine's per_claim_reasoning if captured, else env's cached reasoning
if per_claim_reasoning is None and env._last_reasoning:
per_claim_reasoning = {
ct: r.to_dict() for ct, r in env._last_reasoning.items()
}
yield build_result_html(statement, scenario, obs, result, explanation,
per_claim_reasoning=per_claim_reasoning)
# ─────────────────────────────────────────────────────────────────────────────
# UI Layout
# ─────────────────────────────────────────────────────────────────────────────
CSS = """
@import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700;800&display=swap');
body, .gradio-container { font-family: 'Inter', system-ui, sans-serif !important; background: #f0f4ff !important; }
.gr-button-primary { background: linear-gradient(135deg,#4f8ef7,#7c6af7) !important;
border:none !important; font-weight:700 !important;
font-size:16px !important; padding:14px 36px !important;
border-radius:12px !important; color:white !important; }
.gr-button-primary:hover { opacity:.9 !important; transform:translateY(-1px); }
textarea { font-size:15px !important; border-radius:12px !important;
border:2px solid #e0e7ff !important; padding:14px !important; }
textarea:focus { border-color:#4f8ef7 !important; box-shadow:0 0 0 3px rgba(79,142,247,.15) !important; }
.gr-form { background:transparent !important; }
.gr-box { background:transparent !important; border:none !important; box-shadow:none !important; }
#component-0 { background: #f0f4ff !important; }
"""
EXAMPLES = [
"5G towers confirmed to spread COVID-19. Scientists reveal radiation activates coronavirus. Governments hiding the truth.",
"Doctors confirm: drinking 3 cups of coffee daily prevents Alzheimer's in 100% of patients with zero side effects.",
"New government report proves immigration increases violent crime by 40% in all affected regions. Multiple studies confirm.",
"mRNA vaccines permanently alter human DNA and cause infertility according to multiple top doctors and researchers.",
"There will be a national lockdown starting next week. Government sources confirmed mandatory quarantine for all cities.",
"Eating chocolate every day reduces heart attack risk by 50% — scientists confirm in landmark study.",
]
with gr.Blocks(title="Information Integrity Analyzer") as demo:
gr.HTML("""
🛡️
Information Integrity Analyzer
Paste any statement or claim. The system automatically extracts assertions,
searches real evidence sources, and delivers a moderation verdict with full explanation.
Demo mode — for benchmarking across difficulty tiers run:
python inference.py --benchmark
""")
with gr.Row(elem_id="main-row"):
with gr.Column(scale=1, min_width=640):
statement_box = gr.Textbox(
label="",
placeholder=(
"Paste any statement, post, or claim here…\n\n"
"e.g. \"5G towers confirmed to spread COVID-19. "
"Governments hiding the truth.\""
),
lines=4,
max_lines=8,
show_label=False,
)
analyze_btn = gr.Button(
"🔍 Analyze Statement",
variant="primary",
)
gr.HTML("Try an example
")
with gr.Row():
for ex_text in EXAMPLES[:3]:
btn = gr.Button(ex_text[:45] + "…", size="sm", scale=1)
btn.click(fn=lambda t=ex_text: t, outputs=[statement_box])
with gr.Row():
for ex_text in EXAMPLES[3:]:
btn = gr.Button(ex_text[:45] + "…", size="sm", scale=1)
btn.click(fn=lambda t=ex_text: t, outputs=[statement_box])
result_html = gr.HTML(
value=""
"Results will appear here after analysis.
"
)
analyze_btn.click(
fn=analyze,
inputs=[statement_box],
outputs=[result_html],
)
if __name__ == "__main__":
demo.launch(
server_name="0.0.0.0",
server_port=7860,
show_error=True,
css=CSS,
theme=gr.themes.Soft(
primary_hue="indigo",
secondary_hue="purple",
neutral_hue="slate",
font=[gr.themes.GoogleFont("Inter"), "system-ui", "sans-serif"],
),
)