Skip to content

Training

The training helpers are optional convenience utilities for routine supervised training around a model you constructed yourself. Dataset splits, optimization schedules, and expected benchmark metrics remain explicit experiment choices.

Solver and backward choices belong to the model configuration. For example, SolverConfig(backward_mode="unrolled") differentiates through finite solver steps, while SolverConfig(backward_mode="implicit", backward_solver="gmres") uses the package DEQ/SILVA adjoint path. This path follows the DEQ implicit-gradient formulation [4], with matrix-free GMRES [13]; training and regularization choices can also be compared with [39] and [6].

For supervised data \((x_i,y_i)\), the helper minimizes a batch objective

\[ \mathcal L(\theta,\phi) = \frac{1}{B}\sum_{i=1}^{B} \ell\left(R_\phi(z_i^\star),y_i\right), \qquad z_i^\star=f_\theta(z_i^\star,x_i). \]

Two Usage Modes

Low-level PyTorch remains the most direct path:

model = SILVAGraphPresetNetwork(...)
optimizer = torch.optim.Adam(model.parameters(), lr=0.002)

for batch in train_loader:
    logits = model(**batch.model_kwargs())
    loss = torch.nn.functional.cross_entropy(logits, batch.y)
    loss.backward()
    optimizer.step()

The training engine wraps the same model and loaders:

from silva_networks import TrainConfig, fit_supervised

result = fit_supervised(
    model,
    train_loader,
    val_loader,
    config=TrainConfig(
        task="classification",
        epochs=50,
        lr=0.002,
        weight_decay=1e-5,
        gradient_clipping=1.0,
        checkpoint_path="runs/checkpoint.pt",
    ),
)

Supported batch forms are ordinary (x, y) pairs, mappings with x and y fields, and GraphTensorBatch objects.

Sequence batches use (input, target). Multi-input models such as optical flow use (*model_inputs, target), for example (image1, image2, target_flow). Mappings provide the equivalent named-argument route with y, target, or labels reserved for the supervision tensor.

For objectives that need auxiliary fields or model states, pass step_fn:

def flow_step(model, batch):
    image1, image2, target, valid = batch
    prediction = model(image1, image2)
    loss = silva_endpoint_error(prediction, target, valid)
    return loss, prediction, target

result = fit_supervised(model, loader, config=config, step_fn=flow_step)

The callback receives the device-moved batch and must return scalar loss, prediction, and target tensors. The same callback is accepted by evaluate, which keeps paper-specific losses outside the general engine.

Scale-Aware Training

TrainConfig supports microbatch accumulation and autocast without changing the task loss:

config = TrainConfig(
    task="regression",
    epochs=100,
    optimizer="adamw",
    gradient_accumulation_steps=4,
    mixed_precision="bfloat16",
    checkpoint_path="runs/checkpoint.pt",
    resume=True,
)

For per-device batch (B), accumulation count (K), and process count (P), the effective batch is (BKP). A final partial accumulation group is rescaled to preserve the batch-mean gradient. Distributed wrappers use no_sync() for intermediate microbatches, and distributed samplers receive the current epoch. The checkpoint also stores gradient-scaler state when CUDA float16 is active.

Sharded and distributed loader construction is documented in Scaling Data. The complete model/data/training route is in Full-Scale SILVA.

Classification is not restricted to matrix logits. The default class_dim=-1 supports ordinary (batch, classes) and sequence (batch, length, classes) outputs. Set class_dim=1 for dense image logits with shape (batch, classes, height, width). metric_mode selects whether validation metrics are minimized or maximized; auto maximizes accuracy and custom metrics and minimizes losses and regression errors.

Pass a custom optimizer, scheduler, loss, metric, and epoch hook directly to fit_supervised. For tasks with paper-specific multi-term objectives, such as indexed DEQ-Flow corrections, an ordinary PyTorch loop remains the full-control route around the same package model.

