Keras 中的 CNN 模型条件层

Question

我正在尝试构建一个conditional CNN 模型。模型是，

在我的模型的first stage，我将我的数据提供给Model 1，然后based on the prediction of Model 1，我想train the model to Conditional Cat model or Conditional Dog model，最后，给出条件猫模型或条件狗模型的输出。 我该怎么做？

注意： 我的努力是，

import keras
from keras.layers import *
from keras.models import *
from keras.utils import *

img_rows,img_cols,number_of_class = 256,256,2
input = Input(shape=(img_rows,img_cols,3))

#----------- main model (Model 1) ------------------------------------
conv_01 = Convolution2D(64, 3, 3, activation='relu',name = 'conv_01') (input)
conv_02 = Convolution2D(64, 3, 3, activation='relu',name = 'conv_02') (conv_01)

skip_dog =  conv_02

conv_03 = Convolution2D(64, 3, 3, activation='relu',name = 'conv_03') (conv_02)

skip_cat =  conv_03

conv_04 = Convolution2D(64, 3, 3, activation='relu',name = 'conv_04') (conv_03)


flatten_main_model =  Flatten() (conv_04)
Output_main_model = Dense(units = number_of_class , activation = 'softmax', name = "Output_layer")(flatten_main_model)


#----------- Conditional  Cat model ------------------------------------ 
conv_05 = Convolution2D(64, 3, 3, activation='relu',name = 'conv_05') (skip_cat)
flatten_cat_model =  Flatten() (conv_05)
Output_cat_model = Dense(units = number_of_class , activation = 'softmax', name = "Output_layer_cat")(flatten_cat_model)

#----------- Conditional  Dog model ------------------------------------ 
conv_06 = Convolution2D(64, 3, 3, activation='relu',name = 'conv_06') (skip_dog)
flatten_dog_model =  Flatten() (conv_06)
Output_dog_model = Dense(units = number_of_class , activation = 'softmax', name = "Output_layer_dog")(flatten_dog_model)

#----------------------------- My discrete 3 models --------------------------------
model_01 = Model(inputs = input , outputs = Output_main_model,name = 'model_main')
model_02_1 = Model(inputs = input , outputs = Output_cat_model ,name = 'Conditional_cat_model')
model_02_2 = Model(inputs = input , outputs = Output_dog_model ,name = 'Conditional_dog_model')

如何根据这些条件合并这 3 个模型 (model_01, model_02_1, model_02_2)？

**条件为：**

1. 将数据输入模型model_01
2. 基于model_01 结果馈送数据到model_02_1 or model_02_2
3. 接下来，从model_02_1 or model_02_2预测最终输出

Answer 1

神经网络中的条件问题

将开关或条件（如 if-then-else）作为神经网络的一部分的问题在于，条件并非处处可微。因此，自动微分方法不会直接起作用，解决这个问题非常复杂。更多详情请查看this。

一个捷径是您最终可以独立训练 3 个单独的模型，然后在推理过程中使用条件控制流从它们中进行推理。

#Training - 
model1 = model.fit(all images, P(cat/dog))
model2 = model.fit(all images, P(cat))
model3 = model.fit(all images, P(dog))
final prediction = argmax(model2, model3)

#Inference - 
if model1.predict == Cat: 
    model2.predict
else:
    model3.predict

但我不认为你在寻找那个。 我认为您希望将条件作为计算图本身的一部分。

遗憾的是，据我所知，您没有直接的方法可以将 if-then 条件构建为计算图的一部分。您看到的keras.switch 允许您在训练期间使用张量输出，但不能使用图的层。这就是为什么您会看到它被用作损失函数的一部分，而不是在计算图中（引发输入错误）。

一种可能的解决方案 - 跳过连接和软切换

但是，您可以尝试使用 skip connections 和 soft switching 构建类似的东西。

跳过连接是从前一层到另一层的连接，允许您将信息传递到后续层。这在非常深的网络中很常见，其中来自原始数据的信息随后会丢失。例如，检查U-net 或Resnet，它们使用层之间的跳过连接将信息传递给未来的层。

