vendor/paper2code/scaffolds/train_template.py

"""
{{PAPER_TITLE}} — Training Loop

Paper: https://arxiv.org/abs/{{ARXIV_ID}}
Implements: {{TRAINING_DESCRIPTION}}

Section references:
  {{§SECTION}} — {{DESCRIPTION}}
"""

import math
import yaml
from pathlib import Path
from typing import Optional

import torch
import torch.nn as nn
from torch.utils.data import DataLoader

from src.model import {{MODEL_CLASS}}, ModelConfig
from src.loss import {{LOSS_FN}}
from src.data import {{DATASET_CLASS}}


def load_config(config_path: str = "configs/base.yaml") -> dict:
    """Load training configuration."""
    with open(config_path) as f:
        return yaml.safe_load(f)


def build_optimizer(
    model: nn.Module,
    config: dict,
) -> torch.optim.Optimizer:
    """Build optimizer with parameter groups.
    
    §{{SECTION}} — "{{quote about optimizer}}"
    """
    # [ASSUMPTION] Not applying weight decay to biases and normalization layers
    # Paper does not specify which parameters get weight decay
    no_decay = ["bias", "LayerNorm.weight", "LayerNorm.bias"]
    param_groups = [
        {
            "params": [
                p for n, p in model.named_parameters()
                if not any(nd in n for nd in no_decay) and p.requires_grad
            ],
            "weight_decay": config["training"]["weight_decay"],
        },
        {
            "params": [
                p for n, p in model.named_parameters()
                if any(nd in n for nd in no_decay) and p.requires_grad
            ],
            "weight_decay": 0.0,
        },
    ]

    optimizer_name = config["training"]["optimizer"]
    lr = config["training"]["lr"]

    if optimizer_name == "adam":
        # §{{SECTION}} — optimizer specification
        return torch.optim.Adam(
            param_groups,
            lr=lr,
            betas=tuple(config["training"].get("betas", [0.9, 0.999])),
            eps=config["training"].get("eps", 1e-8),
        )
    elif optimizer_name == "adamw":
        return torch.optim.AdamW(
            param_groups,
            lr=lr,
            betas=tuple(config["training"].get("betas", [0.9, 0.999])),
            eps=config["training"].get("eps", 1e-8),
        )
    elif optimizer_name == "sgd":
        return torch.optim.SGD(
            param_groups,
            lr=lr,
            momentum=config["training"].get("momentum", 0.9),
        )
    else:
        raise ValueError(f"Unknown optimizer: {optimizer_name}")


def build_scheduler(
    optimizer: torch.optim.Optimizer,
    config: dict,
) -> Optional[torch.optim.lr_scheduler._LRScheduler]:
    """Build learning rate scheduler.
    
    §{{SECTION}} — "{{quote about LR schedule}}"
    """
    schedule = config["training"].get("schedule", "constant")
    warmup_steps = config["training"].get("warmup_steps", 0)
    total_steps = config["training"]["total_steps"]

    if schedule == "constant":
        return None
    elif schedule == "cosine":
        def lr_lambda(step):
            if step < warmup_steps:
                return step / max(warmup_steps, 1)
            progress = (step - warmup_steps) / max(total_steps - warmup_steps, 1)
            return 0.5 * (1.0 + math.cos(math.pi * progress))
        return torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
    elif schedule == "linear":
        def lr_lambda(step):
            if step < warmup_steps:
                return step / max(warmup_steps, 1)
            return max(0.0, (total_steps - step) / max(total_steps - warmup_steps, 1))
        return torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
    else:
        raise ValueError(f"Unknown schedule: {schedule}")


def train(config_path: str = "configs/base.yaml"):
    """Main training loop.
    
    {{Describe what this training loop does and which paper sections it follows.}}
    """
    config = load_config(config_path)

    # Device
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    # Build model
    model_config = ModelConfig(
        # REPLACE: populate from config dict
    )
    model = {{MODEL_CLASS}}(model_config).to(device)

    # Build optimizer and scheduler
    optimizer = build_optimizer(model, config)
    scheduler = build_scheduler(optimizer, config)

    # Build loss
    loss_fn = {{LOSS_FN}}()  # REPLACE with actual loss construction

    # Build data — see src/data.py for dataset setup instructions
    # dataset = {{DATASET_CLASS}}(config["data"])
    # dataloader = DataLoader(dataset, batch_size=config["training"]["batch_size"],
    #                         shuffle=True, num_workers=4, pin_memory=True)

    # Training loop
    # §{{SECTION}} — training procedure
    gradient_clip = config["training"].get("gradient_clip", None)
    total_steps = config["training"]["total_steps"]

    model.train()
    step = 0

    print(f"Starting training for {total_steps} steps")
    print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

    # REPLACE: implement the actual training loop
    # for epoch in range(max_epochs):
    #     for batch in dataloader:
    #         batch = {k: v.to(device) for k, v in batch.items()}
    #
    #         optimizer.zero_grad()
    #         output = model(batch["input"])
    #         loss = loss_fn(output, batch["target"])
    #         loss.backward()
    #
    #         if gradient_clip is not None:
    #             torch.nn.utils.clip_grad_norm_(model.parameters(), gradient_clip)
    #
    #         optimizer.step()
    #         if scheduler is not None:
    #             scheduler.step()
    #
    #         step += 1
    #         if step % 100 == 0:
    #             print(f"Step {step}/{total_steps} | Loss: {loss.item():.4f}")
    #
    #         if step >= total_steps:
    #             break


if __name__ == "__main__":
    train()