python - 在模型中使用 run_eagerly=False 时出错。在 Tensorflow 中编译自定义 Keras 模型

问题描述

我正在 Tensorflow 中开发一个自定义模型。我正在尝试从https://arxiv.org/abs/1704.03976实施虚拟对抗训练 (VAT) 模型。该模型在其分类任务中同时使用标记和未标记的数据。因此，在train_step模型中，我需要将batch的数据分为有标签的（0，或1），或者无标签的（-1）。使用 run_eagerly=True 编译模型时，它似乎按预期工作，但是当我使用 run_eagerly=False 时，它给了我以下错误：

ValueError: Number of mask dimensions must be specified, even if some dimensions are None.  E.g. shape=[None] is ok, but shape=None is not.

这似乎是在：

X_l, y_l = tf.boolean_mask(X, tf.logical_not(missing)), tf.boolean_mask(y, tf.logical_not(missing))

我不确定是什么导致了错误，但它似乎与仅在run_eagerly=False. 我需要该boolean_mask功能来区分标记和未标记的数据。我希望有人可以帮助我。为了重现错误，我添加了模型和一个小型仿真示例。设置时，模拟将产生我遇到的错误run_eagerly=False。

提前致谢。

型号定义：

from tensorflow import keras
import tensorflow as tf


metric_acc = keras.metrics.BinaryAccuracy()
metric_loss = keras.metrics.Mean('loss')


class VAT(keras.Model):

    def __init__(self, units_1=16, units_2=16, dropout=0.3, xi=1e-6, epsilon=2.0, alpha=1.0):
        super(VAT, self).__init__()

        # Set model parameters
        self.units_1 = units_1
        self.units_2 = units_2
        self.dropout = dropout
        self.xi = xi
        self.epsilon = epsilon
        self.alpha = alpha

        # First hidden
        self.dense1 = keras.layers.Dense(self.units_1)
        self.activation1 = keras.layers.Activation(tf.nn.leaky_relu)
        self.dropout1 = keras.layers.Dropout(self.dropout)

        # Second hidden
        self.dense2 = keras.layers.Dense(self.units_2)
        self.activation2 = keras.layers.Activation(tf.nn.leaky_relu)
        self.dropout2 = keras.layers.Dropout(self.dropout)

        # Output layer
        self.dense3 = keras.layers.Dense(1)
        self.activation3 = keras.layers.Activation("sigmoid")

    def call(self, inputs, training=None, mask=None):

        x1 = self.dense1(inputs)
        x2 = self.activation1(x1)
        x3 = self.dropout1(x2, training=True)

        x4 = self.dense2(x3)
        x5 = self.activation2(x4)
        x6 = self.dropout2(x5, training=True)

        x7 = self.dense3(x6)
        x8 = self.activation3(x7)

        return x8

    def generate_perturbation(self, inputs):

        # Generate normal vectors
        d = tf.random.normal(shape=tf.shape(inputs))

        # Normalize vectors
        d = tf.math.l2_normalize(d, axis=1)

        # Calculate r
        r = self.xi * d

        # Make predictions
        p = self(inputs, training=True)

        # Tape gradient
        with tf.GradientTape() as tape:
            tape.watch(r)

            # Perturbed predictions
            p_perturbed = self(inputs + r, training=True)

            # Calculate divergence
            D = keras.losses.KLD(p, p_perturbed) + keras.losses.KLD(1 - p, 1 - p_perturbed)

        # Calculate gradient
        gradient = tape.gradient(D, r)

        # Calculate r_vadv
        r_vadv = tf.math.l2_normalize(gradient, axis=1)

        # Return virtual adversarial perturbation
        return r_vadv

    @tf.function
    def train_step(self, data):

        # Unpack data
        X, y = data

        # Missing label boolean indices
        missing = tf.squeeze(tf.equal(y, -1))

        # Split data into labeled and unlabeled data
        X_l, y_l = tf.boolean_mask(X, tf.logical_not(missing)), tf.boolean_mask(y, tf.logical_not(missing))
        X_u = tf.boolean_mask(X, missing)

        # Calculate virtual perturbations for labeled and unlabeled
        r_l = self.generate_perturbation(X_l)
        r_u = self.generate_perturbation(X_u)

        # Tape gradient
        with tf.GradientTape() as model_tape:
            model_tape.watch(self.trainable_variables)

            # Calculate probabilities real data
            prob_l, prob_u = self(X_l, training=True), self(X_u, training=True)

            # Calculate probabilities perturbed data
            prob_r_l, prob_r_u = self(X_l + self.epsilon * r_l, training=True), self(X_u + self.epsilon * r_u, training=True)

            # Calculate loss
            loss = vat_loss(y_l, prob_l, prob_u, prob_r_l, prob_r_u, self.alpha)

        # Calculate gradient
        model_gradient = model_tape.gradient(loss, self.trainable_variables)

        # Update weights
        self.optimizer.apply_gradients(zip(model_gradient, self.trainable_variables))

        # Compute metrics
        metric_acc.update_state(y_l, prob_l)
        metric_loss.update_state(loss)

        return {'loss': metric_loss.result(), 'accuracy': metric_acc.result()}

    @property
    def metrics(self):
        return [metric_loss, metric_acc]


