Блог

Как новичок в Pytorch (из tensorflow), я не уверен в том, как реализовать раннюю остановку. Мои исследования привели меня к выводу, что у pytorch нет собственного способа сделать это. Я также обнаружил torchsample, но по какой-либо причине не могу установить его в своей среде conda. Есть ли простой способ применить раннюю остановку без этого? Вот моя текущая настройка:

 class RegressionDataset(Dataset):
    
    def __init__(self, X_data, y_data):
        self.X_data = X_data
        self.y_data = y_data
        
    def __getitem__(self, index):
        return self.X_data[index], self.y_data[index]
    
    def __len__(self):
        return len(self.X_data)

train_dataset = RegressionDataset(torch.from_numpy(X_train).float(), torch.from_numpy(y_train).float())
val_dataset = RegressionDataset(torch.from_numpy(X_val).float(), torch.from_numpy(y_val).float())
test_dataset = RegressionDataset(torch.from_numpy(X_test).float(), torch.from_numpy(y_test).float())

# Model Params
EPOCHS = 100
BATCH_SIZE = 1000
LEARNING_RATE = 0.001
NUM_FEATURES = np.shape(X_test)[1]

# Initialize Dataloader
train_loader = DataLoader(dataset = train_dataset, batch_size=BATCH_SIZE, shuffle = True)
val_loader = DataLoader(dataset = val_dataset, batch_size=BATCH_SIZE)
test_loader = DataLoader(dataset = test_dataset, batch_size=BATCH_SIZE)

# Define Neural Network Architecture
class MultipleRegression(nn.Module):
    def __init__(self, num_features):
        super(MultipleRegression, self).__init__()
        
        # Define architecture
        self.layer_1 = nn.Linear(num_features, 16)
        self.layer_2 = nn.Linear(16, 32)
        self.layer_3 = nn.Linear(32, 25)
        self.layer_4 = nn.Linear(25, 20)
        self.layer_5 = nn.Linear(20, 16)
        self.layer_out = nn.Linear(16, 1)
        
        self.relu = nn.ReLU() # ReLU applied to all layers
        
        # Initialize weights and biases
        nn.init.xavier_uniform_(self.layer_1.weight)
        nn.init.zeros_(self.layer_1.bias)
        nn.init.xavier_uniform_(self.layer_2.weight)
        nn.init.zeros_(self.layer_2.bias)
        nn.init.xavier_uniform_(self.layer_3.weight)
        nn.init.zeros_(self.layer_3.bias)
        nn.init.xavier_uniform_(self.layer_4.weight)
        nn.init.zeros_(self.layer_4.bias)
        nn.init.xavier_uniform_(self.layer_5.weight)
        nn.init.zeros_(self.layer_5.bias)
        nn.init.xavier_uniform_(self.layer_out.weight)
        nn.init.zeros_(self.layer_out.bias)
        
    def forward(self, inputs):
        x = self.relu(self.layer_1(inputs))
        x = self.relu(self.layer_2(x))
        x = self.relu(self.layer_3(x))
        x = self.relu(self.layer_4(x))
        x = self.relu(self.layer_5(x))
        x = self.layer_out(x)
        return(x)
    
    def predict(self, test_inputs):
        x = self.relu(self.layer_1(test_inputs))
        x = self.relu(self.layer_2(x))
        x = self.relu(self.layer_3(x))
        x = self.relu(self.layer_4(x))
        x = self.relu(self.layer_5(x))
        x = self.layer_out(x)
        return(x)

# Check for GPU
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(device)

model = MultipleRegression(NUM_FEATURES)
model.to(device)
print(model)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr = LEARNING_RATE)

# define dictionary to store loss/epochs for training and validation
loss_stats = {
    "train": [],
    "val": []
    }

# begin training
print("Begin Training")
for e in tqdm(range(1, EPOCHS 1)):
    # Training
    train_epoch_loss = 0
    model.train()

    for X_train_batch, y_train_batch in train_loader:
        X_train_batch, y_train_batch = X_train_batch.to(device), y_train_batch.to(device)
        optimizer.zero_grad()
        
        y_train_pred = model(X_train_batch)
        
        train_loss = criterion(y_train_pred, y_train_batch.unsqueeze(1))
        
        train_loss.backward()
        optimizer.step()
        
        train_epoch_loss  = train_loss.item()
        
        # validation
        with torch.no_grad():
            
            val_epoch_loss = 0
            
            model.eval()
            for X_val_batch, y_val_batch in val_loader:
                X_val_batch, y_val_batch = X_val_batch.to(device), y_val_batch.to(device)
                
                y_val_pred = model(X_val_batch)
                
                val_loss = criterion(y_val_pred, y_val_batch.unsqueeze(1))
                
                val_epoch_loss  = val_loss.item()
    loss_stats["train"].append(train_epoch_loss/len(train_loader))
    loss_stats["val"].append(val_epoch_loss/len(val_loader))
    print(f"Epoch {e}:  Train loss: {train_epoch_loss/len(train_loader):.5f}  Val loss: {val_epoch_loss/len(val_loader):.5f}")

# Visualize loss and accuracy
train_val_loss_df = pd.DataFrame.from_dict(loss_stats).reset_index().melt(id_vars=["index"]).rename(columns = {"index":"epochs"})
plt.figure()
sns.lineplot(data = train_val_loss_df, x = "epochs", y = "value", hue = "variable").set_title("Train-Val Loss/Epoch")

# Test model
y_pred_list = []
with torch.no_grad(): 
    model.eval()
    for X_batch, _ in test_loader:
        X_batch = X_batch.to(device)
        y_test_pred = model(X_batch)
        y_pred_list.append(y_test_pred.cpu().numpy())
y_pred_list = [a.squeeze().tolist() for a in y_pred_list]

y_pred_list = [item for sublist in y_pred_list for item in sublist]
y_pred_list = np.array(y_pred_list)

mse = mean_squared_error(y_test, y_pred_list)
r_square = r2_score(y_test, y_pred_list)
print("Mean Squared Error :", mse)
print("R^2 :", r_square)
 

Основной способ сделать это-отслеживать наилучшие потери при проверке, полученные до сих пор.

Вы можете инициализировать переменную best_loss = 0 перед циклом по эпохам (или вы можете делать другие вещи, например, лучшие потери за эпоху и т. Д.).

После каждого прохождения проверки выполните:

 if val_loss > best_loss:
    best_loss = val_loss
    # At this point also save a snapshot of the current model
    torch.save(model, 'my_model_best_loss.pth')
 

Затем, если результат best_loss не улучшится значительно после некоторого количества тренировочных шагов, или к концу эпохи, или если он val_loss ухудшится, выйдите из цикла и завершите обучение там.

Для реализации таких алгоритмов, как ранняя остановка (и ваш цикл обучения в целом), вам может быть проще попробовать PyTorch Lightning (без привязки, но это намного проще, чем пытаться свернуть все вручную).