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

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

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

• Masking and padding with Keras

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

 

 

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

セットアップ

from __future__ import absolute_import, division, print_function, unicode_literals

import numpy as np

import tensorflow as tf

from tensorflow.keras import layers

 

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

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

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

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

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

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

Keras は共通の長さにシークエンスを切り詰めてパディングする API を提供します : tf.keras.preprocessing.sequence.pad_sequences です。

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

# 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)

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

 

マスキング

総てのサンプルが統一された (= 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)

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

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  True  True  True]
 [ True  True  True  True  True False]
 [ True  True  True False False False]], 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 は入力に対応するマスクを自動的に取得してそれをどのように使用するかを知る任意の層に渡します。

サブクラス化されたモデルや層の call メソッドでは、マスクは自動的には伝播されませんので、それを必要とする任意の層に mask 引数を手動で渡す必要があることに注意してください。詳細は下のセクションを見てください。

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

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

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

inputs = tf.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 = tf.keras.Model(inputs, outputs)

 

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

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

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

こうして、このようなことができます :

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: id=4730, shape=(32, 32), dtype=float32, numpy=
array([[-0.00149354, -0.00657718,  0.0043684 , ...,  0.01915387,
         0.00254279,  0.00201567],
       [-0.00874859,  0.00249364,  0.00269479, ..., -0.01414887,
         0.00511035, -0.00541363],
       [-0.00457095, -0.0097013 , -0.00557693, ...,  0.00384533,
         0.00664415,  0.00333986],
       ...,
       [-0.00762534, -0.00543655,  0.0005238 , ...,  0.01187737,
         0.00214507, -0.00063268],
       [ 0.00428915, -0.00258686,  0.00012214, ...,  0.0064177 ,
         0.00800534,  0.00203928],
       [-0.01474019, -0.00349469, -0.00311312, ..., -0.0064069 ,
         0.00472621,  0.005593  ]], dtype=float32)>

 

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

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

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

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

殆どの層は時間次元を変更しませんので、マスキングについて心配する必要はありません。そのような場合 compute_mask() のデフォルトの動作は単に現在のマスクを通過させるだけです。

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

class TemporalSplit(tf.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(
[[ True  True  True]
 [ True  True False]
 [False False False]], shape=(3, 3), dtype=bool)

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

class CustomEmbedding(tf.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  True  True  True  True  True  True  True  True]
 [False  True False  True  True  True False  True False  True]], shape=(3, 10), dtype=bool)

 

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

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

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

class MaskConsumer(tf.keras.layers.Layer):
  
  def call(self, inputs, mask=None):
    ...

 

要点

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

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

以上