Gin 設定で制御する PyTorch パイプライン構築法
このチュートリアルは、Gin Config を活用して PyTorch の実験パイプラインを構築する方法を示し、コードの安定性を保ちつつ設定変更による自由度を高める実践的なアプローチを提供する。
キーポイント
設定とコードの分離による保守性向上
Gin Config を使用して実験パラメータを宣言的な設定ファイルに移行し、実行コードの変更なしに実験の自由度を大幅に高める手法を紹介している。
スコープ付きバインディングによる柔軟なアーキテクチャ
@gin.configurable を用いて MLP のアーキテクチャ変種やオプティマイザ、スケジューラなどのパラメータをスコープ別に定義し、複雑な構成を管理可能にしている。
ランタイムオーバーライドと設定エクスポート
ソースコードを変更せずに実行時にパラメータを上書きできる機能や、各トレーニングランで生成された正確な解決済み設定を出力する仕組みを実装している。
Ginによる設定可能関数の定義
@gin.configurableデコレータを使用して、シード管理やデータセット生成などの関数を外部設定ファイルから制御可能な形式で定義しています。
スパイラルデータの動的生成と前処理
ノイズ量や回転数などをパラメータ化し、クラスごとの螺旋状データを生成して学習用・検証用に分割する関数を構築しています。
Gin Config と PyTorch の統合
Gin Config を使用して、外部設定ファイルから MLP の構造や最適化パラメータを動的に制御し、実験の再現性を確保します。
非線形データセットとデータローダーのカスタマイズ
スパイラルのような複雑な非線形データを生成し、Gin のバインディングを通じて柔軟に設定可能なデータローダーを構築しています。
重要な引用
"the executable training code remains stable. At the same time, the experimental degrees of freedom are moved into declarative configuration files."
"We use Gin's scoped references to instantiate separate model configurations, runtime bindings to override selected parameters without editing source code..."
@gin.configurable def seed_everything(seed=42):
@gin.configurable(denylist=["x", "y"]) def make_loader(...)
reset Gin's global configuration state so the notebook runs reproducibly
build a configurable DataLoader that Gin can control through external bindings
影響分析・編集コメントを表示
影響分析
この記事は、機械学習の実験環境における「設定管理のベストプラクティス」を具体的に示しており、大規模な実験やチーム開発において再現性を担保し、設定ミスによるトラブルを減らす上で重要な指針となる。特に、コードと設定を明確に分離するアプローチは、複雑なモデルアーキテクチャやハイパーパラメータ探索を行う研究者やエンジニアにとって即座に適用可能な価値がある。
編集コメント
実験の再現性と管理効率を劇的に向上させるための、実務レベルで即戦力となる技術的アプローチが提示されています。コードと設定を分離するこの手法は、複雑化するモデル開発プロセスにおいて不可欠なスキルセットと言えます。
本チュートリアルでは、実行可能なトレーニングコードを安定させたまま、Gin Config を制御する PyTorch の実験パイプラインを実装します。これにより、実験の自由度は宣言型の設定ファイルへと移管されます。
具体的には、非線形スパイラルの二値分類タスクを構築し、スコープ付きアーキテクチャバリアントを持つ構成可能な MLP（多層パーセプトロン）を定義します。また、@gin.configurable バインディングを通じて、オプティマイザ、スケジューラ、損失関数、バッチ処理、シード設定、トレーニングループに関するパラメータを公開します。
Gin のスコープ参照機能を用いて別々のモデル構成をインスタンス化し、ソースコードを編集せずにランタイムバインディングで特定のパラメータを上書きできるようにします。さらに、各トレーニング実行で生成された正確な解決済み設定をキャプチャするための、運用可能な設定エクスポート機能も実装しています。
Gin Config のインストールとスパイラルデータセットの構築
コードをコピーしてブラウザを変更してください
!pip -q install 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.functional as F
from torch.utils.data import TensorDataset, DataLoader
import matplotlib.pyplot as plt
ROOT = Path("/content/gin_config_sharp_tutorial")
CONFIG_DIR = ROOT / "configs"
RUN_DIR = ROOT / "runs"
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
RUN_DIR.mkdir(parents=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.normal(0.0, noise, size=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.normal(0.0, noise, size=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]
split = int(train_fraction * len(x))
x_train, y_train = x[:split], y[:split]
x_val, y_val = x[split:], y[split:]
mean = x_train.mean(axis=0, keepdims=True)
std = x_train.std(axis=0, keepdims=True) + 1e-8
x_train = (x_train - mean) / std
x_val = (x_val - mean) / std
return {
"train": (
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.shape[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,
)
まずは Gin Config をインストールし、実験に必要なコアとなる Python ライブラリである PyTorch と NumPy、そして可視化用のプロットライブラリをインポートします。その後、クリーンなプロジェクトディレクトリ構造を作成し、ノートブックが再現性を持って実行できるよう、Gin のグローバル設定状態をリセットします。
次にシード関数を定義し、非線形なスパイラルデータセットを生成。さらに Gin が外部バインディングを通じて制御可能な DataLoader を構築します。
Gin 対応の MLP、オプティマイザ、スケジューラの定義
Copy CodeCopiedUse a different Browser
このコードは、Gin 設定ファイルで制御される PyTorch パイプラインの構築例を示しています。可変的な MLP 構造、コサイン学習率スケジューリング、そして実行時のパラメータ上書き機能を備えています。
まず、活性化関数を選択する activation_layer 関数から始めます。この関数は引数の名前を小文字に変換し、"relu"、"gelu"、"tanh"、"silu" のいずれかを指定すると対応する PyTorch モジュールを返します。もし不明な名前が渡された場合は、エラーを発生させます。
次に、@gin.configurable デコレータで定義される MLP クラスです。これは可変的な多層パーセプトロン（MLP）を実装しています。初期化パラメータには、入力次元（必須）、隠れ層の次元リスト（デフォルトは [64, 64]）、出力次元（デフォルトは 1）、活性化関数の種類（"gelu"）、ドロップアウト率、レイヤーノーマルの使用有無が含まれます。内部では、指定された隠れ層の数だけループを回し、線形層、必要に応じてレイヤーノーマル、活性化関数、そしてドロップアウトを順に追加してネットワークを構築します。最後に出力層を追加し、これらを nn.Sequential で結合して self.network として保存します。
学習器の作成には make_optimizer 関数が使われます。これも Gin の設定対象ですが、引数 params は除外リスト（denylist）に含まれています。デフォルトでは AdamW オプティマイザが使用され、学習率や重み減衰などのパラメータを指定できます。"sgd" を指定すれば、モーメンタム付きの SGD に切り替えることも可能です。不明な名前が渡された場合はエラーになります。
学習率スケジューリングには make_cosine_scheduler 関数が利用されます。これも Gin で設定可能ですが、オプティマイザ自体は除外リストに含まれています。総エポック数、ウォームアップ期間、最小学習率の比率などを指定できます。内部ロジックでは、ウォームアップ期間中は学習率を線形に増加させ、その後はコサイン関数を用いて徐々に減少させる lr_lambda 関数を定義し、PyTorch の LambdaLR スケジューラとして返します。
損失計算には bce_with_logits_loss 関数が使われます。これはロジットに対するバイナリクロスエントロピーを計算するもので、必要に応じてラベルスムージング（label smoothing）も適用できます。スムージングが有効な場合、ターゲット値は平滑化された値に調整されます。
最後に、モデルの評価を行う evaluate 関数です。この関数は @torch.no_grad() デコレータで囲まれ、推論時の計算グラフの構築を抑制します。モデルを評価モードにし、データローダからバッチを取得して順次処理します。各バッチは指定されたデバイスに転送され、ロジットが計算された後、損失関数によって損失値が算出されます。また、シグモイド関数を適用して確率を求め、0.5 を閾値として予測ラベルを作成し、正解数をカウントします。最終的に、平均損失と精度の辞書を返します。
このように、Gin を活用することで、設定ファイルのみを変更するだけでモデル構造や学習パラメータを柔軟に切り替えることができます。実行時にはコマンドライン引数や環境変数から設定を上書きすることも可能で、実験の効率化と再現性の確保に貢献します。
構成可能なモデルを構築するニューラルネットワークの構成要素と、トレーニングに使用するユーティリティを定義します。MLP クラスを作成し、アーキテクチャ、活性化関数、ドロップアウト、層正規化の動作をハードコードされた値ではなく、Gin によって制御できるようにしています。さらに、オプティマイザ、スケジューラ、損失関数、評価関数も構成可能に実装することで、トレーニングパイプラインがモジュール化され、実験準備が整った状態を維持します。
トレーニングループと実験ランナーの実装
Copy CodeCopiedUse a different Browser
@gin.configurable(
denylist=[
"model",
"optimizer",
"scheduler",
"train_loader",
"val_loader",
"device",
]
)
def fit(
model,
optimizer,
scheduler,
train_loader,
val_loader,
device,
epochs=60,
grad_clip_norm=1.0,
log_every=10,
loss_fn=bce_with_logits_loss,
):
history = []
for epoch in range(1, epochs + 1):
model.train()
for x, y in train_loader:
x = x.to(device)
y = y.to(device)
optimizer.zero_grad(set_to_none=True)
logits = model(x)
loss = loss_fn(logits, y)
loss.backward()
if grad_clip_norm is not None:
nn.utils.clip_grad_norm_(
model.parameters(),
grad_clip_norm,
)
optimizer.step()
if scheduler is not None:
scheduler.step()
train_metrics = evaluate(
model,
train_loader,
loss_fn,
device,
)
val_metrics = evaluate(
model,
val_loader,
loss_fn,
device,
)
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"],
}
history.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 history
@gin.configurable
def run_experiment(
tag=gin.REQUIRED,
model=gin.REQUIRED,
dataset_fn=make_spiral_dataset,
optimizer_factory=make_optimizer,
scheduler_factory=make_cosine_scheduler,
prefer_gpu=True,
):
seed_everything()
device = "cuda" if prefer_gpu and torch.cuda.is_available() else "cpu"
data = dataset_fn()
x_train, y_train = data["train"]
x_val, y_val = data["val"]
train_loader = make_loader(
x_train,
y_train,
shuffle=True,
)
val_loader = make_loader(
x_val,
y_val,
shuffle=False,
)
model = model.to(device)
optimizer = optimizer_factory(model.parameters())
scheduler = None
if scheduler_factory is not None:
scheduler = scheduler_factory(optimizer)
print("\n" + "=" * 80)
print(f"Experiment: {tag}")
print("=" * 80)
print(f"Device: {device}")
print(f"Dataset: {data['metadata']}")
print(f"Parameters: {sum(p.numel() for p in model.parameters()):,}")
history = fit(
model=model,
optimizer=optimizer,
scheduler=scheduler,
train_loader=train_loader,
val_loader=val_loader,
device=device,
)
result = {
"tag": tag,
"device": device,
"metadata": data["metadata"],
"parameters": sum(p.numel() for p in model.parameters()),
"final": history[-1],
"history": history,
}
return result
メインの学習ループを実装します。ここではモデルが順伝播を行い、二値交差エントロピー誤差を計算し、勾配の逆伝播を行って勾配クリッピングを適用した後、パラメータを更新します。各エポック終了後にトレーニングセットと検証セットの両方でモデルを評価し、損失、精度、学習率の履歴を記録します。
その後、Gin によって管理される依存関係を通じてデータセット、モデル、オプティマイザ、スケジューラ、そして学習ループをつなぐトップレベルの実行ランナーを定義します。
スコープ付きバインディングとランタイムオーバーライドを用いた Gin Config ファイルの記述
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.name = '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
fit.epochs = %EPOCHS
fit.grad_clip_norm = 1.0
fit.log_every = 10
fit.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"""
include '{BASE_CONFIG.as_posix()}'
run_experiment.tag = 'compact_gelu_adamw'
run_experiment.model = @compact/MLP()
compact/MLP.hidden_dims = (64, 64, 64)
compact/MLP.dropout = 0.05
compact/MLP.use_layernorm = True
make_optimizer.name = 'adamw'
make_optimizer.lr = 0.003
make_optimizer.weight_decay = 0.001
"""
).strip()
)
WIDE_CONFIG.write_text(
textwrap.dedent(
f"""
include '{BASE_CONFIG.as_posix()}'
run_experiment.tag = 'wide_relu_sgd'
run_experiment.model = @wide/MLP()
wide/MLP.hidden_dims = (128, 128, 128, 64)
wide/MLP.activation = 'relu'
wide/MLP.dropout = 0.02
wide/MLP.use_layernorm = True
make_optimizer.name = '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("fit.epochs =", gin.query_parameter("fit.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"))
try:
gin.bind_parameter("fit.epochs", 999)
except RuntimeError as error:
print("\nConfig lock check:")
print(str(error).splitlines()[0])
result = run_experiment()
tag = result["tag"]
out_dir = RUN_DIR / tag
out_dir.mkdir(parents=True, exist_ok=True)
result_path = out_dir / "result.json"
operative_path = out_dir / "operative_config.gin"
result_path.write_text(json.dumps(result, indent=2))
operative_path.write_text(gin.operative_config_str())
print("\nSaved:")
print(result_path)
print(operative_path)
return result, 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'",
],
)
Python ソースコードを変更せずに、実験を制御する Gin 設定ファイルを作成します。共有ベースの設定を定義した上で、2 つのスコープ限定実験を構成します。1 つはコンパクトな GELU ベースの AdamW モデル、もう 1 つは幅広の ReLU ベースの SGD モデルです。また、ランタイムでのパラメータ上書き、設定値の照会、設定のロック、結果のシリアライズ、再現性のある実験追跡のための運用設定のエクスポート方法も示します。
結果の比較と運用設定のエクスポート
Copy CodeCopiedUse a different Browser
def plot_metric(results, metric, title):
plt.figure(figsize=(9, 4))
for result in results:
epochs = [row["epoch"] for row in result["history"]]
values = [row[metric] for row in result["history"]]
plt.plot(epochs, values, label=result["tag"])
plt.xlabel("Epoch")
plt.ylabel(metric)
plt.title(title)
plt.grid(True, alpha=0.3)
plt.legend()
plt.tight_layout()
plt.show()
plot_metric(
[compact_result, wide_result],
"val_loss",
"Validation Loss Controlled by Gin Config",
)
plot_metric(
[compact_result, wide_result],
"val_accuracy",
"Validation Accuracy Controlled by Gin Config",
)
summary = [
{
"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("Final comparison")
print("=" * 80)
for row in summary:
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 files")
print("=" * 80)
for path in sorted(ROOT.rglob("*")):
if path.is_file():
print(path)
Gin Config を用いた2つの実験における検証損失と検証精度の曲線を可視化しました。また、最終的なパラメータ数、検証損失、検証精度をまとめ、2 つの設定を明確に比較できるようにしています。さらに、実行時に使用された正確な設定を記録するオペレーティブ構成と生成ファイルも出力しています。
結論として、Gin Config が PyTorch プロジェクトにおける制御性、追跡可能性、モジュール性をどのように向上させるかを示す、再現可能な実験管理ワークフローを確立しました。複数のスコープを持つ実験を .gin ファイルから組み合わせて実行し、データセットとエポック設定を統制した下で AdamW と SGD の学習挙動を比較。パース後の構成ロックも検証済みです。後日の確認のためにメトリクスとオペレーティブ構成の両方を保存しました。これにより、モデルアーキテクチャ、最適化戦略、データ生成、トレーニングスケジュールをコア実装を壊すことなく体系的に調整できる、Colab 実験から研究レベルのパイプラインへスケールするためのパターンが得られました。
ノートブック付きの完全なコードはこちらで確認できます。Twitter ではフォローも歓迎しますし、15 万人以上の ML 研究者が集まる SubReddit への参加や、ニュースレターの購読もお忘れなく。あ、Telegram も使っていますか？Telegram でも一緒に活動できますよ。
GitHub リポジトリや Hugging Face ページ、製品リリース、ウェビナーなどのプロモーションを当社と連携して行いたい場合は、ぜひご連絡ください。
この記事は、MarkTechPost にて公開された「可変 MLP バリアント、コサインスケジューリング、ランタイムパラメータの動的変更を備えた Gin 設定制御型 PyTorch パイプラインの構築」シリーズの最終回（12/12）です。
Gin 設定ファイルを活用して PyTorch のパイプラインを制御し、MLP の構造を柔軟に変更可能にし、学習中のスケジューリングをコサインカーブで最適化し、さらに実行時にパラメータを動的に上書きできる仕組みについて解説します。これにより、実験の再現性を高めつつ、迅速なハイパーパラメータチューニングを実現できます。
まず、Gin 設定ファイルの構造と、PyTorch のモデル定義やトレーニングループとの連携方法を確認しましょう。可変 MLP バリアントとは、層の数や幅、活性化関数などを設定ファイルで切り替えることで、同じコードベースで多様なアーキテクチャを試せる機能です。
次に、コサインスケジューリングの導入について触れます。学習率をコサインカーブに沿って変化させることで、収束速度と最終的な精度のバランスを最適化します。従来の定数学習率やステップスケジューリングよりも、より滑らかな学習曲線を得られることが期待されます。
さらに、ランタイムパラメータの動的上書き機能について説明します。トレーニング実行中に設定ファイルを読み直すことなく、プログラム内で指定した値で即座にパラメータを変更できるため、オンラインでの実験やデバッグが容易になります。
最後に、これらすべての機能を統合した実装例を示し、実際の使用シナリオを想定したコードの書き方を解説します。Gin の設定ファイル記法から PyTorch のデータローダー、モデル定義、オプティマイザ、スケジューラまでの一連の流れを、可読性と拡張性を両立させる形で構築する方法を学びます。
本シリーズを通じて、複雑な実験環境でも柔軟かつ堅牢にパイプラインを管理するためのベストプラクティスを身につけてください。次回のシリーズでは、分散学習や大規模モデルへの拡張について取り上げる予定です。
Gin 設定ファイルの記法と、PyTorch モデルとのマッピング方法について詳しく解説します。設定ファイル内で定義された変数が、どのようにして Python コード内のオブジェクトにバインドされるか、その仕組みを理解することが重要です。
可変 MLP バリアントの実装では、設定ファイルで指定されたパラメータに基づいて、動的に層の構成を生成するファクトリ関数を使用します。これにより、コードの変更なしに異なるアーキテクチャを試すことが可能になります。
コサインスケジューリングの実装は、PyTorch の学習率スケジューラクラスを拡張するか、または Gin 設定で指定されたパラメータを用いてカスタムスケジューラを作成することで実現します。学習の進行度に応じて、滑らかに学習率を変化させるロジックを実装します。
ランタイムパラメータの上書き機能は、Gin の override 構文や、プログラム内で設定値を直接参照して更新する仕組みを用いて実装されます。これにより、トレーニング実行中に特定のハイパーパラメータを即座に変更し、その影響を即時に確認できます。
これらの機能を統合したパイプラインの全体像を示すコード例では、データ読み込みからモデル定義、学習ループ、評価までの一連の流れを、Gin 設定ファイルで制御可能な形で記述します。各コンポーネントが独立してテスト可能でありながら、設定ファイルを通じてシームレスに連携する構造を目指します。
実装のポイントは、設定ファイルの変更だけで実験条件を変えられるようにすることと、ランタイムでのパラメータ変更が安全に行えるようにすることです。エラーハンドリングや設定値の妥当性チェックも忘れずに行いましょう。
本シリーズの最後として、これらの技術を実際の研究プロジェクトに適用する際の注意点や、パフォーマンス最適化のヒントについても触れます。Gin 設定ファイルの管理方法や、バージョン管理との連携など、実務で役立つ情報も提供します。
読者の皆様には、本記事で紹介した手法を実際に試していただき、自身の研究や開発に活かしていただければ幸いです。質問やご意見は、コメント欄または Twitter などでお気軽にお寄せください。
次回のシリーズでは、分散学習環境でのパイプライン構築や、大規模モデルへの拡張について取り上げる予定です。引き続きご注目ください。
本シリーズの全記事一覧と、関連するリポジトリへのリンクも合わせてご紹介します。実装コードは GitHub で公開されており、誰でも自由に利用・改変可能です。
以上で「Building a Gin Config Controlled PyTorch Pipeline with Configurable MLP Variants, Cosine Scheduling, and Runtime Parameter Overrides」シリーズ（12/12）を終了します。ご清聴ありがとうございました。
原文を表示
In this tutorial, we implement a Gin Config–controlled PyTorch experiment pipeline in which the executable training code remains stable. At the same time, the experimental degrees of freedom are moved into declarative configuration files. We construct a nonlinear spiral binary classification task, define a configurable MLP with scoped architectural variants, and expose parameters for the optimizer, scheduler, loss, batching, seeding, and training loop via @gin.configurable bindings. We use Gin’s scoped references to instantiate separate model configurations, runtime bindings to override selected parameters without editing source code, and operative config export to capture the exact resolved configuration that produces each training run.
Installing Gin Config and Building the Spiral Dataset
Copy CodeCopiedUse a different Browser
!pip -q install 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.functional as F
from torch.utils.data import TensorDataset, DataLoader
import matplotlib.pyplot as plt
ROOT = Path("/content/gin_config_sharp_tutorial")
CONFIG_DIR = ROOT / "configs"
RUN_DIR = ROOT / "runs"
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
RUN_DIR.mkdir(parents=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.normal(0.0, noise, size=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.normal(0.0, noise, size=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]
split = int(train_fraction * len(x))
x_train, y_train = x[:split], y[:split]
x_val, y_val = x[split:], y[split:]
mean = x_train.mean(axis=0, keepdims=True)
std = x_train.std(axis=0, keepdims=True) + 1e-8
x_train = (x_train - mean) / std
x_val = (x_val - mean) / std
return {
"train": (
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.shape[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 start by installing Gin Config and importing the core Python libraries, PyTorch, NumPy, and the plotting libraries required for the experiment. We create a clean project directory structure and reset Gin’s global configuration state so the notebook runs reproducibly. We then define the seed function, generate a nonlinear spiral dataset, and build a configurable DataLoader that Gin can control through external bindings.
Defining a Gin-Configurable MLP, Optimizer, and Scheduler
Copy CodeCopiedUse a different Browser
def activation_layer(name):
name = name.lower()
if name == "relu":
return nn.ReLU()
if name == "gelu":
return nn.GELU()
if name == "tanh":
return nn.Tanh()
if name == "silu":
return nn.SiLU()
raise ValueError(f"Unknown activation: {name}")
@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,
):
super().__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.network = nn.Sequential(*layers)
def forward(self, x):
return self.network(x)
@gin.configurable(denylist=["params"])
def make_optimizer(
params,
name="adamw",
lr=3e-3,
weight_decay=1e-3,
momentum=0.9,
):
name = name.lower()
if name == "adamw":
return torch.optim.AdamW(
params,
lr=lr,
weight_decay=weight_decay,
)
if name == "sgd":
return torch.optim.SGD(
params,
lr=lr,
momentum=momentum,
weight_decay=weight_decay,
)
raise ValueError(f"Unknown optimizer: {name}")
@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 evaluate(model, loader, loss_fn, device):
model.eval()
total_loss = 0.0
total_correct = 0
total_count = 0
for x, y in loader:
x = x.to(device)
y = y.to(device)
logits = model(x)
loss = loss_fn(logits, y)
probs = torch.sigmoid(logits)
preds = (probs >= 0.5).float()
total_loss += loss.item() * len(x)
total_correct += (preds == y).sum().item()
total_count += len(x)
return {
"loss": total_loss / total_count,
"accuracy": total_correct / total_count,
}
We define the neural network building blocks that form the configurable model and the training utilities. We create an MLP class whose architecture, activation function, dropout, and layer normalization behavior are controlled through Gin rather than hardcoded values. We also implement configurable optimizer, scheduler, loss, and evaluation functions so the training pipeline remains modular and experiment-ready.
Implementing the Training Loop and Experiment Runner
Copy CodeCopiedUse a different Browser
@gin.configurable(
denylist=[
"model",
"optimizer",
"scheduler",
"train_loader",
"val_loader",
"device",
]
)
def fit(
model,
optimizer,
scheduler,
train_loader,
val_loader,
device,
epochs=60,
grad_clip_norm=1.0,
log_every=10,
loss_fn=bce_with_logits_loss,
):
history = []
for epoch in range(1, epochs + 1):
model.train()
for x, y in train_loader:
x = x.to(device)
y = y.to(device)
optimizer.zero_grad(set_to_none=True)
logits = model(x)
loss = loss_fn(logits, y)
loss.backward()
if grad_clip_norm is not None:
nn.utils.clip_grad_norm_(
model.parameters(),
grad_clip_norm,
)
optimizer.step()
if scheduler is not None:
scheduler.step()
train_metrics = evaluate(
model,
train_loader,
loss_fn,
device,
)
val_metrics = evaluate(
model,
val_loader,
loss_fn,
device,
)
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"],
}
history.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 history
@gin.configurable
def run_experiment(
tag=gin.REQUIRED,
model=gin.REQUIRED,
dataset_fn=make_spiral_dataset,
optimizer_factory=make_optimizer,
scheduler_factory=make_cosine_scheduler,
prefer_gpu=True,
):
seed_everything()
device = "cuda" if prefer_gpu and torch.cuda.is_available() else "cpu"
data = dataset_fn()
x_train, y_train = data["train"]
x_val, y_val = data["val"]
train_loader = make_loader(
x_train,
y_train,
shuffle=True,
)
val_loader = make_loader(
x_val,
y_val,
shuffle=False,
)
model = model.to(device)
optimizer = optimizer_factory(model.parameters())
scheduler = None
if scheduler_factory is not None:
scheduler = scheduler_factory(optimizer)
print("\n" + "=" * 80)
print(f"Experiment: {tag}")
print("=" * 80)
print(f"Device: {device}")
print(f"Dataset: {data['metadata']}")
print(f"Parameters: {sum(p.numel() for p in model.parameters()):,}")
history = fit(
model=model,
optimizer=optimizer,
scheduler=scheduler,
train_loader=train_loader,
val_loader=val_loader,
device=device,
)
result = {
"tag": tag,
"device": device,
"metadata": data["metadata"],
"parameters": sum(p.numel() for p in model.parameters()),
"final": history[-1],
"history": history,
}
return result
We implement the main training loop, in which the model performs forward passes, computes binary cross-entropy loss, backpropagates gradients, applies gradient clipping, and updates parameters. We evaluate the model after each epoch on both the training and validation sets, while storing loss, accuracy, and learning rate history. We then define the top-level experiment runner that connects the dataset, model, optimizer, scheduler, and training loop through Gin-managed dependencies.
Writing Gin Config Files with Scoped Bindings and Runtime Overrides
Copy CodeCopiedUse a different Browser
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.name = '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
fit.epochs = %EPOCHS
fit.grad_clip_norm = 1.0
fit.log_every = 10
fit.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"""
include '{BASE_CONFIG.as_posix()}'
run_experiment.tag = 'compact_gelu_adamw'
run_experiment.model = @compact/MLP()
compact/MLP.hidden_dims = (64, 64, 64)
compact/MLP.dropout = 0.05
compact/MLP.use_layernorm = True
make_optimizer.name = 'adamw'
make_optimizer.lr = 0.003
make_optimizer.weight_decay = 0.001
"""
).strip()
)
WIDE_CONFIG.write_text(
textwrap.dedent(
f"""
include '{BASE_CONFIG.as_posix()}'
run_experiment.tag = 'wide_relu_sgd'
run_experiment.model = @wide/MLP()
wide/MLP.hidden_dims = (128, 128, 128, 64)
wide/MLP.activation = 'relu'
wide/MLP.dropout = 0.02
wide/MLP.use_layernorm = True
make_optimizer.name = '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("fit.epochs =", gin.query_parameter("fit.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"))
try:
gin.bind_parameter("fit.epochs", 999)
except RuntimeError as error:
print("\nConfig lock check:")
print(str(error).splitlines()[0])
result = run_experiment()
tag = result["tag"]
out_dir = RUN_DIR / tag
out_dir.mkdir(parents=True, exist_ok=True)
result_path = out_dir / "result.json"
operative_path = out_dir / "operative_config.gin"
result_path.write_text(json.dumps(result, indent=2))
operative_path.write_text(gin.operative_config_str())
print("\nSaved:")
print(result_path)
print(operative_path)
return result, 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 actual Gin configuration files that control the experiment without modifying the Python source code. We define a shared base configuration and then compose two scoped experiments: a compact GELU-based AdamW model and a wider ReLU-based SGD model. We also demonstrate runtime overrides, parameter queries, config locking, result serialization, and operative config export for reproducible experiment tracking.
Comparing Results and Exporting the Operative Config
Copy CodeCopiedUse a different Browser
def plot_metric(results, metric, title):
plt.figure(figsize=(9, 4))
for result in results:
epochs = [row["epoch"] for row in result["history"]]
values = [row[metric] for row in result["history"]]
plt.plot(epochs, values, label=result["tag"])
plt.xlabel("Epoch")
plt.ylabel(metric)
plt.title(title)
plt.grid(True, alpha=0.3)
plt.legend()
plt.tight_layout()
plt.show()
plot_metric(
[compact_result, wide_result],
"val_loss",
"Validation Loss Controlled by Gin Config",
)
plot_metric(
[compact_result, wide_result],
"val_accuracy",
"Validation Accuracy Controlled by Gin Config",
)
summary = [
{
"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("Final comparison")
print("=" * 80)
for row in summary:
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 files")
print("=" * 80)
for path in sorted(ROOT.rglob("*")):
if path.is_file():
print(path)
We visualize the validation loss and validation accuracy curves for both Gin-controlled experiments. We summarize the final parameter counts, validation losses, and validation accuracies to clearly compare the two configurations. We also print the operative configuration and the generated files, which provide a complete record of the exact settings used during execution.
Conclusion
In conclusion, we have a reproducible experiment-management workflow that demonstrates how Gin Config improves control, traceability, and modularity in PyTorch projects. We ran multiple scoped experiments from composed .gin files, compared AdamW and SGD training behavior under controlled dataset and epoch settings, verified configuration locking after parsing, and saved both metrics and operative configs for later inspection. It gives us a pattern for scaling Colab experiments into research-grade pipelines, in which model architecture, optimization strategy, data generation, and training schedules must be systematically adjusted without breaking the core implementation.
Check out the Full Codes with Notebook here. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
The post Building a Gin Config Controlled PyTorch Pipeline with Configurable MLP Variants, Cosine Scheduling, and Runtime Parameter Overrides appeared first on MarkTechPost.
関連記事
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み