""" Defense demo: Multi-Layered Sarcasm-Aware Harmful Content Detection. One codebase, two deployments: - HF Spaces (16GB): live XLM-R sarcasm + live baseline + live Mechanism B - Local laptop (4GB): live XLM-R sarcasm + curated showcase (Qwen models skipped) The app auto-detects what it can load. Set env vars to control: LOAD_QWEN=1 attempt to load the two Qwen classifiers (default on Spaces) SARCASM_MODEL=... HF repo or local dir for the XLM-R sarcasm detector BASELINE_ADAPTER=... / MECHB_ADAPTER=... HF repos or local dirs for adapters Run: python app.py """ import os import json import torch import spaces import gradio as gr import numpy as np # ----------------------------- configuration ------------------------------ # SARCASM_MODEL = os.environ.get("SARCASM_MODEL", "mlaskri/sarcasm-xlmr-hyperparams-only") BASELINE_ADAPTER = os.environ.get("BASELINE_ADAPTER", "mlaskri/baseline-combined-seed42") MECHB_ADAPTER = os.environ.get("MECHB_ADAPTER", "mlaskri/harmful-content-mechanism-b-v2") QWEN_BASE = "Qwen/Qwen2.5-3B-Instruct" LOAD_QWEN = os.environ.get("LOAD_QWEN", "1") == "1" SARCASM_TEMPERATURE = 1.953 # validation-fit temperature (thesis Table 4.5) SARCASM_THRESHOLD = 0.48 # tuned decision threshold BASELINE_THRESHOLD = 0.35 # validation-selected (thesis Table 5.1) MECHB_THRESHOLD = 0.60 # validation-selected (thesis Table 5.4) SHOWCASE_FILE = "curated_examples.json" # ----------------------------- model loading ------------------------------ # print("Loading sarcasm detector (XLM-R)...") from transformers import AutoTokenizer, AutoModelForSequenceClassification sar_tok = AutoTokenizer.from_pretrained(SARCASM_MODEL) sar_model = AutoModelForSequenceClassification.from_pretrained(SARCASM_MODEL) sar_model.eval() # small; runs on CPU inside the GPU-decorated fn or CPU fallback print("Sarcasm detector ready.") qwen_models = {} if LOAD_QWEN: try: from peft import PeftModel print("Loading Qwen classifiers (baseline + Mechanism B)... this takes a few minutes on CPU.") qtok = AutoTokenizer.from_pretrained(QWEN_BASE) if qtok.pad_token is None: qtok.pad_token = qtok.eos_token for name, adapter in [("baseline", BASELINE_ADAPTER), ("mechanism_b", MECHB_ADAPTER)]: base = AutoModelForSequenceClassification.from_pretrained( QWEN_BASE, num_labels=2, torch_dtype=torch.float16, device_map="cpu", # <-- force CPU at load; GPU only inside @spaces.GPU low_cpu_mem_usage=True, ) base.config.pad_token_id = qtok.pad_token_id m = PeftModel.from_pretrained(base, adapter, device_map="cpu") m.eval() qwen_models[name] = m print(f" {name} ready.") except Exception as e: print(f"Qwen loading failed ({e}) -- running in sarcasm-only mode.") qwen_models = {} else: print("LOAD_QWEN=0 -- sarcasm-only mode (local laptop).") LIVE_HARMFUL = bool(qwen_models) # ----------------------------- inference fns ------------------------------ # @torch.no_grad() def sarcasm_score(text): dev = "cuda" if torch.cuda.is_available() else "cpu" model = sar_model.to(dev) enc = sar_tok(text, truncation=True, max_length=128, return_tensors="pt").to(dev) logits = model(**enc).logits # temperature scaling on logits, then softmax p = torch.softmax(logits / SARCASM_TEMPERATURE, dim=-1)[0, 1].item() return p @torch.no_grad() def harmful_pred(name, text): m = qwen_models[name] dev = "cuda" if torch.cuda.is_available() else "cpu" m = m.to(dev) enc = qtok(text, truncation=True, max_length=128, return_tensors="pt").to(dev) logits = m(**enc).logits return torch.softmax(logits.float(), dim=-1)[0, 1].item() def fmt_pct(p): return f"{p:.3f}" @spaces.GPU(duration=90) def analyze(text): text = (text or "").strip() if not text: return "Please enter some text.", "", "" s = sarcasm_score(text) s_label = "SARCASTIC" if s >= SARCASM_THRESHOLD else "not sarcastic" sar_out = (f"**Calibrated sarcasm probability: {fmt_pct(s)}** \n" f"Decision (threshold {SARCASM_THRESHOLD}): **{s_label}** \n" f"*Twitter-XLM-RoBERTa, hyperparameters-only configuration; " f"temperature-scaled (T={SARCASM_TEMPERATURE}).*") if not LIVE_HARMFUL: harm_out = ("*Harmful-content classifiers (3B parameters each) exceed this machine's memory. " "See the Showcase tab for committed evaluation results comparing all mechanisms — " "every number traces to results files in the project repository.*") return sar_out, harm_out, "" pb = harmful_pred("baseline", text) pm = harmful_pred("mechanism_b", text) lb = "HARMFUL" if pb >= BASELINE_THRESHOLD else "non-harmful" lm = "HARMFUL" if pm >= MECHB_THRESHOLD else "non-harmful" harm_out = (f"| Model | p(harmful) | Decision |\n|---|---|---|\n" f"| Baseline (no sarcasm signal) | {fmt_pct(pb)} | **{lb}** |\n" f"| Mechanism B (sarcasm-aware) | {fmt_pct(pm)} | **{lm}** |") note = "" if lb != lm: note = ("**The two models disagree on this example** — the sarcasm-aware model's decision " "differs from the baseline's, illustrating the effect studied in this thesis.") return sar_out, harm_out, note # ----------------------------- showcase tab ------------------------------- # def load_showcase(): if os.path.exists(SHOWCASE_FILE): with open(SHOWCASE_FILE, encoding="utf-8") as f: return json.load(f) return [] SHOWCASE = load_showcase() def showcase_view(idx): if not SHOWCASE: return "No curated examples loaded (curated_examples.json missing)." ex = SHOWCASE[int(idx)] rows = "\n".join( f"| {m['name']} | {m['prob']} | **{m['decision']}** |" for m in ex["models"]) return (f"### {ex['title']}\n\n" f"**Text ({ex['language']}):** {ex['text']}\n\n" f"**True label:** {ex['true_label']} | **Sarcasm score:** {ex['sarcasm_score']}\n\n" f"| Model | p(harmful) | Decision |\n|---|---|---|\n{rows}\n\n" f"**Why this example matters:** {ex['story']}\n\n" f"*Source: {ex['source']} (held-out test set; predictions from committed evaluation artifacts).*") # ----------------------------- UI ----------------------------------------- # with gr.Blocks(title="Sarcasm-Aware Harmful Content Detection — PFE Demo") as demo: gr.Markdown( "# Multi-Layered Multilingual Sarcasm-Aware Harmful Content Detection\n" "**PFE defense demo — ESTIN 2025/2026** · Arabic + English · " + ("**Live mode: full pipeline**" if LIVE_HARMFUL else "**Local mode: live sarcasm layer + committed results**")) with gr.Tab("Live analysis"): inp = gr.Textbox(label="Enter any text (Arabic or English)", lines=3, placeholder="Type here — e.g. a tweet, a comment...") btn = gr.Button("Analyze", variant="primary") out_sar = gr.Markdown(label="Layer 1 — Sarcasm detection") out_harm = gr.Markdown(label="Layer 2 — Harmful-content detection") out_note = gr.Markdown() btn.click(analyze, inputs=inp, outputs=[out_sar, out_harm, out_note]) with gr.Tab("Showcase: findings on the evaluation set"): gr.Markdown("Hand-picked held-out test examples illustrating the thesis's main findings. " "All predictions are committed evaluation artifacts, reproducible from the repository.") if SHOWCASE: slider = gr.Slider(0, len(SHOWCASE) - 1, step=1, value=0, label="Example") view = gr.Markdown(showcase_view(0)) slider.change(showcase_view, inputs=slider, outputs=view) else: gr.Markdown("*curated_examples.json not found — add it next to app.py.*") with gr.Tab("About"): gr.Markdown( "**Architecture:** Layer 1 (Twitter-XLM-RoBERTa sarcasm detector, temperature-calibrated) → " "integration mechanisms → Layer 2 (Qwen2.5-3B + LoRA harmful-content classifier).\n\n" "**Mechanism B** transfers the sarcasm layer's adapter weights into the harmful-content " "classifier's initialization — the best-performing mechanism (macro-F1 0.8366 vs 0.8301 baseline; " "harmful-class McNemar p < 10⁻³⁴).\n\n" "Thresholds shown are the validation-selected values frozen before test evaluation, " "as documented in the thesis.") demo.launch()