Constructing a Gin Config Managed PyTorch Pipeline with Configurable MLP Variants, Cosine Scheduling, and Runtime Parameter Overrides

On this tutorial, we implement a Gin Config–managed PyTorch experiment pipeline wherein the executable coaching code stays steady. On the similar time, the experimental levels of freedom are moved into declarative configuration information. We assemble a nonlinear spiral binary classification process, outline a configurable MLP with scoped architectural variants, and expose parameters for the optimizer, scheduler, loss, batching, seeding, and coaching loop by way of @gin.configurable bindings. We use Gin’s scoped references to instantiate separate mannequin configurations, runtime bindings to override chosen parameters with out enhancing supply code, and operative config export to seize the precise resolved configuration that produces every coaching run.

Putting in Gin Config and Constructing the Spiral Dataset

!pip -q set up gin-config
import os
import json
import math
import random
import textwrap
from pathlib import Path
import gin
import numpy as np
import torch
import torch.nn as nn
import torch.nn.useful as F
from torch.utils.knowledge import TensorDataset, DataLoader
import matplotlib.pyplot as plt
ROOT = Path("/content material/gin_config_sharp_tutorial")
CONFIG_DIR = ROOT / "configs"
RUN_DIR = ROOT / "runs"
CONFIG_DIR.mkdir(mother and father=True, exist_ok=True)
RUN_DIR.mkdir(mother and father=True, exist_ok=True)
gin.clear_config()
@gin.configurable
def seed_everything(seed=42):
   random.seed(seed)
   np.random.seed(seed)
   torch.manual_seed(seed)
   torch.cuda.manual_seed_all(seed)
   return seed
@gin.configurable
def make_spiral_dataset(
   n_per_class=gin.REQUIRED,
   noise=0.18,
   rotations=1.75,
   train_fraction=0.8,
   seed=0,
):
   rng = np.random.default_rng(seed)
   radius_0 = np.linspace(0.05, 1.0, n_per_class)
   theta_0 = rotations * 2 * np.pi * radius_0
   theta_0 += rng.regular(0.0, noise, dimension=n_per_class)
   x0 = np.stack(
       [
           radius_0 * np.cos(theta_0),
           radius_0 * np.sin(theta_0),
       ],
       axis=1,
   )
   radius_1 = np.linspace(0.05, 1.0, n_per_class)
   theta_1 = rotations * 2 * np.pi * radius_1 + np.pi
   theta_1 += rng.regular(0.0, noise, dimension=n_per_class)
   x1 = np.stack(
       [
           radius_1 * np.cos(theta_1),
           radius_1 * np.sin(theta_1),
       ],
       axis=1,
   )
   x = np.concatenate([x0, x1], axis=0).astype(np.float32)
   y = np.concatenate(
       [
           np.zeros((n_per_class, 1)),
           np.ones((n_per_class, 1)),
       ],
       axis=0,
   ).astype(np.float32)
   order = rng.permutation(len(x))
   x = x[order]
   y = y[order]
   cut up = int(train_fraction * len(x))
   x_train, y_train = x[:split], y[:split]
   x_val, y_val = x[split:], y[split:]
   imply = x_train.imply(axis=0, keepdims=True)
   std = x_train.std(axis=0, keepdims=True) + 1e-8
   x_train = (x_train - imply) / std
   x_val = (x_val - imply) / std
   return {
       "prepare": (
           torch.tensor(x_train),
           torch.tensor(y_train),
       ),
       "val": (
           torch.tensor(x_val),
           torch.tensor(y_val),
       ),
       "metadata": {
           "n_train": int(len(x_train)),
           "n_val": int(len(x_val)),
           "n_features": int(x_train.form[1]),
           "noise": float(noise),
           "rotations": float(rotations),
           "seed": int(seed),
       },
   }
@gin.configurable(denylist=["x", "y"])
def make_loader(
   x,
   y,
   batch_size=128,
   shuffle=True,
   seed=0,
):
   generator = torch.Generator()
   generator.manual_seed(seed)
   dataset = TensorDataset(x, y)
   return DataLoader(
       dataset,
       batch_size=batch_size,
       shuffle=shuffle,
       generator=generator,
       drop_last=False,
   )

