みーのぺーじ

みーが趣味でやっているPCやソフトウェアについて.Python, Javascript, Processing, Unityなど.

chainerで回帰順伝播型ニューラルネットワーク

この記事と同じ題材を扱った,より新しい記事を作成しました.

PyTorch で LSTMCell を使ってみる - みーのぺーじ


最近ホットな話題であるdeep learningをやってみようと思って,まずはchainerで回帰の順伝播型ニューラルネットワークを作成し,sin関数を学習させてみた. ニューラルネットワーク入門

ニューラルネットワークを使った機械学習は昔からあるものなのだが,最近はいろいろな学習法が発達してきていて,ライブラリーも充実してきているので,素人のみーでも気軽にニューラルネットワークが使えるような時代が来たわけである.素晴らしい.

みーは以下の書籍とサイトでニューラルネットワークの基礎を勉強した.2日でだいたい理解できる内容だった.

大雑把にまとめると,ニューラルネットワークにサンプルとなるデータを学習させ,任意の入力を与えた時にそれっぽい出力を帰すようにするのが目的である.メリットは,データの入力と出力がわかっていれば,途中でどのような計算をしているのかを考慮する必要がないこと.このために音声認識や画像認識みたいに,人間が入力の評価方法を開発する必要がなくなるので,過去の大量のデータみたいな感じでいろいろな入力を評価する仕事に向いている技術である.

医療の世界には,電子カルテという膨大なビックデータがあるので,これを医療に応用すれば,薬剤の効果を患者のプロファイルから予測する技術とか,抗癌剤の副作用の予測をする技術,医師のカルテ自動記載なんかができるようになって大変便利そうである.20年もしたらこういう技術は実臨床で使われている気がする.

しかし難点もあって,ニューラルネットワークに学習させるのにコツが必要なのである.これはネットワークが多層になればなるほど厄介なのだそうだ.多層のネットワークは自由度が高く,複雑なデータの学習ができるメリットがあるのだが,そもそも学習させるのが難しいわけである.これをなんとかする技術が最近いろいろと出てきているので,よく耳にするようになっているわけである.

deep learningの詳細は上記の書籍とサイトに譲り,早速pythonで実装してみる.

開発環境

簡単そうなchainerを使う.Caffeなどいろいろなライブラリがあるが,サンプルコードの量を見るとchainerが圧倒的に少なかったので,chainerを選択した.やっぱり簡単に使えるというのは嬉しい.

  • Python3.4
  • chainer
  • numpy

chainerとnumpyはpipを使えば簡単にインストールできる.

sudo pip3 install chainer

をターミナルで実行する.

みーはiMac 2011 で開発をしている.chainerはCUDA対応のグラボがあれば高速に動作するみたいなのだが,残念ながらみーはそういったグラボを持ち合わせていないので,仕方なくCPUのみで計算している.趣味でやるレベルならばCPUで十分な気もするが,必要ならばAmazon EC2のGPUインスタンスを使うという方法もあるらしい.http://sla.hatenablog.com/entry/chainer_on_ec2

実装

練習として,回帰の順伝播型ニューラルネットワークを作成し,sin関数を学習させてみた.0からπまでの入力に対し,f(x)=sin(s)の値を帰すようにするわけだ.

入力が1つ,出力が1つ,間にhidden layerを2つほど作った.optimizerはAdam()を使用し,活性化関数はchainer.functions.relu()を用いた.

import json, sys, glob, datetime, math
import numpy as np
import matplotlib.pyplot as plt

import chainer
from chainer import computational_graph as c
from chainer import cuda
import chainer.functions as F
from chainer import optimizers