Training metrics and equilibrium diagnostics answer different questions. Record loss or accuracy from fit_supervised, and request structured model results in a custom step_fn when residuals, convergence flags, or backward linear-solve diagnostics must be retained. The reporting checklist and method sources are in Citation-Aware Reporting and Paper and References.

Small supervised training utilities for package-built SILVA experiments.

The helpers in this module are intentionally generic. They do not encode private paper configurations, dataset splits, or expected metrics. They provide the repeatable training chores around a user-specified PyTorch model and data loader: seeding, device movement, loss/metric evaluation, gradient clipping, checkpoint resume, and a compact history object.

EpochMetrics dataclass

One row in a supervised training history.

Source code in src/silva_networks/training.py
@dataclass(frozen=True)
class EpochMetrics:
    """One row in a supervised training history."""

    epoch: int
    train_loss: float
    val_loss: float | None
    val_metric: float | None
    metric_name: str
    lr: float

EvaluationResult dataclass

Aggregate loss and metric values from one evaluation pass.

Source code in src/silva_networks/training.py
@dataclass(frozen=True)
class EvaluationResult:
    """Aggregate loss and metric values from one evaluation pass."""

    loss: float
    metric: float
    metric_name: str
    num_examples: int

TrainConfig dataclass

Configuration for fit_supervised.

Parameters:

Name Type Description Default
task TaskName

Supervised task family.

'classification'
epochs int

Number of epochs to train.

10
lr float

Optimizer learning rate.

0.001
weight_decay float

Optimizer weight decay.

0.0
optimizer OptimizerName

Optimizer family.

'adam'
momentum float

SGD momentum.

0.9
loss LossName

Loss function. auto maps classification to cross entropy and regression to mean squared error.

'auto'
metric MetricName

Validation metric. auto maps classification to accuracy and regression to mean absolute error.

'auto'
metric_mode MetricMode

Whether a larger or smaller validation metric is better. auto maximizes accuracy and custom metrics and minimizes losses and regression errors.

'auto'
class_dim int

Class-logit axis for classification. The default final axis handles (batch, classes) and sequence logits (batch, length, classes). Use 1 for dense logits (batch, classes, height, width).

-1
gradient_clipping float | None

Optional global norm clipping threshold.

None
gradient_accumulation_steps int

Number of microbatches per optimizer step.

1
mixed_precision PrecisionName

Autocast precision. float16 requires CUDA; bfloat16 is supported on devices with a matching backend.

'none'
scheduler SchedulerName

Learning-rate scheduler family.

'none'
scheduler_step_size int

Epoch period for step scheduling.

10
scheduler_gamma float

Multiplicative step-scheduler factor.

0.5
device str | device | None

PyTorch device string, torch.device, or auto.

'auto'
seed int | None

Optional deterministic seed applied before training.

None
deterministic bool

Whether to request deterministic PyTorch algorithms.

False
checkpoint_path str | Path | None

Optional path for checkpoint save/resume.

None
resume bool

Whether to resume from checkpoint_path when it exists.