下一个问题是切换问题。您想在图表中的 2 条可能路径之间切换。你可以做的是我从this paper 中获得灵感的软切换方法。请注意，为了switch 在两个单词分布之间（一个来自解码器，另一个来自输入），作者将它们与p 和(1-p) 相乘以获得累积分布。这是一个软开关，允许模型从解码器或输入本身中选择下一个预测词。 （当您希望您的聊天机器人说出用户输入的单词作为其响应的一部分时会有所帮助！）

了解这两个概念后，让我们尝试直观地构建我们的架构。

1. 首先我们需要一个单输入多输出图，因为我们正在训练 2 个模型

2. 我们的第一个模型是一个多类分类，它分别预测 Cat 和 Dog 的个体概率。这将通过激活softmax 和categorical_crossentropy 损失进行训练。

3. 接下来，让我们取预测 Cat 概率的 logit，并将卷积层 3 乘以它。这可以通过Lambda 层来完成。

4. 同样，让我们​​取Dog的概率并将其与卷积层2相乘。这可以看作如下-

   • 如果我的第一个模型完美地预测了猫而不是狗，那么计算将是 1*(Conv3) 和 0*(Conv2)。
   • 如果第一个模型完美地预测了狗而不是猫，那么计算将是 0*(Conv3) 和 1*(Conv2)
   • 您可以将其视为soft-switch 或来自LSTM 的forget gate。 forget gate 是一个 sigmoid（0 到 1）输出，它将单元状态相乘以对其进行门控，并允许 LSTM 忘记或记住以前的时间步长。类似的概念在这里！

5. 这些 Conv3 和 Conv2 现在可以进一步处理、展平、连接，并传递到另一个 Dense 层进行最终预测。

这样，如果模型不确定是狗还是猫，conv2 和 conv3 特征都会参与第二个模型的预测。这就是您可以使用skip connections 和soft switch 启发机制为您的网络添加一些条件控制流的方法。

检查我对下面计算图的实现。

from tensorflow.keras import layers, Model, utils
import numpy as np

X = np.random.random((10,500,500,3))
y = np.random.random((10,2))

#Model
inp = layers.Input((500,500,3))

x = layers.Conv2D(6, 3, name='conv1')(inp)
x = layers.MaxPooling2D(3)(x)

c2 = layers.Conv2D(9, 3, name='conv2')(x)
c2 = layers.MaxPooling2D(3)(c2)

c3 = layers.Conv2D(12, 3, name='conv3')(c2)
c3 = layers.MaxPooling2D(3)(c3)

x = layers.Conv2D(15, 3, name='conv4')(c3)
x = layers.MaxPooling2D(3)(x)

x = layers.Flatten()(x)
out1 = layers.Dense(2, activation='softmax', name='first')(x)

c = layers.Lambda(lambda x: x[:,:1])(out1)
d = layers.Lambda(lambda x: x[:,1:])(out1)

c = layers.Multiply()([c3, c])
d = layers.Multiply()([c2, d])

c = layers.Conv2D(15, 3, name='conv5')(c)
c = layers.MaxPooling2D(3)(c)
c = layers.Flatten()(c)

d = layers.Conv2D(12, 3, name='conv6')(d)
d = layers.MaxPooling2D(3)(d)
d = layers.Conv2D(15, 3, name='conv7')(d)
d = layers.MaxPooling2D(3)(d)
d = layers.Flatten()(d)

x = layers.concatenate([c,d])
x = layers.Dense(32)(x)
out2 = layers.Dense(2, activation='softmax',name='second')(x)

model = Model(inp, [out1, out2])
model.compile(optimizer='adam', loss='categorical_crossentropy', loss_weights=[0.5, 0.5])

model.fit(X, [y, y], epochs=5)

utils.plot_model(model, show_layer_names=False, show_shapes=True)

