☰
Keras多输出模型损失函数配置与梯度冲突实战指南
2026/10/22 4:19:36 网站建设 项目流程

1. 多输出模型里损失函数到底该怎么摆

很多人第一次在Keras里给模型挂多个损失函数,脑子里想的都是“我只要把loss写成一个列表,Keras自己会搞定”。结果跑起来要么报形状不匹配,要么某个分支的loss压根没参与梯度更新,要么训练半天发现只有主任务在收敛,辅助任务纹丝不动。问题不在于Keras不支持多损失,而在于大多数人没搞清楚Keras处理多损失时的三条路径:输出层命名、loss字典映射、loss_weights权重分配。这三者只要有一个对不上,整个训练图就会悄悄退化。

先明确一个概念:Keras里的“多个自定义损失函数”其实分两种场景。第一种是单输出多损失,比如一个输出既要算MSE又要算一个正则项,这时候你需要的是把多个loss加起来返回一个标量。第二种是多输出多损失,模型有多个head,每个head配一个独立的loss,这时候才涉及loss列表或字典。很多人把这两种场景混在一起写,代码自然乱。

我见过最典型的错误写法是这样的:

model.compile(optimizer='adam', loss=[my_loss_a, my_loss_b], loss_weights=[0.7, 0.3])

但模型只有一个输出层。这种情况下Keras会尝试把两个loss都作用在同一个输出上,如果两个loss返回的形状不一致,直接报维度错误;如果形状一致,它会做加权求和,但你以为的“两个独立任务”其实变成了“同一个预测值的两个约束”,语义完全变了。

正确的判断逻辑是:先看model.outputs的长度。如果长度是1,那所有loss最终必须聚合成一个标量;如果长度大于1,才用列表或字典给每个输出分配loss。这个判断我建议你在写代码前先用print(len(model.outputs))确认一遍,别凭感觉。

还有一个隐藏坑:当模型有多个输出但只给部分输出配了loss,Keras默认会对没有配loss的输出不做梯度回传。这本身是合理的,但如果你用的是函数式API且某个中间层被多个输出共享,没配loss的那个分支仍然会参与前向计算,只是不产生梯度。这会导致显存占用比预期高,训练速度也上不去。解决办法是显式给每个输出都配loss,哪怕某个分支你暂时不想训练,也给它配一个权重为0的loss,这样Keras的图结构会更干净。

提示:多输出模型里,输出层的name参数非常关键。后面用字典配loss时,key必须和输出层的name完全一致,大小写敏感。

2. 自定义损失函数的三种写法与适用边界

Keras里写自定义损失函数,表面上有三种写法:纯函数式、带额外参数的闭包式、继承Loss类。很多人随手选一种就开始写,结果在保存模型、加载模型、分布式训练时接连翻车。这三种写法不是等价的,它们的序列化能力、参数绑定方式、对add_loss的兼容性都不一样。

2.1 纯函数式:最简单但最受限

纯函数式就是接收y_true和y_pred两个参数,返回一个标量或逐样本的loss张量:

import tensorflow as tf def focal_loss(y_true, y_pred, gamma=2.0, alpha=0.25): y_pred = tf.clip_by_value(y_pred, 1e-7, 1 - 1e-7) ce = -y_true * tf.math.log(y_pred) weight = alpha * tf.pow(1 - y_pred, gamma) return tf.reduce_mean(weight * ce)

这种写法的问题在于gamma和alpha被硬编码在函数签名里,编译时没法从外部传参。如果你在model.compile(loss=focal_loss)时想改gamma,只能重新定义一个新函数。更麻烦的是模型保存后,load_model需要能定位到这个函数,如果函数定义在__main__里且没有注册,加载时会报Unknown loss function。

适用边界:超参数固定、不需要频繁调整、且函数定义在可导入的模块文件里。如果你只是在notebook里快速实验,纯函数式够用;但一旦要交付或部署,建议换写法。

2.2 闭包式:参数灵活但序列化有坑

闭包式是用一个外层函数包住超参数,返回内层loss函数:

def make_focal_loss(gamma=2.0, alpha=0.25): def focal_loss(y_true, y_pred): y_pred = tf.clip_by_value(y_pred, 1e-7, 1 - 1e-7) ce = -y_true * tf.math.log(y_pred) weight = alpha * tf.pow(1 - y_pred, gamma) return tf.reduce_mean(weight * ce) return focal_loss

这样你可以loss=make_focal_loss(gamma=3.0),参数灵活。但保存模型时,Keras序列化的是内层函数的引用,而内层函数是运行时动态创建的,load_model根本找不到它。你会遇到经典的ValueError: Unknown loss function: focal_loss。

解决办法有两个:一是用custom_objects在加载时手动传入,二是改用Loss类。前者在简单场景可行,但如果你有多个闭包loss,custom_objects的字典会变得很难维护。

2.3 继承Loss类:最规范但代码量最大

@tf.keras.utils.register_keras_serializable(package="MyLosses") class FocalLoss(tf.keras.losses.Loss): def __init__(self, gamma=2.0, alpha=0.25, **kwargs): super().__init__(**kwargs) self.gamma = gamma self.alpha = alpha def call(self, y_true, y_pred): y_pred = tf.clip_by_value(y_pred, 1e-7, 1 - 1e-7) ce = -y_true * tf.math.log(y_pred) weight = self.alpha * tf.pow(1 - y_pred, self.gamma) return tf.reduce_mean(weight * ce) def get_config(self): config = super().get_config() config.update({"gamma": self.gamma, "alpha": self.alpha}) return config

这种写法配合register_keras_serializable装饰器,保存和加载都不会出问题。get_config负责把超参数写进模型文件,加载时自动重建。代价是代码量翻倍,而且call方法里不能直接用Python的if判断张量值,必须用tf.cond或tf.where。

我的建议是:实验阶段用闭包式,交付阶段一律换成Loss类。切换成本不高,但能省掉后面一堆序列化相关的调试时间。

写法参数灵活性序列化支持代码量推荐场景
纯函数式低需手动注册最少快速验证
闭包式高差中等实验调参
Loss类高原生支持最多生产交付

3. 多输出模型的loss字典配置与权重调优

当模型有多个输出时,最清晰的配置方式是用字典把输出层name映射到对应的loss:

model.compile( optimizer=tf.keras.optimizers.Adam(1e-3), loss={ "classification_head": FocalLoss(gamma=2.0), "regression_head": tf.keras.losses.Huber(delta=1.0), "auxiliary_head": tf.keras.losses.MeanSquaredError() }, loss_weights={ "classification_head": 1.0, "regression_head": 0.5, "auxiliary_head": 0.2 }, metrics={ "classification_head": ["accuracy"], "regression_head": ["mae"] } )

这里有几个细节值得展开。第一,loss_weights的key必须和loss字典的key完全一致,否则Keras会忽略权重,默认全部按1.0处理。第二,metrics也是按输出层name配置的,如果你只给部分输出配了metrics,训练日志里只会显示配了的那些,没配的不会报错但也不会显示。第三,loss_weights的总和不需要等于1,Keras内部会做归一化处理,但归一化方式是把每个loss乘以对应权重后直接相加,不做除法。所以如果你把所有权重都放大10倍,等效于把学习率放大10倍,这一点很容易被忽略。

权重怎么调?我的经验是分三步走。第一步,先把所有loss_weights设为1.0,跑几个epoch,观察每个loss的初始量级。如果分类loss在2.0左右,回归loss在0.01左右,那回归任务的梯度贡献几乎可以忽略。第二步,根据量级差异反比设置权重,让每个loss的初始贡献大致相当。比如分类loss是2.0,回归loss是0.01,那回归权重可以设成100左右,但不要一步到位,先设成10观察。第三步,根据任务重要性微调。如果分类是主任务,回归是辅助,那在量级对齐的基础上,把分类权重再放大1.5到2倍。

注意:loss_weights调得过大时,某个分支的梯度会主导整个网络,导致共享层被这个任务“绑架”,其他任务的表现反而下降。这种情况在多任务学习里叫“负迁移”,表现是某个任务的验证指标先降后升再降。遇到这种情况,要么降低该任务权重,要么在共享层和任务头之间加梯度裁剪。

