Код: Выделить всё
First sample is using real sample, preventing backward in generator, allowing backward in discriminator.
Second sample is using generated sample, preventing backward in generator, allowing backward in discriminator.
Third sample is using generated sample, allowing backward in generator, preventing backward in discriminator, but redirect gradient to generator
Fourth sample is basically like first sample.
etc...
Код: Выделить всё
#@title Control Flow Layers Definition
@keras_export("keras.layers.StopGradient")
class StopGradient(layers.Layer):
"""Layer to stop gradient flow during backward propagation.
In another word, only allowing forward propagation.
It takes a single tensor, and returns a single tensor (also of the same shape).
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
def call(self, inputs):
return layers.Lambda(lambda x: ops.stop_gradient(x))(inputs)
def compute_output_shape(self, input_shape):
return input_shape
@keras_export("keras.layers.ControlForward")
class ControlForward(layers.Layer):
"""Layer to control which layer should be passed during forward propagation between two layers based on control mask.
It takes as input a list of tensors.
The first tensor is control mask with shape (None, 1), while the second and third tensors are candidates between true or false, respectively.
Both candidates must be the same.
Returns a single tensor (also of the same shape like candidate layer).
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
def call(self, inputs):
mask, layer_if_true, layer_if_false = inputs
# Cast the mask to the same type like candidate layer
mask = layers.Lambda(lambda x: ops.cast(x, dtype=layer_if_true.dtype))(mask)
mask_h = layers.Lambda(lambda x: ops.logical_not(x))(mask)
# Apply the layer if true
layer_if_true = layers.Multiply()([layer_if_true, mask])
# Apply the layer if false
layer_if_false = layers.Multiply()([layer_if_false, mask_h])
return layers.Add()([layer_if_true, layer_if_false])
def compute_output_shape(self, input_shape):
mask_shape, layer_if_true_shape, layer_if_false_shape = input_shape
# Mask shape must be (None, 1)
if mask_shape != (None, 1):
raise ValueError("Control mask shape must be (None, 1), but got {}".format(mask_shape))
# Make sure both layers have the same shape
if layer_if_true_shape != layer_if_false_shape:
raise ValueError("Both candidates must be the same shape., but got {} and {}".format(layer_if_true_shape, layer_if_false_shape))
return layer_if_true_shape
@keras_export("keras.layers.ControlBackward")
class ControlBackward(layers.Layer):
"""Layer to control whether a layer should be passed during backward propagation based on control mask.
This is also called controlled stop gradient, except it contains a control mask.
The control mask is following the principle of OR logic.
Which means if one of them is true, then there must be weight changes in the previous layer.
It takes as input a list of tensors.
The first tensor is control mask with shape (None, 1), while the second is previous layer.
Returns a single tensor (also of the same shape like previous layer).
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
def call(self, inputs):
mask, prev_layer = inputs
# Cast the mask to the same type as prev_layer
mask = layers.Lambda(lambda x: ops.cast(x, dtype=prev_layer.dtype))(mask)
mask_h = layers.Lambda(lambda x: ops.logical_not(x))(mask)
# Apply the stop_gradient function on the masked parts
stopped_gradient_part = layers.Lambda(lambda x: ops.stop_gradient(x))(prev_layer)
# Multiply stopped gradient part with mask_h
stopped_gradient_masked = layers.Multiply()([stopped_gradient_part, mask_h])
# Multiply normal (non-stopped) part with mask
non_stopped_gradient_part = layers.Multiply()([prev_layer, mask])
# Add the stopped and non-stopped parts
return layers.Add()([stopped_gradient_masked, non_stopped_gradient_part])
def compute_output_shape(self, input_shape):
mask_shape, prev_layer_shape = input_shape
# Mask shape must be (None, 1)
if mask_shape != (None, 1):
raise ValueError("Mask control shape must be (None, 1), but got {}".format(mask_shape))
return prev_layer_shape
@keras_export("keras.layers.RedirectGradient")
class RedirectGradient(layers.Layer):
"""
Layer to allow forward propagation through one layer and redirect gradients to another layer.
During forward propagation, the inputs pass through the forward layer.
During backward propagation, the gradients are redirected to the backward layer.
This acts like a gradient flow jumper.
It takes as input a list of tensors.
The first tensor is the forward layer, while the second is the backward layer.
Returns a single tensor (also of the same shape like forward layer).
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
def call(self, inputs):
forward_layer, backward_layer = inputs
# Forward pass: pass the inputs through the forward layer
forward_pass = forward_layer
# Stop gradient for the forward layer (forward_layer should not propagate gradients)
forward_stopped = layers.Lambda(lambda x: ops.stop_gradient(x))(forward_pass)
# Gradients are applied to the backward_layer
# This ensures that only backward_layer gets the gradients during backpropagation
backward_gradients = backward_layer
# Combine forward propagation and backward gradient redirection
return layers.Add()([forward_stopped, backward_gradients])
def compute_output_shape(self, input_shape):
forward_layer_shape, _ = input_shape
return forward_layer_shape
Код: Выделить всё
'''
forward_mask: allow real sample if 1, else generated sample
backward_mask: allow propagation of generator if 1, else stop propagation
'''
latent_input_layer = layers.Input(shape=(784,), name="latent_il")
real_input_layer = layers.Input(shape=(28, 28, 1), name="real_il")
forward_mask = layers.Input(shape=(1,), name="forward_mask")
backward_mask = layers.Input(shape=(1,), name="backward_mask")
backward_mask_inverse = layers.Input(shape=(1,), name="backward_mask_h")
# Generator Model
generator_hidden_layer = layers.Dense(7 * 7 * 28)(latent_input_layer)
generator_hidden_layer = layers.Reshape((7, 7, 28))(generator_hidden_layer)
generator_hidden_layer = layers.Conv2DTranspose(28, kernel_size=4, strides=2, padding="same")(generator_hidden_layer)
generator_hidden_layer = layers.LeakyReLU(negative_slope=0.2)(generator_hidden_layer)
generator_hidden_layer = layers.Conv2DTranspose(64, kernel_size=4, strides=2, padding="same")(generator_hidden_layer)
generator_hidden_layer = layers.LeakyReLU(negative_slope=0.2)(generator_hidden_layer)
generator_output_layer = layers.Conv2D(1, kernel_size=5, padding="same", activation="sigmoid", name='g_ol')(generator_hidden_layer)
generator_output_layer = ControlBackward()([backward_mask, generator_output_layer])
# Candidate Image
img = ControlForward()([forward_mask, real_input_layer, generator_output_layer])
# Discriminator Model
discriminator_hidden_layer = layers.Conv2D(64, kernel_size=4, strides=2, padding="same")(img)
discriminator_hidden_layer = layers.LeakyReLU(negative_slope=0.2)(discriminator_hidden_layer)
discriminator_hidden_layer = layers.Conv2D(128, kernel_size=4, strides=2, padding="same")(discriminator_hidden_layer)
discriminator_hidden_layer = layers.LeakyReLU(negative_slope=0.2)(discriminator_hidden_layer)
discriminator_hidden_layer = layers.Conv2D(128, kernel_size=4, strides=2, padding="same")(discriminator_hidden_layer)
discriminator_hidden_layer = layers.LeakyReLU(negative_slope=0.2)(discriminator_hidden_layer)
discriminator_hidden_layer = layers.Flatten()(discriminator_hidden_layer)
discriminator_hidden_layer = layers.Dropout(0.2)(discriminator_hidden_layer)
discriminator_output_layer = layers.Dense(1, activation="sigmoid", name='d_ol')(discriminator_hidden_layer)
discriminator_output_layer = ControlBackward()([backward_mask_inverse, discriminator_output_layer])
model = Model(inputs=[real_input_layer, latent_input_layer, forward_mask, backward_mask, backward_mask_inverse], outputs=[generator_output_layer, discriminator_output_layer])
model.compile(optimizer="adam", loss=["binary_crossentropy", "binary_crossentropy"], loss_weights=[1, 1])
model.summary()
Код: Выделить всё
# seed 42
np.random.seed(42)
real_samples_7 = np.expand_dims(real_samples_7, axis=-1) # (7000, 28, 28, 1) float32
random_latent_vectors = np.random.normal(size=(real_samples_7.shape[0], real_samples_7.shape[1]*real_samples_7.shape[2])) # (7000, 784) float32
# tile until length of sample
X_forward_mask = np.tile(np.array([1, 0, 0]), (real_samples_7.shape[0])).astype('bool') # (21000,) bool
X_backward_mask = np.tile(np.array([0, 0, 1]), (real_samples_7.shape[0])).astype('bool') # (21000,) bool
X_backward_mask_inverse = np.tile(np.array([1, 1, 0]), (real_samples_7.shape[0])).astype('bool') # (21000,) bool
y_discriminator = np.tile(np.array([1, 0, 1]), (real_samples_7.shape[0])).astype('bool') # (21000,) bool
y_generator = np.array([np.full((28,28,1), scalar) for scalar in y_discriminator]) # (21000, 28, 28, 1) bool
# repeat axis=0 three times because of periodic length
X_real_samples = np.repeat(real_samples_7, 3, axis=0).astype('float32') # (21000, 28, 28, 1) float32
X_random_latent_vectors = np.repeat(random_latent_vectors, 3, axis=0).astype('float32') # (21000, 784) float32
Код: Выделить всё
history = model.fit([X_real_samples, X_random_latent_vectors, X_forward_mask, X_backward_mask, X_backward_mask], [y_generator, y_discriminator], epochs=10, batch_size=6, shuffle=False, verbose=1)
Код: Выделить всё
Epoch 1/10
3500/3500 ━━━━━━━━━━━━━━━━━━━━ 11s 3ms/step - loss: 10.6243
Epoch 2/10
3500/3500 ━━━━━━━━━━━━━━━━━━━━ 21s 3ms/step - loss: 10.6267
Epoch 3/10
3500/3500 ━━━━━━━━━━━━━━━━━━━━ 20s 3ms/step - loss: 10.6278
Epoch 4/10
3500/3500 ━━━━━━━━━━━━━━━━━━━━ 20s 3ms/step - loss: 10.6281
Epoch 5/10
Я также протестировал уровни потока управления, которые работали должным образом.
Так что же не так с моим настроить?
Подробнее здесь: https://stackoverflow.com/questions/790 ... y-in-keras