False
Source code in src/silva_networks/training.py
@dataclass(frozen=True)
class TrainConfig:
    """Configuration for `fit_supervised`.

    Args:
        task: Supervised task family.
        epochs: Number of epochs to train.
        lr: Optimizer learning rate.
        weight_decay: Optimizer weight decay.
        optimizer: Optimizer family.
        momentum: SGD momentum.
        loss: Loss function. `auto` maps classification to cross entropy and
            regression to mean squared error.
        metric: Validation metric. `auto` maps classification to accuracy and
            regression to mean absolute error.
        metric_mode: Whether a larger or smaller validation metric is better.
            `auto` maximizes accuracy and custom metrics and minimizes losses
            and regression errors.
        class_dim: Class-logit axis for classification. The default final axis
            handles `(batch, classes)` and sequence logits `(batch, length,
            classes)`. Use `1` for dense logits `(batch, classes, height, width)`.
        gradient_clipping: Optional global norm clipping threshold.
        gradient_accumulation_steps: Number of microbatches per optimizer step.
        mixed_precision: Autocast precision. ``float16`` requires CUDA;
            ``bfloat16`` is supported on devices with a matching backend.
        scheduler: Learning-rate scheduler family.
        scheduler_step_size: Epoch period for step scheduling.
        scheduler_gamma: Multiplicative step-scheduler factor.
        device: PyTorch device string, `torch.device`, or `auto`.
        seed: Optional deterministic seed applied before training.
        deterministic: Whether to request deterministic PyTorch algorithms.
        checkpoint_path: Optional path for checkpoint save/resume.
        resume: Whether to resume from `checkpoint_path` when it exists.
    """

    task: TaskName = "classification"
    epochs: int = 10
    lr: float = 1e-3
    weight_decay: float = 0.0
    optimizer: OptimizerName = "adam"
    momentum: float = 0.9
    loss: LossName = "auto"
    metric: MetricName = "auto"
    metric_mode: MetricMode = "auto"
    class_dim: int = -1
    gradient_clipping: float | None = None
    gradient_accumulation_steps: int = 1
    mixed_precision: PrecisionName = "none"
    scheduler: SchedulerName = "none"
    scheduler_step_size: int = 10
    scheduler_gamma: float = 0.5
    device: str | torch.device | None = "auto"
    seed: int | None = None
    deterministic: bool = False
    checkpoint_path: str | Path | None = None
    resume: bool = False

    def __post_init__(self) -> None:
        if self.task not in {"classification", "regression"}:
            raise ValueError(f"Unsupported task: {self.task}")
        if self.epochs < 1:
            raise ValueError("epochs must be at least one")
        if self.lr <= 0:
            raise ValueError("lr must be positive")
        if self.weight_decay < 0:
            raise ValueError("weight_decay must be nonnegative")
        if self.gradient_clipping is not None and self.gradient_clipping <= 0:
            raise ValueError("gradient_clipping must be positive")
        if self.gradient_accumulation_steps < 1:
            raise ValueError("gradient_accumulation_steps must be positive")
        if self.mixed_precision not in {"none", "float16", "bfloat16"}:
            raise ValueError("mixed_precision must be none, float16, or bfloat16")
        if self.metric_mode not in {"auto", "min", "max"}:
            raise ValueError(f"Unsupported metric_mode: {self.metric_mode}")
        if self.class_dim == 0:
            raise ValueError("class_dim cannot be the batch dimension")
        if self.scheduler_step_size < 1:
            raise ValueError("scheduler_step_size must be positive")
        if self.scheduler_gamma <= 0:
            raise ValueError("scheduler_gamma must be positive")

TrainResult dataclass

Result returned by fit_supervised.

Source code in src/silva_networks/training.py
@dataclass(frozen=True)
class TrainResult:
    """Result returned by `fit_supervised`."""

    history: list[EpochMetrics]
    best_epoch: int | None
    best_metric: float | None
    metric_name: str
    checkpoint_path: Path | None

evaluate

evaluate(model, data_loader, *, task='classification', loss='auto', metric='auto', loss_fn=None, metric_fn=None, step_fn=None, class_dim=-1, device='auto', mixed_precision='none')

Evaluate a supervised model on an iterable of batches.

