用pytorch完整上面的任务

好的，以下是使用PyTorch实现上述任务的完整代码：

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from sklearn.model_selection import train_test_split

# Define the model
class MultiTaskModel(nn.Module):
    def __init__(self):
        super(MultiTaskModel, self).__init__()
        # x1 input layer
        self.x1_input = nn.Linear(31, 128)
        
        # x2 input layer
        self.x2_conv = nn.Conv1d(20, 64, kernel_size=3, padding=1)
        self.x2_flatten = nn.Flatten()
        
        # x3 input layer
        self.x3_conv = nn.Conv1d(30, 64, kernel_size=3, padding=1)
        self.x3_flatten = nn.Flatten()
        
        # shared fully connected layers for output
        self.fc1 = nn.Linear(13184, 128)
        self.fc2 = nn.Linear(128, 64)
        self.output = nn.Linear(64, 4)
        
    def forward(self, x1, x2, x3):
        # x1 input layer
        x1 = self.x1_input(x1)
        x1 = nn.functional.relu(x1)
        
        # x2 convolution layer
        x2 = self.x2_conv(x2)
        x2 = nn.functional.relu(x2)
        x2 = self.x2_flatten(x2)
        
        # x3 convolution layer
        x3 = self.x3_conv(x3)
        x3 = nn.functional.relu(x3)
        x3 = self.x3_flatten(x3)
        
        # concatenate x1, x2, x3
        concat_layer = torch.cat((x1, x2, x3), dim=1)
        
        # fully connected layers for output
        fc1 = self.fc1(concat_layer)
        fc1 = nn.functional.relu(fc1)
        fc2 = self.fc2(fc1)
        fc2 = nn.functional.relu(fc2)
        output = self.output(fc2)
        output = nn.functional.softmax(output, dim=1)
        
        return output

# Define the training function
def train(model, optimizer, criterion, x1, x2, x3, y, batch_size):
    loss_list = []
    acc_list = []
    total_loss = 0
    total_acc = 0
    
    # Set model to training mode
    model.train()
    
    for i in range(0, len(x1), batch_size):
        # Get batch
        x1_batch = x1[i:i+batch_size]
        x2_batch = x2[i:i+batch_size]
        x3_batch = x3[i:i+batch_size]
        y_batch = y[i:i+batch_size]
        
        # Convert to tensors
        x1_tensor = torch.tensor(x1_batch, dtype=torch.float32)
        x2_tensor = torch.tensor(x2_batch, dtype=torch.float32)
        x3_tensor = torch.tensor(x3_batch, dtype=torch.float32)
        y_tensor = torch.tensor(y_batch, dtype=torch.long)
        
        # Zero the gradients
        optimizer.zero_grad()
        
        # Forward pass
        output = model(x1_tensor, x2_tensor, x3_tensor)
        
        # Calculate loss and accuracy
        loss = criterion(output, y_tensor)
        total_loss += loss.item()
        pred = torch.argmax(output, dim=1)
        acc = (pred == y_tensor).sum().item() / batch_size
        total_acc += acc
        
        # Backward pass
        loss.backward()
        optimizer.step()
        
    # Calculate average loss and accuracy
    avg_loss = total_loss / (len(x1) / batch_size)
    avg_acc = total_acc / (len(x1) / batch_size)
    
    loss_list.append(avg_loss)
    acc_list.append(avg_acc)
    
    return avg_loss, avg_acc, loss_list, acc_list

