Spaces:
Running on Zero
Running on Zero
| """ | |
| DEVIPUTRA AI // PURE PYTORCH & TRANSFORMERS ENGINE | |
| Fine-tunes Multilingual HingHate Transformer on Code-Mixed Hinglish Slangs & Hate Speech. | |
| Features: | |
| - Pure Python & PyTorch (zero external DLL blocks from pandas/sklearn) | |
| - Supports PyTorch CUDA / Tensor Cores / CPU execution | |
| - Stratified 85/15 Train-Val split with F1-Score & Accuracy tracking | |
| - Saves production-ready checkpoint into models/hing_hate_bert_rtx5050/ | |
| """ | |
| import os | |
| import sys | |
| import csv | |
| import random | |
| from typing import Dict, List, Tuple | |
| if hasattr(sys.stdout, 'reconfigure'): | |
| try: | |
| sys.stdout.reconfigure(encoding='utf-8') | |
| except Exception: | |
| pass | |
| import torch | |
| LABEL_TO_ID = {"Normal": 0, "Offensive": 1, "Hate Speech": 2} | |
| ID_TO_LABEL = {0: "Normal", 1: "Offensive", 2: "Hate Speech"} | |
| class HinglishTextDataset(torch.utils.data.Dataset): | |
| def __init__(self, texts: List[str], labels: List[int], tokenizer, max_length: int = 128): | |
| self.texts = list(texts) | |
| self.labels = list(labels) | |
| self.tokenizer = tokenizer | |
| self.max_length = max_length | |
| def __len__(self): | |
| return len(self.texts) | |
| def __getitem__(self, idx): | |
| text = str(self.texts[idx]) | |
| label = self.labels[idx] | |
| encoding = self.tokenizer( | |
| text, | |
| truncation=True, | |
| padding='max_length', | |
| max_length=self.max_length, | |
| return_tensors='pt' | |
| ) | |
| return { | |
| 'input_ids': encoding['input_ids'].squeeze(0), | |
| 'attention_mask': encoding['attention_mask'].squeeze(0), | |
| 'label': torch.tensor(label, dtype=torch.long) | |
| } | |
| def load_data_clean(filepath: str) -> Tuple[List[str], List[int]]: | |
| texts, labels = [], [] | |
| with open(filepath, 'r', encoding='utf-8', errors='ignore') as f: | |
| reader = csv.reader(f) | |
| header = next(reader, None) | |
| for row in reader: | |
| if len(row) >= 2: | |
| txt = row[0].strip() | |
| lbl = row[1].strip() | |
| if txt and lbl in LABEL_TO_ID: | |
| texts.append(txt) | |
| labels.append(LABEL_TO_ID[lbl]) | |
| return texts, labels | |
| def stratified_split(texts: List[str], labels: List[int], val_ratio: float = 0.15, seed: int = 42): | |
| random.seed(seed) | |
| class_groups = {0: [], 1: [], 2: []} | |
| for t, l in zip(texts, labels): | |
| class_groups[l].append(t) | |
| train_texts, train_labels = [], [] | |
| val_texts, val_labels = [], [] | |
| for l, group in class_groups.items(): | |
| random.shuffle(group) | |
| val_count = max(1, int(len(group) * val_ratio)) | |
| val_texts.extend(group[:val_count]) | |
| val_labels.extend([l] * val_count) | |
| train_texts.extend(group[val_count:]) | |
| train_labels.extend([l] * (len(group) - val_count)) | |
| # Shuffle training items | |
| combined = list(zip(train_texts, train_labels)) | |
| random.shuffle(combined) | |
| train_texts, train_labels = zip(*combined) | |
| return list(train_texts), list(train_labels), val_texts, val_labels | |
| def compute_metrics(preds: List[int], targets: List[int]) -> Dict[str, float]: | |
| total = len(targets) | |
| if total == 0: | |
| return {"accuracy": 0.0, "f1": 0.0} | |
| correct = sum(1 for p, y in zip(preds, targets) if p == y) | |
| acc = correct / total | |
| # Weighted F1 | |
| f1_weighted = 0.0 | |
| for c in [0, 1, 2]: | |
| tp = sum(1 for p, y in zip(preds, targets) if p == c and y == c) | |
| fp = sum(1 for p, y in zip(preds, targets) if p == c and y != c) | |
| fn = sum(1 for p, y in zip(preds, targets) if p != c and y == c) | |
| support = sum(1 for y in targets if y == c) | |
| prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0 | |
| rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0 | |
| f1_class = (2 * prec * rec / (prec + rec)) if (prec + rec) > 0 else 0.0 | |
| f1_weighted += f1_class * (support / total) | |
| return {"accuracy": round(acc, 4), "f1": round(f1_weighted, 4)} | |
| def train_rtx5050( | |
| data_path: str = "data/master_train_dataset.csv", | |
| base_model_name: str = "l3cube-pune/hing-bert", | |
| output_dir: str = "models/hing_hate_bert_rtx5050", | |
| epochs: int = 3, | |
| batch_size: int = 16, | |
| lr: float = 2e-5 | |
| ): | |
| print("=" * 70) | |
| print("DEVIPUTRA AI // HINGLISH TRANSFORMER FINE-TUNING PIPELINE") | |
| print("=" * 70) | |
| # 1. Device and Architecture Detection | |
| if torch.cuda.is_available(): | |
| gpu_name = torch.cuda.get_device_name(0) | |
| vram_gb = torch.cuda.get_device_properties(0).total_memory / (1024**3) | |
| device = torch.device("cuda:0") | |
| print(f"[GPU ENGINE ACTIVE]: {gpu_name} ({vram_gb:.2f} GB VRAM)") | |
| else: | |
| device = torch.device("cpu") | |
| print("[DEVICE]: CPU (Optimized for lightweight inference)") | |
| # 2. Load Dataset using pure csv module | |
| if not os.path.exists(data_path): | |
| data_path = os.path.join("data", "sample.csv") | |
| print(f"Loading Hinglish training corpus: {data_path}...") | |
| texts, labels = load_data_clean(data_path) | |
| print(f"Total samples: {len(texts)} samples loaded.") | |
| train_texts, train_labels, val_texts, val_labels = stratified_split(texts, labels, val_ratio=0.15) | |
| print(f"Split: {len(train_texts)} train items | {len(val_texts)} validation items.") | |
| # 3. Load HuggingFace Tokenizer & Model | |
| from transformers import AutoTokenizer, AutoModelForSequenceClassification | |
| from torch.optim import AdamW | |
| print(f"Loading transformer backbone ({base_model_name})...") | |
| tokenizer = AutoTokenizer.from_pretrained(base_model_name) | |
| model = AutoModelForSequenceClassification.from_pretrained( | |
| base_model_name, | |
| num_labels=3, | |
| id2label=ID_TO_LABEL, | |
| label2id=LABEL_TO_ID | |
| ) | |
| model.to(device) | |
| train_ds = HinglishTextDataset(train_texts, train_labels, tokenizer) | |
| val_ds = HinglishTextDataset(val_texts, val_labels, tokenizer) | |
| train_loader = torch.utils.data.DataLoader( | |
| train_ds, batch_size=batch_size, shuffle=True | |
| ) | |
| val_loader = torch.utils.data.DataLoader( | |
| val_ds, batch_size=batch_size, shuffle=False | |
| ) | |
| optimizer = AdamW(model.parameters(), lr=lr, weight_decay=0.01) | |
| best_val_f1 = 0.0 | |
| os.makedirs(output_dir, exist_ok=True) | |
| print("\nSTARTING TRAINING LOOPS...") | |
| for epoch in range(1, epochs + 1): | |
| model.train() | |
| total_loss = 0.0 | |
| for step, batch in enumerate(train_loader): | |
| input_ids = batch['input_ids'].to(device) | |
| attention_mask = batch['attention_mask'].to(device) | |
| target_labels = batch['label'].to(device) | |
| optimizer.zero_grad() | |
| outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=target_labels) | |
| loss = outputs.loss | |
| loss.backward() | |
| optimizer.step() | |
| total_loss += loss.item() | |
| if (step + 1) % 10 == 0 or (step + 1) == len(train_loader): | |
| print(f" Epoch [{epoch}/{epochs}] | Step [{step+1}/{len(train_loader)}] | Batch Loss: {loss.item():.4f}") | |
| avg_train_loss = total_loss / len(train_loader) | |
| # Validation Phase | |
| model.eval() | |
| val_preds, val_targets = [], [] | |
| with torch.no_grad(): | |
| for batch in val_loader: | |
| input_ids = batch['input_ids'].to(device) | |
| attention_mask = batch['attention_mask'].to(device) | |
| target_labels = batch['label'] | |
| outputs = model(input_ids=input_ids, attention_mask=attention_mask) | |
| preds = torch.argmax(outputs.logits, dim=1).cpu().tolist() | |
| val_preds.extend(preds) | |
| val_targets.extend(target_labels.tolist()) | |
| metrics = compute_metrics(val_preds, val_targets) | |
| val_acc = metrics["accuracy"] | |
| val_f1 = metrics["f1"] | |
| print(f"\nEPOCH {epoch} SUMMARY:") | |
| print(f" Train Loss : {avg_train_loss:.4f}") | |
| print(f" Validation Acc : {val_acc * 100:.2f}%") | |
| print(f" Validation F1 : {val_f1:.4f}") | |
| print("-" * 50) | |
| # Save Best Checkpoint | |
| if val_f1 >= best_val_f1: | |
| best_val_f1 = val_f1 | |
| print(f"Saving checkpoint to: '{output_dir}'...") | |
| model.save_pretrained(output_dir) | |
| tokenizer.save_pretrained(output_dir) | |
| print("\nTRAINING COMPLETE!") | |
| print(f"Best Validation F1: {best_val_f1:.4f}") | |
| print(f"Model saved to: {output_dir}") | |
| print("=" * 70) | |
| if __name__ == "__main__": | |
| import argparse | |
| parser = argparse.ArgumentParser(description="Hinglish Fine-Tuning Script") | |
| parser.add_argument("--epochs", type=int, default=2) | |
| parser.add_argument("--batch_size", type=int, default=16) | |
| parser.add_argument("--lr", type=float, default=2e-5) | |
| args = parser.parse_args() | |
| train_rtx5050(epochs=args.epochs, batch_size=args.batch_size, lr=args.lr) | |