基于神经网络的多参数函数F优化实现方案问询

问题描述 #

拥有函数F(args: tuple) => tuple，接收11个参数并返回含4个元素的元组，项目基于tensorflow开发，需用神经网络实现输入数据集的最优结果预测。尝试过自定义优化器，但仅作用于真实输出与预测输出；编写了训练代码，但不知如何通过函数F建立输入与输出的关联。

现有训练代码：

import numpy as np
import tensorflow as tf
from tensorflow import keras
import matplotlib.pyplot as plt

X = np.array(x)  # 输入数据，形状(N, 11)
Y = np.array(y)  # 输出数据，形状(N, 4)

# 定义神经网络模型
model = keras.Sequential([
    keras.layers.Dense(32, activation='relu', input_shape=(11,)), 
    keras.layers.Dense(16, activation='relu'),
    keras.layers.Dense(4)  # 输出层对应4个参数
])

def custom_loss(y_true, y_pred):
   loss = abs(y_true[:, 0] - y_pred[:, 0]) + abs(y_true[:,1]- y_pred[:, 1]) + abs(y_true[:,2]-y_pred[:, 2]) + abs(y_true[:,3]- y_pred[:, 3])
   return loss

model.compile(optimizer='adam', loss=custom_loss)

# 初始化训练损失和参数记录列表
training_loss = []
experimental_parameters = []

# 记录训练损失和参数的回调类
class LossHistory(keras.callbacks.Callback):
   def on_epoch_end(self, epoch, logs={}):
      training_loss.append(logs.get('loss'))
      experimental_parameters.append(model.predict(X))  # 预测并保存参数

# 创建回调实例
history_callback = LossHistory()

# 训练模型并记录数据
model.fit(X, Y, epochs=10000, batch_size=32, callbacks=[history_callback])

# 绘制训练损失曲线
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(training_loss)
plt.xlabel('Epoch')
plt.ylabel('Training Loss')
plt.title('训练损失变化')

# 绘制参数变化曲线
experimental_parameters = np.array(experimental_parameters)
plt.subplot(1, 2, 2)
for i in range(4):
    plt.plot(experimental_parameters[:, i], label=f'参数 {i + 1}')
plt.xlabel('Epoch')
plt.ylabel('参数值')
plt.title('参数随训练周期变化')
plt.legend()
plt.tight_layout()
plt.show()

预测代码：

input_data = np.array([[35.5, 56.94, 10.0, 123.79, 4.7, 220.0, 12.12, 1.41, 1.36, 7.51, 6.95]])
# 使用模型预测
predictions = model.predict(input_data)
print(predictions)

解决方案 #

情况1：函数F可微分（兼容TensorFlow张量运算） #

若F可改写为TensorFlow张量运算形式，可将其融入损失函数或模型结构，让神经网络学习输入与输出通过F建立的关联。

方案1：将F作为损失函数的约束项

假设核心需求是：让输入X经神经网络得到的预测值Y_pred，与X共同传入F后的输出，尽可能贴近X与真实Y_true传入F的输出。可通过自定义训练循环实现：

首先将F改写为TensorFlow兼容版本：

def F_tensorflow(args):
    # args为形状(N,15)的张量，前11列是X，后4列是预测/真实输出
    x_part = args[:, :11]
    y_part = args[:, 11:]
    # 替换为你的F函数逻辑，需用TensorFlow运算实现
    output_0 = tf.reduce_sum(x_part[:, :5], axis=1) + y_part[:, 0]
    output_1 = tf.multiply(x_part[:, 5], y_part[:, 1])
    output_2 = tf.subtract(x_part[:, 6], y_part[:, 2])
    output_3 = tf.divide(x_part[:, 7], y_part[:, 3] + 1e-6)  # 加小值避免除零
    return tf.stack([output_0, output_1, output_2, output_3], axis=1)

然后用自定义训练循环整合损失：

