amitca71/marbertv2-levantine-incitement-detector

Levantine Arabic Incitement Detector

This repository contains a custom fine-tuned MARBERTv2 model for 3-way classification of Levantine Arabic social text:

• normal
• abusive
• incitement

The training setup uses:

• class-balanced cross-entropy
• asymmetric error cost for incitement mistakes
• a small ordinal penalty
• an auxiliary lexicon head based on incitement.csv

Validation Summary

These are the k-fold cross-validation averages used to select this configuration:

Metric	Value
Accuracy	82.40%
F1 Macro	0.8025
F1 Incitement	0.7752

Labels

ID	Label
0	normal
1	abusive
2	incitement

Confusion Matrix

The image below is a training-set sanity check for the final model trained on all data. It is not an unbiased test result.

How to Load

This is a custom model wrapper, so load it with the provided model.pt weights plus the MARBERTv2 encoder and tokenizer.

import torch
import torch.nn as nn
import re
import unicodedata
from transformers import AutoTokenizer, AutoModel
from huggingface_hub import hf_hub_download

# =========================
# 1. CONFIGURATION
# =========================
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("device:", DEVICE)

BASE_MODEL = "UBC-NLP/MARBERTv2"
REPO_ID = "amitca71/marbertv2-levantine-incitement-detector"  # second repo
MAX_LENGTH = 160
USE_NORMALIZED_TEXT_FOR_MODEL = False
POOLING_STRATEGY = "cls"   # this checkpoint expects 768-d, not cls_mean_max

ARABIC_DIACRITICS = re.compile(r'[\u0610-\u061A\u064B-\u065F\u0670\u06D6-\u06ED]')


# =========================
# 2. TEXT NORMALIZATION
# =========================
def normalize_arabic(text: str) -> str:
    text = unicodedata.normalize("NFKC", text or "").strip()
    text = ARABIC_DIACRITICS.sub("", text)
    text = text.replace("أ", "ا").replace("إ", "ا").replace("آ", "ا")
    text = text.replace("ى", "ي").replace("ؤ", "و").replace("ئ", "ي").replace("ة", "ه")
    text = re.sub(r"[^\w\s#@/]", " ", text)
    text = re.sub(r"\s+", " ", text)
    return text.strip().lower()

def text_for_model(text: str, use_normalized: bool = USE_NORMALIZED_TEXT_FOR_MODEL) -> str:
    return normalize_arabic(text) if use_normalized else (text or "").strip()


# =========================
# 3. MODEL DEFINITION
# =========================
class MarbertMultiTask(nn.Module):
    def __init__(self, base_model_name: str, pooling_strategy: str = "cls"):
        super().__init__()
        # IMPORTANT: use "encoder" because the checkpoint keys are encoder.*
        self.encoder = AutoModel.from_pretrained(base_model_name)
        self.pooling_strategy = pooling_strategy

        hidden_size = self.encoder.config.hidden_size  # 768 for MARBERTv2

        if pooling_strategy in {"cls", "mean", "max"}:
            rep_dim = hidden_size
        elif pooling_strategy == "cls_mean_max":
            rep_dim = hidden_size * 3
        else:
            raise ValueError(f"Unknown pooling_strategy: {pooling_strategy}")

        self.classifier = nn.Linear(rep_dim, 3)
        self.lexicon_head = nn.Linear(rep_dim, 1)

    def masked_mean_pool(self, last_hidden_state, attention_mask):
        mask = attention_mask.unsqueeze(-1).expand(last_hidden_state.size()).float()
        masked_embeddings = last_hidden_state * mask
        summed = masked_embeddings.sum(dim=1)
        counts = mask.sum(dim=1).clamp(min=1e-9)
        return summed / counts

    def masked_max_pool(self, last_hidden_state, attention_mask):
        mask = attention_mask.unsqueeze(-1).bool()
        masked = last_hidden_state.masked_fill(~mask, float("-inf"))
        pooled = masked.max(dim=1).values
        pooled[torch.isinf(pooled)] = 0.0
        return pooled

    def forward(self, input_ids, attention_mask, token_type_ids=None):
        outputs = self.encoder(
            input_ids=input_ids,
            attention_mask=attention_mask,
            token_type_ids=token_type_ids,
            return_dict=True,
        )

        last_hidden_state = outputs.last_hidden_state
        cls_vec = last_hidden_state[:, 0, :]
        mean_vec = self.masked_mean_pool(last_hidden_state, attention_mask)
        max_vec = self.masked_max_pool(last_hidden_state, attention_mask)

        if self.pooling_strategy == "cls":
            pooled = cls_vec
        elif self.pooling_strategy == "mean":
            pooled = mean_vec
        elif self.pooling_strategy == "max":
            pooled = max_vec
        elif self.pooling_strategy == "cls_mean_max":
            pooled = torch.cat([cls_vec, mean_vec, max_vec], dim=-1)
        else:
            raise ValueError(f"Unknown pooling_strategy: {self.pooling_strategy}")

        logits = self.classifier(pooled)
        lexicon_logits = self.lexicon_head(pooled)

        return {
            "logits": logits,
            "lexicon_logits": lexicon_logits,
        }


# =========================
# 4. LOAD TOKENIZER + MODEL
# =========================
print("Loading model and tokenizer...")
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)

model = MarbertMultiTask(BASE_MODEL, pooling_strategy=POOLING_STRATEGY)

# Download checkpoint
model_path = hf_hub_download(repo_id=REPO_ID, filename="model.pt")
state_dict = torch.load(model_path, map_location="cpu", weights_only=True)

# Sanity checks
print("Model classifier.in_features:", model.classifier.in_features)
print("Checkpoint classifier.weight shape:", state_dict["classifier.weight"].shape)
print("Model lexicon_head.in_features:", model.lexicon_head.in_features)
print("Checkpoint lexicon_head.weight shape:", state_dict["lexicon_head.weight"].shape)

# This should now load cleanly
model.load_state_dict(state_dict)

model.to(DEVICE)
model.eval()

bundle = {
    "model": model,
    "tokenizer": tokenizer,
    "label_map": {0: "normal", 1: "abusive", 2: "incitement"},
}


# =========================
# 5. PREDICTION FUNCTION
# =========================
def predict_one(bundle, text: str):
    encoded = bundle["tokenizer"](
        text_for_model(text),
        truncation=True,
        padding="max_length",
        max_length=MAX_LENGTH,
        return_tensors="pt",
    )

    encoded = {k: v.to(DEVICE) for k, v in encoded.items()}

    with torch.no_grad():
        out = bundle["model"](**encoded)

    logits = out["logits"]
    lexicon_logits = out["lexicon_logits"]

    probs = torch.softmax(logits, dim=-1).squeeze(0).tolist()
    pred_id = int(torch.argmax(logits, dim=-1).item())

    response = {
        "pred_label": bundle["label_map"][pred_id],
        "confidence": probs[pred_id],
        "prob_normal": probs[0],
        "prob_abusive": probs[1],
        "prob_incitement": probs[2],
        "lexicon_signal_prob": torch.sigmoid(lexicon_logits).squeeze(0).item(),
    }

    return response


# =========================
# 6. TEST
# =========================
sample_text = "انت يا عميل السفارات يا ابن الكلب حسابك عسير"
response = predict_one(bundle, sample_text)

print("\nPrediction Response:")
print(response)

Limitations

• The top class is built from open-source proxies for incitement, not a gold incitement-only annotation project.
• Performance is best on Levantine political/social text and may degrade on other Arabic varieties or platforms.
• The confusion matrix in this card is from the full-data training run and should not be treated as held-out evaluation.