Spaces:
Runtime error
Runtime error
File size: 7,170 Bytes
bde1c71 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 |
import torch
from torch.utils.data import DataLoader
from datasets import concatenate_datasets, load_dataset,Dataset
from typing import Dict, Any, List, Optional
import warnings
from ..base_evaluator import BaseEvaluator
SUPPORTED_DATASETS = {
"tsac": {
"path": "tunis-ai/tsac",
"text_column": "sentence",
"label_column": "target",
"label_map": {0: 0, 1: 1}, # already binary
"trust_remote_code": True,
"split": "test"
},
}
class SentimentAnalysisEvaluator(BaseEvaluator):
"""
Unified evaluator for Tunisian sentiment analysis.
Supports multiple datasets, harmonizes labels to binary (0=neg, 1=pos).
Neutral/mapped-to-invalid labels are filtered out.
"""
def __init__(
self,
datasets: Optional[List[str]] = None,
max_samples_per_dataset: int = 500,
batch_size: int = 16
):
"""
Args:
datasets: List of dataset keys from SUPPORTED_DATASETS.
If None, uses all available.
max_samples_per_dataset: Limit samples per dataset for faster eval.
batch_size: Inference batch size.
"""
if datasets is None:
self.dataset_keys = list(SUPPORTED_DATASETS.keys())
else:
for d in datasets:
if d not in SUPPORTED_DATASETS:
raise ValueError(f"Dataset '{d}' not in supported list: {list(SUPPORTED_DATASETS.keys())}")
self.dataset_keys = datasets
self.max_samples_per_dataset = max_samples_per_dataset
self.batch_size = batch_size
@property
def task_name(self) -> str:
return "Sentiment Analysis"
def load_dataset(self) -> Dataset:
"""Load and harmonize all configured sentiment datasets."""
print("\n=== Loading Tunisian Sentiment Datasets ===")
all_datasets = []
for key in self.dataset_keys:
cfg = SUPPORTED_DATASETS[key]
print(f"\nLoading '{key}': {cfg.get('description', "No description available.")}")
try:
ds = load_dataset(
cfg["path"],
split=cfg["split"],
trust_remote_code=cfg.get("trust_remote_code", False)
)
print(f" Raw size: {len(ds)}")
except Exception as e:
warnings.warn(f"Failed to load {key}: {e}. Skipping.")
continue
# Harmonize to {"text": str, "label": int in {0,1}}
def harmonize(example):
# print(cfg)
try:
text = example[cfg["text_column"]]
orig_label = example[cfg["label_column"]]
if orig_label not in cfg["label_map"]:
return None
new_label = cfg["label_map"][orig_label]
if new_label not in [0, 1]:
return None # skip neutral/invalid
return {"text": text, "label": new_label}
except Exception:
return None
print(" Harmonizing and filtering...")
ds = ds.map(
harmonize,
load_from_cache_file=False,
desc=f"Harmonizing {key}"
)
# print(ds)
print(" Filtering invalid/neutral samples...")
ds = ds.filter(lambda x: x is not None, load_from_cache_file=False)
print(f" Valid binary samples: {len(ds)}")
if self.max_samples_per_dataset and len(ds) > self.max_samples_per_dataset:
ds = ds.select(range(self.max_samples_per_dataset))
print(f" Trimmed to {self.max_samples_per_dataset} samples")
if len(ds) > 0:
all_datasets.append(ds)
if not all_datasets:
raise ValueError("No valid sentiment data found!")
# Combine all datasets
combined = concatenate_datasets(all_datasets)
print(f"\n✅ Total Tunisian sentiment samples: {len(combined)}")
return combined
def _tokenize_batch(self, examples, tokenizer):
return tokenizer(
examples["sentence"],
padding=True,
truncation=True,
max_length=512,
return_tensors=None
)
def _collate_fn(self, batch):
input_ids = torch.stack([torch.tensor(b["input_ids"]) for b in batch])
attention_mask = torch.stack([torch.tensor(b["attention_mask"]) for b in batch])
labels = torch.tensor([b["labels"] for b in batch], dtype=torch.long)
return {
"input_ids": input_ids,
"attention_mask": attention_mask,
"labels": labels
}
def evaluate(self, model, tokenizer, device: str = "cuda") -> Dict[str, Any]:
"""Evaluate model on unified Tunisian sentiment task."""
print(f"\n=== Evaluating {self.task_name} ===")
print(f"Model: {model.__class__.__name__} | Device: {device}")
print(f"Datasets: {self.dataset_keys}")
# Load and prepare data
raw_dataset = self.load_dataset()
tokenized = raw_dataset.map(
lambda ex: self._tokenize_batch(ex, tokenizer),
batched=True,
remove_columns=raw_dataset.column_names
)
tokenized.set_format(type="torch", columns=["input_ids", "attention_mask"])
tokenized = tokenized.add_column("labels", raw_dataset["label"])
print(tokenized.column_names)
dataloader = DataLoader(
tokenized,
batch_size=self.batch_size,
shuffle=False,
collate_fn=self._collate_fn
)
# Inference
model.eval()
all_preds, all_labels = [], []
with torch.no_grad():
for i, batch in enumerate(dataloader):
inputs = {
k: v.to(device) for k, v in batch.items()
if k in ["input_ids", "attention_mask"]
}
labels = batch["labels"].to(device)
outputs = model(**inputs)
logits = outputs.logits if hasattr(outputs, "logits") else outputs[0]
if logits.dim() == 3: # [B, L, C]
logits = logits[:, 0, :]
preds = logits.argmax(dim=-1).cpu().tolist()
trues = labels.cpu().tolist()
all_preds.extend(preds)
all_labels.extend(trues)
# Metrics
correct = sum(p == t for p, t in zip(all_preds, all_labels))
total = len(all_preds)
accuracy = correct / total if total > 0 else 0.0
print(f"\n✅ {self.task_name} Results:")
print(f" Accuracy: {accuracy:.4f} ({correct}/{total})")
return {
"task": self.task_name,
"accuracy": accuracy,
"main_metric": accuracy,
"total_samples": total,
"datasets_used": self.dataset_keys
} |