Keras为ResNet添加FFT Lambda层触发输入张量元数据缺失错误

• 核心触发点：定义Keras输入张量后，你用原生TensorFlow操作覆盖了输入变量引用：x_input = tf.shape(tf.squeeze(x_input))执行后，x_input不再是带Keras层元数据的Input层输出，而是丢失了上游层连接信息的原生TF张量。后续将该张量传入keras.Model作为输入时，Keras无法追溯其来源为keras.layers.Input，直接抛出对应错误。
• 计算图追踪破坏：代码中裸写tf.cast、tf.expand_dims等原生TF操作，没有用Lambda层包裹，会导致Keras无法正常记录层之间的连接关系。
• 逻辑错误：残差块resnode入口额外加入了FFT计算层，和「前端加FFT预处理」的需求冲突，会导致每个残差块都重复执行FFT，破坏输入特征。
• API兼容问题：tf.spectral.fft是TF1.x废弃API，TF2.x中需使用tf.signal.fft。
• 语法错误：resnode函数缩进错误，通道对齐判断、Add层、return语句都写在了函数外部，运行会触发变量未定义问题。

所有张量变换操作均通过Keras层（Lambda层封装原生TF逻辑）实现，保留原始Input张量引用不覆盖，保证Keras能完整追踪从输入到输出的计算链路。该方案中FFT为模型内置层，TF自动实现了FFT、取模、类型转换等操作的梯度，训练时损失可正常反向传播到模型输入端，完全满足需求。

修正后的完整核心代码如下：

import tensorflow as tf
from tensorflow import keras
import keras_contrib

def resnode(x_in,filter_count,kernel_size=3,downsample=False):
    x = x_in
    x = keras_contrib.layers.InstanceNormalization()(x)
    # 注意：如果不需要在残差块内做FFT就删掉下面两行，只保留前端FFT即可
    # x = keras.layers.Lambda(lambda v: tf.signal.fft(tf.cast(v,tf.complex64)))(x) 
    # x = keras.layers.Lambda(lambda v: tf.abs(tf.cast(v,tf.complex64)))(x)    
    x = keras.layers.ReLU()(x)
    x = keras.layers.Conv1D(filter_count,kernel_size,padding='same')(x)
    x = keras.layers.Dropout(0.1)(x)
    x = keras_contrib.layers.InstanceNormalization()(x)
    x = keras.layers.ReLU()(x)
    x = keras.layers.Conv1D(filter_count,kernel_size,padding='same')(x)

    if downsample:
        x_in = keras.layers.AveragePooling1D()(x_in)
        x = keras.layers.AveragePooling1D()(x)
    # 修正缩进：以下逻辑全部放入resnode函数体内
    if x_in.shape[-1] != x.shape[-1]:
        print('convolving x_in',x_in.shape)
        x_in = keras.layers.Conv1D(filter_count,1,padding='same')(x_in)
        print('Newshape x_in',x_in.shape)
    
    x = keras.layers.Add()([x_in,x])
    return x

def createresnet(outputs=4):  
    n_features = N    
    # 定义原始输入张量，禁止后续重新赋值覆盖该变量
    input_tensor = keras.layers.Input(shape=(None,1))

    # 所有TF操作全部用Lambda层包裹，保持Keras计算图连接
    x = keras.layers.Lambda(lambda v: tf.squeeze(v, axis=-1))(input_tensor) # 移除最后一维通道，适配FFT输入维度要求
    x = keras.layers.Lambda(lambda v: tf.signal.fft(tf.cast(v,tf.complex64)))(x) # 执行FFT，替换废弃API
    x = keras.layers.Lambda(lambda v: tf.abs(v))(x) # 取FFT模长
    x = keras.layers.Lambda(lambda v: tf.cast(v, tf.float32))(x) # 转回float32适配卷积层计算
    x = keras.layers.Lambda(lambda v: tf.expand_dims(v, axis=-1))(x) # 恢复通道维度，符合1D卷积输入格式(batch, steps, channels)

    filter_size = 32
    x = resnode(x,filter_size,kernel_size=7)

    filter_size = 32
    x = resnode(x,filter_size,kernel_size=5,downsample=True)
    filter_size = 64
    x = resnode(x,filter_size,kernel_size=5,downsample=True)
    x = resnode(x,filter_size,kernel_size=5,downsample=True)
    filter_size = 128
    x = resnode(x,filter_size,downsample=True)
    x = resnode(x,filter_size,downsample=True)
    filter_size = 128
    x = resnode(x,filter_size,downsample=True)
    x = resnode(x,filter_size,downsample=True)
    filter_size = 128
    x = resnode(x,filter_size,downsample=True)
    x = resnode(x,filter_size,downsample=True)
    filter_size = 128
    x = resnode(x,filter_size,downsample=True)
    x = resnode(x,filter_size,downsample=True)
    filter_size = 64
    x = resnode(x,filter_size,downsample=True)

    # 原代码中if False的冗余块可保留，不影响运行
    if False:
        filter_size = 64
        x = resnode(x,filter_size)
        x = resnode(x,filter_size,downsample=True)
        filter_size = 64
        x = resnode(x,filter_size)
        x = resnode(x,filter_size,downsample=True)
        filter_size = 128
        x = resnode(x,filter_size)
        x = resnode(x,filter_size,downsample=True)
        filter_size = 128
        x = resnode(x,filter_size)
        x = resnode(x,filter_size,downsample=True)
        filter_size = 128
        x = resnode(x,filter_size)
        x = resnode(x,filter_size,downsample=True)
        filter_size = 128
        x = resnode(x,filter_size)
        x = resnode(x,filter_size,downsample=True)

    filter_size = outputs
    x = resnode(x,filter_size,downsample=True)
    x = keras.layers.GlobalAveragePooling1D()(x)
    output_tensor = keras.layers.Activation('softmax')(x)

    # 模型入参传入最原始的input_tensor，不要传中间变换后的张量
    model = keras.Model(inputs=input_tensor,outputs=output_tensor)
    return model

如果需要拆分预处理和主干网络逻辑，也可以把FFT部分单独封装为独立模型，再和ResNet主干串联，梯度可正常在两个子模型间反向传播，写法如下：

# 单独封装FFT预处理模块
fft_preprocess = keras.Sequential([
    keras.layers.Lambda(lambda v: tf.squeeze(v, axis=-1)),
    keras.layers.Lambda(lambda v: tf.signal.fft(tf.cast(v, tf.complex64))),
    keras.layers.Lambda(lambda v: tf.abs(v)),
    keras.layers.Lambda(lambda v: tf.cast(v, tf.float32)),
    keras.layers.Lambda(lambda v: tf.expand_dims(v, axis=-1))
])

def createresnet(outputs=4):
    input_tensor = keras.layers.Input(shape=(None,1))
    x = fft_preprocess(input_tensor)
    # 后续接原有残差块、输出层逻辑即可，和上面修复方案一致
    # ...
    model = keras.Model(inputs=input_tensor, outputs=output_tensor)
    return model

内容的提问来源于stack exchange，提问作者G S