Epoch 1/5
1/1 [==============================] - 1s 1s/step - loss: 0.6819 - first_loss: 0.7424 - second_loss: 0.6214
Epoch 2/5
1/1 [==============================] - 0s 423ms/step - loss: 0.6381 - first_loss: 0.6361 - second_loss: 0.6400
Epoch 3/5
1/1 [==============================] - 0s 442ms/step - loss: 0.6137 - first_loss: 0.6126 - second_loss: 0.6147
Epoch 4/5
1/1 [==============================] - 0s 434ms/step - loss: 0.6214 - first_loss: 0.6159 - second_loss: 0.6268
Epoch 5/5
1/1 [==============================] - 0s 427ms/step - loss: 0.6248 - first_loss: 0.6184 - second_loss: 0.6311

Answer 2

为了构建condition-based CNN，我们可以将整批输入传递给 Model2 中的每个子模型，并根据条件从所有子模型输出中选择所需的输出（您在问题中定义的模型确实如此） )，或者我们可以按照条件的步骤（即您列出的三个条件）选择更快的方式

显示条件机制的示例代码：

# Mimic the test dataset and labels
batch = tf.constant([[1, 2, 3], [2, 3, 1], [3, 1, 2]])
y_all = [tf.one_hot(i, number_of_class, dtype=tf.float32) for i in range(number_of_class)]
# Mimic the outputs of model_01
y_p = tf.constant([[0.9, 0.1], [0.1, 0.9], [0.3, 0.7]])
y_p = tf.one_hot(tf.math.argmax(y_p, axis=1), number_of_class, dtype=tf.float32)
# Mimic the conditions by choose the samples from batch base on if prediction is equal to label wrt each class
for y in y_all:
    condition = tf.reduce_all(tf.math.equal(y_p, y), 1)
    indices = tf.where(condition)
    choosed_inputs = tf.gather_nd(batch, indices)
    print("label:\n{}\ncondition:\n{}\nindices:\n{}\nchoosed_inputs:\n{}\n".format(y, condition, indices, choosed_inputs))

输出：

label:
[1. 0.]
condition:
[ True False False]
indices:
[[0]]
choosed_inputs:
[[1 2 3]]

label:
[0. 1.]
condition:
[False  True  True]
indices:
[[1]
 [2]]
choosed_inputs:
[[2 3 1]
 [3 1 2]]

构建condition-based CNN 模型并以自定义训练方式对其进行训练的示例代码：

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras.layers import *
from tensorflow.keras.models import *
from tensorflow.keras.utils import *
import numpy as np

img_rows, img_cols, number_of_class, batch_size = 256, 256, 2, 64

#----------- main model (Model 1) ------------------------------------
inputs = Input(shape=(img_rows, img_cols, 3))
conv_01 = Convolution2D(64, 3, 3, activation='relu', name = 'conv_01') (inputs)
conv_02 = Convolution2D(64, 3, 3, activation='relu', name = 'conv_02') (conv_01)
skip_dog = conv_02

conv_03 = Convolution2D(64, 3, 3, activation='relu', name = 'conv_03') (conv_02)
skip_cat = conv_03

conv_04 = Convolution2D(64, 3, 3, activation='relu', name = 'conv_04') (conv_03)

flatten_main_model =  Flatten() (conv_04)
Output_main_model = Dense(units = number_of_class , activation = 'softmax', name = "Output_layer")(flatten_main_model)

#----------- Conditional  Cat model ------------------------------------ 
inputs_1 = Input(shape=skip_cat.shape[1:])
conv_05 = Convolution2D(64, 3, 3, activation='relu', name = 'conv_05') (inputs_1)
flatten_cat_model =  Flatten() (conv_05)
Output_cat_model = Dense(units = number_of_class , activation = 'softmax', name = "Output_layer_cat")(flatten_cat_model)

#----------- Conditional  Dog model ------------------------------------ 
inputs_2 = Input(shape=skip_dog.shape[1:])
conv_06 = Convolution2D(64, 3, 3, activation='relu', name = 'conv_06') (inputs_2)
flatten_dog_model =  Flatten() (conv_06)
Output_dog_model = Dense(units = number_of_class , activation = 'softmax', name = "Output_layer_dog")(flatten_dog_model)

#----------------------------- My discrete 3 models --------------------------------
model_01 = Model(inputs = inputs, outputs = [skip_cat, skip_dog, Output_main_model], name = 'model_main')
model_02_1 = Model(inputs = inputs_1, outputs = Output_cat_model, name = 'Conditional_cat_model')
model_02_2 = Model(inputs = inputs_2, outputs = Output_dog_model, name = 'Conditional_dog_model')

