Keras自定义聚类损失层报错：无法解释损失函数标识符

问题背景 #

尝试复现IEEE论文中的RffAe-S自编码器结构，模型采用可分离损失函数，需基于编码器输出均值自定义聚类损失项。创建了兼具全连接功能的RffConnected自定义层，通过add_loss计算聚类损失，但训练两轮后出现报错：

ValueError: Could not interpret loss function identifier: Tensor("rff_connected_137/Const:0", shape=(100,), dtype=float32)

相关代码 #

import matplotlib.pyplot as plot
from mpl_toolkits.axes_grid1 import ImageGrid
import numpy as np
import math
from matplotlib.figure import Figure
import tensorflow as tf
import keras
from keras import layers
import random
import time
from os import listdir

#loads data from a text file
def loadData(basePath, samplesPerFile, sampleRate):
    real = []
    imag = []
    fileOrder = []
    
    for file in listdir(basePath):
        if((file != "READ_ME") and ((file != "READ_ME.txt"))):
            fid = open(basePath + "\\" + file, "r")
            fileOrder.append(file)

            t = 0
            sampleEvery = samplesPerFile / sampleRate
            temp1 = []
            temp2 = []
            times = []
            
            for line in fid.readlines():
                times.append(t)
                samples = line.split("\t")
                temp1.append(float(samples[0]))
                temp2.append(float(samples[1]))
                t = t + sampleEvery
                
            real.append(temp1)
            imag.append(temp2)
            fid.close()

    real = np.array(real)
    imag = np.array(imag)

    return real, imag, times, fileOrder

#####################################################################################################

#Breaks up and randomizes data
def breakUpData(real, imag, times, numPartitions, basePath):
    if(len(real) % numPartitions != 0):
        raise ValueError("Error: The length of the dataset must be divisible by the number of partitions.")

    newReal = []
    newImag = []
    newTimes = []
    fileOrder = listdir(basePath)
    dataFiles = []
    interval = int(len(real[0]) / numPartitions)

    for i in range(0, interval):
        newTimes.append(times[i])

    for i in range(0, len(real)):
        tempI = []
        tempQ = []
        
        for j in range(0, len(real[0])):
            tempI.append(real[i, j])
            tempQ.append(imag[i, j])
            
            if((j + 1) % interval == 0):
                newReal.append(tempI)
                newImag.append(tempQ)
                dataFiles.append(fileOrder[i])
                tempI = []
                tempQ = []

    #randomizes the broken up dataset and the file list
    for i in range(0, len(newReal)):
        r = random.randint(0, len(newReal) - 1)
        tempReal = newReal[i]
        tempImag = newImag[i]
        newReal[i] = newReal[r]
        newImag[i] = newImag[r]
        newReal[r] = tempReal
        newImag[r] = tempImag

        tempFile = dataFiles[i]
        dataFiles[i] = dataFiles[r]
        dataFiles[r] = tempFile

    return newReal, newImag, newTimes, dataFiles

#####################################################################################################
#custom loss layer for the RffAe-S that calculates the clustering loss term
class RffConnected(layers.Layer):
    def __init__(self, output_dim, batchSize, beta, alpha):
        super(RffConnected, self).__init__()
        self.iters = 0.0
        self.beta = beta
        self.alpha = alpha
        self.batchSize = batchSize
        self.output_dim = output_dim
        self.sum = tf.cast(tf.zeros(output_dim, tf.float64), tf.float32)
        self.moving_average = tf.cast(tf.zeros(output_dim, tf.float64), tf.float32)
        self.clusterloss = tf.cast(tf.zeros(output_dim, tf.float64), tf.float32)
        
    def build(self, input_shape):
        self.kernel = self.add_weight(name = 'kernel', \
                shape = (int(input_shape[-1]), self.output_dim), \
                initializer = 'normal', trainable = True)
        super(RffConnected, self).build(int(input_shape[-1]))
        
    def call(self, inputs):        
        #keeps track of training epochs
        self.iters = self.iters + 1
        
        #where this custom layer acts as a normal layer- the loss then uses this
        calc = tf.matmul(inputs, self.kernel)
        
        #cumulative sum of deep encoded features
        self.sum = tf.math.add(self.sum, calc)
        
        #calculate the moving average and loss if we have already trained one batch
        if(self.iters >= self.batchSize):
            self.moving_average = tf.math.divide(self.sum, self.iters)
            self.clusterloss = tf.math.exp(\
                tf.math.multiply(-1 * self.beta, tf.math.reduce_sum(tf.math.square(tf.math.subtract(inputs, self.moving_average)))))
        
        #self.add_loss(tf.math.multiply(self.clusterloss, self.alpha))
        self.add_loss(self.clusterloss.numpy() * self.alpha)
        
        return calc
    

#####################################################################################################

def customloss(y_true, y_pred):
    loss = tf.square(y_true - y_pred)
    print(loss)
    return loss

#####################################################################################################

realTraining = np.array(real[0:2200])
realTesting = np.array(real[2200:-1])
imagTraining = np.array(imag[0:2200])
imagTesting = np.array(imag[2200:-1])

numInputs = len(realTraining[0])

i_sig = keras.Input(shape=(numInputs,))
q_sig = keras.Input(shape=(numInputs,))

iRff = tf.keras.layers.experimental.RandomFourierFeatures(numInputs, \
                                                 kernel_initializer='gaussian', scale=9.0)(i_sig)
rff1 = keras.Model(inputs=i_sig, outputs=iRff)
qRff = tf.keras.layers.experimental.RandomFourierFeatures(numInputs, \
                                                 kernel_initializer='gaussian', scale=9.0)(q_sig)
