Keras 2 : ガイド : マスキングとパディングの理解 (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 10/29/2021 (keras 2.6.0)

* 本ページは、Keras の以下のドキュメントを翻訳した上で適宜、補足説明したものです：

• Understanding masking & padding (Author: Scott Zhu, Francois Chollet)

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

Keras 2 : ガイド : マスキングとパディングの理解

セットアップ

import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

 

イントロダクション

マスキング はシークエンス処理層に入力のある時間ステップは欠損していて、そのためにデータを処理するときにスキップされるべきであることを伝える方法です。

パディング はマスキングの特殊な形式で、そこではマスクされるステップはシークエンスの最初か最後です。パディングはシークエンスデータを連続するバッチにエンコードするための必要性に由来します : バッチの総てのシークエンスを与えられた標準的な長さにするためには、幾つかのシークエンスをパッドしたり切り詰める (= truncate) 必要あります。

 

シークエンス・データをパディング

シークエンス・データを処理するとき、個々のサンプルが異なる長さを持つことは非常に一般的です。次のサンプルを考えます (単語としてトークン化されたテキスト) :

[
  ["Hello", "world", "!"],
  ["How", "are", "you", "doing", "today"],
  ["The", "weather", "will", "be", "nice", "tomorrow"],
]

語彙検索の後、データは整数としてベクトル化されるかもしれません、例えば :

[
  [71, 1331, 4231]
  [73, 8, 3215, 55, 927],
  [83, 91, 1, 645, 1253, 927],
]

データはネストされたリストで、そこでは個々のサンプルはそれぞれ長さ 3, 5 と 6 を持ちます。深層学習モデルのための入力データは (この場合 e.g. (batch_size, 6, vocab_size) の shape の) 単一の tensorでなければならないので、最長の項目よりも短いサンプルはあるプレースホルダー値でパディングされる必要があります (代わりに、短いサンプルをパッドする前に長いサンプルを切る詰めるかもしれません)。

Keras は共通の長さに Python リストを切り詰めてパッドするユティリティ関数を提供します : tf.keras.preprocessing.sequence.pad_sequences です。

raw_inputs = [
    [711, 632, 71],
    [73, 8, 3215, 55, 927],
    [83, 91, 1, 645, 1253, 927],
]

# By default, this will pad using 0s; it is configurable via the
# "value" parameter.
# Note that you could "pre" padding (at the beginning) or
# "post" padding (at the end).
# We recommend using "post" padding when working with RNN layers
# (in order to be able to use the
# CuDNN implementation of the layers).
padded_inputs = tf.keras.preprocessing.sequence.pad_sequences(
    raw_inputs, padding="post"
)
print(padded_inputs)

[[ 711  632   71    0    0    0]
 [  73    8 3215   55  927    0]
 [  83   91    1  645 1253  927]]

 

マスキング

総てのサンプルが一様な (= uniform) 長さを持つ今、モデルはデータのある部分は実際にはパディングで無視されるべきであることを知らされなければなりません。そのメカニズムが マスキング です。

Keras モデルで入力マスクを導入するには 3 つの方法があります :

• keras.layers.Masking 層を追加する。
• keras.layers.Embedding 層を mask_zero=True で configure する。
• mask 引数を手動で渡します、この引数をサポートする層を呼び出すとき (e.g. RNN 層)。

 

マスク生成層 : Embedding と Masking

内部的には、これらの層はマスク tensor (shape (batch, sequence_length) の 2D tensor) を作成し、そしてそれを Masking か Embedding 層により返される tensor 出力にアタッチします。

embedding = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)
masked_output = embedding(padded_inputs)

print(masked_output._keras_mask)

masking_layer = layers.Masking()
# Simulate the embedding lookup by expanding the 2D input to 3D,
# with embedding dimension of 10.
unmasked_embedding = tf.cast(
    tf.tile(tf.expand_dims(padded_inputs, axis=-1), [1, 1, 10]), tf.float32
)

masked_embedding = masking_layer(unmasked_embedding)
print(masked_embedding._keras_mask)

tf.Tensor(
[[ True  True  True False False False]
 [ True  True  True  True  True False]
 [ True  True  True  True  True  True]], shape=(3, 6), dtype=bool)
tf.Tensor(
[[ True  True  True False False False]
 [ True  True  True  True  True False]
 [ True  True  True  True  True  True]], shape=(3, 6), dtype=bool)

プリントされた結果から分かるように、マスクは shape (batch_size, sequence_length) を持つ 2D ブーリアン tensor で、そこでは各個々の False エントリは対応する時間ステップは処理の間無視されるべきであることを示します。

 

関数型 API と Sequential API のマスク伝播

