半监督学习自定义训练循环样本权重失效问题求助

大学课外项目中开发半监督模型，先用model.fit完成监督训练验证了模型有效性。为满足半监督学习灵活性需求，改用自定义训练循环，目前仅模拟model.fit的监督训练阶段，但发现模型无法正确应用样本权重。已尝试简化损失函数，问题仍未解决，相关代码如下：

model = nn.model_lite(ds, shapes=nn.get_shapes(x_train))
#The model is implemented from another script.

#Gewichtung:
def generate_class_weight(class_series, multi_class=True, one_hot_encoded=False):
    if multi_class:
        if one_hot_encoded:
            class_series = np.argmax(class_series, axis=1)

        class_labels = np.unique(class_series)
        class_weights = compute_class_weight(class_weight='balanced', classes=class_labels, y=class_series)

        return dict(zip(class_labels, class_weights))
    else:
        mlb = None
        if not one_hot_encoded:
            mlb = MultiLabelBinarizer()
            class_series = mlb.fit_transform(class_series)

        n_samples = len(class_series)
        n_classes = len(class_series[0])

        class_count = [0]*n_classes

        for classes in class_series:
            for index in range(n_classes):
                if classes[index] != 0:
                    class_count[index] += 2

        class_weights = [n_samples/(n_classes * freq) if freq > 0 else 1 for freq in class_count]
        class_labels = range(len(class_weights)) if mlb is None else mlb.classes_
        return dict(zip(class_labels, class_weights))


class_weights = generate_class_weight(y_train)

Batches = []
unlab_batches = []
y_batch = []
# Test for the Lost-Fuction:
for i in range(batch_len2):
    b =  batched_dict(x_train, i, batch_size)
    Batches.append(b)
    d = y_train[i * batch_size:(i+1)*batch_size]
    y_batch.append(d)
batch = {}
for key in x_train:
    batch[key] = x_train[key][(i+1)*batch_size:]
d = y_train[(i+1)*batch_size:]
y_batch.append(d)

Batches.append(batch)
print(Batches)
print(Batches[0])

for i in range(unlab_len):
    c = batched_dict(x_unlabel, i, batch_size)
    unlab_batches.append(c)

batch = {}
for key in x_unlabel:
    batch[key] = x_unlabel[key][(i+1)*batch_size:len(train_unlabels)]
unlab_batches.append(batch)

print(len(unlab_batches))
print(unlab_batches[0])


def new_loss(model, y, x_train,x_unlabel, training):
    
    sample_weights = (np.ones(y.shape) - y)*class_weights[0] + y*class_weights[1]
    
    
    
    
    y_pred = model(x_train, training=training)
    


    Loss_BCE = tf.keras.losses.binary_crossentropy
    Loss_unlab = self_consistency_loss
    #Loss_ges = Loss_BCE(y, y_pred) + Loss_unlab(y_pred, y, y_unlab)
    Loss_ges = Loss_BCE(y, y_pred)
    
    Loss_ges = Loss_ges * sample_weights
    #print(Loss_ges.shape)
    
    
    return Loss_ges


def lr_schedule(epoch):
    lr = 1e-3
    #lr = 1e-5
    if epoch > 20:
        lr *= 0.01
    elif epoch > 10:
        lr *= 0.1
    return lr


def grad(model, y, x_lab , x_unlab):
    with tf.GradientTape() as tape:
    tape.watch(model.trainable_variables)
    loss_value = new_loss(model, y, x_lab, x_unlab, training=True)
    grads = tape.gradient(loss_value, model.trainable_variables)
    

    return loss_value, grads


optimizer = tf.keras.optimizers.Adam(learning_rate=lr_schedule(0))


result_loss = []
result_acc = []

num_epochs = 60

for epoch in range(num_epochs):
    epoch_loss_average = tf.keras.metrics.Mean()
    epoch_accuracy = tf.keras.metrics.BinaryAccuracy()

    for j in range(3):
        x = Batches[j]
        y = y_batch[j]
        z = unlab_batches[j]
        loss_value, grads = grad(model, y, x, z)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))

        # Track
        epoch_loss_average.update_state(loss_value)
        epoch_accuracy.update_state(y, model(x, training=True))
        

    # The End
    result_loss.append(epoch_loss_average.result())
    result_acc.append(epoch_accuracy.result())
    
    if epoch % 5 == 0:
        print("Epoch {:03d}: Loss: {:.3f}, Accuracy: {:.3%}".format(epoch,
                                                                epoch_loss_average.result(),
                                                                epoch_accuracy.result()))


filepath = path
model.save(filepath)

• 样本权重未转为TensorFlow张量：用numpy操作生成sample_weights，未转为tf.Tensor，导致计算图无法正确追踪权重参与的梯度计算，权重无法生效。
• 损失未做批量平均：直接返回逐样本的加权损失，未对批量内损失取平均，optimizer处理梯度时会把所有样本损失的梯度直接累加，导致梯度规模异常，权重效果被稀释。
• GradientTape缩进错误：tape.watch(model.trainable_variables)不在with tf.GradientTape()块内，无法正确追踪模型可训练变量的梯度计算。
• 学习率未动态更新：仅在初始化optimizer时调用一次lr_schedule(0)，后续epoch未更新学习率，不符合调度逻辑，也可能间接影响模型对权重的学习。

1. 修正样本权重的张量类型与损失计算 #

将sample_weights转为tf.Tensor，并对加权后的损失取批量平均：

def new_loss(model, y, x_train,x_unlabel, training):
    # 用TensorFlow原生操作生成权重并转为对应类型
    sample_weights = tf.cast((tf.ones_like(y) - y)*class_weights[0] + y*class_weights[1], dtype=tf.float32)
    
    y_pred = model(x_train, training=training)
    Loss_BCE = tf.keras.losses.binary_crossentropy
    Loss_ges = Loss_BCE(y, y_pred)
    
    # 加权后计算批量平均损失
    Loss_ges = tf.reduce_mean(Loss_ges * sample_weights)
    
    return Loss_ges

2. 修正GradientTape的缩进 #

把tape.watch放到with块内，确保正确追踪变量梯度：

def grad(model, y, x_lab , x_unlab):
    with tf.GradientTape() as tape:
        tape.watch(model.trainable_variables)
        loss_value = new_loss(model, y, x_lab, x_unlab, training=True)
    grads = tape.gradient(loss_value, model.trainable_variables)
    return loss_value, grads

3. 动态更新学习率 #

每个epoch更新optimizer的学习率，匹配调度逻辑：

for epoch in range(num_epochs):
    # 实时更新当前epoch的学习率
    optimizer.learning_rate.assign(lr_schedule(epoch))
    
    epoch_loss_average = tf.keras.metrics.Mean()
    epoch_accuracy = tf.keras.metrics.BinaryAccuracy()

    # 后续训练代码保持不变

4. 验证批次数据完整性 #

检查Batches和y_batch的生成逻辑，确保每个批次的样本数与标签数完全匹配，可添加日志打印每个批次的shape，避免因数据不完整导致权重计算异常。

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