We begin by putting in Gin Config and importing the core Python libraries, PyTorch, NumPy, and the plotting libraries required for the experiment. We create a clear venture listing construction and reset Gin’s world configuration state so the pocket book runs reproducibly. We then outline the seed perform, generate a nonlinear spiral dataset, and construct a configurable DataLoader that Gin can management by way of exterior bindings.

Defining a Gin-Configurable MLP, Optimizer, and Scheduler

def activation_layer(title):
   title = title.decrease()
   if title == "relu":
       return nn.ReLU()
   if title == "gelu":
       return nn.GELU()
   if title == "tanh":
       return nn.Tanh()
   if title == "silu":
       return nn.SiLU()
   increase ValueError(f"Unknown activation: {title}")
@gin.configurable
class MLP(nn.Module):
   def __init__(
       self,
       input_dim=gin.REQUIRED,
       hidden_dims=(64, 64),
       output_dim=1,
       activation="gelu",
       dropout=0.0,
       use_layernorm=False,
   ):
       tremendous().__init__()
       layers = []
       current_dim = input_dim
       for hidden_dim in hidden_dims:
           layers.append(nn.Linear(current_dim, hidden_dim))
           if use_layernorm:
               layers.append(nn.LayerNorm(hidden_dim))
           layers.append(activation_layer(activation))
           if dropout > 0:
               layers.append(nn.Dropout(dropout))
           current_dim = hidden_dim
       layers.append(nn.Linear(current_dim, output_dim))
       self.community = nn.Sequential(*layers)
   def ahead(self, x):
       return self.community(x)
@gin.configurable(denylist=["params"])
def make_optimizer(
   params,
   title="adamw",
   lr=3e-3,
   weight_decay=1e-3,
   momentum=0.9,
):
   title = title.decrease()
   if title == "adamw":
       return torch.optim.AdamW(
           params,
           lr=lr,
           weight_decay=weight_decay,
       )
   if title == "sgd":
       return torch.optim.SGD(
           params,
           lr=lr,
           momentum=momentum,
           weight_decay=weight_decay,
       )
   increase ValueError(f"Unknown optimizer: {title}")
@gin.configurable(denylist=["optimizer"])
def make_cosine_scheduler(
   optimizer,
   total_epochs=60,
   warmup_epochs=5,
   min_lr_factor=0.05,
):
   def lr_lambda(epoch):
       if epoch < warmup_epochs:
           return float(epoch + 1) / float(max(1, warmup_epochs))
       progress = (epoch - warmup_epochs) / float(
           max(1, total_epochs - warmup_epochs)
       )
       cosine = 0.5 * (1.0 + math.cos(math.pi * progress))
       return min_lr_factor + (1.0 - min_lr_factor) * cosine
   return torch.optim.lr_scheduler.LambdaLR(
       optimizer,
       lr_lambda=lr_lambda,
   )
@gin.configurable
def bce_with_logits_loss(
   logits,
   targets,
   label_smoothing=0.0,
):
   if label_smoothing > 0:
       targets = targets * (1.0 - label_smoothing) + 0.5 * label_smoothing
   return F.binary_cross_entropy_with_logits(logits, targets)
@torch.no_grad()
def consider(mannequin, loader, loss_fn, gadget):
   mannequin.eval()
   total_loss = 0.0
   total_correct = 0
   total_count = 0
   for x, y in loader:
       x = x.to(gadget)
       y = y.to(gadget)
       logits = mannequin(x)
       loss = loss_fn(logits, y)
       probs = torch.sigmoid(logits)
       preds = (probs >= 0.5).float()
       total_loss += loss.merchandise() * len(x)
       total_correct += (preds == y).sum().merchandise()
       total_count += len(x)
   return {
       "loss": total_loss / total_count,
       "accuracy": total_correct / total_count,
   }

We outline the neural community constructing blocks that type the configurable mannequin and the coaching utilities. We create an MLP class whose structure, activation perform, dropout, and layer normalization habits are managed by way of Gin reasonably than hardcoded values. We additionally implement configurable optimizer, scheduler, loss, and analysis features so the coaching pipeline stays modular and experiment-ready.

Implementing the Coaching Loop and Experiment Runner

