今日も窓辺でプログラム

外資系企業勤めのエンジニアが勉強した内容をまとめておくブログ

CNTKでロジスティック回帰を試してみました

はじめに

CNTKのチュートリアルの一つ目の題材がロジスティック回帰だったので、チュートリアルに沿ってロジスティック回帰をしてみます。
今回は、下記のチュートリアルを追いかけたものを日本語で解説しているような記事になります。
CNTK/CNTK_101_LogisticRegression.ipynb at v2.0.beta9.0 · microsoft/CNTK · GitHub

CNTKの環境構築は前回の記事で行っているので、環境が整っていない方はこちらもご参照ください。
www.madopro.net

(2017/1/30追記)
なぜかチュートリアルの日本語版が存在していました。ほかの言語は用意されていないっぽいのに、なぜ。

使用したJupyter Notebook

今回使用したJupyter NotebookをGitHubにアップしておきます。
記事中では、このノートブックの中のCNTKに関連する箇所を取り出して紹介しています。
github.com

Jupyter Notebook自体に馴染みのない方はこの記事もおすすめです。
PythonとJupyter Notebookを使ってデータと遊ぶ方法 - 今日も窓辺でプログラム

サンプルデータの用意

今回はロジスティック回帰を行うので、そのために使用するデータを人為的に用意します。
入力データは2次元で、出力は赤と青の2つのクラスであるようなデータを用意しました。データの生成方法については、Jupyter Notebookにコードがありますのでそちらを参照してください。

f:id:kanohk:20170128223843p:plain

モデルの定義

定義するモデルの形

下記のような形のモデルを定義します。画像は公式チュートリアルのものです。
https://www.cntk.ai/jup/logistic_neuron2.jpg

入力変数

入力は2次元の変数だったので、それを定義します。入力変数の定義にはcntk.ops.input_variable関数を使用します。

input_dim = 2
input = input_variable(input_dim, np.float32)

モデル

ウェイトwとバイアスbがパラメータになります。パラメータはPythonの辞書で持っておきます。

param = {"w": None, "b": None}

では、実際に画像のような形のネットワークを定義します。
下記のlinear_layerという関数を使ってネットワークの出力zを定義しています。

def linear_layer(input_var, output_dim):
    input_dim = input_var.shape[0]
    weight_param = parameter(shape=(input_dim, output_dim))
    bias_param = parameter(shape=(output_dim))
    param["w"], param["b"] = weight_param, bias_param
    return times(input_var, weight_param) + bias_param

output_dim = 2
z = linear_layer(input, output_dim)

TensorFlowを使用したことがあるかたは、cntk.ops.parameterがtf.Variableに対応していると思うと分かりやすいのではないでしょうか。
linear_layerの最後の行のcntk.ops.timesは、tf.matmulですね。

損失関数

損失関数を定義します。損失関数は出力にSoftmaxを噛ませたもののクロスエントロピーとします。これもTensorFlowと似たような名前の関数を1行で呼び出すだけでOKです。

label = input_variable((num_output_classes), np.float32)
loss = cross_entropy_with_softmax(z, label)

エラー率

分類に失敗したサンプルの割合(エラー率)も簡単に求められるようです。たったのこれだけ。

eval_error = classification_error(z, label)

学習用のオブジェクト

学習にははTrainerというオブジェクトを使用します。
モデルや損失関数、オプティマイザなどを引数として渡してあげます。
今回はオプティマイザはSGDを使用しています。

learning_rate = 0.1
lr_schedule = learning_rate_schedule(learning_rate, UnitType.minibatch) 
learner = sgd(z.parameters, lr_schedule)
trainer = Trainer(z, loss, eval_error, [learner])

学習状況可視化用の関数

チュートリアルからコピペしただけなのですが、学習の状況を出力する関数を定義しておきます。
print_training_progress関数にtrainerオブジェクトとミニバッチを渡してあげると、損失やエラー率などを出力してくれ、学習状況を把握するのに便利です。

from cntk.utils import get_train_eval_criterion, get_train_loss

# 移動平均
def moving_average(a, w=10):
    if len(a) < w: 
        return a[:]    
    return [val if idx < w else sum(a[(idx-w):idx])/w for idx, val in enumerate(a)]