optimizer = tf.keras.optimizers.Adam()
epochs = 10000
batch_size = 32
dataset = tf.data.Dataset.from_tensor_slices((X, Y)).shuffle(len(X)).batch(batch_size)
training_loss = []

for epoch in range(epochs):
    epoch_loss = 0.0
    for x_batch, y_batch in dataset:
        with tf.GradientTape() as tape:
            y_pred = model(x_batch, training=True)
            # 计算F的输出差异
            f_pred = F_tensorflow(tf.concat([x_batch, y_pred], axis=1))
            f_true = F_tensorflow(tf.concat([x_batch, y_batch], axis=1))
            # 总损失 = 原始L1损失 + F约束损失
            total_loss = tf.reduce_mean(tf.abs(y_batch - y_pred)) + tf.reduce_mean(tf.abs(f_pred - f_true))
        gradients = tape.gradient(total_loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
        epoch_loss += total_loss.numpy()
    epoch_loss /= len(dataset)
    training_loss.append(epoch_loss)
    if epoch % 100 == 0:
        print(f"Epoch {epoch}, Loss: {epoch_loss:.4f}")

方案2：将F作为模型的自定义层

若F是输入到最终输出的必要变换环节，可将其封装为自定义层加入模型：

class FLayer(tf.keras.layers.Layer):
    def call(self, inputs):
        # inputs为元组，包含原始输入X和神经网络中间输出
        x, intermediate_pred = inputs
        return F_tensorflow(tf.concat([x, intermediate_pred], axis=1))

# 重构模型
input_layer = tf.keras.Input(shape=(11,))
dense1 = tf.keras.layers.Dense(32, activation='relu')(input_layer)
dense2 = tf.keras.layers.Dense(16, activation='relu')(dense1)
intermediate_pred = tf.keras.layers.Dense(4)(dense2)
final_output = FLayer()([input_layer, intermediate_pred])

model = tf.keras.Model(inputs=input_layer, outputs=final_output)
model.compile(optimizer='adam', loss=tf.keras.losses.MeanAbsoluteError())

情况2：函数F是黑盒（不可微分） #

若F无法改写为TensorFlow可微分运算（如外部库函数、原生Python函数），需用强化学习或替代模型方案：

方案1：强化学习框架

将神经网络输出作为动作，F的输出作为奖励信号，让模型学习能使F输出最优结果的参数：

model = keras.Sequential([
    keras.layers.Dense(32, activation='relu', input_shape=(11,)), 
    keras.layers.Dense(16, activation='relu'),
    keras.layers.Dense(4)
])

optimizer = tf.keras.optimizers.Adam()
epochs = 10000
batch_size = 32

for epoch in range(epochs):
    avg_reward = 0.0
    for i in range(0, len(X), batch_size):
        x_batch = X[i:i+batch_size]
        with tf.GradientTape() as tape:
            y_pred = model(x_batch, training=True)
            # 调用黑盒F函数
            f_outputs = []
            for x, yp in zip(x_batch.numpy(), y_pred.numpy()):
                # 根据F的参数定义调整输入（此处假设F接收11个X参数+4个预测参数）
                f_out = F(tuple(np.concatenate([x, yp])))
                f_outputs.append(f_out)
            f_outputs = np.array(f_outputs)
            # 以F输出与目标的差异作为奖励（负损失为奖励）
            reward = -tf.reduce_mean(tf.abs(f_outputs - Y[i:i+batch_size]))
            loss = -reward
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
        avg_reward += reward.numpy()
    avg_reward /= (len(X)//batch_size)
    if epoch % 100 == 0:
        print(f"Epoch {epoch}, 平均奖励: {avg_reward:.4f}")

方案2：训练F的替代模型

先收集大量(X,Y)对，计算F(X,Y)的结果，训练一个神经网络模拟F的行为（替代模型），之后即可用情况1的方法将替代模型融入训练流程。

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