model-training

モデルのトレーニング

このスキルを使用すると、AI エージェントは構造化データセットまたは非構造化データセットを使って機械学習モデルをトレーニングできます。データの読み込みと前処理、モデルアーキテクチャの定義、オプティマイザーと損失関数の設定、検証を伴うトレーニングループの実行、学習率スケジューリングの適用、チェックポイントの保存まで、トレーニングのライフサイクル全体を扱います。PyTorch、TensorFlow、scikit-learn などのフレームワークを使用した、従来型の機械学習とディープラーニングの両方のワークフローに対応します。

ワークフロー

1. データの読み込みと調査： ディスク、データベース、リモートストレージからデータセットを読み込みます。特徴量の分布、クラスバランス、欠損値、データ型を把握するために、データのプロファイリングを行います。クラスの不均衡がある場合は、層化サンプリングを使用して、トレーニングセット、検証セット、テストセットに分割します。

2. 前処理と変換： 正規化、標準化、トークン化（テキストの場合）、データ拡張（画像の場合）などの特徴量エンジニアリングを適用します。推論時にも同じ変換を適用できるよう、再現可能でシリアライズ可能な前処理パイプラインを構築します。

3. モデルアーキテクチャの定義： タスクに適したモデルアーキテクチャを選択または構築します。従来型の機械学習では、勾配ブースティングや SVM などの推定器を選びます。ディープラーニングでは、層、活性化関数、ドロップアウトや重み減衰などの正則化を定義します。転移学習が適用できる場合は、事前学習済みのバックボーンを読み込み、タスク固有のヘッドを追加します。

4. トレーニングの設定： オプティマイザー（Adam、SGD、AdamW）、損失関数（交差エントロピー、MSE、焦点損失）、学習率スケジュール（コサインアニーリング、ステップ減衰、ウォームアップ）、バッチサイズを設定します。GPU でのトレーニング時に、メモリ使用量を削減し計算を高速化するには、torch.amp または tf.keras.mixed_precision を使用した混合精度トレーニングを有効にします。

5. 検証を伴うトレーニングループの実行： 指定されたエポック数だけトレーニングを行い、バッチまたはエポックごとにトレーニング損失とメトリクスを記録します。定期的に検証セットで評価します。検証性能が設定可能なエポック数（patience）にわたって停滞した場合にトレーニングを停止する、早期終了を実装します。

6. チェックポイントの保存とエクスポート： 検証スコアが最高のときと、定期的なタイミングでモデルのチェックポイントを保存します。後続のデプロイに備えて、最終モデルをポータブルな形式（ONNX、TorchScript、SavedModel）でエクスポートします。すべてのハイパーパラメーターとメトリクスを、MLflow や Weights & Biases などの実験トラッカーに記録します。

対応テクノロジー

• フレームワーク： PyTorch、TensorFlow/Keras、scikit-learn、XGBoost、LightGBM
• 分散トレーニング： PyTorch DDP、Horovod、TensorFlow MirroredStrategy
• 実験トラッキング： MLflow、Weights & Biases、TensorBoard
• 混合精度： torch.amp、tf.keras.mixed_precision
• データ読み込み： PyTorch DataLoader、tf.data、pandas、Hugging Face Datasets

使用方法

データセットの場所、目的変数またはタスクの説明、制約（優先するフレームワーク、計算予算、目標メトリクス）をエージェントに伝えてください。エージェントはトレーニングワークフロー全体を実行し、評価メトリクスとともにトレーニング済みモデルの成果物を返します。

例

例 1：PyTorch によるテキスト分類器のトレーニング

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from collections import Counter

# Simulated tokenized text data: 2000 samples, sequence length 50, vocab size 5000
X = torch.randint(0, 5000, (2000, 50))
y_raw = ["positive"] * 1000 + ["negative"] * 1000
le = LabelEncoder()
y = torch.tensor(le.fit_transform(y_raw), dtype=torch.long)

X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, stratify=y, random_state=42)
train_loader = DataLoader(TensorDataset(X_train, y_train), batch_size=64, shuffle=True)
val_loader = DataLoader(TensorDataset(X_val, y_val), batch_size=64)

class TextClassifier(nn.Module):
    def __init__(self, vocab_size=5000, embed_dim=128, num_classes=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, 64, batch_first=True, bidirectional=True)
        self.dropout = nn.Dropout(0.3)
        self.fc = nn.Linear(128, num_classes)

    def forward(self, x):
        x = self.embedding(x)
        _, (hidden, _) = self.lstm(x)
        hidden = torch.cat((hidden[-2], hidden[-1]), dim=1)
        return self.fc(self.dropout(hidden))

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = TextClassifier().to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-2)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=10)
criterion = nn.CrossEntropyLoss()