还有一个实操技巧:用tf.keras.callbacks.Callback在训练过程中动态打印每个loss的值,而不是只看总loss。总loss是加权和,某个分支出问题时总loss可能看起来正常,但分支loss已经发散。下面这个callback我经常用:

class LossLogger(tf.keras.callbacks.Callback): def on_epoch_end(self, epoch, logs=None): logs = logs or {} for key, value in logs.items(): if "loss" in key and "val" not in key: print(f"Epoch {epoch} | {key}: {value:.4f}")

这样你能清楚看到每个head的loss走势,比只看一个total_loss有用得多。

4. 梯度冲突与loss平衡的实战处理

多损失训练最棘手的问题不是配置,而是梯度冲突。两个任务的loss对共享层参数的梯度方向相反时,网络会在两个任务之间反复横跳,表现是总loss震荡不降,或者一个任务好另一个任务差。这个问题在推荐系统、多模态模型里特别常见。

4.1 用GradNorm思路做动态权重

固定权重最大的问题是训练过程中各任务的学习难度在变化,前期简单的任务梯度大,后期困难的任务梯度大,固定权重没法适应。GradNorm的核心思路是:根据每个任务当前的梯度范数和训练速度,动态调整loss权重,让各任务的梯度范数保持在同一量级。

简化版的实现思路是:在每个batch结束后,计算每个任务loss对共享层参数的梯度范数,然后按范数的反比更新权重。完整实现比较复杂,但你可以用一个近似版本:

class DynamicWeightCallback(tf.keras.callbacks.Callback): def __init__(self, shared_layer_name, task_names, alpha=0.1): super().__init__() self.shared_layer_name = shared_layer_name self.task_names = task_names self.alpha = alpha self.weights = {name: 1.0 for name in task_names} def on_train_batch_end(self, batch, logs=None): logs = logs or {} losses = {} for name in self.task_names: key = f"{name}_loss" if key in logs: losses[name] = logs[key] if len(losses) < 2: return mean_loss = sum(losses.values()) / len(losses) for name, loss_val in losses.items(): ratio = loss_val / (mean_loss + 1e-8) self.weights[name] *= (1 - self.alpha * (ratio - 1)) self.weights[name] = max(0.1, min(10.0, self.weights[name]))

这个callback在on_train_batch_end里根据各任务loss与平均loss的比值调整权重,loss高于平均的任务权重降低,低于平均的权重升高。alpha控制调整速度,太大会震荡,太小没效果,一般从0.05到0.1开始试。权重上下限是为了防止某个任务权重被压到0或涨到无穷。

4.2 梯度裁剪与共享层保护

如果动态权重调起来太麻烦,一个更直接的办法是在共享层后面加梯度裁剪。Keras里可以在optimizer层面做全局裁剪:

optimizer = tf.keras.optimizers.Adam( learning_rate=1e-3, clipnorm=1.0 )

clipnorm按梯度向量的L2范数裁剪,clipvalue按每个梯度元素的值裁剪。多任务场景下我更推荐clipnorm,因为它保留了梯度方向,只限制幅度。clipvalue会改变梯度方向,可能导致某些任务学不动。

另一个技巧是给共享层加一个较小的学习率,给任务头加较大的学习率。Keras原生不支持分层学习率,但可以用tf.keras.optimizers.schedules配合自定义训练循环实现。简单做法是共享层用SGD,任务头用Adam,但这样需要写自定义train_step。如果不想写自定义循环,退而求其次的办法是在共享层后面加BatchNormalization,BN层的running statistics会起到一定的稳定作用,缓解梯度冲突带来的震荡。

4.3 任务优先级与课程学习

有时候梯度冲突的根源是任务太难,模型在早期根本学不动。这时候可以用课程学习的思路:前期只训练主任务,等主任务收敛到一定程度后再加入辅助任务。实现方式是用callback在指定epoch后修改model.loss_weights:

class CurriculumCallback(tf.keras.callbacks.Callback): def __init__(self, switch_epoch, new_weights): super().__init__() self.switch_epoch = switch_epoch self.new_weights = new_weights def on_epoch_begin(self, epoch, logs=None): if epoch == self.switch_epoch: self.model.loss_weights = self.new_weights print(f"Epoch {epoch}: loss weights switched to {self.new_weights}")

注意model.loss_weights在编译后修改是否生效取决于TF版本,较新的版本支持动态修改,老版本可能需要重新编译。实测下来TF 2.8以上都可以直接改,但改完后第一个batch的loss可能会跳变,属于正常现象。

5. 保存加载与分布式训练的兼容性细节

自定义loss在单机单卡上跑通不难,难的是保存加载和分布式训练时不掉链子。我踩过的坑里,80%都出在这两个环节。

5.1 保存加载的三种失败模式

第一种是Unknown loss function,原因是loss函数没有注册。解决办法是用@tf.keras.utils.register_keras_serializable装饰Loss类,或者在load_model时传custom_objects。我强烈推荐前者,因为后者在模型被其他代码加载时容易遗漏。

第二种是get_config不完整。如果你在Loss类里加了自定义属性但忘了写进get_config,保存时不会报错,加载后属性会丢失,loss计算结果和训练时不一致。这种bug最隐蔽,因为模型能加载、能推理,只是结果悄悄偏了。检查方法是保存后重新加载,用同一批数据分别跑原模型和加载模型,对比loss值是否完全一致。

第三种是loss_weights丢失。Keras保存模型时会把compile信息写进文件,但如果你在保存后手动改了loss_weights再保存,旧值可能被覆盖。稳妥做法是每次保存前重新compile一次,确保配置是最新的。

# 保存前重新compile,确保配置写入 model.compile(optimizer=optimizer, loss=loss_dict, loss_weights=weights) model.save("my_model.keras") # 加载时如果用了register_keras_serializable,不需要custom_objects loaded = tf.keras.models.load_model("my_model.keras")

5.2 分布式训练下的loss聚合

用MirroredStrategy做多卡训练时,每个卡上的loss是各自计算的,Keras会自动做all-reduce求平均。但如果你在loss函数里用了tf.reduce_sum而不是tf.reduce_mean,多卡下的总loss会被放大N倍(N是卡数),等效于学习率放大N倍。这个坑我在一次四卡训练里踩过,单卡正常,四卡直接发散。

解决办法很简单:loss函数里一律用tf.reduce_mean,让Keras的分布式聚合逻辑处理跨卡平均。如果你确实需要sum,那就在compile时把学习率除以卡数。

另一个分布式相关的细节是loss_weights在MirroredStrategy下的行为。权重是在聚合之后应用的,所以每张卡上的加权方式一致,不需要额外处理。但如果你用了自定义训练循环,需要手动在strategy.run外面做加权,否则每张卡会各自加权再聚合,结果和单卡不一致。

提示:分布式训练时,建议先用单卡跑通完整流程,确认loss收敛后再切多卡。直接上多卡调试,问题定位难度会翻倍。

6. 几个我实际踩过的坑和对应解法

第一个坑:在loss函数里用了Python的if判断张量值。比如if y_pred > 0.5: return ...,这在eager模式下能跑,但一旦用tf.function编译或保存成SavedModel,直接报OperatorNotAllowedInGraphError。正确做法是用tf.where:

# 错误写法 def bad_loss(y_true, y_pred): if tf.reduce_mean(y_pred) > 0.5: return tf.reduce_mean(tf.square(y_true - y_pred)) return tf.reduce_mean(tf.abs(y_true - y_pred)) # 正确写法 def good_loss(y_true, y_pred): mse = tf.reduce_mean(tf.square(y_true - y_pred)) mae = tf.reduce_mean(tf.abs(y_true - y_pred)) condition = tf.reduce_mean(y_pred) > 0.5 return tf.where(condition, mse, mae)

第二个坑:loss函数返回了逐样本的loss向量而不是标量。Keras的compile接口期望loss返回标量,如果你返回形状为(batch_size,)的张量,Keras会做一次隐式reduce_mean,但如果你在loss里已经做了加权,这个隐式平均会打乱权重。表现是训练能跑但收敛异常。检查方法是打印loss函数的返回形状,确保是()而不是(batch_size,)。

