**Assessing classification models** I'm focused on evaluating classification models, likely for sentiment or topic identification. I need to write a robust script using Hugging Face transformers, possibly supporting zero-shot learning. It seems I'll be working with sequence classification models tailored to different domains. If the models are pretrained and not fine-tuned, they won’t have labels. I should consider handling label mapping and maybe even detect text scripts automatically based on the input schema provided.**Normalizing text carefully** I need to be careful with text normalization, especially when dealing with different scripts. For instance, I have to ensure that the Arabic yeh (`\u064a`) is converted properly to the Farsi yeh (`\u06cc`), and I should keep in mind the Urdu letter `ے`. I shouldn't overnormalize because it might be optional. Also, for Roman text, I'll need to preserve lower case and whitespace while mapping common spellings and keeping apostrophes intact. Below is an evaluation program for **single-label text classification**. It compares multiple Hugging Face models across: - Nastaliq/Urdu Arabic-script text - Roman Urdu - Individual domains - Domain × script groups - Worst-group performance and robustness gaps It assumes that the models are already fine-tuned for the same classification task. ### Install dependencies ```bash pip install torch transformers pandas numpy scikit-learn sentencepiece ``` ### `urdu_domain_robustness.py` ```python #!/usr/bin/env python3 """ Evaluate multilingual sequence-classification models on Nastaliq and Roman Urdu. Expected input columns: text, label, domain, script, split The `script` and `split` columns are optional. This script is evaluation-only. For a valid domain-generalization experiment, the evaluated checkpoint should not have been trained on the target test domains. Note: Unicode does not encode the font/calligraphic style "Nastaliq". This script uses "nastaliq" to mean Urdu written in the Arabic/Persian script. Explicit script annotations are preferred over automatic detection. """ from __future__ import annotations import argparse import inspect import json import re import unicodedata import warnings from pathlib import Path from typing import Dict, List, Optional import numpy as np import pandas as pd import torch from sklearn.metrics import ( accuracy_score, balanced_accuracy_score, confusion_matrix, f1_score, log_loss, matthews_corrcoef, precision_recall_fscore_support, ) from transformers import AutoModelForSequenceClassification, AutoTokenizer # Common Urdu/Arabic character normalization. URDU_TRANSLATION = str.maketrans( { "ي": "ی", "ى": "ی", "ك": "ک", } ) # --------------------------------------------------------------------- # Data loading and preprocessing # --------------------------------------------------------------------- def load_table(path: str) -> pd.DataFrame: """Load CSV, TSV, JSON, JSONL, NDJSON, or Parquet data.""" path_obj = Path(path) suffix = path_obj.suffix.lower() if suffix == ".csv": return pd.read_csv(path_obj) if suffix == ".tsv": return pd.read_csv(path_obj, sep="\t") if suffix in {".jsonl", ".ndjson"}: return pd.read_json(path_obj, lines=True) if suffix == ".json": with path_obj.open("r", encoding="utf-8") as f: obj = json.load(f) if isinstance(obj, dict) and "data" in obj: obj = obj["data"] if isinstance(obj, dict): # Support either a single record or a dict of columns. if all(isinstance(v, list) for v in obj.values()): return pd.DataFrame(obj) return pd.DataFrame([obj]) return pd.DataFrame(obj) if suffix == ".parquet": return pd.read_parquet(path_obj) raise ValueError( f"Unsupported input format: {suffix}. " "Use CSV, TSV, JSON, JSONL, NDJSON, or Parquet." ) def canonical_label(value) -> str: """Create a stable representation for labels.""" try: if pd.isna(value): return "" except (TypeError, ValueError): pass text = str(value).strip() # Convert labels such as "1.0" to "1". if re.fullmatch(r"[+-]?\d+\.0+", text): text = str(int(float(text))) return text.casefold() def canonical_script(value) -> str: """Normalize script names used in the input file.""" if value is None: return "" try: if pd.isna(value): return "" except (TypeError, ValueError): pass value = str(value).strip().casefold() value = value.replace("-", "_").replace(" ", "_") aliases = { "nastaliq": "nastaliq", "nastaleeq": "nastaliq", "nastaliq_script": "nastaliq", "urdu": "nastaliq", "urdu_script": "nastaliq", "arabic": "nastaliq", "arabic_script": "nastaliq", "roman": "roman_urdu", "romanurdu": "roman_urdu", "roman_urdu": "roman_urdu", "roman_urdu_script": "roman_urdu", "latin": "roman_urdu", "latin_script": "roman_urdu", } return aliases.get(value, value) def detect_script(text: str) -> str: """ Heuristically detect Arabic-script Urdu versus Roman Urdu. This is only a script detector. It cannot determine whether an Arabic script string is rendered using a particular font such as Nastaliq. """ arabic_letters = 0 latin_letters = 0 for char in str(text): if not char.isalpha(): continue unicode_name = unicodedata.name(char, "") if "ARABIC" in unicode_name: arabic_letters += 1 elif "LATIN" in unicode_name: latin_letters += 1 if arabic_letters == 0 and latin_letters == 0: return "unknown" if arabic_letters > 0 and latin_letters == 0: return "nastaliq" if latin_letters > 0 and arabic_letters == 0: return "roman_urdu" if arabic_letters >= 1.5 * latin_letters: return "nastaliq" if latin_letters >= 1.5 * arabic_letters: return "roman_urdu" return "mixed" def normalize_text(text: str, script: str, enabled: bool) -> str: """ Apply conservative normalization. Normalization is disabled by default so that raw-script robustness can be measured. Enable it with --normalize. """ text = str(text) if not enabled: return text text = unicodedata.normalize("NFKC", text) if script == "nastaliq": text = text.replace("\u0640", "") # Remove Arabic tatweel. text = text.translate(URDU_TRANSLATION) text = re.sub(r"\s+", " ", text).strip() return text def prepare_data( dataframe: pd.DataFrame, args: argparse.Namespace, ): """Validate, filter, and prepare the evaluation dataframe.""" df = dataframe.copy() required_columns = [args.text_col, args.label_col, args.domain_col] missing = [column for column in required_columns if column not in df.columns] if missing: raise ValueError( f"Missing required columns: {missing}. " "The data must contain text, label, and domain columns." ) if args.split_col: if args.split_col not in df.columns: raise ValueError( f"Split column '{args.split_col}' was not found in the input." ) split_values = df[args.split_col].astype(str).str.casefold() df = df.loc[ split_values == args.split_value.casefold() ].copy() df = df.dropna(subset=[args.text_col, args.label_col]).copy() if df.empty: raise ValueError("No examples remain after filtering the input data.") df["__row_id"] = df.index.astype(str) df["__text_original"] = df[args.text_col].astype(str) domains = ( df[args.domain_col] .fillna("unknown") .astype(str) .str.strip() .replace("", "unknown") ) df["__domain"] = domains if args.script_col in df.columns: raw_scripts = df[args.script_col].tolist() else: raw_scripts = [None] * len(df) scripts = [] for text, raw_script in zip(df["__text_original"].tolist(), raw_scripts): supplied_script = canonical_script(raw_script) if supplied_script: scripts.append(supplied_script) else: scripts.append(detect_script(text)) df["__script"] = scripts allowed_scripts = { canonical_script(script_name) for script_name in args.scripts } before_filtering = len(df) df = df.loc[df["__script"].isin(allowed_scripts)].copy() if len(df) < before_filtering: ignored = sorted(set(scripts) - allowed_scripts) if ignored: warnings.warn( f"Ignoring scripts not requested by --scripts: {ignored}" ) if df.empty: raise ValueError( "No examples remain after script filtering. " "Check the script column or use --scripts." ) df["__label_key"] = [ canonical_label(value) for value in df[args.label_col].tolist() ] df = df.loc[df["__label_key"] != ""].copy() if df.empty: raise ValueError("No labeled examples remain.") df["__text_eval"] = [ normalize_text(text, script, args.normalize) for text, script in zip( df["__text_original"].tolist(), df["__script"].tolist(), ) ] # Preserve human-readable labels. display_by_key = {} for raw_label, label_key in zip( df[args.label_col].tolist(), df["__label_key"].tolist(), ): display_by_key.setdefault(label_key, str(raw_label)) df = df.reset_index(drop=True) return df, display_by_key # --------------------------------------------------------------------- # Model label alignment # --------------------------------------------------------------------- def parse_numeric_label(label_key: str) -> Optional[int]: if re.fullmatch(r"[+-]?\d+", label_key): return int(label_key) return None def is_generic_model_label(label: str, class_id: int) -> bool: value = canonical_label(label) return value in { f"label_{class_id}", f"label-{class_id}", f"class_{class_id}", str(class_id), } def get_model_id2label(config) -> Dict[int, str]: """Read the model's ID-to-label mapping.""" result = {} raw_mapping = getattr(config, "id2label", {}) or {} if isinstance(raw_mapping, dict): for key, value in raw_mapping.items(): try: result[int(key)] = str(value) except (TypeError, ValueError): continue if result: return result # Fallback to label2id if necessary. raw_reverse = getattr(config, "label2id", {}) or {} if isinstance(raw_reverse, dict): for label, key in raw_reverse.items(): try: result[int(key)] = str(label) except (TypeError, ValueError): continue return result def parse_label_map(raw_value: Optional[str]) -> Optional[Dict[str, int]]: """ Parse either: --label-map '{"negative": 0, "positive": 1}' or a JSON file containing the same mapping. """ if not raw_value: return None raw_value = raw_value.strip() if raw_value.startswith("{"): mapping = json.loads(raw_value) else: path = Path(raw_value) if path.exists(): mapping = json.loads( path.read_text(encoding="utf-8") ) else: mapping = json.loads(raw_value) if not isinstance(mapping, dict): raise ValueError("--label-map must contain a JSON object.") result = {} for label, class_id in mapping.items(): result[canonical_label(label)] = int(class_id) return result def resolve_label_mapping( data_label_keys: List[str], config, explicit_mapping: Optional[Dict[str, int]], ): """ Align dataset labels with model output IDs. Priority: 1. Explicit --label-map 2. Semantic labels in model.config.id2label 3. Numeric dataset labels 4. Alphabetical fallback if the number of labels matches The alphabetical fallback should not be used unless the model's class order is known. """ number_of_labels = int(getattr(config, "num_labels", 0) or 0) if number_of_labels < 2: raise ValueError( "This program supports single-label classification with at " "least two output classes." ) if explicit_mapping is not None: missing = [ label for label in data_label_keys if label not in explicit_mapping ] if missing: raise ValueError( f"--label-map is missing dataset labels: {missing}" ) mapping = { label: int(explicit_mapping[label]) for label in data_label_keys } source = "explicit" else: model_id2label = get_model_id2label(config) semantic_mapping = {} for class_id, label in model_id2label.items(): if not is_generic_model_label(label, class_id): semantic_mapping[canonical_label(label)] = class_id if all(label in semantic_mapping for label in data_label_keys): mapping = { label: int(semantic_mapping[label]) for label in data_label_keys } source = "model_config" else: numeric_values = [ parse_numeric_label(label) for label in data_label_keys ] if ( all(value is not None for value in numeric_values) and all( 0 <= int(value) < number_of_labels for value in numeric_values ) ): mapping = { label: int(value) for label, value in zip( data_label_keys, numeric_values, ) } source = "numeric_dataset_labels" elif len(data_label_keys) == number_of_labels: ordered = sorted(data_label_keys) mapping = { label: class_id for class_id, label in enumerate(ordered) } source = "alphabetical_fallback" warnings.warn( "The model has generic labels such as LABEL_0 and " "LABEL_1. Dataset labels were assigned alphabetically. " "Use --label-map to guarantee the correct class order." ) else: raise ValueError( "Could not align dataset labels with model output IDs. " "Use --label-map, for example: " '\'{"negative": 0, "positive": 1}\'' ) if len(set(mapping.values())) != len(mapping): raise ValueError( "Two dataset labels map to the same model class ID." ) invalid_ids = [ class_id for class_id in mapping.values() if class_id < 0 or class_id >= number_of_labels ] if invalid_ids: raise ValueError( f"Invalid model class IDs in label mapping: {invalid_ids}" ) return mapping, source # --------------------------------------------------------------------- # Metrics # --------------------------------------------------------------------- def finite(value): """Convert NumPy scalars to JSON-friendly floats.""" try: value = float(value) except (TypeError, ValueError): return None if not np.isfinite(value): return None return value def expected_calibration_error( y_true: np.ndarray, y_pred: np.ndarray, probabilities: np.ndarray, number_of_bins: int = 10, ): """Compute a simple expected calibration error.""" if len(y_true) == 0: return None confidence = probabilities.max(axis=1) correct = (y_true == y_pred).astype(float) bin_edges = np.linspace(0.0, 1.0, number_of_bins + 1) ece = 0.0 for bin_index in range(number_of_bins): lower = bin_edges[bin_index] upper = bin_edges[bin_index + 1] if bin_index == number_of_bins - 1: mask = (confidence >= lower) & (confidence <= upper) else: mask = (confidence >= lower) & (confidence < upper) if np.any(mask): ece += ( np.mean(mask) * abs( np.mean(confidence[mask]) - np.mean(correct[mask]) ) ) return finite(ece) def compute_metrics( y_true: np.ndarray, y_pred: np.ndarray, probabilities: np.ndarray, number_of_model_classes: int, target_label_ids: List[int], label_names: Dict[int, str], ): """ Compute aggregate and per-class metrics. Macro-F1 is calculated over labels present in the evaluation dataset. The confusion matrix and log loss use all model output classes. """ target_label_ids = [int(value) for value in target_label_ids] precision, recall, f1, support = ( precision_recall_fscore_support( y_true, y_pred, labels=target_label_ids, average=None, zero_division=0, ) ) try: balanced_accuracy = balanced_accuracy_score(y_true, y_pred) except ValueError: balanced_accuracy = None try: mcc = matthews_corrcoef(y_true, y_pred) except ValueError: mcc = None try: if number_of_model_classes >= 2: loss = log_loss( y_true, probabilities, labels=list(range(number_of_model_classes)), ) else: loss = None except ValueError: loss = None metrics = { "n": int(len(y_true)), "accuracy": finite(accuracy_score(y_true, y_pred)), "macro_precision": finite(np.mean(precision)), "macro_recall": finite(np.mean(recall)), "macro_f1": finite(np.mean(f1)), "weighted_f1": finite( f1_score( y_true, y_pred, labels=target_label_ids, average="weighted", zero_division=0, ) ), "balanced_accuracy": finite(balanced_accuracy), "matthews_corrcoef": finite(mcc), "log_loss": finite(loss), "ece": expected_calibration_error( y_true, y_pred, probabilities, ), "n_classes_evaluated": int(len(target_label_ids)), } per_class = [] for position, class_id in enumerate(target_label_ids): per_class.append( { "class_id": int(class_id), "label": label_names.get( class_id, f"class_{class_id}", ), "precision": finite(precision[position]), "recall": finite(recall[position]), "f1": finite(f1[position]), "support": int(support[position]), } ) confusion = confusion_matrix( y_true, y_pred, labels=list(range(number_of_model_classes)), ).tolist() return metrics, per_class, confusion def make_group_record( group_type: str, group_values: Dict[str, str], indices: np.ndarray, y_true: np.ndarray, y_pred: np.ndarray, probabilities: np.ndarray, number_of_model_classes: int, target_label_ids: List[int], label_names: Dict[int, str], ): metrics, per_class, confusion = compute_metrics( y_true=y_true[indices], y_pred=y_pred[indices], probabilities=probabilities[indices], number_of_model_classes=number_of_model_classes, target_label_ids=target_label_ids, label_names=label_names, ) return { "group_type": group_type, "group": group_values, "metrics": metrics, "per_class": per_class, "confusion_matrix": confusion, } def summarize_robustness( summary_rows: List[Dict], minimum_group_size: int, ): """ Summarize worst-group performance and performance gaps. Groups below minimum_group_size are excluded from worst-group summaries, but their raw metrics remain in summary.csv and the model JSON file. """ result = { "min_group_size": int(minimum_group_size), } overall = next( ( row for row in summary_rows if row["group_type"] == "overall" ), None, ) if overall is not None: for metric in ["accuracy", "macro_f1", "weighted_f1"]: result[f"overall_{metric}"] = overall.get(metric) for group_type in ["domain", "script", "domain_script"]: eligible = [ row for row in summary_rows if row["group_type"] == group_type and int(row["n"]) >= minimum_group_size ] result[f"{group_type}_n_eligible"] = len(eligible) for metric in ["accuracy", "macro_f1", "balanced_accuracy"]: values = [ float(row[metric]) for row in eligible if row.get(metric) is not None ] if not values: continue result[f"{group_type}_mean_{metric}"] = finite( np.mean(values) ) result[f"{group_type}_best_{metric}"] = finite( max(values) ) result[f"{group_type}_worst_{metric}"] = finite( min(values) ) result[f"{group_type}_gap_{metric}"] = finite( max(values) - min(values) ) # Compare pooled Nastaliq and Roman Urdu performance. eligible_script_rows = [ row for row in summary_rows if row["group_type"] == "script" and int(row["n"]) >= minimum_group_size ] by_script = { row["script"]: row for row in eligible_script_rows } if "nastaliq" in by_script and "roman_urdu" in by_script: nastaliq_f1 = by_script["nastaliq"].get("macro_f1") roman_f1 = by_script["roman_urdu"].get("macro_f1") if nastaliq_f1 is not None and roman_f1 is not None: result["nastaliq_minus_roman_urdu_macro_f1"] = finite( nastaliq_f1 - roman_f1 ) result["nastaliq_roman_urdu_macro_f1_gap"] = finite( abs(nastaliq_f1 - roman_f1) ) return result # --------------------------------------------------------------------- # Model evaluation # --------------------------------------------------------------------- def get_device(requested: str) -> torch.device: if requested.casefold() == "auto": if torch.cuda.is_available(): return torch.device("cuda") if ( hasattr(torch.backends, "mps") and torch.backends.mps.is_available() ): return torch.device("mps") return torch.device("cpu") device = torch.device(requested) if device.type == "cuda" and not torch.cuda.is_available(): raise RuntimeError("CUDA was requested but is not available.") return device def get_allowed_forward_arguments(model): """ Remove tokenizer fields not accepted by a model's forward method. """ try: signature = inspect.signature(model.forward) parameters = signature.parameters.values() if any( parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in parameters ): return None return set(signature.parameters.keys()) except (TypeError, ValueError): return None def evaluate_model( model_name: str, df: pd.DataFrame, display_by_key: Dict[str, str], args: argparse.Namespace, device: torch.device, explicit_label_map: Optional[Dict[str, int]], ): print(f"\nLoading model: {model_name}") tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForSequenceClassification.from_pretrained( model_name ) if getattr(model.config, "problem_type", None) == ( "multi_label_classification" ): raise ValueError( f"{model_name} is configured for multi-label classification. " "This program expects single-label classification." ) if tokenizer.pad_token is None: if tokenizer.eos_token is None: raise ValueError( f"Tokenizer for {model_name} has no pad or EOS token." ) tokenizer.pad_token = tokenizer.eos_token model.config.pad_token_id = tokenizer.pad_token_id data_label_keys = list( dict.fromkeys(df["__label_key"].tolist()) ) label_mapping, mapping_source = resolve_label_mapping( data_label_keys=data_label_keys, config=model.config, explicit_mapping=explicit_label_map, ) work = df.copy() work["__true_id"] = [ int(label_mapping[label_key]) for label_key in work["__label_key"].tolist() ] y_true = work["__true_id"].to_numpy(dtype=int) texts = work["__text_eval"].tolist() model.to(device) model.eval() allowed_arguments = get_allowed_forward_arguments(model) probability_batches = [] for start in range(0, len(texts), args.batch_size): batch_texts = texts[start : start + args.batch_size] encoded = tokenizer( batch_texts, padding=True, truncation=True, max_length=args.max_length, return_tensors="pt", ) if allowed_arguments is not None: encoded = { key: value for key, value in encoded.items() if key in allowed_arguments } encoded = { key: value.to(device) for key, value in encoded.items() } with torch.inference_mode(): output = model(**encoded) logits = output.logits if logits.ndim != 2: raise ValueError( f"{model_name} did not return 2D classification logits." ) probabilities = torch.softmax( logits.float(), dim=-1, ).cpu().numpy() probability_batches.append(probabilities) probabilities = np.concatenate(probability_batches, axis=0) number_of_model_classes = int(probabilities.shape[1]) if number_of_model_classes < 2: raise ValueError( f"{model_name} returned only one output class. " "This program supports classification, not regression." ) invalid_mapping_ids = [ class_id for class_id in label_mapping.values() if class_id >= number_of_model_classes ] if invalid_mapping_ids: raise ValueError( f"Label mapping contains IDs not returned by {model_name}: " f"{invalid_mapping_ids}" ) y_pred = probabilities.argmax(axis=1).astype(int) target_label_ids = sorted(set(y_true.tolist())) model_id2label = get_model_id2label(model.config) label_names = { int(model_id): display_by_key[data_label_key] for data_label_key, model_id in label_mapping.items() } # Add predictions to a working dataframe. work["__pred_id"] = y_pred work["__confidence"] = probabilities.max(axis=1) records = [] # Overall metrics. records.append( make_group_record( group_type="overall", group_values={ "domain": "ALL", "script": "ALL", }, indices=np.arange(len(work), dtype=int), y_true=y_true, y_pred=y_pred, probabilities=probabilities, number_of_model_classes=number_of_model_classes, target_label_ids=target_label_ids, label_names=label_names, ) ) # Domain, script, and domain-by-script metrics. group_specs = [ ("domain", ["__domain"]), ("script", ["__script"]), ("domain_script", ["__domain", "__script"]), ] for group_type, columns in group_specs: grouped = work.groupby( columns, dropna=False, sort=True, ) for keys, part in grouped: if not isinstance(keys, tuple): keys = (keys,) group_values = { "domain": "ALL", "script": "ALL", } for column, key in zip(columns, keys): output_key = ( "domain" if column == "__domain" else "script" ) group_values[output_key] = str(key) records.append( make_group_record( group_type=group_type, group_values=group_values, indices=part.index.to_numpy(dtype=int), y_true=y_true, y_pred=y_pred, probabilities=probabilities, number_of_model_classes=number_of_model_classes, target_label_ids=target_label_ids, label_names=label_names, ) ) summary_rows = [] for record in records: row = { "model": model_name, "group_type": record["group_type"], } row.update(record["group"]) row.update(record["metrics"]) summary_rows.append(row) robustness = summarize_robustness( summary_rows=summary_rows, minimum_group_size=args.min_group_size, ) def display_prediction(class_id: int) -> str: if class_id in label_names: return label_names[class_id] return model_id2label.get( class_id, f"class_{class_id}", ) predictions = pd.DataFrame( { "row_id": work["__row_id"].tolist(), "text": work["__text_original"].tolist(), "model_input": work["__text_eval"].tolist(), "domain": work["__domain"].tolist(), "script": work["__script"].tolist(), "true_label": [ display_by_key[key] for key in work["__label_key"].tolist() ], "true_label_id": y_true.tolist(), "predicted_label": [ display_prediction(int(class_id)) for class_id in y_pred ], "predicted_label_id": y_pred.tolist(), "confidence": work["__confidence"].tolist(), } ) details = { "model": model_name, "n_examples": int(len(work)), "label_mapping_source": mapping_source, "data_labels": [ { "key": label_key, "display": display_by_key[label_key], "model_id": int(label_mapping[label_key]), } for label_key in data_label_keys ], "model_id2label": { str(class_id): label for class_id, label in model_id2label.items() }, "groups": records, "robustness": robustness, } del model del tokenizer if device.type == "cuda": torch.cuda.empty_cache() return { "details": details, "predictions": predictions, "summary_rows": summary_rows, } # --------------------------------------------------------------------- # Command-line interface # --------------------------------------------------------------------- def build_argument_parser(): parser = argparse.ArgumentParser( description=( "Evaluate multilingual classification models on " "Nastaliq and Roman Urdu across domains." ) ) parser.add_argument( "--data", required=True, help="CSV, JSONL, JSON, TSV, or Parquet input file.", ) parser.add_argument( "--models", nargs="+", required=True, help=( "Hugging Face model names or local model directories. " "Models must have sequence-classification heads." ), ) parser.add_argument( "--output-dir", default="urdu_robustness_results", ) parser.add_argument("--text-col", default="text") parser.add_argument("--label-col", default="label") parser.add_argument("--domain-col", default="domain") parser.add_argument("--script-col", default="script") parser.add_argument( "--split-col", default=None, help="Optional split column, such as train/dev/test.", ) parser.add_argument( "--split-value", default="test", help="Value selected from --split-col. Default: test.", ) parser.add_argument( "--scripts", nargs="+", default=["nastaliq", "roman_urdu"], help="Scripts to evaluate.", ) parser.add_argument( "--normalize", action="store_true", help="Apply conservative Unicode and Urdu normalization.", ) parser.add_argument( "--label-map", default=None, help=( "JSON string or JSON file mapping dataset labels to model IDs. " 'Example: \'{"negative": 0, "positive": 1}\'' ), ) parser.add_argument("--batch-size", type=int, default=16) parser.add_argument("--max-length", type=int, default=256) parser.add_argument( "--min-group-size", type=int, default=10, help=( "Minimum examples for worst-group summaries. " "All groups are still written to summary.csv." ), ) parser.add_argument( "--device", default="auto", help="auto, cpu, cuda, cuda:0, or mps.", ) parser.add_argument( "--no-predictions", action="store_true", help="Do not write per-example prediction files.", ) return parser def safe_model_filename(model_name: str) -> str: name = model_name.rstrip("/").replace("/", "__") name = re.sub(r"[^A-Za-z0-9_.-]+", "_", name) return name or "model" def format_metric(value) -> str: if value is None: return "n/a" return f"{float(value):.4f}" def main(): parser = build_argument_parser() args = parser.parse_args() if args.batch_size < 1: raise ValueError("--batch-size must be positive.") if args.max_length < 1: raise ValueError("--max-length must be positive.") dataframe = load_table(args.data) df, display_by_key = prepare_data(dataframe, args) explicit_label_map = parse_label_map(args.label_map) device = get_device(args.device) output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) print(f"Loaded {len(df)} examples.") print(f"Domains: {sorted(df['__domain'].unique().tolist())}") print(f"Scripts: {sorted(df['__script'].unique().tolist())}") print(f"Device: {device}") all_summary_rows = [] robustness_rows = [] used_filenames = set() for model_name in args.models: result = evaluate_model( model_name=model_name, df=df, display_by_key=display_by_key, args=args, device=device, explicit_label_map=explicit_label_map, ) filename = safe_model_filename(model_name) original_filename = filename counter = 2 while filename in used_filenames: filename = f"{original_filename}_{counter}" counter += 1 used_filenames.add(filename) details_path = output_dir / f"{filename}.json" details_path.write_text( json.dumps( result["details"], ensure_ascii=False, indent=2, ), encoding="utf-8", ) if not args.no_predictions: prediction_path = ( output_dir / f"{filename}_predictions.csv" ) result["predictions"].to_csv( prediction_path, index=False, encoding="utf-8", ) all_summary_rows.extend(result["summary_rows"]) robustness_row = {"model": model_name} robustness_row.update(result["details"]["robustness"]) robustness_rows.append(robustness_row) overall = next( row for row in result["summary_rows"] if row["group_type"] == "overall" ) worst_group = result["details"]["robustness"].get( "domain_script_worst_macro_f1" ) print( f"{model_name}: " f"overall macro-F1={format_metric(overall['macro_f1'])}, " f"worst domain×script macro-F1=" f"{format_metric(worst_group)}" ) summary_df = pd.DataFrame(all_summary_rows) summary_df.to_csv( output_dir / "summary.csv", index=False, encoding="utf-8", ) robustness_df = pd.DataFrame(robustness_rows) robustness_df.to_csv( output_dir / "robustness.csv", index=False, encoding="utf-8", ) print(f"\nWrote results to: {output_dir.resolve()}") print(" summary.csv - metrics for every group") print(" robustness.csv - worst-group scores and gaps") print(" *.json - per-class metrics and confusion matrices") if __name__ == "__main__": main() ``` ### Example input ```csv text,label,domain,script,split "یہ فلم بہت اچھی ہے",positive,reviews,nastaliq,test "یہ فلم بالکل بری ہے",negative,reviews,nastaliq,test "ye film bohat achi hai",positive,reviews,roman_urdu,test "ye film bilkul buri hai",negative,reviews,roman_urdu,test "یہ خبر بہت اہم ہے",positive,news,nastaliq,test "ye khabar bohat aham hai",positive,news,roman_urdu,test ``` ### Run the benchmark ```bash python urdu_domain_robustness.py \ --data test.csv \ --models path/to/model_1 path/to/model_2 \ --split-col split \ --split-value test \ --output-dir results \ --batch-size 16 \ --max-length 256 \ --min-group-size 10 ``` With explicit label ordering: ```bash python urdu_domain_robustness.py \ --data test.csv \ --models path/to/model \ --label-map '{"negative": 0, "positive": 1}' \ --output-dir results ``` The most useful fields in `robustness.csv` are: - `domain_script_worst_macro_f1` - `domain_script_gap_macro_f1` - `nastaliq_roman_urdu_macro_f1_gap` Higher worst-group macro-F1 and smaller gaps indicate better robustness. For a valid domain-generalization test, evaluate on domains that were held out during model training; a random row split can leak domain-specific vocabulary.

Urdu Comparison