@gin.configurable(
   denylist=[
       "model",
       "optimizer",
       "scheduler",
       "train_loader",
       "val_loader",
       "device",
   ]
)
def match(
   mannequin,
   optimizer,
   scheduler,
   train_loader,
   val_loader,
   gadget,
   epochs=60,
   grad_clip_norm=1.0,
   log_every=10,
   loss_fn=bce_with_logits_loss,
):
   historical past = []
   for epoch in vary(1, epochs + 1):
       mannequin.prepare()
       for x, y in train_loader:
           x = x.to(gadget)
           y = y.to(gadget)
           optimizer.zero_grad(set_to_none=True)
           logits = mannequin(x)
           loss = loss_fn(logits, y)
           loss.backward()
           if grad_clip_norm just isn't None:
               nn.utils.clip_grad_norm_(
                   mannequin.parameters(),
                   grad_clip_norm,
               )
           optimizer.step()
       if scheduler just isn't None:
           scheduler.step()
       train_metrics = consider(
           mannequin,
           train_loader,
           loss_fn,
           gadget,
       )
       val_metrics = consider(
           mannequin,
           val_loader,
           loss_fn,
           gadget,
       )
       lr = optimizer.param_groups[0]["lr"]
       row = {
           "epoch": epoch,
           "lr": lr,
           "train_loss": train_metrics["loss"],
           "train_accuracy": train_metrics["accuracy"],
           "val_loss": val_metrics["loss"],
           "val_accuracy": val_metrics["accuracy"],
       }
       historical past.append(row)
       if epoch == 1 or epoch % log_every == 0 or epoch == epochs:
           print(
               f"epoch={epoch:03d} | "
               f"lr={lr:.6f} | "
               f"train_loss={row['train_loss']:.4f} | "
               f"train_acc={row['train_accuracy']:.3f} | "
               f"val_loss={row['val_loss']:.4f} | "
               f"val_acc={row['val_accuracy']:.3f}"
           )
   return historical past
@gin.configurable
def run_experiment(
   tag=gin.REQUIRED,
   mannequin=gin.REQUIRED,
   dataset_fn=make_spiral_dataset,
   optimizer_factory=make_optimizer,
   scheduler_factory=make_cosine_scheduler,
   prefer_gpu=True,
):
   seed_everything()
   gadget = "cuda" if prefer_gpu and torch.cuda.is_available() else "cpu"
   knowledge = dataset_fn()
   x_train, y_train = knowledge["train"]
   x_val, y_val = knowledge["val"]
   train_loader = make_loader(
       x_train,
       y_train,
       shuffle=True,
   )
   val_loader = make_loader(
       x_val,
       y_val,
       shuffle=False,
   )
   mannequin = mannequin.to(gadget)
   optimizer = optimizer_factory(mannequin.parameters())
   scheduler = None
   if scheduler_factory just isn't None:
       scheduler = scheduler_factory(optimizer)
   print("n" + "=" * 80)
   print(f"Experiment: {tag}")
   print("=" * 80)
   print(f"Machine: {gadget}")
   print(f"Dataset: {knowledge['metadata']}")
   print(f"Parameters: {sum(p.numel() for p in mannequin.parameters()):,}")
   historical past = match(
       mannequin=mannequin,
       optimizer=optimizer,
       scheduler=scheduler,
       train_loader=train_loader,
       val_loader=val_loader,
       gadget=gadget,
   )
   consequence = {
       "tag": tag,
       "gadget": gadget,
       "metadata": knowledge["metadata"],
       "parameters": sum(p.numel() for p in mannequin.parameters()),
       "closing": historical past[-1],
       "historical past": historical past,
   }
   return consequence

We implement the primary coaching loop, wherein the mannequin performs ahead passes, computes binary cross-entropy loss, backpropagates gradients, applies gradient clipping, and updates parameters. We consider the mannequin after every epoch on each the coaching and validation units, whereas storing loss, accuracy, and studying price historical past. We then outline the top-level experiment runner that connects the dataset, mannequin, optimizer, scheduler, and coaching loop by way of Gin-managed dependencies.

Writing Gin Config Information with Scoped Bindings and Runtime Overrides

