Kerasでちょっと難しいModelやTrainingを実装するときのTips

テクノロジー

Kerasのコードはシンプルでモジュール性が高いのでシンプルに記述可能で、理解しやすく使いやすいです。ただし、標準で用意されている以外のLayerや学習をさせようとすると、あまりサンプルがなくてどう書いていいかわからなくなることが多いです。最近いくつか変わったModelを書いた時に学んだTipsを備忘録も兼ねて共有します。

目次

Tips

Functional APIを使おう


Kerasには2通りのModelの書き方があります。
Sequencial ModelFunctional API Model です。

Sequential Modelは

model = Sequential()

model.add(Dense(32, input_dim=784))

model.add(Activation('relu'))
という書き方で、最初これを見て「Kerasすげーわかりやすい!」と思った方も多いのではないでしょうか。

それとは別に

inputs = Input(shape=(784,))

x = Dense(64, activation='relu')(inputs)

x = Dense(64, activation='relu')(x)

predictions = Dense(10, activation='softmax')(x)

model = Model(input=inputs, output=predictions)
という書き方があります。これは LayerInstance(InputTensor) -> OutputTensor というリズムで書いていく書き方です。

Dense(64, activation='relu')(x) というのが Pythonっぽい言語に慣れていない人には違和感があるかもしれませんが、
Dense(64, activation='relu') の部分で Dense Class の Instance を作って、それにたいして DenseInstance(x) しているだけです。

dense = Dense(64, activation='relu')

x = dense(x)
と意味は同じです。

流れとしては、 入力となるLayer 出力となるLayerを決めて、それを Model Classに渡してあげるということです。
入力となるLayerが実データなら Input classを使って指定しておきます(Placeholderのようなもの)。

ここで意識しておきたいことは、 LayerInstance毎にWeightを保持している ということです。
つまり 同じLayerInstanceを使うとWeightを共有している ということになります。
意図して共有する場合はもちろん、意図しない共有にも気をつけましょう。

この書き方なら、同じOutputTensorを別のLayerに入力することが簡単にできます。
記述量は対して変わらないですし、だんだん慣れてきたらこちらのFunctional API で書く練習をして、今後の難しいModelへ備えておくことをおすすめします。

複数LayerのWeightを共有したい場合は Container を使うと便利


異なる入力Layer と 異なる出力Layer を持つが、中身のNetworkとWeightは共有したい、という場合があります。
その場合は、 Container クラスで まとめておくと 取り回しがよくなります。
Container は Layer のサブクラスなので、Layerと同様に 同じContainerInstanceを使うとWeightを共有している ことになります。

例えば、

inputs = Input(shape=(784,))

x = Dense(64, activation='relu')(inputs)

x = Dense(64, activation='relu')(x)

predictions = Dense(10, activation='softmax')(x)

shared_layers = Container(inputs, predictions, name="shared_layers")
というような shared_layers はあたかも一つのLayerのように扱えます。

Container 自体は基本的には独自のWeightを持たず、あくまで他のLayerを束ねる役割を果たします。

逆にWeightを共有したくない場合は、Containerを共有せずに、個別にLayerInstanceを連ねるようにするしないといけません。

「LayerのOutput」と「生のTensor」は似て非なるもの


独自の計算やTensor変換を書いているとしばしば、

TypeError: ('Not a Keras tensor:', Elemwise{add,no_inplace}.0)

というエラーを目にします。

これは大抵、LayerInstance の 入力に「他のLayerのOutput」ではなく「生のTensor」を入れてしまう場合に起こります。
例えば、

from keras import backend as K
inputs = Input((10, ))

x = K.relu(inputs * 2 + 1)

x = Dense(64, activation='relu')(x)

などとするとそういうことが起こります。
よくはわかりませんが、LayerのOutputは KerasTensorという内部的にShapeを持ったObjectで、K.hogehoge などの計算結果とは異なるもののようです。

そのような場合は、下記のLambdaを使ってあげると上手くいきます(強引に _keras_shape を埋めようとしないほうが無難です ^^;)。

Lambdaを使った簡易変換は便利


例えば、 10要素のVectorを前半の5個、後半の5個に分けたいとします。
前述の通り

inputs = Input((10, ))

x0_4 = inputs[:5]

x5_9 = inputs[5:]

d1 = Dense(10)(x0_4)

d2 = Dense(10)(x5_9)
などとするとエラーになります。

そこで

inputs = Input((10, ))

x0_4 = Lambda(lambda x: x[:, :5], output_shape=(5, ))(inputs)

x5_9 = Lambda(lambda x: x[:, 5:], output_shape=lambda input_shape: (None, int(input_shape[1]/2), ))(inputs)

d1 = Dense(10)(x0_4)

d2 = Dense(10)(x5_9)
というように Lambda classでWrapしてあげるとうまくいきます。
ここでは少しポイントがあります。

Lambdaの内部ではSample次元を含めたTensor計算式を書く必要がある


Kerasでは一貫して最初の次元がSample次元(batch_sizeの次元)になっています。
LambdaなどのLayerを実装するときには、内部ではそのSample次元を含めた計算式を書きます。
なので lambda x: x[:5] ではなく lambda x: x[:, :5] と書く必要があります。

入力のShapeと出力のShapeが異なるときはoutput_shapeを指定する


