TensorFlow 2.4 : ガイド : Keras :- Keras でマスキングとパディング (翻訳/解説)

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

* 本ページは、TensorFlow org サイトの Guide – Keras の以下のページを翻訳した上で
適宜、補足説明したものです：

Masking and padding with Keras

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

★ 無料セミナー実施中 ★ クラスキャット主催人工知能 & ビジネス Web セミナー

人工知能とビジネスをテーマにウェビナー (WEB セミナー) を定期的に開催しています。スケジュールは弊社公式 Web サイトでご確認頂けます。

お住まいの地域に関係なく Web ブラウザからご参加頂けます。事前登録 が必要ですのでご注意ください。
Windows PC のブラウザからご参加が可能です。スマートデバイスもご利用可能です。

クラスキャットは人工知能・テレワークに関する各種サービスを提供しております :

人工知能研究開発支援	人工知能研修サービス	テレワーク & オンライン授業を支援
PoC(概念実証)を失敗させないための支援 (本支援はセミナーに参加しアンケートに回答した方を対象としています。

◆ お問合せ : 本件に関するお問い合わせ先は下記までお願いいたします。

株式会社クラスキャット セールス・マーケティング本部セールス・インフォメーション

E-Mail：sales-info@classcat.com ; WebSite: https://www.classcat.com/

Facebook: https://www.facebook.com/ClassCatJP/

ガイド : Keras :- Keras でマスキングとパディング

セットアップ

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

イントロダクション

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

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

Let’s take a close look.

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

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

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

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

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

データはネストされたリストで、そこでは個々のサンプルはそれぞれ長さ 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 エントリは対応する時間ステップは処理の間無視されるべきであることを示します。

Functional API と Sequential API のマスク伝播

Functional 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),]
)

これはまた次の Functional 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([[ 5.17572416e-03, -2.62944121e-03, -1.35105345e-02, ...,
        -4.60889516e-03,  5.47572691e-03,  4.72571421e-03],
       [ 1.64351589e-03, -4.22070269e-03, -5.25859464e-03, ...,
         2.60032900e-03,  1.49587099e-03, -4.24058782e-03],
       [-4.32722829e-03,  1.06376233e-02,  1.42529635e-02, ...,
         2.60018371e-03,  7.20684696e-03,  1.94990877e-02],
       ...,
       [ 4.70825378e-03, -1.18944524e-02, -1.36345504e-02, ...,
        -1.12335877e-02,  3.96840260e-05, -3.31538590e-03],
       [-1.87381962e-03,  4.51480271e-03, -1.13129790e-03, ...,
         4.01536841e-03,  1.47789170e-03,  7.81966280e-03],
       [-6.34184107e-03, -1.26939616e-03,  2.88227363e-03, ...,
        -4.33679670e-03, -6.08236482e-03, -4.22849880e-05]], 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 False  True  True]
 [ True  True False  True  True  True  True  True False  True]
 [False  True False  True False  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: KerasTensor(type_spec=TensorSpec(shape=(None, None), dtype=tf.bool, name=None), name='Placeholder_1:0')

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

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

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

下の単純なサンプルがここにあります : 層は入力シークエンスの時間次元に渡り 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 層が当てはまります。
Functional API と Sequential API では、マスク情報は自動的に伝播されます。
スタンドアロンな方法で層を使用するとき、層に mask 引数を手動で渡すことができます。
現在のマスクを変更する層を容易に書くことができます、それは新しいマスクを生成するか、入力に関連するマスクを消費します。

以上

2021年1月
月	火	水	木	金	土	日
				1	2	3
4	5	6	7	8	9	10
11	12	13	14	15	16	17
18	19	20	21	22	23	24
25	26	27	28	29	30	31