Kerasのモデル・レイヤー周りの話

これはKerasアドベントカレンダー2017 16日目の記事です。

こんにちは。アイパー隊長です。

今年４月に転職して、３ヶ月ぐらいKerasと毎日をともにしてきました。

モデルを構築したり、学習したり、学習途中をデバッグしたり、学習結果を確認したりと。

その中で色々学んだので、それを書きなぐっておきます。

環境

バックエンドはTensorflowを使います。

• Tensorflow
• Keras

学習したモデルの中間レイヤーのアウトプットの確認

Kerasのドキュメントにあります、「中間レイヤーの出力を得るには？」に書いてある通り、途中のレイヤーのインプットとアウトプットを関数化してレイヤーのアウトプットの値を確認することができます。

下記はドキュメントのコード

from keras.models import Model

model = ...  # create the original model

layer_name = 'my_layer'
intermediate_layer_model = Model(inputs=model.input,
                                 outputs=model.get_layer(layer_name).output)
intermediate_output = intermediate_layer_model.predict(data)

モデルオブジェクトから、layers配列から各レイヤーのオブジェクトを取得することもできます。でも、おすすめは名前をつけて管理しやすい↑がいいかと。

model = ...

for layer in model.layers:
    print(layer.name)

このレイヤーオブジェクトに重みがあるなら、get_weights()取得することができます。

model = ...

for layer in model.layers:
     w = layer.get_weights()
     if w:
         # なんらかの処理

さて、話を題に戻します。このレイヤーオブジェクトの入力と、出力のメンバー変数を使って、中間レイヤーの出力を得る関数のようなものを定義します。この時使うのが、from keras import backendというクラスです。

from keras.models import Sequential
from keras.layers.core import Dense
from keras import backend as K
import numpy as np

# 100個の入力を受け取り、出力が２つのモデル
model = Sequential()
model.add(Dense(128, activation='relu', input_shape=(100,)))
model.add(Dense(128, activation='relu'))
model.add(Dense(2, activation='softmax'))

# １つ目の全結合レイヤー（Dense）へ値を渡し、２つ目の全結合レイヤーの出力を取得できる関数を定義する
_input = model.layers[0].input
_output = model.layers[1].output
func = K.function([_input, K.learning_phase()], [_output])

# 利用例
X = np.random.randint(1, 100, (100))
# リストでラップされて返ってくるので、要素番号を指定しています
output_value = func([X, 0])[0]

K.learning_phase()という関数も入力として定義しています。これは、BatchNormalizationであったり、Dropoutなど学習時とテストで振る舞いが違うレイヤーを利用する・しないを指定するために定義します。関数の第二引数に0を渡すとテスト時（利用しない）の結果が返り、1を渡すと学習時（利用する）の結果が返ります。

また、ディープラーニングで学習を進めていると、面倒なのが各レイヤーのデバッグでした。それぞれ、どういう重みを持っているか -> 学習しているのか、を確認するには、この中間レイヤーの出力確認方法が使えます。

さらに、応用でまれに凝った学習の手法を使ったり（分散学習とか）、モデルの組み方がが間違ってしまい（ここでいう間違いとはプログラムとして動作するが、計算が間違えていることを指します）途中でnanが出力されてしまい、学習がうまくいかなかったケースがありました。

手動で確認するのはとても面倒だったので、下記のようにコードを組んでデバッグしていました。

model = ... #構築したモデルオブジェクト

nans = []
input_value = np.random.randint(1, 100, (1, 100))
_input = model.layers[0].input

# モデルレイヤーの重み、出力にnanが存在していないか確認する
for index, layer in enumerate(model.layers):
    w = layer.get_weights()
    if not w: continue
    for _w in w:
        nans.append(np.isnan(_w).any())

    # １番目のレイヤーの入力・出力で関数の定義は行わない
    if index == 0: continue
    _output = layer.output
    func = K.function([_input, K.learning_phase()], [_output])
    nans.append(np.isnan(func([input_value, 0])[0]).any())
    nans.append(np.isnan(func([input_value, 1])[0]).any())

# True が出力されたらnanがある（探す）
print(np.array(nans).any())

複雑なモデルを組む

カスタマイズして使いたい場合は、簡単に利用できるSequentialモデルではなく、FunctionalAPIを利用しましょう。Sequentialの場合はぶっちゃけ、どっちも簡単に利用できますが、Functionalのほうが、返り値の値を修正したりとかできるのでカスタマイズの面ではよい組み方だと思います。

# from keras.models import Model
from keras.layers import Input, Dense

# Sequential
# model = Sequential()
# model.add(Dense(128, activation='relu', input_shape=(100,)))
# model.add(Dense(128, activation='relu'))
# model.add(Dense(2, activation='softmax'))

