Смотрите основные доклады, презентации продуктов, семинары и многое другое на Google I / O См. Плейлист

Маскирование и заполнение Keras

Посмотреть на TensorFlow.org Запускаем в Google Colab Посмотреть исходный код на GitHub Скачать блокнот

Настраивать

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

Вступление

Маскирование - это способ сообщить слоям обработки последовательности, что определенные временные интервалы на входе отсутствуют и, следовательно, их следует пропускать при обработке данных.

Заполнение - это особая форма маскирования, при которой замаскированные шаги находятся в начале или в конце последовательности. Заполнение происходит из-за необходимости кодировать данные последовательности в непрерывные пакеты: для того, чтобы все последовательности в пакете соответствовали заданной стандартной длине, необходимо дополнить или усечь некоторые последовательности.

Давайте внимательно посмотрим.

Данные последовательности заполнения

При обработке данных последовательности очень часто отдельные образцы имеют разную длину. Рассмотрим следующий пример (текст размечен как слова):

[
  ["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 соответственно. Поскольку входные данные для модели глубокого обучения должны быть одним тензором (в данном случае формы, например (batch_size, 6, vocab_size) ), образцы, которые короче самого длинного элемента, должны быть дополнены некоторым значением-заполнителем (в качестве альтернативы, одним может также обрезать длинные сэмплы перед заполнением коротких сэмплов).

Keras предоставляет служебную функцию для усечения и tf.keras.preprocessing.sequence.pad_sequences списков 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]]

Маскировка

Теперь, когда все образцы имеют одинаковую длину, модель должна быть проинформирована о том, что некоторая часть данных на самом деле является заполнением и ее следует игнорировать. Этот механизм маскируется .

Есть три способа ввести маски ввода в модели Keras:

  • Добавьте слой keras.layers.Masking .
  • Настройка keras.layers.Embedding слой с mask_zero=True .
  • Передайте аргумент mask вручную при вызове слоев, которые поддерживают этот аргумент (например, слоев RNN).

Слои, генерирующие маску: Embedding и Masking

Под капотом эти слои будут создавать тензор маски (2D-тензор с формой (batch, sequence_length) ) и присоединять его к выходным данным тензора, возвращаемым слоем Masking или Embedding .

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)

Как видно из распечатанного результата, маска представляет собой двумерный логический тензор с формой (batch_size, sequence_length) , где каждая отдельная False запись указывает, что соответствующий временной шаг следует игнорировать во время обработки.

Распространение маски в функциональном API и последовательном API

При использовании функционального API или последовательного API маска, сгенерированная уровнем Embedding или Masking будет распространяться по сети для любого уровня, который может их использовать (например, слоев RNN). Keras автоматически выберет маску, соответствующую входу, и передаст ее любому слою, который знает, как ее использовать.

Например, в следующей последовательной модели слой 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)

Передача тензоров масок непосредственно слоям

Слои, которые могут обрабатывать маски (например, слой LSTM ), имеют аргумент mask в их методе __call__ .

Между тем, слои, которые создают маску (например, Embedding ), предоставляют compute_mask(input, previous_mask) который вы можете вызвать.

Таким образом, вы можете передать вывод метода compute_mask() слоя, __call__ методу __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([[-0.0071368 ,  0.00202324,  0.00393163, ..., -0.00365972,
        -0.00194294, -0.00275828],
       [ 0.00865301, -0.00411554, -0.00328279, ...,  0.00395685,
         0.01023738, -0.0013066 ],
       [ 0.0115475 , -0.00367757, -0.0049072 , ...,  0.00312295,
         0.00557074,  0.00681297],
       ...,
       [ 0.00537544, -0.00517081,  0.00668133, ...,  0.00428408,
         0.00251086, -0.00211114],
       [ 0.00286667, -0.00301991, -0.0095289 , ...,  0.00381294,
         0.00675705, -0.00599195],
       [-0.0045211 ,  0.0019338 , -0.00031986, ...,  0.00275819,
        -0.00126366, -0.00347176]], dtype=float32)>

Поддержка маскировки в ваших пользовательских слоях

Иногда вам может потребоваться написать слои, которые генерируют маску (например, Embedding ), или слои, которым необходимо изменить текущую маску.

Например, любой слой, который создает тензор с измерением времени, отличным от его входных данных, такой как слой Concatenate который объединяется по измерению времени, должен будет изменить текущую маску, чтобы нижележащие слои могли правильно принимать замаскированные временные шаги в учетная запись.

Для этого ваш слой должен реализовать метод 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 False  True]
 [ True  True  True  True  True  True  True  True  True  True]
 [ True  True  True False  True  True  True  True  True  True]], shape=(3, 10), dtype=bool)

Включение маски распространения на совместимых слоях

Большинство слоев не изменяют временное измерение, поэтому нет необходимости изменять текущую маску. Тем не менее, они могут по-прежнему хотеть иметь возможность распространить текущую маску без изменений на следующий слой. Это добровольное поведение. По умолчанию пользовательский слой уничтожает текущую маску (поскольку платформа не имеет возможности определить, безопасно ли распространение маски).

Если у вас есть настраиваемый слой, который не изменяет измерение времени, и если вы хотите, чтобы он мог распространять текущую маску ввода, вы должны установить 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')

Написание слоев, которым нужна информация о маске

Некоторые слои являются потребителями масок: они принимают аргумент mask в call и используют его, чтобы определить, пропустить ли определенные временные шаги.

Чтобы написать такой слой, вы можете просто добавить аргумент mask=None в свою подпись call . Маска, связанная с входами, будет передаваться на ваш слой всякий раз, когда она доступна.

Вот простой пример ниже: слой, который вычисляет softmax по временному измерению (ось 1) входной последовательности, отбрасывая замаскированные временные шаги.

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_exp * 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. Резюмируем:

  • «Маскирование» - это то, как слои могут узнать, когда пропускать / игнорировать определенные временные шаги во входных данных последовательности.
  • Некоторые слои маски-генераторы: Embedding может генерировать маску из входных значений (если mask_zero=True ), и так может Masking слой.
  • Некоторые слои являются потребителями маски: они предоставляют аргумент mask в своем методе __call__ . Так обстоит дело со слоями RNN.
  • В функциональном API и последовательном API информация о маске распространяется автоматически.
  • При автономном использовании слоев вы можете передать аргументы mask слоям вручную.
  • Вы можете легко написать слои, которые изменяют текущую маску, генерируют новую маску или используют маску, связанную с входами.