Реализация ввода маскировки и прокладки в моделях Tensorflow Keras
19 июля 2025 г.Обзор контента
- Настраивать
- Введение
- Данные последовательности
- Маскировка
- Слои, генерирующие маски: внедрение и маскировка
- Распространение маски в функциональном API и последовательном API
- Передача тензоров маски непосредственно к слоям
- Поддержка маскировки в ваших пользовательских слоях
- Выбор распространения маски на совместимые слои
- Написание слоев, которые нуждаются в информации о маске
- Краткое содержание
Настраивать
import numpy as np
import tensorflow as tf
import keras
from 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)В этом случае), образцы, которые являются короче, чем самый длинный элемент, необходимо сочетать с некоторой стоимостью заполнителя (в качестве альтернативы, можно также усечь длинные образцы перед прокладками коротких образцов).
Керас обеспечивает функцию утилиты для списков усечения и падки Python до общей длины:tf.keras.utils.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 use "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.utils.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)
Как вы можете видеть из печатного результата, маска представляет собой 2D -логический тензор с формой(batch_size, sequence_length), где каждый человекFalseВход указывает, что соответствующий временный пакет следует игнорировать во время обработки.
Распространение маски в функциональном API и последовательном API
При использовании функционального API или последовательного API, маска, сгенерированнаяEmbeddingилиMaskingУровень будет распространяться через сеть для любого уровня, который способен их использовать (например, слои RNN). Керас автоматически принесет маску, соответствующую входу, и передаст ее на любой слой, который знает, как ее использовать.
Например, в следующей последовательной модели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__Метод слоя, требующего маски, например:
class MyLayer(layers.Layer):
def __init__(self, **kwargs):
super().__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([[ 1.1063378e-04, -5.7033719e-03, 3.0645048e-03, ...,
3.6328615e-04, -2.8766368e-03, -1.3289017e-03],
[-9.2790304e-03, -1.5139847e-02, 5.7660388e-03, ...,
3.5337124e-03, 4.0699611e-03, -3.9524431e-04],
[-3.4190060e-03, 7.9529232e-04, 3.7830453e-03, ...,
-6.8300538e-04, 4.7965860e-03, 4.4357078e-03],
...,
[-4.3796434e-04, 3.5149506e-03, 5.0854073e-03, ...,
6.3023632e-03, -4.6664057e-03, -2.1111544e-03],
[ 1.2171637e-03, -1.8671650e-03, 8.6708134e-03, ...,
-2.6730294e-03, -1.6238958e-03, 5.9354519e-03],
[-7.1832030e-03, -6.0863695e-03, 4.3814078e-05, ...,
3.8765911e-03, -1.7828923e-03, -2.3530782e-03]], 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().__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 False True False True True True]
[ True False True False True True True True True True]], shape=(3, 10), dtype=bool)
Примечание:Для получения более подробной информации об ограничениях формата, связанных с маскировкой, см.
Выбор распространения маски на совместимые слои
Большинство слоев не изменяют размер времени, поэтому не нужно изменять текущую маску. Однако они все еще могут хотеть иметь возможностьраспространятьТекущая маска, неизменная, до следующего слоя.Это поведение, поведение.По умолчанию пользовательский слой разрушит текущую маску (поскольку фреймворк не может сказать, безопасно ли распространение маски).
Если у вас есть пользовательский слой, который не изменяет размер времени, и если вы хотите, чтобы он мог распространять текущую входную маску, вам следует установитьself.supports_masking = TrueВ конструкторе слоя. В этом случае поведение по умолчаниюcompute_mask()это просто пройти текущую маску через.
Вот пример слоя, который является белым списком для распространения маски:
@keras.saving.register_keras_serializable()
class MyActivation(keras.layers.Layer):
def __init__(self, **kwargs):
super().__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) входной последовательности, при этом отбрасывая маскированные временные рамки.
@keras.saving.register_keras_serializable()
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)))
Краткое содержание
Это все, что вам нужно знать о накладке и маскировании в керах. Чтобы подтвердить:
- «Маскировка» - это то, как слои могут знать, когда пропустить / игнорировать определенные временные рамки при входах последовательностей.
- Некоторые слои являются генераторами маски:
Embeddingможет генерировать маску из входных значений (еслиmask_zero=True), как иMaskingслой. - Некоторые слои являются масками-потребителями: они выставляют
maskаргумент в их__call__метод Это относится к слоям RNN. - В функциональном API и последовательном API информация о маске распространяется автоматически.
- При использовании слоев отдельно вы можете пройти
maskАргументы на слои вручную. - Вы можете легко написать слои, которые изменяют текущую маску, которые генерируют новую маску или потребляют маску, связанную с входами.
Первоначально опубликовано на
Оригинал