第三个坑:多个loss共享同一个可训练变量时,梯度累加导致更新量翻倍。比如两个loss都对同一个embedding层求梯度,Keras会把两个梯度相加后更新。这本身是合理的,但如果你期望的是两个loss分别更新,就需要用tf.GradientTape手动控制。大多数情况下梯度相加是想要的行为,但如果你发现某个共享层更新过快,可以检查是不是多个loss都在对它产生梯度。

第四个坑:metrics和loss用了同一个函数但期望不同行为。比如你把focal loss同时配成loss和metric,metric会显示focal loss的值,但这个值受loss_weights影响,而metric本身不应该被加权。结果是metric数值和实际任务表现对不上。正确做法是loss和metric分开定义,metric用标准的accuracy或mae。

坑表现根因解法
Python if判断张量Graph模式报错图模式不支持Python控制流改用tf.where
loss返回向量收敛异常隐式reduce打乱权重确保返回标量
共享层梯度翻倍更新过快多loss梯度累加检查是否预期行为
metric被加权指标失真metric受loss_weights影响loss和metric分开

7. 一套可直接复用的多损失模板

把上面所有经验整合成一个模板,你可以直接抄去改。这个模板覆盖了Loss类定义、多输出配置、动态权重、保存加载四个环节。

import tensorflow as tf @tf.keras.utils.register_keras_serializable(package="MyLosses") class WeightedFocalLoss(tf.keras.losses.Loss): def __init__(self, gamma=2.0, alpha=0.25, name="weighted_focal_loss", **kwargs): super().__init__(name=name, **kwargs) self.gamma = gamma self.alpha = alpha def call(self, y_true, y_pred): y_pred = tf.clip_by_value(y_pred, 1e-7, 1 - 1e-7) ce = -y_true * tf.math.log(y_pred) weight = self.alpha * tf.pow(1 - y_pred, self.gamma) return tf.reduce_mean(weight * ce) def get_config(self): config = super().get_config() config.update({"gamma": self.gamma, "alpha": self.alpha}) return config def build_model(): inputs = tf.keras.Input(shape=(128,), name="input") shared = tf.keras.layers.Dense(256, activation="relu", name="shared")(inputs) shared = tf.keras.layers.BatchNormalization(name="shared_bn")(shared) shared = tf.keras.layers.Dropout(0.3)(shared) cls_out = tf.keras.layers.Dense(10, activation="softmax", name="cls_head")(shared) reg_out = tf.keras.layers.Dense(1, name="reg_head")(shared) return tf.keras.Model(inputs=inputs, outputs=[cls_out, reg_out]) model = build_model() loss_dict = { "cls_head": WeightedFocalLoss(gamma=2.0, alpha=0.25), "reg_head": tf.keras.losses.Huber(delta=1.0) } loss_weights = { "cls_head": 1.0, "reg_head": 0.5 } model.compile( optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3, clipnorm=1.0), loss=loss_dict, loss_weights=loss_weights, metrics={"cls_head": ["accuracy"], "reg_head": ["mae"]} ) model.summary() # 保存 model.save("multi_loss_model.keras") # 加载,因为用了register_keras_serializable,不需要custom_objects loaded_model = tf.keras.models.load_model("multi_loss_model.keras")

这个模板里几个关键点再强调一遍:clipnorm=1.0防止梯度爆炸,BatchNormalization稳定共享层,register_keras_serializable保证序列化,loss_weights按任务重要性分配。你拿到后只需要改输入形状、输出维度、loss类型和权重值。

如果训练过程中发现某个head的loss不降,先检查三件事:该head的输出层name是否和loss字典的key一致、该head的loss是否返回标量、该head的梯度是否被其他head的梯度淹没。这三件事查完,90%的问题都能定位。

最后分享一个我常用的调试技巧:在训练前用model.compute_loss手动算一次loss,传入随机生成的y_true和y_pred,看返回的loss字典结构是否符合预期。这一步能在正式训练前暴露大部分配置错误,比跑几个epoch再发现问题省时间得多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询