import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader import torch.nn.functional as F from torch.cuda.amp import GradScaler, autocast import os import json import argparse import time import math import glob from typing import Dict, List, Optional from tqdm import tqdm import numpy as np import gc import logging from collections import defaultdict import multiprocessing # Import custom modules try: from model_slm import MixtureOfRecursions, count_parameters, TextGenerator from custom_tokenizer import TechnicalTokenizer except ImportError as e: raise ImportError(f"Failed to import custom modules: {e}") # Constants for configuration DEFAULT_MAX_LENGTH = 128 DEFAULT_MAX_EXAMPLES = 50000 DEFAULT_D_MODEL = 384 DEFAULT_N_LAYERS = 6 DEFAULT_N_HEADS = 6 DEFAULT_EPOCHS = 15 DEFAULT_BATCH_SIZE = 16 DEFAULT_LEARNING_RATE = 5e-4 DEFAULT_GRAD_ACCUM_STEPS = 1 DEFAULT_EVAL_EVERY = 500 DEFAULT_WARMUP_RATIO = 0.05 DEFAULT_CHECKPOINT_DIR = "checkpoints" DEFAULT_LOG_LEVEL = "INFO" # Set up logging logging.basicConfig( level=DEFAULT_LOG_LEVEL, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("training.log") ] ) logger = logging.getLogger(__name__) class FastTechnicalTextDataset(Dataset): """Optimized dataset for fast loading and processing of technical text.""" def __init__( self, data_file: str, tokenizer: TechnicalTokenizer, max_length: int = DEFAULT_MAX_LENGTH, max_examples: int = DEFAULT_MAX_EXAMPLES ): """ Initialize the dataset with optimized loading. Args: data_file (str): Path to the training data file. tokenizer (TechnicalTokenizer): Tokenizer for encoding text. max_length (int): Maximum sequence length. max_examples (int): Maximum number of examples to load. Raises: FileNotFoundError: If the data file does not exist. ValueError: If max_length or max_examples is invalid. """ if not os.path.exists(data_file): raise FileNotFoundError(f"Data file not found: {data_file}") if max_length <= 0 or max_examples <= 0: raise ValueError("max_length and max_examples must be positive") self.tokenizer = tokenizer self.max_length = max_length self.pad_token_id = tokenizer.vocab.get('', 0) self.max_examples = max_examples self.examples = [] logger.info(f"Loading dataset from {data_file} with max_length={max_length}, max_examples={max_examples}") start_time = time.time() self._fast_load_data(data_file) self._tensorize_data() logger.info(f"Loaded {len(self.examples)} examples in {time.time() - start_time:.1f}s") if torch.cuda.is_available(): torch.cuda.empty_cache() gc.collect() def _fast_load_data(self, data_file: str) -> None: """Load and filter data efficiently.""" logger.info("Reading and filtering data...") with open(data_file, 'r', encoding='utf-8') as f: lines = f.readlines() logger.info(f"File contains {len(lines)} lines") good_examples = [] seen_hashes = set() for line in lines[:self.max_examples * 3]: line = line.strip() if ( 50 <= len(line) <= 400 and line.count(' ') >= 8 and not line.lower().startswith(('http', 'www', 'ftp')) and line.count('.') <= len(line) * 0.1 ): line_hash = hash(line[:100]) if line_hash not in seen_hashes: seen_hashes.add(line_hash) good_examples.append(line) if len(good_examples) >= self.max_examples: break logger.info(f"Filtered to {len(good_examples)} quality examples") batch_size = 1000 for i in range(0, len(good_examples), batch_size): batch = good_examples[i:i + batch_size] for line in batch: try: if not line.endswith('<|endoftext|>'): line += ' <|endoftext|>' tokens = self.tokenizer.encode_ids(line, add_special_tokens=True) if 30 <= len(tokens) <= self.max_length: if len(tokens) < self.max_length: tokens.extend([self.pad_token_id] * (self.max_length - len(tokens))) self.examples.append(tokens) except Exception as e: logger.warning(f"Failed to process line: {e}") continue if i % 5000 == 0: logger.info(f"Processed {len(self.examples)} examples...") logger.info(f"Final dataset size: {len(self.examples)} examples") def _tensorize_data(self) -> None: """Pre-tensorize data for faster training.""" logger.info("Pre-tensorizing data...") seq_len = self.max_length - 1 tensorized_examples = [] for tokens in self.examples: if len(tokens) != self.max_length: continue input_ids = torch.tensor(tokens[:-1], dtype=torch.long) targets = torch.tensor(tokens[1:], dtype=torch.long) original_len = next((i for i, x in enumerate(tokens) if x == self.pad_token_id), self.max_length) mask_len = min(original_len, seq_len) attention_mask = torch.zeros(seq_len, dtype=torch.long) attention_mask[:mask_len] = 1 tensorized_examples.append({ 'input_ids': input_ids, 'targets': targets, 'attention_mask': attention_mask }) self.examples = tensorized_examples logger.info("Data pre-tensorized successfully") def __len__(self) -> int: """Return the number of examples in the dataset.""" return len(self.examples) def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]: """Return a single example from the dataset.""" return self.examples[idx] class FastCosineScheduler: """Cosine learning rate scheduler with warmup.""" def __init__(self, optimizer: optim.Optimizer, total_steps: int, warmup_ratio: float = DEFAULT_WARMUP_RATIO): """ Initialize the cosine scheduler. Args: optimizer (optim.Optimizer): Optimizer to schedule. total_steps (int): Total training steps. warmup_ratio (float): Ratio of steps for warmup phase. Raises: ValueError: If total_steps or warmup_ratio is invalid. """ if total_steps <= 0 or not 0 <= warmup_ratio <= 1: raise ValueError("total_steps must be positive and warmup_ratio must be in [0, 1]") self.optimizer = optimizer self.total_steps = total_steps self.warmup_steps = int(total_steps * warmup_ratio) self.base_lr = optimizer.param_groups[0]['lr'] self.step_count = 0 def step(self) -> float: """ Update the learning rate. Returns: float: Current learning rate. """ self.step_count += 1 if self.step_count <= self.warmup_steps: lr = self.base_lr * self.step_count / self.warmup_steps else: progress = (self.step_count - self.warmup_steps) / (self.total_steps - self.warmup_steps) lr = self.base_lr * 0.5 * (1 + math.cos(math.pi * progress)) for param_group in self.optimizer.param_groups: param_group['lr'] = lr return lr class UltraFastTrainer: """Trainer optimized for fast training of transformer models.""" def __init__( self, model: nn.Module, tokenizer: TechnicalTokenizer, train_dataset: FastTechnicalTextDataset, val_dataset: Optional[FastTechnicalTextDataset] = None, config: Optional[Dict] = None ): """ Initialize the trainer. Args: model (nn.Module): The transformer model to train. tokenizer (TechnicalTokenizer): Tokenizer for encoding/decoding. train_dataset (FastTechnicalTextDataset): Training dataset. val_dataset (Optional[FastTechnicalTextDataset]): Validation dataset. config (Optional[Dict]): Training configuration. Raises: ValueError: If config contains invalid parameters. """ self.model = model self.tokenizer = tokenizer self.train_dataset = train_dataset self.val_dataset = val_dataset self.config = config or {} self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.model.to(self.device) self._validate_config() self._fast_init_weights() self._setup_fast_optimizer() epochs = self.config.get('epochs', DEFAULT_EPOCHS) batch_size = self.config.get('batch_size', DEFAULT_BATCH_SIZE) total_steps = len(train_dataset) // batch_size * epochs self.scheduler = FastCosineScheduler(self.optimizer, total_steps) self.scaler = GradScaler() if self.device.type == 'cuda' else None self.global_step = 0 self.best_loss = float('inf') self.grad_accum_steps = self.config.get('gradient_accumulation_steps', DEFAULT_GRAD_ACCUM_STEPS) self.eval_every = self.config.get('eval_every', DEFAULT_EVAL_EVERY) def _validate_config(self) -> None: """Validate training configuration.""" if self.config.get('batch_size', DEFAULT_BATCH_SIZE) <= 0: raise ValueError("batch_size must be positive") if self.config.get('epochs', DEFAULT_EPOCHS) <= 0: raise ValueError("epochs must be positive") if self.config.get('learning_rate', DEFAULT_LEARNING_RATE) <= 0: raise ValueError("learning_rate must be positive") if self.config.get('gradient_accumulation_steps', DEFAULT_GRAD_ACCUM_STEPS) <= 0: raise ValueError("gradient_accumulation_steps must be positive") def _fast_init_weights(self) -> None: """Initialize model weights.""" def fast_init(module: nn.Module) -> None: if isinstance(module, nn.Linear): nn.init.normal_(module.weight, std=0.02) if module.bias is not None: nn.init.zeros_(module.bias) elif isinstance(module, nn.Embedding): nn.init.normal_(module.weight, std=0.02) self.model.apply(fast_init) logger.info("Model weights initialized") def _setup_fast_optimizer(self) -> None: """Set up AdamW optimizer.""" lr = self.config.get('learning_rate', DEFAULT_LEARNING_RATE) params = [p for p in self.model.parameters() if p.requires_grad] self.optimizer = optim.AdamW(params, lr=lr, betas=(0.9, 0.99), weight_decay=0.01, eps=1e-6) logger.info(f"Optimizer initialized with learning rate: {lr}") def compute_fast_loss(self, logits: torch.Tensor, targets: torch.Tensor, mask: torch.Tensor) -> torch.Tensor: """ Compute masked cross-entropy loss. Args: logits (torch.Tensor): Model output logits of shape (batch_size, seq_len, vocab_size). targets (torch.Tensor): Target token IDs of shape (batch_size, seq_len). mask (torch.Tensor): Attention mask of shape (batch_size, seq_len). Returns: torch.Tensor: Computed loss. """ logits_flat = logits.view(-1, logits.size(-1)) targets_flat = targets.view(-1) mask_flat = mask.view(-1).bool() if not mask_flat.any(): return torch.tensor(0.0, device=logits.device, requires_grad=True) return F.cross_entropy(logits_flat[mask_flat], targets_flat[mask_flat]) def train_epoch_fast(self, epoch: int, dataloader: DataLoader) -> Dict[str, float]: """ Train for one epoch. Args: epoch (int): Current epoch number. dataloader (DataLoader): Training data loader. Returns: Dict[str, float]: Training metrics (loss, perplexity, epoch_time_min). """ self.model.train() total_loss = 0 num_batches = 0 start_time = time.time() progress_bar = tqdm(dataloader, desc=f"Epoch {epoch}", leave=False, miniters=50) for batch_idx, batch in enumerate(progress_bar): input_ids = batch['input_ids'].to(self.device, non_blocking=True) targets = batch['targets'].to(self.device, non_blocking=True) mask = batch['attention_mask'].to(self.device, non_blocking=True) with autocast(enabled=self.device.type == 'cuda'): logits, comp_loss = self.model(input_ids, mask) lm_loss = self.compute_fast_loss(logits, targets, mask) total_loss_step = lm_loss + 0.0001 * comp_loss if self.grad_accum_steps > 1: total_loss_step = total_loss_step / self.grad_accum_steps if self.scaler: self.scaler.scale(total_loss_step).backward() if (batch_idx + 1) % self.grad_accum_steps == 0: self.scaler.unscale_(self.optimizer) torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0) self.scaler.step(self.optimizer) self.scaler.update() self.optimizer.zero_grad(set_to_none=True) self.scheduler.step() self.global_step += 1 else: total_loss_step.backward() if (batch_idx + 1) % self.grad_accum_steps == 0: torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0) self.optimizer.step() self.optimizer.zero_grad(set_to_none=True) self.scheduler.step() self.global_step += 1 total_loss += lm_loss.item() num_batches += 1 if batch_idx % 100 == 0: current_loss = total_loss / num_batches progress_bar.set_postfix({'loss': f"{current_loss:.3f}", 'ppl': f"{math.exp(min(current_loss, 10)):.1f}"}) if batch_idx % 200 == 0 and batch_idx > 0 and self.device.type == 'cuda': torch.cuda.empty_cache() avg_loss = total_loss / max(num_batches, 1) return { 'loss': avg_loss, 'perplexity': math.exp(min(avg_loss, 10)), 'epoch_time_min': (time.time() - start_time) / 60 } def validate_fast(self, dataloader: DataLoader) -> Dict[str, float]: """ Validate the model on the validation dataset. Args: dataloader (DataLoader): Validation data loader. Returns: Dict[str, float]: Validation metrics (loss, perplexity). """ self.model.eval() total_loss = 0 num_batches = 0 max_val_batches = min(100, len(dataloader)) with torch.no_grad(): for batch_idx, batch in enumerate(dataloader): if batch_idx >= max_val_batches: break input_ids = batch['input_ids'].to(self.device, non_blocking=True) targets = batch['targets'].to(self.device, non_blocking=True) mask = batch['attention_mask'].to(self.device, non_blocking=True) with autocast(enabled=self.device.type == 'cuda'): logits, _ = self.model(input_ids, mask) loss = self.compute_fast_loss(logits, targets, mask) total_loss += loss.item() num_batches += 1 avg_loss = total_loss / max(num_batches, 1) return {'loss': avg_loss, 'perplexity': math.exp(min(avg_loss, 10))} def save_checkpoint_fast(self, epoch: int, metrics: Dict, save_dir: str = DEFAULT_CHECKPOINT_DIR) -> Optional[str]: """ Save a checkpoint if the loss improves. Args: epoch (int): Current epoch number. metrics (Dict): Training and validation metrics. save_dir (str): Directory to save checkpoints. Returns: Optional[str]: Path to the saved checkpoint or None. """ os.makedirs(save_dir, exist_ok=True) val_loss = metrics.get('val_loss', metrics.get('loss', float('inf'))) if val_loss < self.best_loss: self.best_loss = val_loss checkpoint = { 'epoch': epoch, 'model_state_dict': self.model.state_dict(), 'optimizer_state_dict': self.optimizer.state_dict(), 'metrics': metrics, 'scaler_state_dict': self.scaler.state_dict() if self.scaler else None } best_path = os.path.join(save_dir, "best_model.pt") torch.save(checkpoint, best_path) logger.info(f"New best checkpoint saved: {best_path}, Loss: {val_loss:.4f}") return best_path return None def train_ultra_fast(self, num_epochs: int = DEFAULT_EPOCHS, batch_size: int = DEFAULT_BATCH_SIZE) -> List[Dict]: """ Train the model with optimized settings. Args: num_epochs (int): Number of training epochs. batch_size (int): Batch size for training. Returns: List[Dict]: Training history with metrics for each epoch. """ logger.info(f"Starting ultra-fast training: {num_epochs} epochs, batch_size={batch_size}") logger.info("Target: Loss < 2.0, PPL < 12, Time: 4-5 hours") train_loader = DataLoader( self.train_dataset, batch_size=batch_size, shuffle=True, num_workers=min(multiprocessing.cpu_count(), 4), pin_memory=self.device.type == 'cuda', persistent_workers=True, drop_last=True ) val_loader = None if self.val_dataset: val_loader = DataLoader( self.val_dataset, batch_size=batch_size * 2, shuffle=False, num_workers=min(multiprocessing.cpu_count() // 2, 2), pin_memory=self.device.type == 'cuda' ) total_start_time = time.time() history = [] for epoch in range(1, num_epochs + 1): logger.info(f"Starting epoch {epoch}/{num_epochs}") train_metrics = self.train_epoch_fast(epoch, train_loader) val_metrics = {} if val_loader and (epoch % 2 == 0 or epoch == num_epochs): val_metrics = self.validate_fast(val_loader) epoch_time = train_metrics['epoch_time_min'] * 60 epoch_info = { 'epoch': epoch, 'train_loss': train_metrics['loss'], 'train_ppl': train_metrics['perplexity'], 'epoch_time_min': train_metrics['epoch_time_min'] } if val_metrics: epoch_info.update({'val_loss': val_metrics['loss'], 'val_ppl': val_metrics['perplexity']}) history.append(epoch_info) elapsed_hours = (time.time() - total_start_time) / 3600 remaining_hours = elapsed_hours * (num_epochs - epoch) / max(epoch, 1) logger.info(f"Epoch {epoch} results:") logger.info(f" Epoch time: {epoch_time/60:.1f} min") logger.info(f" Total elapsed: {elapsed_hours:.1f}h") logger.info(f" Est. remaining: {remaining_hours:.1f}h") logger.info(f" Train Loss: {train_metrics['loss']:.4f}") logger.info(f" Train PPL: {train_metrics['perplexity']:.1f}") if val_metrics: logger.info(f" Val Loss: {val_metrics['loss']:.4f}") logger.info(f" Val PPL: {val_metrics['perplexity']:.1f}") current_loss = val_metrics.get('loss', train_metrics['loss']) current_ppl = val_metrics.get('perplexity', train_metrics['perplexity']) if current_loss < 2.0 and current_ppl < 12: logger.info(f"Targets achieved: Loss={current_loss:.4f} < 2.0, PPL={current_ppl:.1f} < 12") combined_metrics = {**train_metrics} if val_metrics: combined_metrics.update({f"val_{k}": v for k, v in val_metrics.items()}) self.save_checkpoint_fast(epoch, combined_metrics) if self.device.type == 'cuda': torch.cuda.empty_cache() gc.collect() if current_loss < 1.8 and current_ppl < 10: logger.info("Early stopping: Excellent performance achieved!") break total_time = (time.time() - total_start_time) / 3600 logger.info(f"Training completed in {total_time:.1f} hours") logger.info(f"Best loss: {self.best_loss:.4f}") return history def run_ultra_fast_training() -> int: """ Run the ultra-fast training pipeline. Returns: int: Exit code (0 for success, 1 for failure). """ parser = argparse.ArgumentParser(description="Ultra-Fast Training for MixtureOfRecursions Model") parser.add_argument("--train_file", default=None, help="Path to training data file") parser.add_argument("--val_file", default=None, help="Path to validation data file") parser.add_argument("--tokenizer_dir", default="tokenizer", help="Directory for tokenizer files") parser.add_argument("--max_examples", type=int, default=DEFAULT_MAX_EXAMPLES, help="Maximum number of training examples") parser.add_argument("--d_model", type=int, default=DEFAULT_D_MODEL, help="Model embedding dimension") parser.add_argument("--n_layers", type=int, default=DEFAULT_N_LAYERS, help="Number of transformer layers") parser.add_argument("--n_heads", type=int, default=DEFAULT_N_HEADS, help="Number of attention heads") parser.add_argument("--max_seq_len", type=int, default=DEFAULT_MAX_LENGTH, help="Maximum sequence length") parser.add_argument("--epochs", type=int, default=DEFAULT_EPOCHS, help="Number of training epochs") parser.add_argument("--batch_size", type=int, default=DEFAULT_BATCH_SIZE, help="Batch size for training") parser.add_argument("--learning_rate", type=float, default=DEFAULT_LEARNING_RATE, help="Learning rate") parser.add_argument("--gradient_accumulation_steps", type=int, default=DEFAULT_GRAD_ACCUM_STEPS, help="Gradient accumulation steps") parser.add_argument("--eval_every", type=int, default=DEFAULT_EVAL_EVERY, help="Evaluate every N steps") args = parser.parse_args() torch.manual_seed(42) np.random.seed(42) logger.info("Starting ultra-fast training pipeline") if args.train_file is None: patterns = ["*train*.txt", "*_train.txt"] files = [] for pattern in patterns: files.extend(glob.glob(pattern)) files.extend(glob.glob(os.path.join("split_data", pattern))) files.extend(glob.glob(os.path.join("data", pattern))) if files: args.train_file = files[0] logger.info(f"Found training file: {args.train_file}") else: logger.error("No training files found!") return 1 try: tokenizer = TechnicalTokenizer() tokenizer.load(args.tokenizer_dir) logger.info(f"Tokenizer loaded with vocab size: {tokenizer.get_vocab_size()}") except Exception as e: logger.error(f"Failed to load tokenizer: {e}") return 1 logger.info("Creating training dataset...") try: train_dataset = FastTechnicalTextDataset( args.train_file, tokenizer, args.max_seq_len, args.max_examples ) except Exception as e: logger.error(f"Failed to create training dataset: {e}") return 1 val_dataset = None if args.val_file and os.path.exists(args.val_file): try: val_dataset = FastTechnicalTextDataset( args.val_file, tokenizer, args.max_seq_len, max_examples=5000 ) logger.info("Validation dataset created") except Exception as e: logger.warning(f"Failed to create validation dataset: {e}") try: model = MixtureOfRecursions( vocab_size=tokenizer.get_vocab_size(), d_model=args.d_model, n_layers=args.n_layers, n_heads=args.n_heads, max_seq_len=args.max_seq_len - 1, padding_idx=tokenizer.vocab.get('', 0) ) logger.info("Model initialized") except Exception as e: logger.error(f"Failed to initialize model: {e}") return 1 config = { 'learning_rate': args.learning_rate, 'gradient_accumulation_steps': args.gradient_accumulation_steps, 'eval_every': args.eval_every, 'batch_size': args.batch_size, 'epochs': args.epochs } try: trainer = UltraFastTrainer(model, tokenizer, train_dataset, val_dataset, config) results = trainer.train_ultra_fast(args.epochs, args.batch_size) with open('ultra_fast_results.json', 'w') as f: json.dump(results, f, indent=2) logger.info("Training results saved to ultra_fast_results.json") return 0 except Exception as e: logger.error(f"Training failed: {e}") return 1 if __name__ == "__main__": exit(run_ultra_fast_training())