from __future__ import annotations import os import torch from torch.optim import AdamW from torch.utils.data import DataLoader from tqdm import tqdm from yaml_bert.config import YamlBertConfig from yaml_bert.dataset import YamlDataset, collate_fn from yaml_bert.model import YamlBertModel class YamlBertTrainer: """Training loop with hybrid loss from two prediction heads.""" def __init__( self, config: YamlBertConfig, model: YamlBertModel, dataset: YamlDataset, checkpoint_dir: str | None = None, checkpoint_every: int = 1, resume_from: str | None = None, ) -> None: self.config = config self.model = model self.dataset = dataset self.checkpoint_dir = checkpoint_dir self.checkpoint_every = checkpoint_every self.resume_from = resume_from self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") def train(self) -> list[float]: from datetime import datetime self.model.to(self.device) self.model.train() print(f"Training started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") num_params = sum(p.numel() for p in self.model.parameters()) print(f"Model parameters: {num_params:,}") print(f"Config: d_model={self.config.d_model}, layers={self.config.num_layers}, heads={self.config.num_heads}") print(f"Device: {self.device}") optimizer = AdamW(self.model.parameters(), lr=self.config.lr, weight_decay=0.01) start_epoch = 0 if self.resume_from: checkpoint = torch.load(self.resume_from, map_location=self.device) self.model.load_state_dict(checkpoint["model_state_dict"]) optimizer.load_state_dict(checkpoint["optimizer_state_dict"]) start_epoch = checkpoint["epoch"] print(f"Resumed from epoch {start_epoch}") # num_workers parallelizes dataset.__getitem__ across CPU cores so # GPU isn't blocked waiting for the next batch. v6.1 used the default # (0 = main process only), which was OK when __getitem__ was cheap. # With the v7-era growth in vocab + optional tree_distances compute, # this is now the bottleneck — without workers, GPU sits at 0% util. import os num_workers = min(8, max(2, (os.cpu_count() or 4) // 2)) dataloader = DataLoader( self.dataset, batch_size=self.config.batch_size, shuffle=True, collate_fn=collate_fn, num_workers=num_workers, persistent_workers=True, pin_memory=True, ) epoch_losses: list[float] = [] for epoch in range(start_epoch, self.config.num_epochs): total_loss: float = 0.0 num_batches: int = 0 running_breakdown: dict[str, float] = {} pbar = tqdm(dataloader, desc=f"Epoch {epoch+1}/{self.config.num_epochs}", leave=True) for batch in pbar: batch = {k: v.to(self.device) for k, v in batch.items()} optimizer.zero_grad() simple_logits, kind_logits = self.model( token_ids=batch["token_ids"], node_types=batch["node_types"], depths=batch["depths"], sibling_indices=batch["sibling_indices"], padding_mask=batch["padding_mask"], tree_distances=batch.get("tree_distances"), ) loss, breakdown = self.model.compute_loss( simple_logits, batch["simple_labels"], kind_logits, batch["kind_labels"], ) if torch.isnan(loss): continue # skip bad batches loss.backward() torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() for k, v in breakdown.items(): running_breakdown[k] = running_breakdown.get(k, 0.0) + v num_batches += 1 postfix = {"loss": f"{total_loss/num_batches:.4f}"} for k in ["simple", "kind"]: if k in running_breakdown: postfix[k] = f"{running_breakdown[k]/num_batches:.4f}" pbar.set_postfix(**postfix) avg_loss: float = total_loss / max(num_batches, 1) epoch_losses.append(avg_loss) breakdown_str: str = " | ".join( f"{k}: {v/max(num_batches,1):.4f}" for k, v in sorted(running_breakdown.items()) ) print(f"Epoch {epoch+1}/{self.config.num_epochs} — loss: {avg_loss:.4f} ({breakdown_str})") if self.checkpoint_dir and (epoch+1) % self.checkpoint_every == 0: self._save_checkpoint(epoch+1, optimizer) if self.checkpoint_dir: self._save_checkpoint(self.config.num_epochs, optimizer) return epoch_losses def _save_checkpoint(self, epoch: int, optimizer: AdamW) -> None: os.makedirs(self.checkpoint_dir, exist_ok=True) path = os.path.join(self.checkpoint_dir, f"yaml_bert_v4_epoch_{epoch}.pt") torch.save({ "epoch": epoch, "model_state_dict": self.model.state_dict(), "optimizer_state_dict": optimizer.state_dict(), "tree_pos_variant": self.config.tree_pos_variant.value, }, path) print(f"Checkpoint saved: {path}")