Maskowanie i wyściółka Keras

Zadbaj o dobrą organizację dzięki kolekcji Zapisuj i kategoryzuj treści zgodnie ze swoimi preferencjami.

Zobacz na TensorFlow.org Uruchom w Google Colab Wyświetl źródło na GitHub Pobierz notatnik

Ustawiać

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

Wstęp

Maskowanie jest sposobem przetwarzania sekwencji powiedzieć, że pewne warstwy timesteps w wejście brakuje, a zatem powinny być pomijane podczas przetwarzania danych.

Wyściółka jest szczególną formą maskowania, w którym etapy są zamaskowane na początku lub na końcu sekwencji. Dopełnienie wynika z potrzeby kodowania danych sekwencji w sąsiednich partiach: aby wszystkie sekwencje w paczce pasowały do ​​określonej standardowej długości, konieczne jest wypełnienie lub obcięcie niektórych sekwencji.

Przyjrzyjmy się bliżej.

Dopełnienie danych sekwencji

Podczas przetwarzania danych sekwencji bardzo często zdarza się, że poszczególne próbki mają różne długości. Rozważmy następujący przykład (tekst tokenizowany jako słowa):

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

Po wyszukaniu słownika dane mogą być zwektoryzowane jako liczby całkowite, np.:

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

Dane są listą zagnieżdżoną, w której poszczególne próbki mają odpowiednio długość 3, 5 i 6. Ponieważ dane wejściowe do głębokiego modelu uczenia się musi być pojedynczym tensor (kształt np (batch_size, 6, vocab_size) w tym przypadku), próbki, które są krótsze niż najdłuższa poz konieczności zostać wypełniona jakąś wartość zastępczego (alternatywnie, jeden może również obcinać długie próbki przed dopełnieniem krótkich próbek).

Keras zapewnia funkcję użytkową do ucięcia i podkładki list Pythona do wspólnej długości: 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]]

Maskowanie

Teraz, gdy wszystkie próbki mają jednakową długość, model musi zostać poinformowany, że jakaś część danych jest w rzeczywistości wypełniająca i należy ją zignorować. Mechanizm ten jest maskowanie.

Istnieją trzy sposoby wprowadzenia masek wprowadzania w modelach Keras:

  • Dodaj keras.layers.Masking warstwę.
  • Konfigurowanie keras.layers.Embedding warstwę z mask_zero=True .
  • Przepuścić mask argumentu ręcznie przy wywołaniu warstwy, które obsługują ten argument (np warstw RNN).

Maska warstwy generujące: Embedding i Masking

W obszarze okapu, warstwy te będą tworzyć tensora maski (tensor 2D kształtu (batch, sequence_length) ) i zamocować go na wyjście tensora zwróconego przez Masking lub Embedding warstwy.

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)

Jak widać z wydrukowanym wyniku, maska jest 2D logiczna tensor z kształtem (batch_size, sequence_length) , gdzie każda jednostka False wpis oznacza, że odpowiadająca kroku czasu powinny być ignorowane podczas przetwarzania.

Propagacja masek w interfejsie API funkcjonalnym i sekwencyjnym

Przy użyciu API funkcyjną lub sekwencyjne API maska generowany przez Embedding lub Masking warstwy będzie propagowane przez sieć dla każdej warstwy, który jest zdolny do ich stosowania (na przykład, warstwy RNN). Keras automatycznie pobierze maskę odpowiadającą danemu wejściowemu i przekaże ją dowolnej warstwie, która wie, jak jej użyć.

Na przykład, w następujący sekwencyjnego modelu LSTM warstwa automatycznie otrzyma maskę, co oznacza, że będzie ignorować wartości wypełnienia:

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

Dotyczy to również następującego modelu funkcjonalnego interfejsu 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)

Przekazywanie tensorów maski bezpośrednio do warstw

Warstwy mogą obsługiwać maski (takie jak LSTM warstwy) mają mask argument ich __call__ metody.

Tymczasem warstwy, które produkują maski (np Embedding ) wystawiać compute_mask(input, previous_mask) metoda, którą można nazwać.

