如何为TensorFlow ResNet50回归模型加入数值型输入数据？

核心思路 #

通过多输入模型架构实现：一个分支处理图像特征（基于ResNet），另一个分支处理数值型特征（全连接层），最后将两个分支的特征拼接后共同完成回归任务。以下是完整的修改步骤和代码：

1. 数据预处理：合并与特征工程 #

首先将图像标签CSV和数值特征CSV合并，同时对数值特征做标准化/编码：

• 连续特征（年龄、体重）：做标准化（减去均值除以标准差）
• 类别特征（性别）：做独热编码或标签编码

import pandas as pd
import matplotlib.pyplot as plt
import tensorflow as tf
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.applications.resnet_rs import ResNetRS50
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, GlobalMaxPooling2D, Dense, Dropout, Concatenate
from tensorflow.keras.optimizers import Adam
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
import numpy as np

# 加载数据：合并标签与数值特征
labels_df = pd.read_csv('[Target Dataframe path]')
numerical_df = pd.read_csv('[Numerical Feature CSV path]')
# 以ID为关联键合并两个表
full_df = pd.merge(labels_df, numerical_df, on='ID', how='inner')

# 定义特征列：替换为你的实际特征名
continuous_features = ['年龄', '体重']
categorical_features = ['性别']

# 构建特征预处理管道
preprocessor = ColumnTransformer(
    transformers=[
        ('num', StandardScaler(), continuous_features),
        ('cat', OneHotEncoder(sparse_output=False), categorical_features)
    ])

# 拟合预处理管道（仅用训练集数据，避免数据泄露）
full_df['split'] = np.random.choice(['training', 'validation'], size=len(full_df), p=[0.8, 0.2])
train_df = full_df[full_df['split'] == 'training']
preprocessor.fit(train_df[continuous_features + categorical_features])

# 生成预处理后的数值特征
full_df['processed_numerical'] = list(preprocessor.transform(full_df[continuous_features + categorical_features]))

2. 自定义多输入数据生成器 #

替换原有的数据加载函数，生成器需同时返回图像、数值特征和标签：

def load_multi_input_data(df, image_dir, target_size=(224,224), batch_size=32, seed=1234):
    datagen = ImageDataGenerator(rescale=1./255)  # 图像归一化
    
    # 仅生成图像的生成器
    image_generator = datagen.flow_from_dataframe(
        dataframe=df,
        directory=image_dir,
        x_col='ID',
        y_col=None,
        target_size=target_size,
        batch_size=batch_size,
        class_mode=None,
        seed=seed
    )
    
    # 提取数值特征和标签数组
    numerical_features = np.array(df['processed_numerical'].tolist())
    labels = df['Value'].values
    
    # 自定义多输入生成器
    def generator():
        while True:
            images = next(image_generator)
            # 获取当前batch的索引
            batch_idx = image_generator.batch_index * batch_size
            if batch_idx >= len(df):
                batch_idx = 0
            # 匹配当前batch的数值特征和标签
            batch_num = numerical_features[batch_idx:batch_idx+batch_size]
            batch_labels = labels[batch_idx:batch_idx+batch_size]
            yield [images, batch_num], batch_labels
    
    steps_per_epoch = len(df) // batch_size
    return generator(), steps_per_epoch

# 加载训练/验证数据
train_data, train_steps = load_multi_input_data(train_df, '[Images folder path]')
test_df = full_df[full_df['split'] == 'validation']
test_data, test_steps = load_multi_input_data(test_df, '[Images folder path]')

3. 构建多输入模型 #

改用Keras函数式API，定义两个输入分支：

def create_model(image_input_shape=(224,224,3), numerical_feature_dim=None):
    # 图像分支
    image_input = Input(shape=image_input_shape)
    backbone = ResNetRS50(input_tensor=image_input, weights='imagenet', include_top=False)
    backbone.trainable = False  # 先冻结预训练权重，后续可微调
    x = backbone.output
    x = Dropout(0.3)(x)
    x = GlobalMaxPooling2D()(x)
    image_features = Dense(128, activation='relu')(x)
    
    # 数值特征分支
    numerical_input = Input(shape=(numerical_feature_dim,))
    y = Dense(64, activation='relu')(numerical_input)
    y = Dropout(0.2)(y)
    numerical_features = Dense(32, activation='relu')(y)
    
    # 拼接双分支特征
    combined = Concatenate()([image_features, numerical_features])
    z = Dense(64, activation='relu')(combined)
    output = Dense(1, activation='linear')(z)
    
    # 构建并编译模型
    model = Model(inputs=[image_input, numerical_input], outputs=output)
    optimizer = Adam(learning_rate=0.0003)
    model.compile(optimizer=optimizer, loss='mae', metrics=['mae'])
    print(model.summary())
    return model

# 获取预处理后数值特征的维度
num_features = len(preprocessor.transform(train_df[continuous_features + categorical_features])[0])
model = create_model(numerical_feature_dim=num_features)

4. 训练多输入模型 #

调整训练函数适配多输入数据：

def train_model(model, train_data, train_steps, test_data, test_steps, epochs=100):
    history = model.fit(
        train_data,
        validation_data=test_data,
        steps_per_epoch=train_steps,
        validation_steps=test_steps,
        epochs=epochs,
        verbose=2
    )
    
    # 可视化损失曲线
    training_loss = history.history['loss']
    test_loss = history.history['val_loss']
    epoch_count = range(1, len(training_loss) + 1)
    
    plt.plot(epoch_count, training_loss, 'r--')
    plt.plot(epoch_count, test_loss, 'b-')
    plt.legend(['训练损失', '验证损失'])
    plt.xlabel('轮次')
    plt.ylabel('损失')
    plt.show()
    
    return model

# 启动训练
model = train_model(model, train_data, train_steps, test_data, test_steps)

关键注意事项 #

• 数据泄露：预处理管道必须仅用训练集数据拟合，绝对不能使用测试集数据，否则会严重影响模型泛化能力。
• 模型微调：若训练后期精度提升缓慢，可解冻ResNet的顶层若干层权重，用更小的学习率（如1e-5）继续训练。
• 特征权重：可通过注意力机制或特征重要性分析，查看数值特征对最终预测结果的贡献程度。

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