関数型 API や Sequential API を使用するとき、Embedding や Masking 層により生成されたマスクはそれらを使用できる任意の層 (例えば、RNN 層) に対してネットワークを通して伝播されます。Keras は入力に対応するマスクを自動的に取得してそれを使用する方法知る任意の層に渡します。

例えば、次の Sequential モデルでは、LSTM 層は自動的にマスクを受け取ります、それはそれがパッドされた値を無視することを意味します :

model = keras.Sequential(
    [layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True), layers.LSTM(32),]
)

これはまた次の関数型 API モデルにも当てはまります :

inputs = keras.Input(shape=(None,), dtype="int32")
x = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)(inputs)
outputs = layers.LSTM(32)(x)

model = keras.Model(inputs, outputs)

 

マスク tensor を直接的に層に渡す

(LSTM 層のような) マスクを処理できる層は __call__ メソッドで mask 引数を持ちます。

一方で、マスクを生成する層 (e.g. Embedding) は呼び出せる compute_mask(input, previous_mask) メソッドを公開します。

こうして、マスク生成層の compute_mask() メソッドの出力をマスク消費層の __call__ メソッドに渡すことができます、このようにです :

class MyLayer(layers.Layer):
    def __init__(self, **kwargs):
        super(MyLayer, self).__init__(**kwargs)
        self.embedding = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)
        self.lstm = layers.LSTM(32)

    def call(self, inputs):
        x = self.embedding(inputs)
        # Note that you could also prepare a `mask` tensor manually.
        # It only needs to be a boolean tensor
        # with the right shape, i.e. (batch_size, timesteps).
        mask = self.embedding.compute_mask(inputs)
        output = self.lstm(x, mask=mask)  # The layer will ignore the masked values
        return output


layer = MyLayer()
x = np.random.random((32, 10)) * 100
x = x.astype("int32")
layer(x)

<tf.Tensor: shape=(32, 32), dtype=float32, numpy=
array([[ 9.3598114e-03, -5.4868571e-03, -1.2649748e-02, ...,
         1.3104092e-03, -1.8691338e-03,  1.6320259e-03],
       [-6.0183648e-03, -4.9164523e-03,  3.0082103e-03, ...,
         1.7394881e-03,  9.1036235e-04, -1.2966867e-02],
       [ 6.0863183e-03,  1.3509918e-03, -7.1913302e-03, ...,
         3.9419280e-03,  2.9930705e-03,  3.4562423e-04],
       ...,
       [-5.7978416e-04, -1.8325391e-03, -2.0467002e-04, ...,
        -3.9534271e-03, -2.2688047e-04,  1.2577593e-03],
       [ 2.4689233e-03, -3.6403039e-04,  7.7487719e-05, ...,
         1.0208538e-03,  2.3937733e-03, -4.4873711e-03],
       [ 2.6551904e-03, -1.8738948e-03, -1.9827935e-04, ...,
        -3.3328766e-03,  1.0988748e-06,  1.4491909e-04]], dtype=float32)>

 

カスタム層でマスキングをサポートする

時に貴方は (Embedding のような) マスクを生成する層や、現在のマスクを変更する必要がある層を書く必要があるかもしれません。

例えば、時間次元上で連結する Concatenate 層のような、入力とは異なる時間次元を持つ tensor を生成する任意の層は現在のマスクを変更する必要があります、その結果下流の層はマスクされた時間ステップを正しく考慮に入れることができます。

これを行なうために、貴方の層は layer.compute_mask() メソッドを実装するべきです、これは入力と現在のマスクが与えられたとき新しいマスクを生成します。

ここに現在のマスクを変更する必要がある TemporalSplit 層のサンプルがあります。

class TemporalSplit(keras.layers.Layer):
    """Split the input tensor into 2 tensors along the time dimension."""

    def call(self, inputs):
        # Expect the input to be 3D and mask to be 2D, split the input tensor into 2
        # subtensors along the time axis (axis 1).
        return tf.split(inputs, 2, axis=1)

    def compute_mask(self, inputs, mask=None):
        # Also split the mask into 2 if it presents.
        if mask is None:
            return None
        return tf.split(mask, 2, axis=1)


first_half, second_half = TemporalSplit()(masked_embedding)
print(first_half._keras_mask)
print(second_half._keras_mask)

tf.Tensor(
[[ True  True  True]
 [ True  True  True]
 [ True  True  True]], shape=(3, 3), dtype=bool)
tf.Tensor(
[[False False False]
 [ True  True False]
 [ True  True  True]], shape=(3, 3), dtype=bool)

ここに入力値からマスクを生成できる CustomEmbedding 層の別のサンプルがあります :

