Keras中如何避免Sequential模型调用时输入形状被自动挤压的问题

问题原因分析 #

你遇到的错误本质是Sequential模型的输入处理逻辑和你的预期不匹配，再加上代码中存在一个小的参数遗漏bug：

1. Sequential形状匹配冲突：你的self.R Sequential模型第一层定义为Input(shape=(1,))，意味着它期望接收批量的一维特征（形状(batch_size, 1)）。但在call方法中，你传入的是扩展后的三维张量(None, n_channels, 1)。Keras的Sequential模型默认会将批量维度之后的所有维度合并为特征维度，因此它会自动把(None, 128, 1)挤压为(None, 128)，这就和Input(shape=(1,))要求的特征维度为1矛盾，从而触发报错。
2. 参数遗漏bug：你的__init__方法参数列表里没有d，但内部却直接使用了self.d = d，这会导致NameError，必须先修正这个问题。

解决方案 #

根据你的需求（对每个通道的1维输入独立应用提升网络R），这里提供两种简洁的修正方案：

方案1：使用TimeDistributed包装Sequential模型 #

TimeDistributed层的核心作用是将一个模型/层应用到输入的子维度（这里的n_channels就是我们要遍历的子维度），确保每个(1,)的子输入都被单独处理：

class FONet(keras.Model):
    """
    DeepONet模型
    输入假设为传感器点列表上的编码函数
    """
    def __init__(self, lifting_dims, d, act):  # 新增d作为参数传入
        super().__init__()
        self.d = d
        self.L = len(d)-1

        # 定义内部的提升网络（处理单个1维输入）
        inner_lifter = tf.keras.Sequential()
        inner_lifter.add(tf.keras.layers.Input(shape=(1,)))
        for dim in lifting_dims:
            print(f"{dim=}")
            inner_lifter.add(tf.keras.layers.Dense(dim, activation=act))
        inner_lifter.add(tf.keras.layers.Dense(d[0]))
        
        # 用TimeDistributed包装，让它遍历每个通道的输入
        self.R = tf.keras.layers.TimeDistributed(inner_lifter)
    
    def call(self, inputs):
        inputs = tf.expand_dims(inputs, axis=-1)  # 形状变为(None, n_channels, 1)
        print(f"{inputs.shape=}")
        lifted = self.R(inputs)  # 输出形状为(None, n_channels, d[0])
        return lifted

这样修改后，当你传入(None, n_channels, 1)的张量时，TimeDistributed会自动遍历n_channels维度，将每个(1,)的子张量传入内部的inner_lifter，完全符合你的预期，不会再出现形状不匹配的错误。

方案2：使用tf.vectorized_map显式遍历通道元素 #

如果你不想使用TimeDistributed，可以在call方法中用tf.vectorized_map高效地将self.R应用到每个通道的输入上：

class FONet(keras.Model):
    """
    DeepONet模型
    输入假设为传感器点列表上的编码函数
    """
    def __init__(self, lifting_dims, d, act):  # 新增d作为参数传入
        super().__init__()
        self.d = d
        self.L = len(d)-1

        self.R = tf.keras.Sequential()
        self.R.add(tf.keras.layers.Input(shape=(1,)))
        for dim in lifting_dims:
            print(f"{dim=}")
            self.R.add(tf.keras.layers.Dense(dim, activation=act))
        self.R.add(tf.keras.layers.Dense(d[0]))
    
    def call(self, inputs):
        inputs = tf.expand_dims(inputs, axis=-1)  # 形状变为(None, n_channels, 1)
        print(f"{inputs.shape=}")
        
        # 交换维度，让通道维度成为批量维度，处理后再换回原顺序
        inputs_swapped = tf.transpose(inputs, perm=[1, 0, 2])  # 形状变为(n_channels, None, 1)
        lifted_swapped = tf.vectorized_map(lambda x: self.R(x), inputs_swapped)  # 输出(n_channels, None, d[0])
        lifted = tf.transpose(lifted_swapped, perm=[1, 0, 2])  # 换回(None, n_channels, d[0])
        
        return lifted

tf.vectorized_map会在计算图层面高效地执行遍历，既保证了计算效率，又不会破坏Keras的自动求导机制。

额外注意事项 #

1. 输出与标签形状匹配：你的y.shape是(1028, 128)，而模型修正后的输出形状是(1028, 128, d[0])。如果你的标签y是每个通道的标量值，需要确保d[0] = 1，或者将y的形状调整为(1028, 128, 1)，否则MSE损失函数会因为形状不匹配而报错。
2. 验证模型结构：初始化模型后，可以通过fonet_model.summary()和fonet_model.R.summary()确认各层的输入输出形状是否符合预期。

测试修正后的代码 #

最后，初始化模型时记得传入d参数（根据你的需求定义d的值）：

lift_dims = [200, 200]
d = [64, 32]  # 示例d的值，可根据实际需求调整
fonet_model = FONet(lift_dims, d, tf.keras.activations.sigmoid)

# 编译与训练
lr = 1e-3
ep = 250
loss = tf.keras.losses.MSE
optimizer = tf.keras.optimizers.Adam(learning_rate=lr)
fonet_model.compile(loss=loss, optimizer=optimizer)

# 注意：这里将y扩展一维以匹配模型输出形状
history = fonet_model.fit(X, tf.expand_dims(y, axis=-1), verbose=1, epochs=ep, batch_size=32, shuffle=True)

备注：内容来源于stack exchange，提问作者The_Docc