Source code in src/silva_networks/training.py
def evaluate(
    model: nn.Module,
    data_loader: Iterable[Any],
    *,
    task: TaskName = "classification",
    loss: LossName = "auto",
    metric: MetricName = "auto",
    loss_fn: nn.Module | None = None,
    metric_fn: MetricFunction | None = None,
    step_fn: BatchStep | None = None,
    class_dim: int = -1,
    device: str | torch.device | None = "auto",
    mixed_precision: PrecisionName = "none",
) -> EvaluationResult:
    """Evaluate a supervised model on an iterable of batches."""

    resolved = resolve_device(device)
    _validate_precision(mixed_precision, resolved)
    model.to(resolved)
    was_training = model.training
    model.eval()
    metric_name = "custom" if metric_fn is not None else _resolve_metric_name(task, metric)
    total_loss = 0.0
    total_metric = 0.0
    total_examples = 0
    try:
        with torch.no_grad():
            for batch in data_loader:
                moved = _move_batch(batch, resolved)
                with _autocast_context(resolved, mixed_precision):
                    loss_value, prediction, target = _batch_step_values(
                        model,
                        moved,
                        task=task,
                        loss=loss,
                        loss_fn=loss_fn,
                        class_dim=class_dim,
                        step_fn=step_fn,
                    )
                batch_metric = (
                    _as_metric_tensor(metric_fn(prediction, target), prediction)
                    if metric_fn is not None
                    else loss_value
                    if metric_name == "loss" and step_fn is not None
                    else _metric_value(
                        prediction,
                        target,
                        task,
                        metric_name,
                        selected_loss=loss,
                        loss_fn=loss_fn,
                        class_dim=class_dim,
                    )
                )
                count = _num_examples(target)
                total_loss += float(loss_value.detach().cpu()) * count
                total_metric += float(batch_metric.detach().cpu()) * count
                total_examples += count
    finally:
        model.train(was_training)
    if total_examples == 0:
        raise ValueError("data_loader produced no examples")
    return EvaluationResult(
        loss=total_loss / total_examples,
        metric=total_metric / total_examples,
        metric_name=metric_name,
        num_examples=total_examples,
    )

fit_supervised

fit_supervised(model, train_loader, val_loader=None, *, config=None, loss_fn=None, optimizer=None, scheduler=None, metric_fn=None, epoch_hook=None, step_fn=None)

Train a user-constructed PyTorch model on supervised batches.

Batches may be ordinary (x, y) pairs, (*model_inputs, target) tuples, dictionaries with x/y keys, or GraphTensorBatch objects. Dictionary and graph batches are passed to models as keyword arguments, which keeps graph, molecular, flow, and custom SILVA models usable without package-specific experiment configs.

