確率的勾配降下法 | MNIST データセット | 早期終了

はじめに

確率的勾配降下法（Stochastic Gradient Descent）は、損失関数を最小化するために使用される一般的な最適化手法です。この手法は、確率的に各反復でサンプルを選択することで、勾配降下法を段階的に実行します。この方法は、特に線形モデルのフィッティングに効率的です。ただし、各反復で収束が保証されるわけではなく、各反復で損失関数が必ずしも減少するとは限りません。この場合、損失関数の収束を監視するのは難しい場合があります。この実験では、検証スコアの収束を監視するアプローチである早期終了戦略を探ります。scikit-learn ライブラリのSGDClassifierモデルと MNIST データセットを使用して、早期終了を使用した場合と早期終了を使用せずに構築したモデルと比較して、ほぼ同じ精度を達成し、学習時間を大幅に短縮できる方法を示します。

VM のヒント

VM の起動が完了したら、左上隅をクリックしてノートブックタブに切り替え、Jupyter Notebook を使って練習しましょう。

時々、Jupyter Notebook が読み込み終了するまで数秒待つ必要がある場合があります。Jupyter Notebook の制限により、操作の検証を自動化することはできません。

学習中に問題に遭遇した場合は、Labby にお問い合わせください。セッション後にフィードバックを提供してください。すぐに問題を解決いたします。

必要なライブラリと MNIST データセットを読み込む

最初のステップは、必要なライブラリとデータセットを読み込むことです。pandas、numpy、matplotlib、およびscikit-learnライブラリを使用します。また、scikit-learn のfetch_openml関数を使って MNIST データセットを読み込みます。

import time
import sys
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn import linear_model
from sklearn.datasets import fetch_openml
from sklearn.model_selection import train_test_split
from sklearn.utils._testing import ignore_warnings
from sklearn.exceptions import ConvergenceWarning
from sklearn.utils import shuffle

## MNIST データセットを読み込む
def load_mnist(n_samples=None, class_0="0", class_1="8"):
    """MNIST を読み込み、2 つのクラスを選択し、シャッフルして n_samples のみを返す。"""
    ## http://openml.org/d/554からデータを読み込む
    mnist = fetch_openml("mnist_784", version=1, as_frame=False, parser="pandas")

    ## 2 値分類のために 2 つのクラスのみを取り出す
    mask = np.logical_or(mnist.target == class_0, mnist.target == class_1)

    X, y = shuffle(mnist.data[mask], mnist.target[mask], random_state=42)
    if n_samples is not None:
        X, y = X[:n_samples], y[:n_samples]
    return X, y

X, y = load_mnist(n_samples=10000)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.5, random_state=0)

推定器と早期終了戦略を定義する

次のステップは、推定器と早期終了戦略を定義することです。scikit-learn のSGDClassifierモデルを使用します。3 つの異なる停止基準を定義します：停止基準なし、学習損失、検証スコア。fit_and_score関数を使用して、推定器をトレーニングセットに適合させ、両方のセットでスコア付けします。

@ignore_warnings(category=ConvergenceWarning)
def fit_and_score(estimator, max_iter, X_train, X_test, y_train, y_test):
    """推定器をトレーニングセットに適合させ、両方のセットでスコア付けする"""
    estimator.set_params(max_iter=max_iter)
    estimator.set_params(random_state=0)

    start = time.time()
    estimator.fit(X_train, y_train)

    fit_time = time.time() - start
    n_iter = estimator.n_iter_
    train_score = estimator.score(X_train, y_train)
    test_score = estimator.score(X_test, y_test)

    return fit_time, n_iter, train_score, test_score

## 比較する推定器を定義する
estimator_dict = {
    "停止基準なし": linear_model.SGDClassifier(n_iter_no_change=3),
    "学習損失": linear_model.SGDClassifier(
        early_stopping=False, n_iter_no_change=3, tol=0.1
    ),
    "検証スコア": linear_model.SGDClassifier(
        early_stopping=True, n_iter_no_change=3, tol=0.0001, validation_fraction=0.2
    ),
}

推定器を学習して評価する

次のステップは、各停止基準を使用して推定器を学習して評価することです。各推定器と停止基準を反復処理するためのループを使用し、さらに別のループを使用して異なる最大反復回数を反復処理します。その後、結果を簡単にプロットするために pandas のデータフレームに格納します。

results = []
for estimator_name, estimator in estimator_dict.items():
    print(estimator_name + ": ", end="")
    for max_iter in range(1, 50):
        print(".", end="")
        sys.stdout.flush()

        fit_time, n_iter, train_score, test_score = fit_and_score(
            estimator, max_iter, X_train, X_test, y_train, y_test
        )

        results.append(
            (estimator_name, max_iter, fit_time, n_iter, train_score, test_score)
        )
    print("")

## 結果を簡単にプロットするために pandas のデータフレームに変換する
columns = [
    "停止基準",
    "max_iter",
    "学習時間 (秒)",
    "n_iter_",
    "トレーニングスコア",
    "テストスコア",
]
results_df = pd.DataFrame(results, columns=columns)

結果をプロットする

最後のステップは、結果をプロットすることです。学習とテストのスコア、および反復回数と学習時間をプロットするために 2 つのサブプロットを使用します。各推定器と停止基準に対して異なる線のスタイルを使用します。

## プロットする内容を定義する
lines = "停止基準"
x軸 = "max_iter"
スタイル = ["-.", "--", "-"]

## 最初のプロット：学習とテストのスコア
fig, axes = plt.subplots(nrows=1, ncols=2, sharey=True, figsize=(12, 4))
for ax, y軸 in zip(axes, ["学習スコア", "テストスコア"]):
    for スタイル, (基準, group_df) in zip(スタイル, results_df.groupby(lines)):
        group_df.plot(x=x軸, y=y軸, label=基準, ax=ax, style=スタイル)
    ax.set_title(y軸)
    ax.legend(title=lines)
fig.tight_layout()

## 2 番目のプロット：n_iter と学習時間
fig, axes = plt.subplots(nrows=1, ncols=2, figsize=(12, 4))
for ax, y軸 in zip(axes, ["n_iter_", "学習時間 (秒)"]):
    for スタイル, (基準, group_df) in zip(スタイル, results_df.groupby(lines)):
        group_df.plot(x=x軸, y=y軸, label=基準, ax=ax, style=スタイル)
    ax.set_title(y軸)
    ax.legend(title=lines)
fig.tight_layout()

plt.show()

まとめ

この実験では、損失関数を最小化するために確率的勾配降下法を使用する際に、検証スコアに基づいて収束を監視するための早期終了戦略を検討しました。scikit-learn のSGDClassifierモデルと MNIST データセットを使用して、早期終了を使用した場合と早期終了を使用せずに構築されたモデルと比較して、ほぼ同じ精度を達成し、学習時間を大幅に短縮できる方法を示しました。3 つの異なる停止基準：停止基準なし、学習損失、検証スコアを定義し、各停止基準を使用して推定器を学習および評価するためのループを使用しました。その後、各推定器と停止基準に対して異なる線のスタイルを使用して結果をプロットしました。

確率的勾配降下法の早期終了