W ten sposób można przejść wyjście compute_mask() metodą warstwy maski w celu produkcji __call__ sposobu maski czasochłonne warstwy, na przykład:

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([[-3.6287602e-04,  8.8942451e-03, -4.5623952e-03, ...,
         3.6509466e-04, -4.3871473e-03, -1.7532009e-03],
       [ 2.6261162e-03, -2.5420082e-03,  7.6517118e-03, ...,
         5.8210879e-03, -1.5617531e-03, -1.7562184e-03],
       [ 6.8687932e-03,  1.2330032e-03, -1.2028826e-02, ...,
         2.0486799e-03,  5.7172528e-03,  2.6641595e-03],
       ...,
       [-3.4327951e-04,  1.3967649e-03, -1.2102776e-02, ...,
         3.8406218e-03, -2.3374180e-03, -4.9669710e-03],
       [-2.3023323e-03,  1.8474255e-03,  2.7329330e-05, ...,
         6.1798934e-03,  4.2709545e-04,  3.9026213e-03],
       [ 7.4090287e-03,  1.9879336e-03, -2.0261200e-03, ...,
         8.2100276e-03,  8.7051848e-03,  9.9167246e-03]], dtype=float32)>

Wspieranie maskowania w niestandardowych warstwach

Czasami trzeba napisać, że generują warstw maski (jak Embedding ), lub warstw, które trzeba zmodyfikować aktualną maskę.

Na przykład, każda warstwa powoduje tensor o innym wymiarze czasowym niż jego wejścia, takich jak Concatenate warstwą Łączy on wymiaru czasu, potrzebne do modyfikacji aktualnej maski tak, żeby dalszych warstw będzie w stanie właściwie się zamaskowane timesteps do rachunek.

Aby to zrobić, Twój warstwa powinna wdrożyć layer.compute_mask() metodę, która produkuje nową maskę danego wejścia i aktualną maskę.

Oto przykład TemporalSplit warstwy, która musi zmienić aktualny maskę.

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)

Inny przykład z CustomEmbedding warstwy, która jest w stanie wytworzyć maskę z wartości wejściowych

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

Włączenie maskowania propagacji na kompatybilnych warstwach

Większość warstw nie modyfikuje wymiaru czasu, więc nie ma potrzeby modyfikowania bieżącej maski. Mogą one jednak nadal chcą, aby móc propagować aktualną maskę niezmienione do następnej warstwy. To jest zachowanie opt-in. Domyślnie warstwa niestandardowa zniszczy bieżącą maskę (ponieważ framework nie ma możliwości określenia, czy propagacja maski jest bezpieczna).

Jeśli masz warstwę niestandardową że nie zmienia wymiar czasu, a jeśli chcesz, aby móc propagować aktualną maskę wprowadzania, należy ustawić self.supports_masking = True w konstruktorze warstwy. W tym przypadku domyślne zachowanie compute_mask() jest po prostu przejść przez obecną maskę.

Oto przykład warstwy umieszczonej na białej liście do propagacji maski:

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)

Teraz można korzystać z tej niestandardowej warstwy w między warstwą generowania maski (jak Embedding ) oraz warstwę maski czasochłonne (np LSTM ), i to minie maskę wzdłuż tak, że osiągnie warstwę maski czasochłonne.

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

Pisanie warstw, które wymagają informacji o masce

Niektórzy konsumenci są maski warstwy: oni zaakceptować mask argument w call i używać go w celu ustalenia, czy pominąć pewne etapy czasowe.

Aby napisać taką warstwę, można po prostu dodać mask=None argumentu w swoim call podpisu. Maska powiązana z danymi wejściowymi zostanie przekazana do Twojej warstwy, gdy tylko będzie dostępna.

Oto prosty przykład poniżej: warstwa, która oblicza softmax w wymiarze czasowym (oś 1) sekwencji wejściowej, jednocześnie odrzucając zamaskowane kroki czasowe.

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

Streszczenie

To wszystko, co musisz wiedzieć o wyściółce i maskowaniu w Keras. Przypomnę:

  • „Maskowanie” to sposób, w jaki warstwy mogą wiedzieć, kiedy pominąć / zignorować określone kroki czasowe w wprowadzanych sekwencjach.
  • Niektóre warstwy masek są generatory: Embedding może generować maski z wartości wejściowych (jeśli mask_zero=True ) itd CAN Masking warstwy.
  • Niektóre warstwy są maski konsumenci: oni narazić mask argument w ich __call__ metody. Tak jest w przypadku warstw RNN.
  • W interfejsach Functional API i Sequential API informacje o masce są propagowane automatycznie.
  • Podczas korzystania z warstw w sposób samodzielny, można przekazać mask argumentów warstw ręcznie.
  • Możesz łatwo pisać warstwy, które modyfikują bieżącą maskę, generują nową maskę lub zużywają maskę powiązaną z danymi wejściowymi.