class CustomEmbedding(keras.layers.Layer):
    def __init__(self, input_dim, output_dim, mask_zero=False, **kwargs):
        super(CustomEmbedding, self).__init__(**kwargs)
        self.input_dim = input_dim
        self.output_dim = output_dim
        self.mask_zero = mask_zero

    def build(self, input_shape):
        self.embeddings = self.add_weight(
            shape=(self.input_dim, self.output_dim),
            initializer="random_normal",
            dtype="float32",
        )

    def call(self, inputs):
        return tf.nn.embedding_lookup(self.embeddings, inputs)

    def compute_mask(self, inputs, mask=None):
        if not self.mask_zero:
            return None
        return tf.not_equal(inputs, 0)


layer = CustomEmbedding(10, 32, mask_zero=True)
x = np.random.random((3, 10)) * 9
x = x.astype("int32")

y = layer(x)
mask = layer.compute_mask(x)

print(mask)

tf.Tensor(
[[ True  True  True  True  True  True  True  True  True  True]
 [ True  True  True  True  True  True  True  True  True  True]
 [ True  True  True  True  True  True  True  True  True  True]], shape=(3, 10), dtype=bool)

 

互換な層上でマスク伝播にオプトインする

殆どの層は時間次元を変更しませんので、現在のマスクを変更する必要はありません。けれども、それらは依然として現在のマスクを、変更されないまま、次の層に 伝播 できることを望むかもしれません。これはオプトイン (= opt-in) 動作です。デフォルトでは、カスタム層は現在のマスクを壊します (何故ならばフレームワークはマスクの伝播が行なうに安全であるか識別する方法を持たないからです)。

時間次元を変更しないカスタム層を持つ場合、そしてそれに現在の入力マスクを伝播できることを望む場合、層コンストラクタで self.supports_masking = True を設定するべきです。この場合、compute_mask() のデフォルト動作は単に現在のマスクをパススルーすることです。

ここにマスク伝播のためにホワイトリストに登録された層のサンプルがあります :

class MyActivation(keras.layers.Layer):
    def __init__(self, **kwargs):
        super(MyActivation, self).__init__(**kwargs)
        # Signal that the layer is safe for mask propagation
        self.supports_masking = True

    def call(self, inputs):
        return tf.nn.relu(inputs)

今では (Embedding のような) マスク生成層と (LSTM のような) マスク消費層を媒介するこのカスタム層を使用できて、そしてそれは一緒にマスクを渡します、その結果それはマスク消費層に到達します。

inputs = keras.Input(shape=(None,), dtype="int32")
x = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)(inputs)
x = MyActivation()(x)  # Will pass the mask along
print("Mask found:", x._keras_mask)
outputs = layers.LSTM(32)(x)  # Will receive the mask

model = keras.Model(inputs, outputs)

Mask found: Tensor("embedding_4/NotEqual:0", shape=(None, None), dtype=bool)

 

マスク情報を必要とする層を書く

いくつかの層はマスク消費者です : それらは call で mask 引数を受け取りそしてそれを特定の時間ステップをスキップするかどうかを決定するために使用します。

そのような層を書くために、単純に call シグネチャで mask=None 引数を追加できます。入力に関連するマスクはそれが利用可能なときにいつでも層に渡されます。

下に単純なサンプルがあります : 層は入力シークエンスの時間次元 (軸 1) に渡り softmax を計算する一方で、マスクされた時間ステップは破棄します。

class TemporalSoftmax(keras.layers.Layer):
    def call(self, inputs, mask=None):
        broadcast_float_mask = tf.expand_dims(tf.cast(mask, "float32"), -1)
        inputs_exp = tf.exp(inputs) * broadcast_float_mask
        inputs_sum = tf.reduce_sum(inputs * broadcast_float_mask, axis=1, keepdims=True)
        return inputs_exp / inputs_sum


inputs = keras.Input(shape=(None,), dtype="int32")
x = layers.Embedding(input_dim=10, output_dim=32, mask_zero=True)(inputs)
x = layers.Dense(1)(x)
outputs = TemporalSoftmax()(x)

model = keras.Model(inputs, outputs)
y = model(np.random.randint(0, 10, size=(32, 100)), np.random.random((32, 100, 1)))

 

要点

それが Keras でマスキングについて知る必要があることの総てです。まとめると :

• 「マスキング」は層がシークエンス入力である時間ステップをいつスキップ/無視するかを知ることができる方法です。
• 幾つかの層はマスク生成者 (-generator) です : Embedding は (mask_zero=True の場合) 入力値からマスクを生成できて、Masking 層もそれができます。
• 幾つかの層はマスク消費者 (-consumer) です : それらは __call__ メソッドで mask 引数を公開します。これは RNN 層が当てはまります。
• 関数型 API と Sequential API では、マスク情報は自動的に伝播されます。
• スタンドアロンな方法で層を使用するとき、層に mask 引数を手動で渡すことができます。
• 現在のマスクを変更する層を容易に書くことができます、それは新しいマスクを生成するか、入力に関連するマスクを消費します。

以上