BASE_CONFIG = CONFIG_DIR / "base.gin"
COMPACT_CONFIG = CONFIG_DIR / "compact_adamw.gin"
WIDE_CONFIG = CONFIG_DIR / "wide_sgd.gin"
BASE_CONFIG.write_text(
   textwrap.dedent(
       """
       SEED = 123
       N_PER_CLASS = 900
       EPOCHS = 50
       BATCH = 128
       seed_everything.seed = %SEED
       make_spiral_dataset.n_per_class = %N_PER_CLASS
       make_spiral_dataset.noise = 0.20
       make_spiral_dataset.rotations = 1.85
       make_spiral_dataset.train_fraction = 0.80
       make_spiral_dataset.seed = %SEED
       make_loader.batch_size = %BATCH
       make_loader.seed = %SEED
       MLP.input_dim = 2
       MLP.output_dim = 1
       MLP.activation = 'gelu'
       MLP.dropout = 0.05
       MLP.use_layernorm = True
       make_optimizer.title="adamw"
       make_optimizer.lr = 0.003
       make_optimizer.weight_decay = 0.001
       make_optimizer.momentum = 0.9
       make_cosine_scheduler.total_epochs = %EPOCHS
       make_cosine_scheduler.warmup_epochs = 5
       make_cosine_scheduler.min_lr_factor = 0.05
       bce_with_logits_loss.label_smoothing = 0.02
       match.epochs = %EPOCHS
       match.grad_clip_norm = 1.0
       match.log_every = 10
       match.loss_fn = @bce_with_logits_loss
       run_experiment.dataset_fn = @make_spiral_dataset
       run_experiment.optimizer_factory = @make_optimizer
       run_experiment.scheduler_factory = @make_cosine_scheduler
       run_experiment.prefer_gpu = True
       """
   ).strip()
)
COMPACT_CONFIG.write_text(
   textwrap.dedent(
       f"""
       embody '{BASE_CONFIG.as_posix()}'
       run_experiment.tag = 'compact_gelu_adamw'
       run_experiment.mannequin = @compact/MLP()
       compact/MLP.hidden_dims = (64, 64, 64)
       compact/MLP.dropout = 0.05
       compact/MLP.use_layernorm = True
       make_optimizer.title="adamw"
       make_optimizer.lr = 0.003
       make_optimizer.weight_decay = 0.001
       """
   ).strip()
)
WIDE_CONFIG.write_text(
   textwrap.dedent(
       f"""
       embody '{BASE_CONFIG.as_posix()}'
       run_experiment.tag = 'wide_relu_sgd'
       run_experiment.mannequin = @huge/MLP()
       huge/MLP.hidden_dims = (128, 128, 128, 64)
       huge/MLP.activation = 'relu'
       huge/MLP.dropout = 0.02
       huge/MLP.use_layernorm = True
       make_optimizer.title="sgd"
       make_optimizer.lr = 0.035
       make_optimizer.momentum = 0.92
       make_optimizer.weight_decay = 0.0005
       bce_with_logits_loss.label_smoothing = 0.0
       """
   ).strip()
)
def run_from_gin_file(config_path, runtime_bindings=None):
   runtime_bindings = runtime_bindings or []
   gin.clear_config()
   gin.parse_config_files_and_bindings(
       config_files=[str(config_path)],
       bindings=runtime_bindings,
       skip_unknown=False,
       finalize_config=True,
   )
   print("nLoaded config file:")
   print(config_path)
   print("nSelected queried parameters:")
   print("match.epochs =", gin.query_parameter("match.epochs"))
   print("make_loader.batch_size =", gin.query_parameter("make_loader.batch_size"))
   print("make_spiral_dataset.noise =", gin.query_parameter("make_spiral_dataset.noise"))
   strive:
       gin.bind_parameter("match.epochs", 999)
   besides RuntimeError as error:
       print("nConfig lock examine:")
       print(str(error).splitlines()[0])
   consequence = run_experiment()
   tag = consequence["tag"]
   out_dir = RUN_DIR / tag
   out_dir.mkdir(mother and father=True, exist_ok=True)
   result_path = out_dir / "consequence.json"
   operative_path = out_dir / "operative_config.gin"
   result_path.write_text(json.dumps(consequence, indent=2))
   operative_path.write_text(gin.operative_config_str())
   print("nSaved:")
   print(result_path)
   print(operative_path)
   return consequence, operative_path
