使用TensorFlow预测曲线拟合初始值的自定义损失函数实现问题

你的错误根源是TensorFlow图模式下，y_pred是符号化Tensor，无法直接转换为NumPy数组传给SciPy的curve_fit；同时SciPy操作不在TensorFlow计算图中，无法自动计算梯度，导致反向传播失败。以下是三种可行的解决办法：

方案1：用tf.py_function包装SciPy调用（兼容Keras fit循环） #

tf.py_function允许在TensorFlow图中执行普通Python代码，自动处理Tensor与NumPy的转换，但需注意梯度只能通过数值近似获取，适合对训练效率要求不高的场景。

修改后的代码：

import scipy
import tensorflow as tf
import numpy as np

def linear_function(x, m, n):
    return m*x + n

def fit_and_compute_loss(x_data_np, y_data_np, y_pred_np, y_true_np):
    # SciPy曲线拟合
    result_from_prediction, _ = scipy.optimize.curve_fit(linear_function,
                                                         x_data_np,
                                                         y_data_np,
                                                         p0=y_pred_np)
    # 计算MSE损失
    return np.mean((y_true_np - result_from_prediction) ** 2)

def custom_loss(x):
    def loss(y_true, y_pred):
        # 用tf.py_function包装Python逻辑，转换Tensor与NumPy
        loss_values = tf.py_function(
            func=lambda yt, yp: np.array([
                fit_and_compute_loss(
                    *np.split(x[i], 2),
                    yp[i].numpy(),
                    yt[i].numpy()
                ) for i in range(len(x))
            ], dtype=np.float32),
            inp=[y_true, y_pred],
            Tout=tf.float32
        )
        return tf.reduce_mean(loss_values)
    return loss

# 数据与模型定义不变
x = np.array([[1, 2, 3, 1, 2, 3], [1, 2, 3, 2, 4, 6], [1, 2, 3, 3, 6, 9]])
y_true = np.array([[1,0],[2,0],[3,0]])

model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, input_shape=(x.shape[1],)),
    tf.keras.layers.Dense(2)
])

model.compile(optimizer='adam', loss=custom_loss(x))
model.fit(x=x, y=y_true, epochs=10, batch_size=32)

方案2：手动编写训练循环（完全可控） #

脱离Keras的fit方法，手动处理Tensor转NumPy、损失计算与参数更新，适合需要精细控制训练流程的场景。

代码示例：

import scipy
import tensorflow as tf
import numpy as np

def linear_function(x, m, n):
    return m*x + n

# 数据准备
x = np.array([[1, 2, 3, 1, 2, 3], [1, 2, 3, 2, 4, 6], [1, 2, 3, 3, 6, 9]])
y_true = np.array([[1,0],[2,0],[3,0]])
dataset = tf.data.Dataset.from_tensor_slices((x, y_true)).batch(32)

# 模型与优化器定义
model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, input_shape=(x.shape[1],)),
    tf.keras.layers.Dense(2)
])
optimizer = tf.keras.optimizers.Adam()

# 手动训练循环
epochs = 10
for epoch in range(epochs):
    total_loss = 0.0
    for batch_x, batch_y_true in dataset:
        with tf.GradientTape() as tape:
            # 预测初始参数
            y_pred = model(batch_x, training=True)
            # 转换为NumPy数组
            y_pred_np = y_pred.numpy()
            batch_y_true_np = batch_y_true.numpy()
            batch_x_np = batch_x.numpy()
            
            # 逐个样本计算损失
            batch_loss = []
            for i in range(len(batch_x_np)):
                x_data, y_data = np.split(batch_x_np[i], 2)
                result, _ = scipy.optimize.curve_fit(linear_function,
                                                    x_data,
                                                    y_data,
                                                    p0=y_pred_np[i])
                batch_loss.append(np.mean((batch_y_true_np[i] - result) ** 2))
            
            # 转换为Tensor计算总损失
            loss = tf.convert_to_tensor(np.mean(batch_loss), dtype=tf.float32)
        
        # 计算梯度并更新参数
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
        
        total_loss += loss.numpy()
    
    print(f"Epoch {epoch+1}, Loss: {total_loss/len(dataset):.4f}")

方案3：用TensorFlow重写拟合逻辑（最优方案） #

如果你的拟合函数（比如线性函数）可以用TensorFlow运算实现，直接在图中完成拟合，能自动计算精确梯度，训练效率最高。

代码示例：

import tensorflow as tf
import numpy as np

def linear_function(x, m, n):
    return m*x + n

def tf_linear_fit(x_data, y_data, p0):
    # TensorFlow实现线性最小二乘拟合
    x_tensor = tf.convert_to_tensor(x_data, dtype=tf.float32)
    y_tensor = tf.convert_to_tensor(y_data, dtype=tf.float32)
    # 构造设计矩阵 [x, 1]
    A = tf.stack([x_tensor, tf.ones_like(x_tensor)], axis=1)
    # 求解最小二乘解
    m, n = tf.linalg.lstsq(A, y_tensor[:, tf.newaxis], l2_regularizer=1e-6)
    return tf.squeeze(tf.stack([m, n], axis=1))

def custom_loss(x):
    def loss(y_true, y_pred):
        loss_values = []
        for i in range(len(x)):
            x_data, y_data = np.split(x[i], 2)
            # TensorFlow内拟合
            result_from_prediction = tf_linear_fit(x_data, y_data, y_pred[i])
            # 计算MSE损失
            loss_values.append(tf.reduce_mean(tf.square(y_true[i] - result_from_prediction)))
        return tf.reduce_mean(loss_values)
    return loss

# 数据与模型定义不变
x = np.array([[1, 2, 3, 1, 2, 3], [1, 2, 3, 2, 4, 6], [1, 2, 3, 3, 6, 9]])
y_true = np.array([[1,0],[2,0],[3,0]])

model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, input_shape=(x.shape[1],)),
    tf.keras.layers.Dense(2)
])

model.compile(optimizer='adam', loss=custom_loss(x))
model.fit(x=x, y=y_true, epochs=10, batch_size=32)

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