rff2 = keras.Model(inputs=q_sig, outputs=qRff)
combined = layers.Concatenate()([iRff, qRff])
combineRff = tf.keras.layers.experimental.RandomFourierFeatures(4 * numInputs, \
                                                                kernel_initializer='gaussian', scale=10.0)(combined)
    
preprocess = keras.Model(inputs=[iRff, qRff], outputs=combineRff)

preprocessedTraining = preprocess.predict([realTraining, imagTraining])
preprocessedTesting = preprocess.predict([realTesting, imagTesting])

##################  Entering Encoder  ######################
encoderIn = keras.Input(shape=(4*numInputs,))

clusterLossLayer = RffConnected(100, 30, 1.00, 100.00)(encoderIn)

encoder = keras.Model(inputs=encoderIn, outputs=clusterLossLayer)
##################  Entering Decoder  ######################

connected2 = layers.Dense(125, activation="sigmoid")(clusterLossLayer)
relu1 = layers.ReLU()(connected2)
dropout = layers.Dropout(0.2)(relu1)
reshape1 = layers.Reshape((25, 5, 1))(dropout)
bn1 = layers.BatchNormalization()(reshape1)
trans1 = layers.Conv2DTranspose(1, (4, 2))(bn1)
ups1 = layers.UpSampling2D(size=(2, 1))(trans1)
relu2 = layers.ReLU()(ups1)
bn2 = layers.BatchNormalization()(relu2)
trans2 = layers.Conv2DTranspose(1, (4, 2))(bn2)
ups2 = layers.UpSampling2D(size=(2, 1))(trans2)
relu3 = layers.ReLU()(ups2)
bn3 = layers.BatchNormalization()(relu3)
trans3 = layers.Conv2DTranspose(1, (5, 2))(bn3)
ups3 = layers.UpSampling2D(size=(2, 1))(trans3)
relu4 = layers.ReLU()(ups3)
bn4 = layers.BatchNormalization()(relu4)
trans4 = layers.Conv2DTranspose(1, (7, 1))(bn4)
reshape2 = layers.Reshape((4*numInputs, 1, 1))(trans4)

autoencoder = keras.Model(inputs=encoderIn, outputs=reshape2)
encoded_input = keras.Input(shape=(None, 100))
decoder_layer = autoencoder.layers[-1]

autoencoder.compile(optimizer='adam', loss=[autoencoder.losses[-1], customloss], metrics=['accuracy', 'accuracy'])

autoencoder.fit(preprocessedTraining, preprocessedTraining, epochs=100, batch_size=20, shuffle=True, validation_data=(preprocessedTesting, preprocessedTesting))

错误原因及修复方案 #

核心错误：损失函数传递与计算图破坏 #

1. 模型编译时的损失参数错误
   autoencoder.compile中直接传入autoencoder.losses[-1]，这是一个Tensor对象，而Keras期望损失函数为可调用对象或字符串标识符，导致无法解析。

   修复：自定义层通过add_loss添加的损失会被模型自动收集，无需手动在compile中指定，只需定义重建损失：

   autoencoder.compile(optimizer='adam', loss=customloss, metrics=['accuracy'])
   

2. 自定义层中破坏计算图
   使用self.clusterloss.numpy()将张量转为numpy数组，切断了梯度追踪，导致训练时无法构建有效计算图。

   修复：去掉.numpy()，直接使用张量运算，同时修正聚类损失的计算逻辑（需输出标量损失）：

   # 自定义层__init__中初始化聚类损失为标量
   self.clusterloss = tf.cast(0.0, tf.float32)
   
   # call方法中修正损失计算与添加
   if self.iters >= self.batchSize:
       self.moving_average = tf.math.divide(self.sum, self.iters)
       # 基于编码器输出calc计算损失，对batch做平均得到标量
       batch_loss = tf.math.exp(
           tf.math.multiply(-1 * self.beta, tf.math.reduce_sum(tf.math.square(tf.math.subtract(calc, self.moving_average)), axis=1))
       )
       self.clusterloss = tf.reduce_mean(batch_loss)
       self.add_loss(tf.math.multiply(self.clusterloss, self.alpha))
   

其他潜在问题优化 #

• 预处理模型输入定义错误
  preprocess模型的输入应为原始信号[i_sig, q_sig]，而非RFF输出：

  preprocess = keras.Model(inputs=[i_sig, q_sig], outputs=combineRff)
  

• 移动平均计数逻辑错误
  当前self.iters按样本计数，而非batch计数，需改为按batch更新，并使用tf.Variable存储可训练状态：

  # __init__中改为可训练变量
  self.iters = tf.Variable(0.0, trainable=False)
  self.sum = tf.Variable(tf.zeros(self.output_dim, tf.float32), trainable=False)
  
  # call方法中按batch更新
  def call(self, inputs, training=False):
      calc = tf.matmul(inputs, self.kernel)
      if training:
          self.iters.assign_add(1.0)
          self.sum.assign_add(tf.reduce_sum(calc, axis=0))
          # 后续损失计算逻辑...
      return calc
  

• 解码器输出形状不匹配
  解码器输出(4*numInputs, 1, 1)与输入(4*numInputs,)形状不符，修正Reshape层：

  reshape2 = layers.Reshape((4*numInputs,))(trans4)
  

• 数据打乱效率优化
  替换低效的逐元素交换，使用np.random.permutation：

  perm = np.random.permutation(len(newReal))
  newReal = np.array(newReal)[perm].tolist()
  newImag = np.array(newImag)[perm].tolist()
  dataFiles = np.array(dataFiles)[perm].tolist()
  

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