best_val_acc, patience, patience_counter = 0.0, 3, 0
for epoch in range(10):
    model.train()
    for xb, yb in train_loader:
        xb, yb = xb.to(device), yb.to(device)
        loss = criterion(model(xb), yb)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    scheduler.step()

    model.eval()
    correct, total = 0, 0
    with torch.no_grad():
        for xb, yb in val_loader:
            xb, yb = xb.to(device), yb.to(device)
            correct += (model(xb).argmax(1) == yb).sum().item()
            total += yb.size(0)
    val_acc = correct / total
    print(f"Epoch {epoch+1}: val_acc={val_acc:.4f}")

    if val_acc > best_val_acc:
        best_val_acc = val_acc
        torch.save(model.state_dict(), "best_model.pt")
        patience_counter = 0
    else:
        patience_counter += 1
        if patience_counter >= patience:
            print("Early stopping triggered.")
            break

例 2：Hugging Face Transformers による事前学習済みモデルのファインチューニング

from datasets import load_dataset
from transformers import AutoTokenizer, AutoModelForSequenceClassification, TrainingArguments, Trainer
import numpy as np
from sklearn.metrics import accuracy_score, f1_score

dataset = load_dataset("imdb")
tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")

def tokenize(batch):
    return tokenizer(batch["text"], padding="max_length", truncation=True, max_length=256)

tokenized = dataset.map(tokenize, batched=True)
tokenized.set_format("torch", columns=["input_ids", "attention_mask", "label"])

model = AutoModelForSequenceClassification.from_pretrained("distilbert-base-uncased", num_labels=2)

def compute_metrics(eval_pred):
    preds = np.argmax(eval_pred.predictions, axis=1)
    return {"accuracy": accuracy_score(eval_pred.label_ids, preds), "f1": f1_score(eval_pred.label_ids, preds)}

training_args = TrainingArguments(
    output_dir="./results", num_train_epochs=3, per_device_train_batch_size=16,
    per_device_eval_batch_size=32, eval_strategy="epoch", save_strategy="epoch",
    load_best_model_at_end=True, metric_for_best_model="f1", fp16=True,
    learning_rate=2e-5, weight_decay=0.01, warmup_steps=500, logging_steps=100,
)

trainer = Trainer(model=model, args=training_args, train_dataset=tokenized["train"],
                  eval_dataset=tokenized["test"], compute_metrics=compute_metrics)
trainer.train()
trainer.save_model("./best_model")

ベストプラクティス

• 不均衡なデータセットを扱う場合は、必ず層化して分割してください。各分割に実際のクラス分布が反映されるようにします。
• 事前学習済みモデルをファインチューニングする場合は、学習率のウォームアップを使用してください。学習の初期段階での壊滅的忘却を防げます。
• GPU で学習する場合は、混合精度（fp16 または bf16）を有効にしてください。メモリ使用量をおよそ半分に抑え、スループットを向上させられます。
• すべての情報を実験トラッカーに記録してください。ハイパーパラメーター、エポックごとのメトリクス、システムリソースの使用状況、学習コードの git ハッシュを記録します。
• チェックポイントをこまめに保存し、常に検証結果が最良のチェックポイントを保持してください。クラッシュや過学習によって進捗が失われるのを防げます。
• 学習にもハイパーパラメーターの選定にも使用していないホールドアウトセットで検証してください。汎化性能を偏りなく推定できます。

エッジケース

• 小規模なデータセット（サンプル数 < 1000）： 単一の学習／検証分割ではなく、k 分割交差検証を使用してください。スクラッチから学習するよりも、転移学習や事前学習済みモデルを優先してください。
• 極端なクラス不均衡（比率 > 100:1）： オーバーサンプリング（SMOTE）、クラス重み付き損失関数、またはフォーカルロスを使用してください。評価には正解率ではなく、F1、適合率・再現率 AUC、またはマシューズ相関係数を使用してください。
• 学習の発散または NaN 損失： 学習率を下げ、勾配クリッピング（torch.nn.utils.clip_grad_norm_）を適用し、無限値などのデータの問題を確認するか、数値的不安定性を切り分けるために混合精度を無効にしてください。
• メモリ不足エラー： バッチサイズを小さくし、勾配累積を有効にし、混合精度を使用するか、勾配チェックポイントに切り替えて、計算量と引き換えにメモリ使用量を抑えてください。
• 非定常データ（概念ドリフト）： 新しいデータで定期的に再学習し、ランダム分割ではなく時間ベースの学習／検証分割を使用して、性能低下を検出できるよう本番メトリクスを監視してください。