class RegressionFNN:
    def __init__(self, n_units=64, batchsize=100):
        self.n_units = n_units
        self.batchsize = batchsize
        self.plotcount=0

    def load(self, train_x, train_y):
        if len(train_x)!=len(train_y):
            raise ValueError
        self.N = len(train_y)
        self.I = 1
        self.x_train = np.array(train_x, np.float32).reshape(len(train_x),1)
        self.y_train = np.array(train_y, np.float32).reshape(len(train_y),)
        #
        self.model = chainer.FunctionSet(l1=F.Linear(self.I, self.n_units),
                                        l2=F.Linear(self.n_units, self.n_units),
                                        l3=F.Linear(self.n_units, self.n_units),
                                        l4=F.Linear(self.n_units, 1))
        #
        self.optimizer = optimizers.Adam()
        self.optimizer.setup(self.model.collect_parameters())


    def forward(self, x_data, y_data, train=True):
        x = chainer.Variable(x_data)
        t = chainer.Variable(y_data)
        h1 = F.relu(self.model.l1(x))
        h2 = F.relu(self.model.l2(h1))
        h3 = F.relu(self.model.l3(h2))
        y = F.reshape(self.model.l4(h3), (len(y_data), ))
        return F.mean_squared_error(y, t), y

    def calc(self, n_epoch):
        for epoch in range(n_epoch):
            perm = np.random.permutation(self.N)
            sum_loss = 0
            #
            for i in range(0, self.N, self.batchsize):
                x_batch = self.x_train[perm[i:i + self.batchsize]]
                y_batch = self.y_train[perm[i:i + self.batchsize]]
                #
                self.optimizer.zero_grads()
                loss, y = self.forward(x_batch, y_batch)
                loss.backward()
                self.optimizer.update()
                #
                sum_loss += float(loss.data) * len(y_batch)
            #  
            print('epoch = {}, train mean loss={}\r'.format(epoch, sum_loss / self.N), end="")

    def getY(self, test_x, test_y):
        if len(test_x)!=len(test_y):
            raise ValueError
        x_test = np.array(test_x, np.float32).reshape(len(test_x),1)
        y_test = np.array(test_y, np.float32).reshape(len(test_y),)
        loss, y = self.forward(x_test, y_test, train=False)
        #
        with open("output/{}.csv".format(self.plotcount), "w") as f:
            f.write("\n".join(["{},{}".format(ux,uy) for ux,uy in zip(y.data, y_test)]))

        #
        plt.clf()
        plt.plot(x_test, y_test, color="b", label="original")
        plt.plot(x_test, y.data, color="g", label="RegFNN")
        plt.legend(loc = 'lower left')
        plt.ylim(-1.2,1.2)
        plt.grid(True)
        plt.savefig("output/{}.png".format(self.plotcount))
        self.plotcount+=1


def f(d):
    return [math.sin(u) for u in d]
if __name__=="__main__":
    rf = RegressionFNN(n_units=64, batchsize=314)
    d = [u*0.02 for u in range(314)]
    rf.load(d,f(d))
    rf.calc(1000) # epoch
    rf.getY(d,f(d))
    print("\ndone.")

回帰なので,誤差関数にmean_squared_error()を使っているのと,学習の評価のためにグラフを出力して視覚的に学習の様子が分かるようにした.epochの数を変えて実行たところ,以下のようにsin関数に少しずつ近づいているのがよく分かった.

epoch = 100

epoch = 300

epoch = 800

epoch = 2800

というわけで,sinを計算するニューラルネットワークがびっくりするほど簡単に完成した.

ちなみに,学習で与えた入力以外の入力に対してどんな出力が帰ってくるのだろう?という疑問を検証してみた.0から2πまでを314分割した入力のみで学習させ,0から4.1πまでを314分割した入力を与えた時の出力をグラフにしてみた.4.1πというのは,学習させたデータと少しずれた値を得るために中途半端な数字を設定しただけで大きな意味は無い.

2πまではかなりきれいに学習できているが,それ以降は直線になってしまっている.このネットワークでは学習に使った入力とはかなり異なる値を与えても,結果を類推することはできないようだ.

まとめ

これを作成して分かったことは,ニューラルネットワークのプログラミング自体は全く簡単であり,chainerで更に簡単に作れるようになっていることと,学習方法の選択には経験則のノウハウがあって,どの方法を選択するかが重要であるということだ.実際,上記のサンプルのoptimizerをAdam()から別のものに変更するだけで学習速度や精度が全く変わってくるし,どんなネットワークを使うかでも学習の様子が異なることはすぐに分かる.

以上から,ニューラルネットワークを上手に使うには,適切なデータを大量に用意し,適切な方法とネットワークを用いて学習させるのがポイントのようである.

ニューラルネットワーク,面白い.