# Define the testing function
def test(model, criterion, x1, x2, x3, y, batch_size):
    loss_list = []
    acc_list = []
    total_loss = 0
    total_acc = 0
    
    # Set model to evaluation mode
    model.eval()
    
    with torch.no_grad():
        for i in range(0, len(x1), batch_size):
            # Get batch
            x1_batch = x1[i:i+batch_size]
            x2_batch = x2[i:i+batch_size]
            x3_batch = x3[i:i+batch_size]
            y_batch = y[i:i+batch_size]

            # Convert to tensors
            x1_tensor = torch.tensor(x1_batch, dtype=torch.float32)
            x2_tensor = torch.tensor(x2_batch, dtype=torch.float32)
            x3_tensor = torch.tensor(x3_batch, dtype=torch.float32)
            y_tensor = torch.tensor(y_batch, dtype=torch.long)

            # Forward pass
            output = model(x1_tensor, x2_tensor, x3_tensor)

            # Calculate loss and accuracy
            loss = criterion(output, y_tensor)
            total_loss += loss.item()
            pred = torch.argmax(output, dim=1)
            acc = (pred == y_tensor).sum().item() / batch_size
            total_acc += acc
        
        # Calculate average loss and accuracy
        avg_loss = total_loss / (len(x1) / batch_size)
        avg_acc = total_acc / (len(x1) / batch_size)
        
        loss_list.append(avg_loss)
        acc_list.append(avg_acc)
        
        return avg_loss, avg_acc, loss_list, acc_list

# Load the data
x1 = np.random.randn(71000, 31)
x2 = np.random.randn(71000, 9, 20)
x3 = np.random.randn(71000, 9, 30)
y = np.random.randint(0, 4, size=(71000,))

# Split the data into training and validation sets
x1_train, x1_val, x2_train, x2_val, x3_train, x3_val, y_train, y_val = train_test_split(x1, x2, x3, y, test_size=0.2)

# Define hyperparameters
batch_size = 128
lr = 0.001
num_epochs = 10

# Create the model, optimizer, and criterion
model = MultiTaskModel()
optimizer = optim.Adam(model.parameters(), lr=lr)
criterion = nn.CrossEntropyLoss()

# Train and test the model
train_loss_list = []
train_acc_list = []
val_loss_list = []
val_acc_list = []

for epoch in range(num_epochs):
    print('Epoch: {}/{}'.format(epoch+1, num_epochs))
    
    # Train the model
    train_loss, train_acc, train_loss_list_epoch, train_acc_list_epoch = train(model, optimizer, criterion, x1_train, x2_train, x3_train, y_train, batch_size)
    train_loss_list += train_loss_list_epoch
    train_acc_list += train_acc_list_epoch
    print('Training loss: {:.4f}, Training accuracy: {:.4f}'.format(train_loss, train_acc))
    
    # Test the model
    val_loss, val_acc, val_loss_list_epoch, val_acc_list_epoch = test(model, criterion, x1_val, x2_val, x3_val, y_val, batch_size)
    val_loss_list += val_loss_list_epoch
    val_acc_list += val_acc_list_epoch
    print('Validation loss: {:.4f}, Validation accuracy: {:.4f}\n'.format(val_loss, val_acc))

在上述代码中，我们首先定义了一个名为`MultiTaskModel`的类，该类继承自`nn.Module`。在类的`__init__`方法中，我们定义了三个输入层，分别对应x1、x2、x3，然后将x2、x3输入到共享的卷积层进行卷积操作。接着，我们将x1、x2、x3连接起来，用全连接层生成目标数据。在`forward`方法中，我们实现了模型的前向传播过程。

接下来，我们定义了`train`函数和`test`函数，分别用于训练和测试模型。在`train`函数中，我们首先将模型设为训练模式，然后将输入数据按batch分割，并将每个batch的数据转换成PyTorch张量。接着，我们将输入数据传递给模型进行前向传播，并计算损失和准确率。最后，我们执行反向传播和参数更新。在`test`函数中，我们首先将模型设为评估模式，然后按照与`train`函数相同的方式计算损失和准确率。

在主函数中，我们首先加载数据，然后将数据分为训练集和验证集。接着，我们定义了一些超参数，包括batch大小、学习率和训练轮数。然后，我们创建了模型、优化器和损失函数，并在每个epoch中训练模型和测试模型。最后，我们将训练和验证的损失和准确率记录下来。