compact_result, compact_operative = run_from_gin_file(
   COMPACT_CONFIG,
   runtime_bindings=[
       "fit.epochs = 45",
       "make_spiral_dataset.noise = 0.18",
       "run_experiment.tag = 'compact_gelu_adamw_runtime_override'",
   ],
)
wide_result, wide_operative = run_from_gin_file(
   WIDE_CONFIG,
   runtime_bindings=[
       "fit.epochs = 45",
       "make_spiral_dataset.noise = 0.18",
       "run_experiment.tag = 'wide_relu_sgd_runtime_override'",
   ],
)

We create the precise Gin configuration information that management the experiment with out modifying the Python supply code. We outline a shared base configuration after which compose two scoped experiments: a compact GELU-based AdamW mannequin and a wider ReLU-based SGD mannequin. We additionally exhibit runtime overrides, parameter queries, config locking, consequence serialization, and operative config export for reproducible experiment monitoring.

Evaluating Outcomes and Exporting the Operative Config

def plot_metric(outcomes, metric, title):
   plt.determine(figsize=(9, 4))
   for end in outcomes:
       epochs = [row["epoch"] for row in consequence["history"]]
       values = [row[metric] for row in consequence["history"]]
       plt.plot(epochs, values, label=consequence["tag"])
   plt.xlabel("Epoch")
   plt.ylabel(metric)
   plt.title(title)
   plt.grid(True, alpha=0.3)
   plt.legend()
   plt.tight_layout()
   plt.present()
plot_metric(
   [compact_result, wide_result],
   "val_loss",
   "Validation Loss Managed by Gin Config",
)
plot_metric(
   [compact_result, wide_result],
   "val_accuracy",
   "Validation Accuracy Managed by Gin Config",
)
abstract = [
   {
       "tag": compact_result["tag"],
       "params": compact_result["parameters"],
       "val_loss": compact_result["final"]["val_loss"],
       "val_accuracy": compact_result["final"]["val_accuracy"],
   },
   {
       "tag": wide_result["tag"],
       "params": wide_result["parameters"],
       "val_loss": wide_result["final"]["val_loss"],
       "val_accuracy": wide_result["final"]["val_accuracy"],
   },
]
print("n" + "=" * 80)
print("Ultimate comparability")
print("=" * 80)
for row in abstract:
   print(
       f"{row['tag']} | "
       f"params={row['params']:,} | "
       f"val_loss={row['val_loss']:.4f} | "
       f"val_acc={row['val_accuracy']:.3f}"
   )
print("n" + "=" * 80)
print("Compact experiment operative config preview")
print("=" * 80)
print(compact_operative.read_text()[:2500])
print("n" + "=" * 80)
print("Generated information")
print("=" * 80)
for path in sorted(ROOT.rglob("*")):
   if path.is_file():
       print(path)

We visualize the validation loss and validation accuracy curves for each Gin-controlled experiments. We summarize the ultimate parameter counts, validation losses, and validation accuracies to obviously evaluate the 2 configurations. We additionally print the operative configuration and the generated information, which give an entire report of the precise settings used throughout execution.

Conclusion

In conclusion, we have now a reproducible experiment-management workflow that demonstrates how Gin Config improves management, traceability, and modularity in PyTorch tasks. We ran a number of scoped experiments from composed .gin information, in contrast AdamW and SGD coaching habits beneath managed dataset and epoch settings, verified configuration locking after parsing, and saved each metrics and operative configs for later inspection. It offers us a sample for scaling Colab experiments into research-grade pipelines, wherein mannequin structure, optimization technique, knowledge era, and coaching schedules should be systematically adjusted with out breaking the core implementation.

Try the Full Codes with Pocket book right here. Additionally, be at liberty to comply with us on Twitter and don’t neglect to affix our 150k+ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you’ll be able to be a part of us on telegram as nicely.

Have to accomplice with us for selling your GitHub Repo OR Hugging Face Web page OR Product Launch OR Webinar and so forth.? Join with us

Sana Hassan, a consulting intern at Marktechpost and dual-degree pupil at IIT Madras, is keen about making use of know-how and AI to deal with real-world challenges. With a eager curiosity in fixing sensible issues, he brings a recent perspective to the intersection of AI and real-life options.