def vat_loss(y_l, prob_l, prob_u, prob_r_l, prob_r_u, alpha):
    N_l = tf.cast(tf.size(prob_l), dtype=tf.dtypes.float32)
    N_u = tf.cast(tf.size(prob_u), dtype=tf.dtypes.float32)

    if tf.equal(N_l, 0):
        # No labeled examples: get contribution from unlabeled data using perturbations
        R_vadv = tf.reduce_sum(
            keras.losses.KLD(prob_u, prob_r_u)
            + keras.losses.KLD(1 - prob_u, 1 - prob_r_u)
        )

        return alpha * R_vadv / N_u

    elif tf.equal(N_u, 0):

        # No unlabeled examples: get contribution from labeled data
        R = tf.reduce_sum(keras.losses.binary_crossentropy(y_l, prob_l))

        R_vadv = tf.reduce_sum(
            keras.losses.KLD(prob_l, prob_r_l)
            + keras.losses.KLD(1 - prob_l, 1 - prob_r_l)
        )

        return R / N_l + alpha * R_vadv / N_l

    else:
        # Get contribution from labeled data
        R = tf.reduce_sum(keras.losses.binary_crossentropy(y_l, prob_l))

        # Get contribution from labeled and unlabeled data using perturbations
        R_vadv = tf.reduce_sum(
            keras.losses.KLD(prob_l, prob_r_l)
            + keras.losses.KLD(1 - prob_l, 1 - prob_r_l)
        ) + tf.reduce_sum(
            keras.losses.KLD(prob_u, prob_r_u)
            + keras.losses.KLD(1 - prob_u, 1 - prob_r_u)
        )

        return R / N_l + alpha * R_vadv / (N_l + N_u)

模拟示例： 为了显示模型/代码按预期工作（使用时run_eagerly=True，我做了一个模拟示例。在此示例中，我在观察被标记/未标记时产生偏差。下图说明了模型使用的标记观察（黄色或紫色）和未标记的观察值（蓝色）。

VAT 的准确度约为 0.75，而参考模型的准确度约为 0.58。这些精度是在没有超参数调整的情况下产生的。

from modules.vat import VAT

import numpy as np
from sklearn import datasets
from sklearn.model_selection import train_test_split
import tensorflow as tf
from tensorflow import keras
import matplotlib.pyplot as plt

def create_biased_sample(x, proportion_labeled):
    labeled =  np.random.choice([True, False], p=[proportion_labeled, 1-proportion_labeled])
    if x[0] < 0.0:
        return False
    elif x[0] > 1.0:
        return False
    else:
        return labeled


# Simulation parameters
N = 2000
proportion_labeled = 0.15

# Model training parameters
BATCH_SIZE = 128
BUFFER_SIZE = 60000
EPOCHS = 100

# Generate a dataset
X, y = datasets.make_moons(n_samples=N, noise=.05, random_state=3)
X, y = X.astype('float32'), y.astype('float32')
y = y.reshape(-1, 1)


# Split in train and test
X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=0.5)

# Simulate missing labels
sample_biased = lambda x: create_biased_sample(x, proportion_labeled)
labeled = np.array([sample_biased(k) for k in X_train])
y_train[~ labeled] = -1

# Estimate VAT model
vat = VAT(dropout=0.2, units_1=16, units_2=16, epsilon=0.5)
vat.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.01), run_eagerly=True)
vat.fit(X_train, y_train, batch_size=BATCH_SIZE, epochs=EPOCHS, shuffle=True)

# Estimate a reference model
reference = keras.models.Sequential([
    keras.layers.Input(shape=(2,)),
    keras.layers.Dense(16),
    keras.layers.Activation(tf.nn.leaky_relu),
    keras.layers.Dropout(0.2),
    keras.layers.Dense(16),
    keras.layers.Activation(tf.nn.leaky_relu),
    keras.layers.Dropout(0.2),
    keras.layers.Dense(1),
    keras.layers.Activation("sigmoid")
])
reference.compile(optimizer=keras.optimizers.Adam(learning_rate=0.01), loss=keras.losses.binary_crossentropy, run_eagerly=False)
reference.fit(X_train[y_train.flatten() != -1, :], y_train[y_train.flatten() != -1], batch_size=BATCH_SIZE, epochs=EPOCHS, shuffle=True)

# Calculate out-of-sample accuracies
test_acc_vat = tf.reduce_mean(keras.metrics.binary_accuracy(y_test, vat(X_test, training=False)))
test_acc_reference = tf.reduce_mean(keras.metrics.binary_accuracy(y_test, reference(X_test, training=False)))

# Print results
print('Test accuracy of VAT: {}'.format(test_acc_vat))
print('Test accuracy of reference model: {}'.format(test_acc_reference))

# Plot scatter
plt.scatter(X_test[:, 0], X_test[:, 1])
plt.scatter(X_train[y_train.flatten() != -1, 0], X_train[y_train.flatten() != -1, 1], c=y_train.flatten()[y_train.flatten() != -1])

标签： pythontensorflowkerasdeep-learningeager-execution

python - 在模型中使用 run_eagerly=False 时出错。在 Tensorflow 中编译自定义 Keras 模型

问题描述

解决方案

推荐阅读