output_shape は入出力のShapeが同じ場合は省略できますが、異なる場合は必ず指定します。
output_shapeの引数はTupleやFunctionが指定可能ですが、TupleのときはSample次元を含めない、FunctionのときはSample次元を含める ようにします。
Functionのときは、基本的にSample次元はNoneにしておけばOKです。
また Functionで指定する場合の引数であるinput_shape ですが、これにはSample次元が含まれているので注意が必要です。

カスタムなLoss FunctionはSample別にLossを返す


ModelのcompileメソッドでLoss関数を指定することができ、自分でカスタムしたLoss関数も指定可能です。
Functionの形状ですが、 y_true, y_pred の2つを引数にとって、Sample数の数だけ数値を返すようにします。
例えば以下のようになります。

def generator_loss(y_true, y_pred): # y_true's shape=(batch_size, row, col, ch)

return K.mean(K.abs(y_pred - y_true), axis=[1, 2, 3])

LayerじゃないところからLoss関数に式を追加したい場合


Layerから渡したい場合は、素直にLayer#add_loss を呼び出せば良いですが、Layerではないところから渡すのは少しむずかしいです(というか正しいやり方がわかりません)。

Loss Function以外のLossの計算式は、Model Instanceのcompileが実行されるタイミングで、Model#losses により各Layerから収集されます(regularizerなどから)。
つまり、なんとかしてここに渡せば良いわけです。
例えば、 ContainerやModel を継承して、 #losses を Overriedeするとかするとなんとか渡せます。
VATModel を作ったときはこの方法で渡しました。

学習時にパラメータを更新しつつLossに反映した場合


学習時のLoss計算を以前のLoss結果を反映した値にしたい場合があります。
例えば、下記のようなBEGANの計算の DiscriminatorLoss のような場合です。
https://github.com/mokemokechicken/keras_BEGAN/blob/master/src/began/training.py#L104

学習時のパラメータ更新は、
Model Instanceのcompileが実行されるタイミングで、Model#updatesによって渡されるパラメータ情報が使われます。
Loss FunctionからそのModel#updatesにデータを渡す手段が通常では存在しないので(たぶん)、少し工夫します。

と考えて、以下のようにすると一応可能です。

class DiscriminatorLoss:

__name__ = 'discriminator_loss'
def __init__(self, lambda_k=0.001, gamma=0.5):

self.lambda_k = lambda_k

self.gamma = gamma

self.k_var = K.variable(0, dtype=K.floatx(), name="discriminator_k")

self.m_global_var = K.variable(0, dtype=K.floatx(), name="m_global")

self.loss_real_x_var = K.variable(0, name="loss_real_x") # for observation

self.loss_gen_x_var = K.variable(0, name="loss_gen_x") # for observation

self.updates = []


def __call__(self, y_true, y_pred): # y_true, y_pred shape: (BS, row, col, ch * 2)

data_true, generator_true = y_true[:, :, :, 0:3], y_true[:, :, :, 3:6]

data_pred, generator_pred = y_pred[:, :, :, 0:3], y_pred[:, :, :, 3:6]

loss_data = K.mean(K.abs(data_true - data_pred), axis=[1, 2, 3])

loss_generator = K.mean(K.abs(generator_true - generator_pred), axis=[1, 2, 3])

ret = loss_data - self.k_var * loss_generator


# for updating values in each epoch, use `updates` mechanism

# DiscriminatorModel collects Loss Function's updates attributes

mean_loss_data = K.mean(loss_data)

mean_loss_gen = K.mean(loss_generator)


# update K

new_k = self.k_var + self.lambda_k * (self.gamma * mean_loss_data - mean_loss_gen)

new_k = K.clip(new_k, 0, 1)

self.updates.append(K.update(self.k_var, new_k))


# calculate M-Global

m_global = mean_loss_data + K.abs(self.gamma * mean_loss_data - mean_loss_gen)

self.updates.append(K.update(self.m_global_var, m_global))


# let loss_real_x mean_loss_data

self.updates.append(K.update(self.loss_real_x_var, mean_loss_data))


# let loss_gen_x mean_loss_gen

self.updates.append(K.update(self.loss_gen_x_var, mean_loss_gen))


return ret


class DiscriminatorModel(Model):

"""Model which collects updates from loss_func.updates"""


@property

def updates(self):

updates = super().updates

if hasattr(self, 'loss_functions'):

for loss_func in self.loss_functions:

if hasattr(loss_func, 'updates'):

updates += loss_func.updates

return updates


discriminator = DiscriminatorModel(all_input, all_output, name="discriminator")

discriminator.compile(optimizer=Adam(), loss=DiscriminatorLoss())

さいごに


たぶん、最後の2つはしばらくしたら不要になるようなTipsかもしれないですね。
Kerasはソースコードも追いやすいので、意外となんとかなります。
あと、 Keras2になってエラーメッセージがよりわかりやすくなったので、Debugするときには助かっています。

まずはお気軽にお問い合わせください

「Sprocket」の費用や導入スケジュール、また、御社の顧客体験の向上やコンバージョンの最適化、Web接客ツールの比較検討においてご不明な点がございましたら、お気軽にお問い合わせください。(無料)

03-6303-4123

受付時間:平日10時~12時/13時~17時