如何将自定义变换层接入TensorFlow模型并实现逐行独立处理？

需要将基于NumPy实现的自定义变换层接入TensorFlow Sequential模型，当前代码报错TypeError: CustomLayer.__init__() missing 1 required positional argument: 'x'。核心需求：

• 自定义层独立处理输入的每一行数据（输入形状(13,1)）
• 先对每行添加高斯噪声，再执行指定的信号变换
• 层固定参数为step=2、mean=0、std_dev=0.03

原代码及报错信息如下：

原NumPy实现的CustomLayer类 #

class CustomLayer:
    def __init__(self, x, step, mean, std_dev):
        self.x = x
        self.step = step
        self.mean= mean
        self.std_dev = std_dev
        self.size = len(self.x)
    
    def transform_signal(self, y):
        indices_transform = np.arange(0,len(y), self.step)
        a1 = y[indices_transform]
        a2 = a1[1:]
        a1 = a1[:-1]

        return np.repeat( (a1 + (a2-a1)/2), self.step )

    def gaussian_noise(self):
        return np.random.normal(loc=self.mean, scale=self.std_dev, size=self.size)
    
    def transform(self):
        x_gauss_noise = self.x + self.gaussian_noise()
        return self.transform_signal(y=x_gauss_noise)

示例数据集 #

arr = np.expand_dims( np.array([ [1, 5, 7, 8, 10, 11, 11.5, 12, 12.5, 13, 13.2, 13.8, 14.3],
                                 [11, 15, 17, 18, 110, 111, 111.5, 112, 112.5, 113, 113.2, 113.8, 114.3],
                                 [2, 6, 8, 9, 11, 12, 12.5, 13, 13.5, 14, 15.2, 14.8, 15.3]]), axis=2 )

报错的TensorFlow模型 #

import tensorflow as tf

from tensorflow.keras.layers import (Conv1D,
                                     Dense)

model = tf.keras.Sequential([
    CustomLayer(mean=0, std_dev=0.03, step=2),
    Conv1D(filters=2,
           kernel_size=5,
           padding="same",
           activation="relu"),
    Dense(units=1, activation='relu')
])

错误信息 #

TypeError: CustomLayer.__init__() missing 1 required positional argument: 'x'

要在TensorFlow中使用自定义层，必须继承tf.keras.layers.Layer，遵循Keras层的生命周期规范：__init__只定义配置参数，输入数据在call方法中处理，同时用TensorFlow原生操作替代NumPy操作以兼容计算图。

修改后的自定义层代码 #

import tensorflow as tf

class CustomLayer(tf.keras.layers.Layer):
    def __init__(self, step, mean=0.0, std_dev=0.03, **kwargs):
        super(CustomLayer, self).__init__(**kwargs)
        self.step = step
        self.mean = mean
        self.std_dev = std_dev

    def call(self, inputs, training=None):
        # inputs形状：(batch_size, seq_len, channels)，此处seq_len=13，channels=1
        # 1. 训练阶段添加高斯噪声，推理阶段跳过
        if training:
            noise = tf.random.normal(shape=tf.shape(inputs), mean=self.mean, stddev=self.std_dev)
            x_gauss_noise = inputs + noise
        else:
            x_gauss_noise = inputs

        # 2. 实现信号变换逻辑（替换NumPy操作为TensorFlow API）
        seq_len = tf.shape(x_gauss_noise)[1]
        indices_transform = tf.range(0, seq_len, self.step, dtype=tf.int32)
        a1 = tf.gather(x_gauss_noise, indices_transform, axis=1)
        
        a1_part = a1[:, :-1, :]
        a2_part = a1[:, 1:, :]
        mid_values = a1_part + (a2_part - a1_part) / 2
        repeated_values = tf.repeat(mid_values, repeats=self.step, axis=1)

        # 处理序列长度不整除step的情况（原示例13→12）
        output_len = tf.shape(repeated_values)[1]
        if output_len < seq_len - 1:
            last_value = tf.tile(a2_part[:, -1:, :], multiples=[1, seq_len - 1 - output_len, 1])
            repeated_values = tf.concat([repeated_values, last_value], axis=1)
        
        return repeated_values

    def compute_output_shape(self, input_shape):
        # 明确输出形状，帮助模型推断后续层输入
        return (input_shape[0], input_shape[1] - 1, input_shape[2])

修改后的TensorFlow模型 #

model = tf.keras.Sequential([
    CustomLayer(step=2, mean=0, std_dev=0.03, input_shape=(13,1)),
    Conv1D(filters=2,
           kernel_size=5,
           padding="same",
           activation="relu"),
    Dense(units=1, activation='relu')
])

# 验证模型结构
model.summary()

关键修改说明 #

1. 继承tf.keras.layers.Layer：这是TensorFlow自定义层的必要前提，确保层能被Sequential模型识别
2. __init__移除x参数：仅保留层的配置参数（step、mean、std_dev），输入数据在call方法中接收（符合Keras层的设计规范）
3. 替换NumPy操作为TensorFlow API：使用tf.random.normal、tf.range、tf.gather等替代NumPy函数，保证可微分、兼容GPU加速和计算图
4. 支持训练/推理模式：通过training参数控制噪声添加逻辑，符合Keras层的标准行为
5. 明确输出形状：重写compute_output_shape让模型正确推断后续层的输入形状

测试验证 #

import numpy as np

# 转换示例数据为TensorFlow张量
arr = np.expand_dims( np.array([ [1, 5, 7, 8, 10, 11, 11.5, 12, 12.5, 13, 13.2, 13.8, 14.3],
                                 [11, 15, 17, 18, 110, 111, 111.5, 112, 112.5, 113, 113.2, 113.8, 114.3],
                                 [2, 6, 8, 9, 11, 12, 12.5, 13, 13.5, 14, 15.2, 14.8, 15.3]]), axis=2 )
tf_arr = tf.convert_to_tensor(arr, dtype=tf.float32)

# 前向传播测试
output = model(tf_arr, training=True)
print("输入形状:", tf_arr.shape)
print("输出形状:", output.shape)

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