import torch
import gradio as gr
import faiss
import numpy as np
import wikipediaapi
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
BitsAndBytesConfig,
pipeline,
)
from peft import PeftModel
from sentence_transformers import SentenceTransformer
# ── Config ──────────────────────────────────────────────
BASE_MODEL = "HuggingFaceH4/zephyr-7b-beta"
PEFT_MODEL = "ShehlaKanwal/counterspeech-zephyr" # your HF repo
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
# ── Load model ──────────────────────────────────────────
print("Loading model...")
bnb = BitsAndBytesConfig(
load_in_4bit = True,
bnb_4bit_quant_type = "nf4",
bnb_4bit_compute_dtype = torch.float16,
bnb_4bit_use_double_quant = True,
)
base_model = AutoModelForCausalLM.from_pretrained(
BASE_MODEL, quantization_config=bnb, device_map="auto"
)
model = PeftModel.from_pretrained(base_model, PEFT_MODEL)
tok = AutoTokenizer.from_pretrained(PEFT_MODEL)
print("Model loaded!")
# ── Translation pipeline ────────────────────────────────
translator = pipeline(
"translation",
model = "Helsinki-NLP/opus-mt-mul-en",
device = 0 if torch.cuda.is_available() else -1,
)
# ── Embedder + FAISS ────────────────────────────────────
print("Building knowledge base...")
embedder = SentenceTransformer("intfloat/multilingual-e5-small")
TOPICS = [
"Antisemitism", "Jewish people", "History of the Jews",
"Holocaust", "Jews in the United States",
"Islamophobia", "Islam", "Islamic Golden Age",
"Muslim contributions to medieval Europe",
"Immigration", "Refugee", "Economic impact of immigration",
"Hate speech", "Racism", "Religious discrimination",
]
wiki_en = wikipediaapi.Wikipedia(language="en", user_agent="CounterspeechRAG/1.0")
wiki_es = wikipediaapi.Wikipedia(language="es", user_agent="CounterspeechRAG/1.0")
all_passages, all_topics = [], []
def fetch(topic, lang="en"):
wiki = wiki_en if lang == "en" else wiki_es
page = wiki.page(topic)
if not page.exists():
return []
text = page.summary[:2000]
sentences = text.replace("! ", ". ").replace("? ", ". ").split(". ")
sentences = [s.strip() for s in sentences if len(s.strip()) > 30]
passages = []
for i in range(0, len(sentences) - 1, 2):
p = sentences[i] + ". "
if i + 1 < len(sentences):
p += sentences[i + 1] + "."
passages.append(p.strip())
return passages
for t in TOPICS:
for lang in ["en", "es"]:
ps = fetch(t, lang)
all_passages.extend(ps)
all_topics.extend([t] * len(ps))
embs = embedder.encode(all_passages, batch_size=32, normalize_embeddings=True)
index = faiss.IndexFlatIP(embs.shape[1])
index.add(embs.astype("float32"))
print(f"Knowledge base ready — {index.ntotal} passages")
# ── Generation helpers ──────────────────────────────────
SYSTEM = "You are a factual counter-narrative generator. Provide only 2 factual sentences that directly counter the hate speech. No personal opinions. No conversational tone. Facts only."
def generate(prompt):
inputs = tok(prompt, return_tensors="pt", truncation=True, max_length=512).to(DEVICE)
with torch.no_grad():
out = model.generate(
**inputs,
max_new_tokens = 60,
do_sample = False,
repetition_penalty = 1.5,
pad_token_id = tok.eos_token_id,
eos_token_id = tok.eos_token_id,
)
text = tok.decode(out[0], skip_special_tokens=True)
resp = text.split("<|assistant|>")[-1].strip()
if "." in resp:
resp = resp[:resp.rfind(".") + 1]
return resp
def translate(text):
try:
return translator(text, max_length=400)[0]["translation_text"]
except:
return text
def counterspeech(hate_speech, mode):
if not hate_speech.strip():
return "", "", ""
# Baseline
b_prompt = f"<|system|>\n{SYSTEM}\n<|user|>\nCounter this hate speech with facts only: {hate_speech}\n<|assistant|>\n"
baseline = generate(b_prompt)
baseline_en = translate(baseline)
# RAG
q_emb = embedder.encode([hate_speech], normalize_embeddings=True).astype("float32")
_, idx = index.search(q_emb, 2)
knowledge = " ".join([all_passages[i] for i in idx[0]])[:300]
r_prompt = f"<|system|>\n{SYSTEM}\n<|user|>\nKnowledge: {knowledge}\nCounter this hate speech with facts only: {hate_speech}\n<|assistant|>\n"
rag = generate(r_prompt)
rag_en = translate(rag)
if mode == "Baseline only":
return baseline_en, "", ""
elif mode == "RAG only":
return "", rag_en, f"Retrieved: {knowledge[:200]}..."
else:
return baseline_en, rag_en, f"Retrieved: {knowledge[:200]}..."
# ── Gradio UI ───────────────────────────────────────────
css = """
body { font-family: 'Segoe UI', sans-serif; }
.gr-button-primary { background: linear-gradient(135deg,#6c63ff,#8b85ff) !important; border:none !important; }
footer { display: none !important; }
"""
with gr.Blocks(css=css, title="CounterSpeech AI") as demo:
gr.HTML("""
CounterSpeech AI
Zephyr-7B fine-tuned + Wikipedia RAG · COLING 2025 MCG Shared Task
""")
with gr.Row():
with gr.Column(scale=3):
hs_input = gr.Textbox(
label = "Hate Speech Input",
placeholder = "Type or paste hate speech here...",
lines = 3,
)
mode = gr.Radio(
choices = ["Both", "Baseline only", "RAG only"],
value = "Both",
label = "Mode",
)
gr.Examples(
examples = [
["Jews control Hollywood and the media."],
["Muslims don't contribute anything to our culture."],
["Immigrants are stealing our jobs."],
["Los judíos controlan Hollywood y los medios."],
["Women are too emotional to be good leaders."],
],
inputs = hs_input,
)
btn = gr.Button("Generate Counter-Narrative", variant="primary", size="lg")
with gr.Row():
with gr.Column():
gr.HTML("Baseline Output — Zephyr-7B Fine-tuned
")
baseline_out = gr.Textbox(label="", lines=4, interactive=False)
with gr.Column():
gr.HTML("RAG Enhanced Output — + Wikipedia Knowledge
")
rag_out = gr.Textbox(label="", lines=4, interactive=False)
knowledge_out = gr.Textbox(label="Wikipedia Knowledge Used (RAG)", lines=2, interactive=False)
gr.HTML("""
Evaluation Results — Baseline vs RAG (30 validation examples)
BERTScore F1
64.65
65.42 ▲
Red = Baseline Green = RAG RAG wins 5/7 metrics
""")
btn.click(
fn = counterspeech,
inputs = [hs_input, mode],
outputs = [baseline_out, rag_out, knowledge_out],
)
demo.launch()