# ロスや分類のエラーなどの状況を表示
def print_training_progress(trainer, mb, frequency, verbose=1):
    training_loss, eval_error = "NA", "NA"

    if mb % frequency == 0:
        training_loss = get_train_loss(trainer)
        eval_error = get_train_eval_criterion(trainer)
        if verbose: 
            print ("Minibatch: {0}, Loss: {1:.4f}, Error: {2:.2f}".format(mb, training_loss, eval_error))
        
    return mb, training_loss, eval_error

学習を走らせる

これで学習をする準備が整いました。早速学習をさせてみます。コードは次のようになります。

# 学習データのパラメータ
minibatch_size = 25
num_samples_to_train = 20000
num_minibatches_to_train = int(num_samples_to_train  / minibatch_size)
training_progress_output_freq = 50

plotdata = {"batchsize":[], "loss":[], "error":[]}

for i in range(0, num_minibatches_to_train):
    features, labels = generate_random_data_sample(minibatch_size, input_dim, num_output_classes)
    
    # trainerにミニバッチを与えていきます。TensorFlowのsess.run()でfeed_dictを使ってデータを与えるのと非常に似ていますね。
    trainer.train_minibatch({input : features, label : labels})
    batchsize, loss, error = print_training_progress(trainer, i, 
                                                     training_progress_output_freq, verbose=1)

forループの中でミニバッチのデータを用意し、trainer.train_minibatch()でミニバッチを与えて学習させています。以前cntk.ops.input_variableで定義したinputとlabelという変数に実際のデータを与えているのも、この引数({input : features, label : labels})で行っています。
TensorFlowでsess.run(feed_dict={...})というように、tf.placeholderの実際の値を渡してあげたりしますが、それと非常によく似た作りになっていますね。

上記のコードを実行すると、このように学習が進んでいく様子が確認できます。

Minibatch: 0, Loss: 0.6931, Error: 0.68
Minibatch: 50, Loss: 0.4191, Error: 0.20
Minibatch: 100, Loss: 0.3790, Error: 0.12
Minibatch: 150, Loss: 0.3418, Error: 0.04
Minibatch: 200, Loss: 0.3858, Error: 0.16
Minibatch: 250, Loss: 0.2826, Error: 0.08
Minibatch: 300, Loss: 0.2458, Error: 0.04
Minibatch: 350, Loss: 0.2294, Error: 0.00
Minibatch: 400, Loss: 0.3004, Error: 0.08
Minibatch: 450, Loss: 0.3086, Error: 0.12
Minibatch: 500, Loss: 0.2640, Error: 0.04
Minibatch: 550, Loss: 0.3452, Error: 0.20
Minibatch: 600, Loss: 0.3129, Error: 0.12
Minibatch: 650, Loss: 0.2016, Error: 0.12
Minibatch: 700, Loss: 0.3199, Error: 0.08
Minibatch: 750, Loss: 0.3353, Error: 0.16

学習の過程を可視化すると次のようになります。損失も収束してそうなことが確認できます。
(この可視化部分のコードは、Jupyter Notebookを参照してください。自分で損失を保存して、matplotlibでグラフを書いているだけです。)
f:id:kanohk:20170128231056p:plain

テストデータで精度検証

テストデータでのエラー率を知りたい場合には、trainer.test_minibatch()を次のように使用します。

# テストデータを用意
test_minibatch_size = 25
features, labels = generate_random_data_sample(test_minibatch_size, input_dim, num_output_classes)

# 学習済みのモデルを使ってエラー率を計算
trainer.test_minibatch({input : features, label : labels})

実際にモデルが予測したクラスを取得したい場合はこんな感じに書けます。

out = softmax(z)
result = out.eval({input : features})

print("Label    :", np.argmax(labels[:25],axis=1))
print("Predicted:", np.argmax(result[:25, 0, :],axis=1))

出力結果はこんな形です。

Label    : [1 0 0 1 1 1 0 1 1 0 1 1 1 0 1 0 1 1 0 0 1 0 0 0 1]
Predicted: [1 0 0 0 0 0 0 1 1 0 1 1 1 0 1 0 1 1 0 0 1 0 0 0 1]

ちなみに、今回学習したパラメータを可視化するとこんな形になります。
学習が終わり、ある程度正しく分類できそうな直線が引かれていることが確認できるかと思います。
f:id:kanohk:20170128231604p:plain