Source code in src/silva_networks/training.py
def fit_supervised(
    model: nn.Module,
    train_loader: Iterable[Any],
    val_loader: Iterable[Any] | None = None,
    *,
    config: TrainConfig | None = None,
    loss_fn: nn.Module | None = None,
    optimizer: torch.optim.Optimizer | None = None,
    scheduler: torch.optim.lr_scheduler.LRScheduler | None = None,
    metric_fn: MetricFunction | None = None,
    epoch_hook: EpochHook | None = None,
    step_fn: BatchStep | None = None,
) -> TrainResult:
    """Train a user-constructed PyTorch model on supervised batches.

    Batches may be ordinary `(x, y)` pairs, `(*model_inputs, target)` tuples,
    dictionaries with `x`/`y` keys, or `GraphTensorBatch` objects. Dictionary
    and graph batches are passed to models as keyword arguments, which keeps
    graph, molecular, flow, and custom SILVA models usable without
    package-specific experiment configs.
    """

    cfg = config or TrainConfig()
    if cfg.epochs < 1:
        raise ValueError("epochs must be at least one")
    if cfg.seed is not None:
        seed_everything(cfg.seed, deterministic=cfg.deterministic)

    resolved = resolve_device(cfg.device)
    _validate_precision(cfg.mixed_precision, resolved)
    model.to(resolved)
    opt = optimizer or _make_optimizer(model, cfg)
    scaler_enabled = cfg.mixed_precision == "float16" and resolved.type == "cuda"
    if hasattr(torch.amp, "GradScaler"):
        scaler = torch.amp.GradScaler("cuda", enabled=scaler_enabled)
    else:
        scaler = torch.cuda.amp.GradScaler(enabled=scaler_enabled)
    active_scheduler = scheduler if scheduler is not None else _make_scheduler(opt, cfg)
    checkpoint_path = Path(cfg.checkpoint_path) if cfg.checkpoint_path is not None else None
    start_epoch = 1
    history: list[EpochMetrics] = []
    best_epoch: int | None = None
    best_metric: float | None = None
    metric_name = "custom" if metric_fn is not None else _resolve_metric_name(cfg.task, cfg.metric)

    if checkpoint_path is not None and cfg.resume and checkpoint_path.exists():
        payload = torch.load(checkpoint_path, map_location=resolved, weights_only=False)
        model.load_state_dict(payload["model_state"])
        opt.load_state_dict(payload["optimizer_state"])
        if active_scheduler is not None and payload.get("scheduler_state") is not None:
            active_scheduler.load_state_dict(payload["scheduler_state"])
        if scaler.is_enabled() and payload.get("scaler_state") is not None:
            scaler.load_state_dict(payload["scaler_state"])
        history = [EpochMetrics(**row) for row in payload.get("history", [])]
        best_epoch = payload.get("best_epoch")
        best_metric = payload.get("best_metric")
        start_epoch = int(payload.get("epoch", 0)) + 1
        _restore_rng_state(payload.get("rng_state"))

    for epoch in range(start_epoch, cfg.epochs + 1):
        sampler = getattr(train_loader, "sampler", None)
        if sampler is not None and hasattr(sampler, "set_epoch"):
            sampler.set_epoch(epoch)
        train_loss = _train_one_epoch(
            model,
            train_loader,
            opt,
            cfg,
            loss_fn,
            step_fn,
            resolved,
            scaler,
        )
        val_loss: float | None = None
        val_metric: float | None = None
        if val_loader is not None:
            val = evaluate(
                model,
                val_loader,
                task=cfg.task,
                loss=cfg.loss,
                metric=cfg.metric,
                loss_fn=loss_fn,
                metric_fn=metric_fn,
                step_fn=step_fn,
                class_dim=cfg.class_dim,
                device=resolved,
                mixed_precision=cfg.mixed_precision,
            )
            val_loss = val.loss
            val_metric = val.metric
            if best_metric is None or _is_better(
                val_metric,
                best_metric,
                metric_name,
                cfg.metric_mode,
            ):
                best_metric = val_metric
                best_epoch = epoch
        row = EpochMetrics(
            epoch=epoch,
            train_loss=train_loss,
            val_loss=val_loss,
            val_metric=val_metric,
            metric_name=metric_name,
            lr=float(opt.param_groups[0]["lr"]),
        )
        history.append(row)
        if epoch_hook is not None:
            epoch_hook(row, model, opt)
        if active_scheduler is not None:
            active_scheduler.step()
        if checkpoint_path is not None:
            _save_checkpoint(
                checkpoint_path,
                model,
                opt,
                active_scheduler,
                epoch,
                history,
                best_epoch,
                best_metric,
                scaler,
            )

    return TrainResult(
        history=history,
        best_epoch=best_epoch,
        best_metric=best_metric,
        metric_name=metric_name,
        checkpoint_path=checkpoint_path,
    )

seed_everything

seed_everything(seed, *, deterministic=False)

Seed Python, NumPy, PyTorch CPU, and available PyTorch accelerators.

Source code in src/silva_networks/training.py
def seed_everything(seed: int, *, deterministic: bool = False) -> int:
    """Seed Python, NumPy, PyTorch CPU, and available PyTorch accelerators."""

    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    if deterministic:
        torch.use_deterministic_algorithms(True, warn_only=True)
        cudnn = getattr(torch.backends, "cudnn", None)
        if cudnn is not None:
            cudnn.benchmark = False
            cudnn.deterministic = True
    return seed

Where to Go Next

Question Page
Can I execute fitting, evaluation, checkpointing, and resume? Training Helpers Validation Notebook
What evidence should a trained experiment report? Reconstructing Paper Experiments
Which measured outputs are published? Results
How do I scale data and execution? Full-Scale SILVA