# Get one hot vectors for all the labels
y_all = [tf.one_hot(i, number_of_class, dtype=tf.float32) for i in range(number_of_class)]
sub_models_all = [model_02_1, model_02_2]
sub_models_trainable_variables = [model_01.trainable_variables[:6] + model_02_1.trainable_variables, 
                                  model_01.trainable_variables[:4] + model_02_2.trainable_variables]

cce = keras.losses.CategoricalCrossentropy()
optimizer_01 = keras.optimizers.Adam(learning_rate=1e-3, name='Adam_01')
optimizer_02 = keras.optimizers.Adam(learning_rate=2e-3, name='Adam_02')

@tf.function
def train_step(batch_imgs, labels):
    with tf.GradientTape(persistent=True) as tape:
        model_01_outputs = model_01(batch_imgs)
        y_p_01 = model_01_outputs[-1]
        loss_01 = cce(labels, y_p_01)

        # Convert outputs of model_01 from float in (0, 1) to one hot vectors, no gradients flow back from here
        y_p_01 = tf.one_hot(tf.math.argmax(y_p_01, axis=1), number_of_class, dtype=tf.float32)
        loss_02_all = []
        for i in range(number_of_class):
            condition = tf.reduce_all(tf.math.equal(y_p_01, y_all[i]), 1)
            indices = tf.where(condition)
            choosed_inputs = tf.gather_nd(model_01_outputs[i], indices)
            # Note here the inputs batch size for each sub-model is dynamic
            y_p_02 = sub_models_all[i](choosed_inputs)
            y_t = tf.gather_nd(labels, indices)
            loss_02 = cce(y_t, y_p_02)
            loss_02_all.append(loss_02)

    grads_01 = tape.gradient(loss_01, model_01.trainable_variables)
    optimizer_01.apply_gradients(zip(grads_01, model_01.trainable_variables))

    for i in range(number_of_class):
        grads_02 = tape.gradient(loss_02_all[i], sub_models_trainable_variables[i])
        optimizer_02.apply_gradients(zip(grads_02, sub_models_trainable_variables[i]))

    return loss_01, loss_02_all

def training():
    for j in range(10):
        random_imgs = np.random.rand(batch_size, img_rows, img_cols, 3)
        random_labels = np.eye(number_of_class)[np.random.choice(number_of_class, batch_size)]
        loss_01, loss_02_all = train_step(random_imgs, random_labels)
        print("Step: {}, Loss_01: {}, Loss_02_all: {}".format(j, loss_01, loss_02_all))

输出类似于：

Step: 0, Loss_01: 0.6966696977615356, Loss_02_1: 0.0, Loss_02_2: 0.6886894702911377
Step: 1, Loss_01: 0.6912064552307129, Loss_02_1: 0.6968430280685425, Loss_02_2: 0.6911896467208862
Step: 2, Loss_01: 0.6910352110862732, Loss_02_1: 0.698455274105072, Loss_02_2: 0.6935626864433289
Step: 3, Loss_01: 0.6955667734146118, Loss_02_1: 0.6843984127044678, Loss_02_2: 0.6953505277633667
Step: 4, Loss_01: 0.6941269636154175, Loss_02_1: 0.673763632774353, Loss_02_2: 0.6994296908378601
Step: 5, Loss_01: 0.6872361898422241, Loss_02_1: 0.6769005060195923, Loss_02_2: 0.6907837390899658
Step: 6, Loss_01: 0.6931678056716919, Loss_02_1: 0.7674703598022461, Loss_02_2: 0.6935689449310303
Step: 7, Loss_01: 0.6976977586746216, Loss_02_1: 0.7503389120101929, Loss_02_2: 0.7076178789138794
Step: 8, Loss_01: 0.6932153105735779, Loss_02_1: 0.7428234219551086, Loss_02_2: 0.6935019493103027
Step: 9, Loss_01: 0.693305253982544, Loss_02_1: 0.6476342082023621, Loss_02_2: 0.6916818618774414