【深度学习】多目标融合算法（二）：底部共享多任务模型（Shared-Bottom Multi-task Model）

导入 pytorch 库
定义 SharedBottomMultiTaskModel 类 继承自 nn.Module:
    定义 __init__ 方法 参数 (self, 输入维度, 隐藏层1大小, 隐藏层2大小, 隐藏层3大小, 输出任务1维度, 输出任务2维度):
        初始化共享底部的三层全连接层
        初始化任务1的三层全连接层
        初始化任务2的三层全连接层

    定义 forward 方法 参数 (self, 输入数据):
        计算输入数据通过共享底部后的输出
        从共享底部输出分别计算任务1和任务2的结果
        返回任务1和任务2的结果

生成虚拟样本数据:
    创建训练集和测试集

实例化模型对象
定义损失函数和优化器
训练循环:
    前向传播: 获取预测值
    计算每个任务的损失
    反向传播和优化

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

class SharedBottomMultiTaskModel(nn.Module):
    def __init__(self, input_dim, hidden1_dim, hidden2_dim, hidden3_dim, output_task1_dim, output_task2_dim):
        super(SharedBottomMultiTaskModel, self).__init__()
        # 定义共享底部的三层全连接层
        self.shared_bottom = nn.Sequential(
            nn.Linear(input_dim, hidden1_dim),
            nn.ReLU(),
            nn.Linear(hidden1_dim, hidden2_dim),
            nn.ReLU(),
            nn.Linear(hidden2_dim, hidden3_dim),
            nn.ReLU()
        )

        # 定义任务1的三层全连接层
        self.task1_head = nn.Sequential(
            nn.Linear(hidden3_dim, hidden2_dim),
            nn.ReLU(),
            nn.Linear(hidden2_dim, output_task1_dim)
        )

        # 定义任务2的三层全连接层
        self.task2_head = nn.Sequential(
            nn.Linear(hidden3_dim, hidden2_dim),
            nn.ReLU(),
            nn.Linear(hidden2_dim, output_task2_dim)
        )

    def forward(self, x):
        # 计算输入数据通过共享底部后的输出
        shared_output = self.shared_bottom(x)

        # 从共享底部输出分别计算任务1和任务2的结果
        task1_output = self.task1_head(shared_output)
        task2_output = self.task2_head(shared_output)

        return task1_output, task2_output

# 构造虚拟样本数据
torch.manual_seed(42)  # 设置随机种子以保证结果可重复
input_dim = 10
task1_dim = 3
task2_dim = 2
num_samples = 1000
X_train = torch.randn(num_samples, input_dim)
y_train_task1 = torch.randn(num_samples, task1_dim)  # 假设任务1的输出维度为task1_dim
y_train_task2 = torch.randn(num_samples, task2_dim)  # 假设任务2的输出维度为task2_dim

# 创建数据加载器
train_dataset = TensorDataset(X_train, y_train_task1, y_train_task2)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)

# 实例化模型对象
model = SharedBottomMultiTaskModel(input_dim, 64, 32, 16, task1_dim, task2_dim)

# 定义损失函数和优化器
criterion_task1 = nn.MSELoss()
criterion_task2 = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 训练循环
num_epochs = 10
for epoch in range(num_epochs):
    model.train()
    running_loss = 0.0

    for batch_idx, (X_batch, y_task1_batch, y_task2_batch) in enumerate(train_loader):
        # 前向传播: 获取预测值
        outputs_task1, outputs_task2 = model(X_batch)

        # 计算每个任务的损失
        loss_task1 = criterion_task1(outputs_task1, y_task1_batch)
        loss_task2 = criterion_task2(outputs_task2, y_task2_batch)
        #print(f'loss_task1:{loss_task1},loss_task2:{loss_task2}')
        total_loss = loss_task1 + loss_task2

        # 反向传播和优化
        optimizer.zero_grad()
        total_loss.backward()
        optimizer.step()

        running_loss += total_loss.item()

    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {running_loss/len(train_loader):.4f}')

# 模型预测
model.eval()
with torch.no_grad():
    test_input = torch.randn(1, input_dim)  # 构造一个测试样本
    pred_task1, pred_task2 = model(test_input)

    print(f'任务1预测结果: {pred_task1}')
    print(f'任务2预测结果: {pred_task2}')