Deeptanshuu's picture
Upload folder using huggingface_hub
d187b57 verified
# train.py
import pandas as pd
import torch
import logging
import os
import gc
import wandb
from datetime import datetime
import signal
import atexit
import sys
from pathlib import Path
import numpy as np
import warnings
import json
from tqdm import tqdm
import torch.nn as nn
import torch.nn.functional as F
import time
from transformers import (
XLMRobertaTokenizer
)
from torch.utils.data import DataLoader
from model.evaluation.evaluate import ToxicDataset
from model.training_config import MetricsTracker, TrainingConfig
from model.data.sampler import MultilabelStratifiedSampler
from model.language_aware_transformer import LanguageAwareTransformer
from torch.optim.lr_scheduler import CosineAnnealingWarmRestarts
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f'logs/train_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Set environment variables if not already set
os.environ['TF_CPP_MIN_LOG_LEVEL'] = os.environ.get('TF_CPP_MIN_LOG_LEVEL', '2')
warnings.filterwarnings("ignore", message="Was asked to gather along dimension 0")
warnings.filterwarnings("ignore", message="AVX2 detected")
# Initialize global variables with None
_model = None
_optimizer = None
_scheduler = None
_cleanup_handlers = []
def register_cleanup(handler):
"""Register cleanup handlers that will be called on exit"""
_cleanup_handlers.append(handler)
def cleanup():
"""Cleanup function to be called on exit"""
global _model, _optimizer, _scheduler
print("\nPerforming cleanup...")
for handler in _cleanup_handlers:
try:
handler()
except Exception as e:
print(f"Warning: Cleanup handler failed: {str(e)}")
if torch.cuda.is_available():
try:
torch.cuda.empty_cache()
except Exception as e:
print(f"Warning: Could not clear CUDA cache: {str(e)}")
try:
if _model is not None:
del _model
if _optimizer is not None:
del _optimizer
if _scheduler is not None:
del _scheduler
except Exception as e:
print(f"Warning: Error during cleanup: {str(e)}")
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
# Register cleanup handlers
atexit.register(cleanup)
def signal_handler(signum, frame):
print(f"\nReceived signal {signum}. Cleaning up...")
cleanup()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
def init_model(config):
"""Initialize model with error handling"""
global _model
try:
_model = LanguageAwareTransformer(
num_labels=config.num_labels,
hidden_size=config.hidden_size,
num_attention_heads=config.num_attention_heads,
model_name=config.model_name,
dropout=config.model_dropout
)
assert config.hidden_size == 1024, "XLM-R hidden size must be 1024"
assert _model.base_model.config.num_attention_heads == 16, "Head count mismatch"
if config.freeze_layers > 0:
for param in list(_model.base_model.parameters())[:8]:
param.requires_grad = False
assert not any([p.requires_grad for p in _model.base_model.parameters()][:8]), "First 8 layers should be frozen"
# Enhanced gradient checkpointing setup
if config.activation_checkpointing:
logger.info("Enabling gradient checkpointing for memory efficiency")
_model.gradient_checkpointing = True
_model.base_model.gradient_checkpointing_enable()
_model.base_model._set_gradient_checkpointing(enable=True)
# Verify checkpointing is enabled
assert _model.base_model.is_gradient_checkpointing, "Gradient checkpointing failed to enable"
_model = _model.to(config.device)
return _model
except Exception as e:
logger.error(f"Fatal error initializing model: {str(e)}")
raise
def get_grad_stats(model):
"""Calculate gradient statistics for monitoring"""
try:
grad_norms = []
grad_means = []
grad_maxs = []
grad_mins = []
param_names = []
for name, param in model.named_parameters():
if param.grad is not None:
grad = param.grad
grad_norm = grad.norm().item()
grad_norms.append(grad_norm)
grad_means.append(grad.mean().item())
grad_maxs.append(grad.max().item())
grad_mins.append(grad.min().item())
param_names.append(name)
if grad_norms:
return {
'grad/max_norm': max(grad_norms),
'grad/min_norm': min(grad_norms),
'grad/mean_norm': sum(grad_norms) / len(grad_norms),
'grad/max_value': max(grad_maxs),
'grad/min_value': min(grad_mins),
'grad/mean_value': sum(grad_means) / len(grad_means),
'grad/largest_layer': param_names[grad_norms.index(max(grad_norms))],
'grad/smallest_layer': param_names[grad_norms.index(min(grad_norms))]
}
return {}
except Exception as e:
logger.warning(f"Error calculating gradient stats: {str(e)}")
return {}
class LanguageAwareFocalLoss(nn.Module):
def __init__(self, reduction='mean'):
super().__init__()
self.reduction = reduction
def forward(self, inputs, targets, lang_weights=None, alpha=None, gamma=None):
"""
Compute focal loss with language-aware weighting and per-class parameters
Args:
inputs: Model predictions [batch_size, num_classes]
targets: Target labels [batch_size, num_classes]
lang_weights: Optional language weights [batch_size, num_classes]
alpha: Optional class-wise weight factor [num_classes] or [batch_size, num_classes]
gamma: Optional focusing parameter [num_classes] or [batch_size, num_classes]
"""
if alpha is None:
alpha = torch.full_like(inputs, 0.25)
if gamma is None:
gamma = torch.full_like(inputs, 2.0)
# Ensure alpha and gamma have correct shape [batch_size, num_classes]
if alpha.dim() == 1:
alpha = alpha.unsqueeze(0).expand(inputs.size(0), -1)
if gamma.dim() == 1:
gamma = gamma.unsqueeze(0).expand(inputs.size(0), -1)
# Compute binary cross entropy without reduction
bce_loss = F.binary_cross_entropy_with_logits(
inputs, targets, reduction='none'
)
# Compute probabilities for focusing
pt = torch.exp(-bce_loss) # [batch_size, num_classes]
# Compute focal weights with per-class gamma
focal_weights = (1 - pt) ** gamma # [batch_size, num_classes]
# Apply alpha weighting per-class
weighted_focal_loss = alpha * focal_weights * bce_loss
# Apply language-specific weights if provided
if lang_weights is not None:
weighted_focal_loss = weighted_focal_loss * lang_weights
# Reduce if needed
if self.reduction == 'mean':
return weighted_focal_loss.mean()
elif self.reduction == 'sum':
return weighted_focal_loss.sum()
return weighted_focal_loss
def training_step(batch, model, optimizer, scheduler, config, scaler, batch_idx):
"""Execute a single training step with gradient accumulation"""
# Move batch to device
batch = {k: v.to(config.device) if isinstance(v, torch.Tensor) else v
for k, v in batch.items()}
# Calculate language weights and focal parameters
lang_weights = None
alpha = None
gamma = None
if hasattr(config, 'lang_weights') and config.lang_weights is not None:
weight_dict = config.lang_weights.get_weights_for_batch(
[lang.item() for lang in batch['lang']],
batch['labels'],
config.device
)
lang_weights = weight_dict['weights'] # [batch_size, num_classes]
alpha = weight_dict['alpha'] # [num_classes]
gamma = weight_dict['gamma'] # [num_classes]
else:
# Default focal parameters if no language weights
num_classes = batch['labels'].size(1)
alpha = torch.full((num_classes,), 0.25, device=config.device)
gamma = torch.full((num_classes,), 2.0, device=config.device)
# Forward pass
with config.get_autocast_context():
outputs = model(
input_ids=batch['input_ids'],
attention_mask=batch['attention_mask'],
labels=batch['labels'],
lang_ids=batch['lang']
)
# Calculate loss with per-class focal parameters
loss_fct = LanguageAwareFocalLoss()
loss = loss_fct(
outputs['logits'],
batch['labels'].float(),
lang_weights=lang_weights,
alpha=alpha,
gamma=gamma
)
outputs['loss'] = loss
# Check for numerical instability
if torch.isnan(loss).any() or torch.isinf(loss).any():
logger.error(f"Numerical instability detected! Loss: {loss.item()}")
logger.error(f"Batch stats - input_ids shape: {batch['input_ids'].shape}, labels shape: {batch['labels'].shape}")
if lang_weights is not None:
logger.error(f"Weights stats - min: {lang_weights.min():.3f}, max: {lang_weights.max():.3f}")
logger.error(f"Focal params - gamma range: [{gamma.min():.3f}, {gamma.max():.3f}], alpha range: [{alpha.min():.3f}, {alpha.max():.3f}]")
optimizer.zero_grad()
return None
# Scale loss for gradient accumulation
if config.grad_accum_steps > 1:
loss = loss / config.grad_accum_steps
# Backward pass with scaled loss
scaler.scale(loss).backward()
# Only update weights after accumulating enough gradients
if (batch_idx + 1) % config.grad_accum_steps == 0:
# Log gradient stats before clipping
if batch_idx % 100 == 0:
grad_stats = get_grad_stats(model)
if grad_stats:
logger.debug("Gradient stats before clipping:")
for key, value in grad_stats.items():
logger.debug(f"{key}: {value}")
# Gradient clipping
if config.max_grad_norm > 0:
# Unscale gradients before clipping
scaler.unscale_(optimizer)
grad_norm = torch.nn.utils.clip_grad_norm_(
model.parameters(),
config.max_grad_norm
)
if grad_norm.isnan() or grad_norm.isinf():
logger.warning(f"Gradient norm is {grad_norm}, skipping optimizer step")
optimizer.zero_grad()
return loss.item() * config.grad_accum_steps # Return unscaled loss for logging
# Optimizer step with scaler
scaler.step(optimizer)
scaler.update()
# Zero gradients after optimizer step
optimizer.zero_grad(set_to_none=True) # More efficient than zero_grad()
# Step scheduler after optimization
scheduler.step()
# Log gradient stats after update
if batch_idx % 100 == 0:
grad_stats = get_grad_stats(model)
if grad_stats:
logger.debug("Gradient stats after update:")
for key, value in grad_stats.items():
logger.debug(f"{key}: {value}")
# Return the original (unscaled) loss for logging
return loss.item() * config.grad_accum_steps if config.grad_accum_steps > 1 else loss.item()
def save_checkpoint(model, optimizer, scheduler, metrics, config, epoch):
"""Save model checkpoint with versioning and timestamps"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Create base checkpoint directory
base_dir = Path('weights/toxic_classifier_xlm-roberta-large')
base_dir.mkdir(parents=True, exist_ok=True)
# Create versioned checkpoint directory
checkpoint_dir = base_dir / f"checkpoint_epoch{epoch:02d}_{timestamp}"
checkpoint_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Saving checkpoint to {checkpoint_dir}")
try:
# Save model state
model_save_path = checkpoint_dir / 'pytorch_model.bin'
torch.save(model.state_dict(), model_save_path)
logger.info(f"Saved model state to {model_save_path}")
# Save training state
training_state = {
'epoch': epoch,
'optimizer_state': optimizer.state_dict(),
'scheduler_state': scheduler.state_dict(),
'metrics': {
'train_loss': metrics.train_losses[-1] if metrics.train_losses else None,
'best_auc': metrics.best_auc,
'timestamp': timestamp
}
}
state_save_path = checkpoint_dir / 'training_state.pt'
torch.save(training_state, state_save_path)
logger.info(f"Saved training state to {state_save_path}")
# Save config
config_save_path = checkpoint_dir / 'config.json'
with open(config_save_path, 'w') as f:
json.dump(config.to_serializable_dict(), f, indent=2)
logger.info(f"Saved config to {config_save_path}")
# Save checkpoint metadata
metadata = {
'timestamp': timestamp,
'epoch': epoch,
'model_size': os.path.getsize(model_save_path) / (1024 * 1024), # Size in MB
'git_commit': os.environ.get('GIT_COMMIT', 'unknown'),
'training_metrics': {
'loss': metrics.train_losses[-1] if metrics.train_losses else None,
'best_auc': metrics.best_auc
}
}
meta_save_path = checkpoint_dir / 'metadata.json'
with open(meta_save_path, 'w') as f:
json.dump(metadata, f, indent=2)
logger.info(f"Saved checkpoint metadata to {meta_save_path}")
# Only create symlink after all files are saved successfully
latest_path = base_dir / 'latest'
if latest_path.exists():
latest_path.unlink() # Remove existing symlink if it exists
# Create relative symlink
os.symlink(checkpoint_dir.name, latest_path)
logger.info(f"Updated 'latest' symlink to point to {checkpoint_dir.name}")
# Cleanup old checkpoints if needed
keep_last_n = 3 # Keep last 3 checkpoints
all_checkpoints = sorted([d for d in base_dir.iterdir() if d.is_dir() and d.name.startswith('checkpoint')])
if len(all_checkpoints) > keep_last_n:
for old_checkpoint in all_checkpoints[:-keep_last_n]:
try:
import shutil
shutil.rmtree(old_checkpoint)
logger.info(f"Removed old checkpoint: {old_checkpoint}")
except Exception as e:
logger.warning(f"Failed to remove old checkpoint {old_checkpoint}: {str(e)}")
logger.info(f"Successfully saved checkpoint for epoch {epoch + 1}")
return checkpoint_dir
except Exception as e:
logger.error(f"Error saving checkpoint: {str(e)}")
logger.error("Checkpoint save failed with traceback:", exc_info=True)
# If checkpoint save fails, ensure we don't leave a broken symlink
latest_path = base_dir / 'latest'
if latest_path.exists():
latest_path.unlink()
raise
def train(model, train_loader, config):
"""Train the model"""
global _model, _optimizer, _scheduler
_model = model
logger.info("Initializing training components...")
logger.info(f"Using gradient accumulation with {config.grad_accum_steps} steps")
logger.info(f"Effective batch size: {config.batch_size * config.grad_accum_steps}")
# Initialize gradient scaler for mixed precision
logger.info("Setting up gradient scaler...")
scaler = torch.cuda.amp.GradScaler(enabled=config.use_amp)
logger.info("Creating optimizer...")
optimizer = torch.optim.AdamW(
config.get_param_groups(model),
weight_decay=config.weight_decay
)
_optimizer = optimizer
# Calculate total steps for cosine scheduler
total_steps = (len(train_loader) // config.grad_accum_steps) * config.epochs
warmup_steps = int(total_steps * config.warmup_ratio)
logger.info(f"Training schedule: {total_steps} total steps, {warmup_steps} warmup steps")
logger.info(f"Actual number of batches per epoch: {len(train_loader)}")
# Initialize cosine scheduler with warm restarts
logger.info("Creating learning rate scheduler...")
scheduler = CosineAnnealingWarmRestarts(
optimizer,
T_0=total_steps // config.num_cycles,
T_mult=1,
eta_min=config.lr * config.min_lr_ratio
)
_scheduler = scheduler
# Initialize metrics tracker
metrics = MetricsTracker()
logger.info("Starting training loop...")
# Training loop
model.train()
# Verify data loader is properly initialized
try:
logger.info("Verifying data loader...")
test_batch = next(iter(train_loader))
logger.info(f"Data loader test successful. Batch keys: {list(test_batch.keys())}")
logger.info(f"Input shape: {test_batch['input_ids'].shape}")
logger.info(f"Label shape: {test_batch['labels'].shape}")
except Exception as e:
logger.error(f"Data loader verification failed: {str(e)}")
raise
for epoch in range(config.epochs):
epoch_loss = 0
num_batches = 0
logger.info(f"Starting epoch {epoch + 1}/{config.epochs}")
# Create progress bar with additional metrics
progress_bar = tqdm(
train_loader,
desc=f"Epoch {epoch + 1}/{config.epochs}",
dynamic_ncols=True, # Adapt to terminal width
leave=True # Keep progress bar after completion
)
optimizer.zero_grad(set_to_none=True) # More efficient gradient clearing
logger.info("Iterating through batches...")
batch_start_time = time.time()
for batch_idx, batch in enumerate(progress_bar):
try:
# Log first batch details
if batch_idx == 0:
logger.info("Successfully loaded first batch")
logger.info(f"Batch shapes - input_ids: {batch['input_ids'].shape}, "
f"attention_mask: {batch['attention_mask'].shape}, "
f"labels: {batch['labels'].shape}")
logger.info(f"Memory usage: {torch.cuda.memory_allocated() / 1024**2:.1f}MB")
# Execute training step
loss = training_step(batch, model, optimizer, scheduler, config, scaler, batch_idx)
if loss is not None:
epoch_loss += loss
num_batches += 1
# Calculate batch processing time
batch_time = time.time() - batch_start_time
# Format loss string outside of the postfix dict
loss_str = "N/A" if loss is None else f"{loss:.4f}"
# Update progress bar with detailed metrics
progress_bar.set_postfix({
'loss': loss_str,
'lr': f"{scheduler.get_last_lr()[0]:.2e}",
'batch_time': f"{batch_time:.2f}s",
'processed': f"{(batch_idx + 1) * config.batch_size}"
})
# Log to wandb with more frequent updates
if (batch_idx + 1) % max(1, config.grad_accum_steps // 2) == 0:
try:
wandb.log({
'batch_loss': loss if loss is not None else 0,
'learning_rate': scheduler.get_last_lr()[0],
'batch_time': batch_time,
'gpu_memory': torch.cuda.memory_allocated() / 1024**2
})
except Exception as e:
logger.warning(f"Could not log to wandb: {str(e)}")
# More frequent logging for debugging
if batch_idx % 10 == 0:
loss_debug_str = "N/A" if loss is None else f"{loss:.4f}"
logger.debug(
f"Batch {batch_idx}/{len(train_loader)}: "
f"Loss={loss_debug_str}, "
f"Time={batch_time:.2f}s"
)
# Memory management
if batch_idx % config.gc_frequency == 0:
if torch.cuda.is_available():
torch.cuda.empty_cache()
batch_start_time = time.time()
except Exception as e:
logger.error(f"Error in batch {batch_idx}: {str(e)}")
logger.error("Batch contents:")
for k, v in batch.items():
if isinstance(v, torch.Tensor):
logger.error(f"{k}: shape={v.shape}, dtype={v.dtype}, device={v.device}")
else:
logger.error(f"{k}: type={type(v)}")
if torch.cuda.is_available():
logger.error(f"GPU Memory: {torch.cuda.memory_allocated() / 1024**2:.1f}MB")
continue
# Calculate average epoch loss
avg_epoch_loss = epoch_loss / num_batches if num_batches > 0 else float('inf')
metrics.update_train(avg_epoch_loss)
logger.info(f"Epoch {epoch + 1} completed. Average loss: {avg_epoch_loss:.4f}")
# Save checkpoint
try:
save_checkpoint(model, optimizer, scheduler, metrics, config, epoch)
logger.info(f"Saved checkpoint for epoch {epoch + 1}")
except Exception as e:
logger.error(f"Could not save checkpoint: {str(e)}")
# Log epoch metrics
try:
wandb.log({
'epoch': epoch + 1,
'epoch_loss': avg_epoch_loss,
'best_auc': metrics.best_auc,
'learning_rate': scheduler.get_last_lr()[0],
'gpu_memory': torch.cuda.memory_allocated() / 1024**2 if torch.cuda.is_available() else 0
})
except Exception as e:
logger.error(f"Could not log epoch metrics to wandb: {str(e)}")
def create_dataloaders(train_dataset, val_dataset, config):
"""Create DataLoader with simplified settings"""
logger.info("Creating data loader...")
# Create sampler
train_sampler = MultilabelStratifiedSampler(
labels=train_dataset.labels,
groups=train_dataset.langs,
batch_size=config.batch_size
)
# Create DataLoader with minimal settings
train_loader = DataLoader(
train_dataset,
batch_size=config.batch_size,
sampler=train_sampler,
num_workers=0, # Disable multiprocessing for now
pin_memory=torch.cuda.is_available(),
drop_last=False
)
# Verify DataLoader
logger.info("Testing DataLoader...")
try:
test_batch = next(iter(train_loader))
logger.info("DataLoader test successful")
return train_loader
except Exception as e:
logger.error(f"DataLoader test failed: {str(e)}")
raise
def main():
try:
# Set environment variables for CUDA and multiprocessing
os.environ['CUDA_LAUNCH_BLOCKING'] = '1'
os.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'max_split_size_mb:128'
os.environ['OMP_NUM_THREADS'] = '1' # Limit OpenMP threads
os.environ['MKL_NUM_THREADS'] = '1' # Limit MKL threads
logger.info("Initializing training configuration...")
# Initialize config first
config = TrainingConfig()
# Initialize CUDA settings
if torch.cuda.is_available():
# Disable TF32 on Ampere GPUs
torch.backends.cuda.matmul.allow_tf32 = False
torch.backends.cudnn.allow_tf32 = False
# Set deterministic mode
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
# Clear CUDA cache
torch.cuda.empty_cache()
# Set device to current CUDA device
torch.cuda.set_device(torch.cuda.current_device())
logger.info(f"Using CUDA device: {torch.cuda.get_device_name()}")
logger.info("Configured CUDA settings for stability")
# Initialize wandb
try:
wandb.init(
project="toxic-comment-classification",
name=f"toxic-{datetime.now().strftime('%Y%m%d-%H%M%S')}",
config=config.to_serializable_dict()
)
logger.info("Initialized wandb logging")
except Exception as e:
logger.warning(f"Could not initialize wandb: {str(e)}")
global _model, _optimizer, _scheduler
_model = None
_optimizer = None
_scheduler = None
logger.info("Loading datasets...")
try:
train_df = pd.read_csv("dataset/split/train.csv")
logger.info(f"Loaded train dataset with {len(train_df)} samples")
except Exception as e:
logger.error(f"Error loading datasets: {str(e)}")
raise
try:
logger.info("Creating tokenizer and dataset...")
tokenizer = XLMRobertaTokenizer.from_pretrained(config.model_name)
train_dataset = ToxicDataset(train_df, tokenizer, config)
logger.info("Dataset creation successful")
except Exception as e:
logger.error(f"Error creating datasets: {str(e)}")
raise
logger.info("Creating data loaders...")
train_loader = create_dataloaders(train_dataset, None, config)
logger.info("Initializing model...")
model = init_model(config)
logger.info("Starting training...")
train(model, train_loader, config)
except KeyboardInterrupt:
print("\nTraining interrupted by user")
cleanup()
except Exception as e:
print(f"Error during training: {str(e)}")
import traceback
traceback.print_exc()
raise
finally:
if wandb.run is not None:
try:
wandb.finish()
except Exception as e:
print(f"Warning: Could not finish wandb run: {str(e)}")
cleanup()
if __name__ == "__main__":
# Set global PyTorch settings
torch.set_num_threads(1) # Limit CPU threads
np.set_printoptions(precision=4, suppress=True)
torch.set_printoptions(precision=4, sci_mode=False)
try:
main()
except Exception as e:
print(f"Fatal error: {str(e)}")
cleanup()
sys.exit(1)