Keras多损失函数配置及MSE保留问题求助（含报错排查）

问题背景 #

需要在Keras模型中结合5种损失约束：

• 避免预测结果与往期结果重复
• 预测结果总和需在指定区间[V, W]内
• 限制00-49区间的预测数字出现次数不超过X次
• 限制50-99区间的预测数字出现次数不超过Y次
• 保留默认MSE损失函数

原代码执行model.fit时触发错误：

ValueError: Unknown loss function: combined_loss_fn. Please ensure this object is passed to the custom_objects argument.

同时需要支持从数据库动态获取损失函数中的限制参数。

错误根源分析 #

1. 损失函数签名不兼容：Keras要求自定义损失函数仅接收y_true和y_pred两个参数，原fn_never_repeat额外添加的previous_data参数无法被框架识别。
2. 计算图兼容性问题：fn_quadrant中使用Python原生if语句，无法在TensorFlow计算图中执行，必须替换为框架支持的条件操作。
3. 函数命名冲突：fn_combined_loss内部重新定义了同名变量fn_never_repeat，导致递归调用错误。
4. 编译配置错误：编译模型时直接传递函数对象即可，无需使用字符串指定，custom_objects仅在模型保存/加载时需要。

修正方案与完整实现 #

核心优化思路 #

• 用闭包传递额外参数（往期数据、动态约束值），保证损失函数符合Keras的签名要求。
• 所有逻辑判断改用TensorFlow/Keras后端函数，确保能在计算图中运行。
• 动态参数通过闭包注入，支持从数据库实时获取最新约束值。

完整代码实现 #

import tensorflow as tf
from tensorflow.keras import backend as K
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense

# 模拟从数据库获取动态参数（实际替换为数据库调用）
def get_dynamic_params():
    return {
        "sum_min": 133,
        "sum_max": 249,
        "count_0_49_max": 10,
        "count_50_99_max": 10,
        "loss_weights": {
            "repeat": 0.25,
            "sum": 0.25,
            "quadrant": 0.25,
            "mse": 0.25
        }
    }

# 闭包生成：避免与往期结果重复的损失函数
def create_never_repeat_loss(previous_data):
    def never_repeat_loss(y_true, y_pred):
        # 计算预测与往期数据的差异，差异越小惩罚越高（避免重复）
        diff = K.mean(K.square(y_pred - previous_data), axis=-1)
        return K.mean(1.0 / (diff + K.epsilon()))
    return never_repeat_loss

# 闭包生成：预测总和区间约束的损失函数
def create_sum_loss(sum_min, sum_max):
    def sum_loss(y_true, y_pred):
        predicted_sum = K.sum(y_pred)
        # 超出区间的部分直接作为损失
        loss = K.maximum(sum_min - predicted_sum, 0) + K.maximum(predicted_sum - sum_max, 0)
        return loss
    return sum_loss

# 闭包生成：区间数字计数约束的损失函数
def create_quadrant_loss(count_0_49_max, count_50_99_max):
    def quadrant_loss(y_true, y_pred):
        count_0_to_49 = K.sum(K.cast(K.less(y_pred, 50), K.floatx()))
        count_50_to_99 = K.sum(K.cast(K.greater_equal(y_pred, 50), K.floatx()))
        
        # 用TensorFlow条件操作替代Python原生if
        penalty_0_49 = K.switch(
            K.greater(count_0_to_49, count_0_49_max),
            K.square(count_0_to_49 - count_0_49_max),
            0.0
        )
        penalty_50_99 = K.switch(
            K.greater(count_50_to_99, count_50_99_max),
            K.square(count_50_to_99 - count_50_99_max),
            0.0
        )
        return penalty_0_49 + penalty_50_99
    return quadrant_loss

# 闭包生成：组合所有损失函数
def create_combined_loss(previous_data, dynamic_params):
    never_repeat_loss = create_never_repeat_loss(previous_data)
    sum_loss = create_sum_loss(dynamic_params["sum_min"], dynamic_params["sum_max"])
    quadrant_loss = create_quadrant_loss(dynamic_params["count_0_49_max"], dynamic_params["count_50_99_max"])
    weights = dynamic_params["loss_weights"]
    
    def combined_loss(y_true, y_pred):
        mse_loss = K.mean(K.square(y_true - y_pred))
        repeat_loss = never_repeat_loss(y_true, y_pred)
        sum_penalty = sum_loss(y_true, y_pred)
        quadrant_penalty = quadrant_loss(y_true, y_pred)
        
        return (weights["mse"] * mse_loss +
                weights["repeat"] * repeat_loss +
                weights["sum"] * sum_penalty +
                weights["quadrant"] * quadrant_penalty)
    return combined_loss

# ---------------------- 模型使用示例 ----------------------
# 模拟往期历史数据（实际替换为真实数据）
previous_data = K.constant([[12, 34, 56, 78, 90], [23, 45, 67, 89, 10]])
# 获取动态约束参数
params = get_dynamic_params()
# 创建最终组合损失函数
custom_loss = create_combined_loss(previous_data, params)

# 构建示例模型
model = Sequential()
model.add(Dense(64, activation='relu', input_shape=(10,)))
model.add(Dense(5))  # 假设输出5个0-99的数字

# 编译模型：直接传入损失函数对象
model.compile(optimizer=Adam(learning_rate=0.0001), loss=custom_loss)

# 模拟训练数据
x_train = tf.random.normal((100, 10))
y_train = tf.random.uniform((100, 5), minval=0, maxval=100)

# 启动训练
model.fit(x_train, y_train, epochs=10, batch_size=32)

关键注意事项 #

• 模型保存/加载：若需保存模型，加载时必须通过custom_objects注册自定义损失函数，示例：

  model.save("lottery_model.h5")
  loaded_model = tf.keras.models.load_model("lottery_model.h5", custom_objects={"combined_loss": custom_loss})
  

• 损失权重调整：可根据实际需求修改loss_weights中的比例，平衡不同约束的优先级。
• 数值稳定性：使用K.epsilon()避免除零错误，保证计算图稳定运行。

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