如何合并两个Keras模型并在中间重塑以实现联合训练？

我有两个Keras模型：

• 编码器模型：将256×192的图像编码为128维向量，输入形状(None, 192, 256, 1)，输出形状(None, 128)（None代表批量大小）
• 比较器模型：对比来自编码器的两个向量，输入形状(None, 256)，输出形状(None, 1)

需要合并成一个模型，输入形状(None, 2, 192, 256, 1)（2代表待对比的两张图像），输出形状(None, 1)，同时保留原两个模型以便后续单独使用，且能用常规fit方法训练。

尝试代码 #

class _layer_1(keras.layers.Layer):
    def call(self, x):
        return tf.reshape(x, (x.shape[0] * x.shape[1],) + x.shape[2:])

class _layer_2(keras.layers.Layer):
    def call(self, x):
        return tf.reshape(x, (x.shape[0] // 2, x.shape[1] * 2) + x.shape[2:])

model = keras.Sequential([
    keras.Input(shape=(2, 192, 256, 1)),
    _layer_1(),
    encoder_model,
    _layer_2(),
    comparator_model
])

model.compile(optimizer=keras.optimizers.Adam(learning_rate=3e-5), loss='mean_squared_error')

报错信息 #

Traceback (most recent call last):
  File "comparator_train.py", line 61, in <module>
    get_models()[2].fit(np.random.sample((30, 2, 192, 256, 1),), np.random.sample((30, 1),))
  File "comparator_train.py", line 54, in get_models
    comparator_model
  File "/opt/miniconda3/lib/python3.7/site-packages/tensorflow/python/training/tracking/base.py", line 587, in _method_wrapper
    result = method(self, *args, **kwargs)
  File "/opt/miniconda3/lib/python3.7/site-packages/keras/utils/traceback_utils.py", line 67, in error_handler
    raise e.with_traceback(filtered_tb) from None
  File "/tmp/__autograph_generated_filecoqup163.py", line 12, in tf__call
    retval_ = ag__.converted_call(ag__.ld(tf).reshape, (ag__.ld(x), (((ag__.ld(x).shape[0] * ag__.ld(x).shape[1]),) + ag__.ld(x).shape[2:])), None, fscope)
TypeError: Exception encountered when calling layer "private_layer_1" (type _layer_1).

in user code:

    File "comparator_train.py", line 43, in call  *
        return tf.reshape(x, (x.shape[0] * x.shape[1],) + x.shape[2:])

    TypeError: unsupported operand type(s) for *: 'NoneType' and 'int'


Call arguments received by layer "private_layer_1" (type _layer_1):
  • x=tf.Tensor(shape=(None, 2, 192, 256, 1), dtype=float32)

附编码器与比较器完整代码 #

encoder_model = keras.Sequential([
    keras.layers.Input(shape=(192, 256, 1)),
    keras.layers.SeparableConv2D(filters=128, kernel_size=12, strides=2, padding='same', activation=keras.layers.LeakyReLU(alpha=0.1)),
    keras.layers.BatchNormalization(),
    keras.layers.SeparableConv2D(filters=256, kernel_size=12, strides=3, padding='same', activation=keras.layers.LeakyReLU(alpha=0.1)),
    keras.layers.BatchNormalization(),
    keras.layers.SeparableConv2D(filters=256, kernel_size=(12, 12), strides=2, padding='same', activation=keras.layers.LeakyReLU(alpha=0.1)),
    keras.layers.BatchNormalization(),
    keras.layers.Dropout(0.8),
    keras.layers.SeparableConv2D(filters=128, kernel_size=(8, 11), strides=1, padding='valid', activation=keras.layers.LeakyReLU(alpha=0.1)),
    keras.layers.Dropout(0.14),
    keras.layers.MaxPooling2D(pool_size=(3, 3)),
    keras.layers.Flatten(),
    keras.layers.Dense(units=128, activation='sigmoid')
])

comparator_model = keras.Sequential([
    keras.layers.Input(shape=(256,)),
    keras.layers.Dense(units=256, activation=keras.layers.LeakyReLU(alpha=0.1)),
    keras.layers.Dense(units=256, activation=keras.layers.LeakyReLU(alpha=0.1)),
    keras.layers.Dense(units=1, activation='sigmoid')
])

错误核心原因 #

报错是因为静态形状（x.shape）中的批量维度为None，属于NoneType，无法直接和整数进行乘法/除法运算。自定义层中使用x.shape[0]获取的是静态形状，而动态计算时需要用tf.shape(x)获取张量的动态形状值。

推荐实现方式：使用Keras Functional API #

Functional API比Sequential更灵活，能直观处理多输入/分支逻辑，且无需手动做复杂的形状重塑：

import tensorflow as tf
from tensorflow import keras

# 定义合并模型的输入
input_pair = keras.Input(shape=(2, 192, 256, 1))

# 将输入拆分为两张独立图像
img1 = input_pair[:, 0, :, :, :]  # 形状：(None, 192, 256, 1)
img2 = input_pair[:, 1, :, :, :]  # 形状：(None, 192, 256, 1)

# 用同一个编码器处理两张图像（共享权重）
encoded1 = encoder_model(img1)
encoded2 = encoder_model(img2)

# 拼接两个编码向量，得到比较器需要的(None, 256)输入
combined_feat = keras.layers.concatenate([encoded1, encoded2], axis=1)

# 比较器输出结果
output = comparator_model(combined_feat)

# 构建合并模型
combined_model = keras.Model(inputs=input_pair, outputs=output)

# 编译模型
combined_model.compile(optimizer=keras.optimizers.Adam(learning_rate=3e-5), loss='mean_squared_error')

验证测试 #

用随机数据测试模型是否正常运行：

import numpy as np

# 生成随机输入和标签
x = np.random.random((30, 2, 192, 256, 1))
y = np.random.random((30, 1))

# 训练测试
combined_model.fit(x, y, epochs=1, batch_size=2)

# 验证原模型是否可单独使用
test_img = np.random.random((1, 192, 256, 1))
encoder_output = encoder_model.predict(test_img)
print("编码器输出形状:", encoder_output.shape)  # 应为(1, 128)

test_combined = np.random.random((1, 256))
comparator_output = comparator_model.predict(test_combined)
print("比较器输出形状:", comparator_output.shape)  # 应为(1, 1)

若坚持修复原自定义层方案 #

如果一定要用Sequential+自定义层的方式，需修改自定义层使用动态形状：

class _layer_1(keras.layers.Layer):
    def call(self, x):
        # 获取动态形状
        batch_size = tf.shape(x)[0]
        num_imgs = tf.shape(x)[1]
        new_shape = tf.concat([[batch_size * num_imgs], tf.shape(x)[2:]], axis=0)
        return tf.reshape(x, new_shape)

class _layer_2(keras.layers.Layer):
    def call(self, x):
        batch_size = tf.shape(x)[0] // 2
        feat_dim = tf.shape(x)[1]
        new_shape = tf.concat([[batch_size], [feat_dim * 2]], axis=0)
        return tf.reshape(x, new_shape)

# 构建模型
model = keras.Sequential([
    keras.Input(shape=(2, 192, 256, 1)),
    _layer_1(),
    encoder_model,
    _layer_2(),
    comparator_model
])

model.compile(optimizer=keras.optimizers.Adam(learning_rate=3e-5), loss='mean_squared_error')

不过这种方式不如Functional API直观，形状变换时容易出错，优先推荐第一种方案。

内容的提问来源于stack exchange，提问作者Усердный бобёр