# Functional API
_input = Input(shape=(32,))
x = Dense(32, activation='relu')(_input)
x = Dense(32, activation='relu')(x)
_output = Dense(2, activation='softmax')(x)
model = Model(inputs=_input, outputs=_output)

これが使えるようになると、インプットを複数を受け取る、出力を３つ出すような複雑なモデルが組めるようになります。

from keras.models import Model
from keras.layers import Input
from keras.layers.core import Dense
from keras.layers.merge import Multiply

_input1 = Input(shape=(100,))
x1 = Dense(128, activation='relu')(_input1)
x1 = Dense(128, activation='relu')(x1)

_input2 = Input(shape=(25,))
x2 = Dense(128, activation='relu')(_input2)
x2 = Dense(128, activation='relu')(x2)

_input3 = Input(shape=(30,))
x3 = Dense(128, activation='relu')(_input3)

x = Multiply()([x1, x2, x3])
_output = Dense(5, activation='relu')(x)
model = Model(inputs=[_input1, _input2, _input3], outputs=_output)

こんな感じ。可視化してみましょう。

from IPython.display import SVG
from keras.utils.vis_utils import model_to_dot

SVG(model_to_dot(model).create(prog='dot', format='svg'))

X1 = np.random.randint(1, 100, (1, 100))
X2 = np.random.randint(1, 100, (1, 25))
X3 = np.random.randint(1, 100, (1, 30))
model.predict([X1, X2, X3])
# array([[  9235.22558594,   3631.79345703,  33012.4921875 ,  25466.93554688, 1908.30749512]], dtype=float32)

自作レイヤーを作る

個人的には、KerasはTensorflowを便利にしてくれる大きなラッパーってイメージで使っています。Tensorflowを叩いたことある人ならわかると思いますが、最初のとっつきづらがすごく大変でした。スレッド？キュー？叩いたけど結果が帰ってこないよ！！などなど。

諦めて、Kerasに切り替えたこともありますが、やっぱりすごく使いやすいですね。本当にいいものだと思います。

さて、そんなKerasに用意されていないレイヤーを自作する場合どうすれば良いか。

ドキュメントにはこう書いてある。

from keras import backend as K
from keras.engine.topology import Layer
import numpy as np

class MyLayer(Layer):

    def __init__(self, output_dim, **kwargs):
        self.output_dim = output_dim
        super(MyLayer, self).__init__(**kwargs)

    def build(self, input_shape):
        # Create a trainable weight variable for this layer.
        self.kernel = self.add_weight(name='kernel',
                                      shape=(input_shape[1], self.output_dim),
                                      initializer='uniform',
                                      trainable=True)
        super(MyLayer, self).build(input_shape)  # Be sure to call this somewhere!

    def call(self, x):
        return K.dot(x, self.kernel)

    def compute_output_shape(self, input_shape):
        return (input_shape[0], self.output_dim)

また、簡易的に作成するなら、Lambdaレイヤーを使いましょう。こいつを使うことで、簡単な処理をレイヤー化することができます。

from keras.models import Model
from keras.layers import Input
from keras.layers.core import Dense, Lambda

_input = Input(shape=(100,))
x = Dense(128, activation='relu')(_input)
x = Dense(128, activation='relu')(x)
x = Lambda(lambda x: x + 2)(x)  # 入力値に+2するだけのレイヤー
_output = Dense(2, activation='softmax')(x)
model = Model(inputs=_input, outputs=_output)

簡単な処理であればLambdaレイヤーで、重みをもたせたいならレイヤークラスを作成しましょう。

Lambdaレイヤーについては、Tensorflowを触ったことがある人なら「？　なんでレイヤーにする必要があるの？　実装すればいいじゃん。」となると思います。つまり

from keras.models import Model
from keras.layers import Input
from keras.layers.core import Dense, Lambda
import tensorflow as tf

_input = Input(shape=(100,))
x = Dense(128, activation='relu')(_input)
x = Dense(128, activation='relu')(x)
x = tf.add(x, 2)  # Lambdaレイヤーを使わずに普通に計算する
_output = Dense(2, activation='softmax')(x)
model = Model(inputs=_input, outputs=_output)

しかし、これはエラーになります。

AttributeError: 'Tensor' object has no attribute '_keras_history'

Kerasでは、下記関数になげてTrueが返ってこない値はエラーになります。

from keras import backend as K

# これでTrueならないやつはレイヤーの引数・モデル構築に使えない
K.is_keras_tensor(tf.add(x, 2))  # -> False

なので、おとなしくLambdaでラップしてあげましょう。

• レイヤークラスの引数にレイヤーのinput、outputのTensor以外の値をいれるとエラー
• レイヤーのoutputをレイヤークラスの引数に突っ込むと、get_config()あたりのcopy.deepcopy()でエラーが発生する
• モデルの計算はレイヤーの中以外でやると、keras._historyなんちゃらのエラーが発生する

雑間

これからもKerasで遊ぶぞ！！！（仕事も）