• 机器学习的实用程序

    设置

    我们现在必须为可重复性设置很多种子，所以让我们将它们全部包装在一个函数中。

    1. import numpy as np
    2. import pandas as pd
    3. import random
    4. import torch
    5. import torch.nn as nn
    SEED = 1234
    1. def set_seeds(seed=1234):
    2.     """Set seeds for reproducibility."""
    3.     np.random.seed(seed)
    4.     random.seed(seed)
    5.     torch.manual_seed(seed)
    6.     torch.cuda.manual_seed(seed)
    7.     torch.cuda.manual_seed_all(seed) # multi-GPU
    1. # Set seeds for reproducibility
    2. set_seeds(seed=SEED)
    1. # Set device
    2. cuda = True
    3. device = torch.device("cuda" if (
    4.     torch.cuda.is_available() and cuda) else "cpu")
    5. torch.set_default_tensor_type("torch.FloatTensor")
    6. if device.type == "cuda":
    7.     torch.set_default_tensor_type("torch.cuda.FloatTensor")
    8. print (device)

    加载数据

    我们将使用之前课程中相同的螺旋数据集来演示我们的实用程序。

    1. import matplotlib.pyplot as plt
    2. import pandas as pd
    1. # Load data
    2. url = "https://raw.githubusercontent.com/GokuMohandas/Made-With-ML/main/datasets/spiral.csv"
    3. df = pd.read_csv(url, header=0) # load
    4. df = df.sample(frac=1).reset_index(drop=True) # shuffle
    5. df.head()
    X1X2color
    00.1067370.114197c1
    10.311513-0.664028c1
    20.019870-0.703126c1
    3-0.0540170.508159c3
    4-0.127751-0.011382c3
    1. # Data shapes
    2. X = df[["X1", "X2"]].values
    3. y = df["color"].values
    4. print ("X: ", np.shape(X))
    5. print ("y: ", np.shape(y))
    1. X: (1500, 2)
    2. y: (1500,)
    1. # Visualize data
    2. plt.title("Generated non-linear data")
    3. colors = {"c1": "red", "c2": "yellow", "c3": "blue"}
    4. plt.scatter(X[:, 0], X[:, 1], c=[colors[_y] for _y in y], edgecolors="k", s=25)
    5. plt.show()

    拆分数据

    1. import collections
    2. from sklearn.model_selection import train_test_split
    1. TRAIN_SIZE = 0.7
    2. VAL_SIZE = 0.15
    3. TEST_SIZE = 0.15
    1. def train_val_test_split(X, y, train_size):
    2.     """Split dataset into data splits."""
    3.     X_train, X_, y_train, y_ = train_test_split(X, y, train_size=TRAIN_SIZE, stratify=y)
    4.     X_val, X_test, y_val, y_test = train_test_split(X_, y_, train_size=0.5, stratify=y_)
    5.     return X_train, X_val, X_test, y_train, y_val, y_test
    1. # Create data splits
    2. X_train, X_val, X_test, y_train, y_val, y_test = train_val_test_split(
    3.     X=X, y=y, train_size=TRAIN_SIZE)
    4. print (f"X_train: {X_train.shape}, y_train: {y_train.shape}")
    5. print (f"X_val: {X_val.shape}, y_val: {y_val.shape}")
    6. print (f"X_test: {X_test.shape}, y_test: {y_test.shape}")
    7. print (f"Sample point: {X_train[0]} → {y_train[0]}")
    1. X_train: (1050, 2), y_train: (1050,)
    2. X_val: (225, 2), y_val: (225,)
    3. X_test: (225, 2), y_test: (225,)
    4. 采样点：[-0.63919105 -0.69724176] → c1

    标签编码

    接下来，我们将定义 aLabelEncoder将我们的文本标签编码为唯一索引。我们不再使用 scikit-learn 的 LabelEncoder，因为我们希望能够以我们想要的方式保存和加载我们的实例。

    import itertools
    1. class LabelEncoder(object):
    2.     """Label encoder for tag labels."""
    3.     def __init__(self, class_to_index={}):
    4.         self.class_to_index = class_to_index or {}  # mutable defaults ;)
    5.         self.index_to_class = {v: k for k, v in self.class_to_index.items()}
    6.         self.classes = list(self.class_to_index.keys())
    7.  
    8.     def __len__(self):
    9.         return len(self.class_to_index)
    10.  
    11.     def __str__(self):
    12.         return f""
    13.  
    14.     def fit(self, y):
    15.         classes = np.unique(y)
    16.         for i, class_ in enumerate(classes):
    17.             self.class_to_index[class_] = i
    18.         self.index_to_class = {v: k for k, v in self.class_to_index.items()}
    19.         self.classes = list(self.class_to_index.keys())
    20.         return self
    21.  
    22.     def encode(self, y):
    23.         encoded = np.zeros((len(y)), dtype=int)
    24.         for i, item in enumerate(y):
    25.             encoded[i] = self.class_to_index[item]
    26.         return encoded
    27.  
    28.     def decode(self, y):
    29.         classes = []
    30.         for i, item in enumerate(y):
    31.             classes.append(self.index_to_class[item])
    32.         return classes
    33.  
    34.     def save(self, fp):
    35.         with open(fp, "w") as fp:
    36.             contents = {'class_to_index': self.class_to_index}
    37.             json.dump(contents, fp, indent=4, sort_keys=False)
    38.  
    39.     @classmethod
    40.     def load(cls, fp):
    41.         with open(fp, "r") as fp:
    42.             kwargs = json.load(fp=fp)
    43.         return cls(**kwargs)
    1. # Encode
    2. label_encoder = LabelEncoder()
    3. label_encoder.fit(y_train)
    4. label_encoder.class_to_index
    {“c1”：0，“c2”：1，“c3”：2}
    1. # Convert labels to tokens
    2. print (f"y_train[0]: {y_train[0]}")
    3. y_train = label_encoder.encode(y_train)
    4. y_val = label_encoder.encode(y_val)
    5. y_test = label_encoder.encode(y_test)
    6. print (f"y_train[0]: {y_train[0]}")
    1. y_train[0]：c1
    2. y_train[0]：0
    1. # Class weights
    2. counts = np.bincount(y_train)
    3. class_weights = {i: 1.0/count for i, count in enumerate(counts)}
    4. print (f"counts: {counts}\nweights: {class_weights}")
    1. 计数：[350 350 350]
    2. 权重：{0: 0.002857142857142857, 1: 0.002857142857142857, 2: 0.002857142857142857}

    标准化数据

    我们需要标准化我们的数据（零均值和单位方差），以便特定特征的大小不会影响模型如何学习其权重。我们只会对输入 X 进行标准化，因为我们的输出 y 是类值。我们将编写自己的StandardScaler类，以便稍后在推理过程中轻松保存和加载它。

    1. class StandardScaler(object):
    2.     def __init__(self, mean=None, std=None):
    3.         self.mean = np.array(mean)
    4.         self.std = np.array(std)
    5.  
    6.     def fit(self, X):
    7.         self.mean =  np.mean(X_train, axis=0)
    8.         self.std = np.std(X_train, axis=0)
    9.  
    10.     def scale(self, X):
    11.         return (X - self.mean) / self.std
    12.  
    13.     def unscale(self, X):
    14.         return (X * self.std) + self.mean
    15.  
    16.     def save(self, fp):
    17.         with open(fp, "w") as fp:
    18.             contents = {"mean": self.mean.tolist(), "std": self.std.tolist()}
    19.             json.dump(contents, fp, indent=4, sort_keys=False)
    20.  
    21.     @classmethod
    22.     def load(cls, fp):
    23.         with open(fp, "r") as fp:
    24.             kwargs = json.load(fp=fp)
    25.         return cls(**kwargs)
    1. # Standardize the data (mean=0, std=1) using training data
    2. X_scaler = StandardScaler()
    3. X_scaler.fit(X_train)
    1.  
    2. # Apply scaler on training and test data (don't standardize outputs for classification)
    3. X_train = X_scaler.scale(X_train)
    4. X_val = X_scaler.scale(X_val)
    5. X_test = X_scaler.scale(X_test)
    1. # Check (means should be ~0 and std should be ~1)
    2. print (f"X_test[0]: mean: {np.mean(X_test[:, 0], axis=0):.1f}, std: {np.std(X_test[:, 0], axis=0):.1f}")
    3. print (f"X_test[1]: mean: {np.mean(X_test[:, 1], axis=0):.1f}, std: {np.std(X_test[:, 1], axis=0):.1f}")
    1. X_test[0]：平均值：0.1，标准：0.9
    2. X_test[1]：平均值：0.0，标准：1.0

    数据加载器

    我们将把我们的数据放入 aDataset并使用 aDataLoader来有效地创建用于训练和评估的批次。

    1. import torch
    2.  
    3. # Seed seed for reproducibility
    4. torch.manual_seed(SEED)
    1. class Dataset(torch.utils.data.Dataset):
    2.     def __init__(self, X, y):
    3.         self.X = X
    4.         self.y = y
    5.  
    6.     def __len__(self):
    7.         return len(self.y)
    8.  
    9.     def __str__(self):
    10.         return f""
    11.  
    12.     def __getitem__(self, index):
    13.         X = self.X[index]
    14.         y = self.y[index]
    15.         return [X, y]
    16.  
    17.     def collate_fn(self, batch):
    18.         """Processing on a batch."""
    19.         # Get inputs
    20.         batch = np.array(batch)
    21.         X = np.stack(batch[:, 0], axis=0)
    22.         y = batch[:, 1]
    23.  
    24.         # Cast
    25.         X = torch.FloatTensor(X.astype(np.float32))
    26.         y = torch.LongTensor(y.astype(np.int32))
    27.  
    28.         return X, y
    29.  
    30.     def create_dataloader(self, batch_size, shuffle=False, drop_last=False):
    31.         return torch.utils.data.DataLoader(
    32.             dataset=self, batch_size=batch_size, collate_fn=self.collate_fn,
    33.             shuffle=shuffle, drop_last=drop_last, pin_memory=True)

    我们真的不需要collate_fn这里，但我们想让它透明，因为当我们想要对我们的批处理进行特定处理时（例如填充），我们将需要它。

    1. # Create datasets
    2. train_dataset = Dataset(X=X_train, y=y_train)
    3. val_dataset = Dataset(X=X_val, y=y_val)
    4. test_dataset = Dataset(X=X_test, y=y_test)
    5. print ("Datasets:\n"
    6.     f"  Train dataset:{train_dataset.__str__()}\n"
    7.     f"  Val dataset: {val_dataset.__str__()}\n"
    8.     f"  Test dataset: {test_dataset.__str__()}\n"
    9.     "Sample point:\n"
    10.     f"  X: {train_dataset[0][0]}\n"
    11.     f"  y: {train_dataset[0][1]}")

    到目前为止，我们使用批量梯度下降来更新我们的权重。这意味着我们使用整个训练数据集计算了梯度。我们也可以使用随机梯度下降 (SGD) 更新我们的权重，我们一次传入一个训练示例。当前的标准是小批量梯度下降，它在批量和 SGD 之间取得平衡，我们使用 n ( BATCH_SIZE) 个样本的小批量更新权重。这是DataLoader对象派上用场的地方。

    1. # Create dataloaders
    2. batch_size = 64
    3. train_dataloader = train_dataset.create_dataloader(batch_size=batch_size)
    4. val_dataloader = val_dataset.create_dataloader(batch_size=batch_size)
    5. test_dataloader = test_dataset.create_dataloader(batch_size=batch_size)
    6. batch_X, batch_y = next(iter(train_dataloader))
    7. print ("Sample batch:\n"
    8.     f"  X: {list(batch_X.size())}\n"
    9.     f"  y: {list(batch_y.size())}\n"
    10.     "Sample point:\n"
    11.     f"  X: {batch_X[0]}\n"
    12.     f"  y: {batch_y[0]}")
    1. 样品批次：
    2.   X: [64, 2]
    3.   和： [64]
    4. 采样点：
    5.   X：张量（[-1.4736，-1.6742]）
    6.   和：0

    设备

    到目前为止，我们一直在 CPU 上运行我们的操作，但是当我们有大型数据集和更大的模型要训练时，我们可以通过在 GPU 上并行化张量操作而受益。在此笔记本中，您可以通过转到下拉菜单中的Runtime> Change runtime type> 选择来使用 GPU。我们可以使用以下代码行使用什么设备：GPUHardware accelerator

    1. # Set CUDA seeds
    2. torch.cuda.manual_seed(SEED)
    3. torch.cuda.manual_seed_all(SEED) # multi-GPU
    1.  
    2. # Device configuration
    3. device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    4. print (device)

    模型

    让我们初始化我们将用来展示训练实用程序功能的模型。

    1. import math
    2. from torch import nn
    3. import torch.nn.functional as F
    1. INPUT_DIM = X_train.shape[1] # 2D
    2. HIDDEN_DIM = 100
    3. DROPOUT_P = 0.1
    4. NUM_CLASSES = len(label_encoder.classes)
    5. NUM_EPOCHS = 10
    1. class MLP(nn.Module):
    2.     def __init__(self, input_dim, hidden_dim, dropout_p, num_classes):
    3.         super(MLP, self).__init__()
    4.         self.fc1 = nn.Linear(input_dim, hidden_dim)
    5.         self.dropout = nn.Dropout(dropout_p)
    6.         self.fc2 = nn.Linear(hidden_dim, num_classes)
    7.  
    8.     def forward(self, inputs):
    9.         x_in, = inputs
    10.         z = F.relu(self.fc1(x_in))
    11.         z = self.dropout(z)
    12.         z = self.fc2(z)
    13.         return z
    1. # Initialize model
    2. model = MLP(
    3.     input_dim=INPUT_DIM, hidden_dim=HIDDEN_DIM,
    4.     dropout_p=DROPOUT_P, num_classes=NUM_CLASSES)
    5. model = model.to(device) # set device
    6. print (model.named_parameters)

    培训

    到目前为止，我们一直在编写仅使用训练数据拆分进行训练的训练循环，然后我们对测试集进行评估。但实际上，我们会遵循这个过程：

    1. 在训练数据拆分的一个时期使用小批量进行训练。
    2. 评估验证拆分的损失并使用它来调整超参数（例如学习率）。
    3. 训练结束后（通过停滞的改进、期望的性能等），在测试（保留）数据拆分上评估您的训练模型。

    我们将创建一个Trainer类来组织所有这些过程。

    该类中的第一个函数train_step将使用来自训练数据拆分的一个时期的批次训练模型。

    1. def train_step(self, dataloader):
    2.     """Train step."""
    3.     # Set model to train mode
    4.     self.model.train()
    5.     loss = 0.0
    6.  
    7.     # Iterate over train batches
    8.     for i, batch in enumerate(dataloader):
    9.  
    10.         # Step
    11.         batch = [item.to(self.device) for item in batch]  # Set device
    12.         inputs, targets = batch[:-1], batch[-1]
    13.         self.optimizer.zero_grad()  # Reset gradients
    14.         z = self.model(inputs)  # Forward pass
    15.         J = self.loss_fn(z, targets)  # Define loss
    16.         J.backward()  # Backward pass
    17.         self.optimizer.step()  # Update weights
    18.  
    19.         # Cumulative Metrics
    20.         loss += (J.detach().item() - loss) / (i + 1)
    21.  
    22.     return loss

    接下来，我们将定义eval_step将用于处理验证和测试数据拆分的哪个。这是因为它们都不需要梯度更新并显示相同的指标。

    1. def eval_step(self, dataloader):
    2.     """Validation or test step."""
    3.     # Set model to eval mode
    4.     self.model.eval()
    5.     loss = 0.0
    6.     y_trues, y_probs = [], []
    7.  
    8.     # Iterate over val batches
    9.     with torch.inference_mode():
    10.         for i, batch in enumerate(dataloader):
    11.  
    12.             # Step
    13.             batch = [item.to(self.device) for item in batch]  # Set device
    14.             inputs, y_true = batch[:-1], batch[-1]
    15.             z = self.model(inputs)  # Forward pass
    16.             J = self.loss_fn(z, y_true).item()
    17.  
    18.             # Cumulative Metrics
    19.             loss += (J - loss) / (i + 1)
    20.  
    21.             # Store outputs
    22.             y_prob = F.softmax(z).cpu().numpy()
    23.             y_probs.extend(y_prob)
    24.             y_trues.extend(y_true.cpu().numpy())
    25.  
    26.     return loss, np.vstack(y_trues), np.vstack(y_probs)

    最后一个函数是predict_step用于推理的函数。eval_step除了我们不计算任何指标外，它与 非常相似。我们传递可以用来生成性能分数的预测。

    1. def predict_step(self, dataloader):
    2.     """Prediction step."""
    3.     # Set model to eval mode
    4.     self.model.eval()
    5.     y_probs = []
    6.  
    7.     # Iterate over val batches
    8.     with torch.inference_mode():
    9.         for i, batch in enumerate(dataloader):
    10.  
    11.             # Forward pass w/ inputs
    12.             inputs, targets = batch[:-1], batch[-1]
    13.             z = self.model(inputs)
    14.  
    15.             # Store outputs
    16.             y_prob = F.softmax(z).cpu().numpy()
    17.             y_probs.extend(y_prob)
    18.  
    19.     return np.vstack(y_probs)

    LR调度器

    随着我们的模型开始优化并表现更好，损失将减少，我们需要进行较小的调整。如果我们继续使用固定的学习率，我们就会来回过冲。因此，我们将在优化器中添加一个学习率调度程序，以在训练期间调整我们的学习率。有许多调度程序可供选择，但一种流行的调度程序是ReduceLROnPlateau在度量（例如验证损失）停止改进时降低学习率。factor=0.1在下面的示例中，当我们的兴趣指标( ) 连续三个 ( ) 时期self.scheduler.step(val_loss)停止下降 ( ) 时，我们将学习率降低 0.1 ( )。mode="min"patience=3

    1.  
    2. # Initialize the LR scheduler
    3. scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
    4.     optimizer, mode="min", factor=0.1, patience=3)
    5. ...
    6. train_loop():
    7.     ...
    8.     # Steps
    9.     train_loss = trainer.train_step(dataloader=train_dataloader)
    10.     val_loss, _, _ = trainer.eval_step(dataloader=val_dataloader)
    11.     self.scheduler.step(val_loss)
    12.     ...

    提前停止

    我们永远不应该为任意数量的 epoch 训练我们的模型，而是应该有明确的停止标准（即使你被计算资源引导）。常见的停止标准包括验证性能在某些时期 ( patience) 中停滞不前、达到预期性能等。

    1. # Early stopping
    2. if val_loss < best_val_loss:
    3.     best_val_loss = val_loss
    4.     best_model = trainer.model
    5.     _patience = patience  # reset _patience
    6. else:
    7.     _patience -= 1
    8. if not _patience:  # 0
    9.     print("Stopping early!")
    10.     break

    训练

    现在让我们将所有这些放在一起来训练我们的模型。

    from torch.optim import Adam
    1. LEARNING_RATE = 1e-2
    2. NUM_EPOCHS = 100
    3. PATIENCE = 3
    1. # Define Loss
    2. class_weights_tensor = torch.Tensor(list(class_weights.values())).to(device)
    3. loss_fn = nn.CrossEntropyLoss(weight=class_weights_tensor)
    1. # Define optimizer & scheduler
    2. optimizer = Adam(model.parameters(), lr=LEARNING_RATE)
    3. scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
    4.     optimizer, mode="min", factor=0.1, patience=3)
    1. class Trainer(object):
    2.     def __init__(self, model, device, loss_fn=None, optimizer=None, scheduler=None):
    3.  
    4.         # Set params
    5.         self.model = model
    6.         self.device = device
    7.         self.loss_fn = loss_fn
    8.         self.optimizer = optimizer
    9.         self.scheduler = scheduler
    10.  
    11.     def train_step(self, dataloader):
    12.         """Train step."""
    13.         # Set model to train mode
    14.         self.model.train()
    15.         loss = 0.0
    16.  
    17.         # Iterate over train batches
    18.         for i, batch in enumerate(dataloader):
    19.  
    20.             # Step
    21.             batch = [item.to(self.device) for item in batch]  # Set device
    22.             inputs, targets = batch[:-1], batch[-1]
    23.             self.optimizer.zero_grad()  # Reset gradients
    24.             z = self.model(inputs)  # Forward pass
    25.             J = self.loss_fn(z, targets)  # Define loss
    26.             J.backward()  # Backward pass
    27.             self.optimizer.step()  # Update weights
    28.  
    29.             # Cumulative Metrics
    30.             loss += (J.detach().item() - loss) / (i + 1)
    31.  
    32.         return loss
    33.  
    34.     def eval_step(self, dataloader):
    35.         """Validation or test step."""
    36.         # Set model to eval mode
    37.         self.model.eval()
    38.         loss = 0.0
    39.         y_trues, y_probs = [], []
    40.  
    41.         # Iterate over val batches
    42.         with torch.inference_mode():
    43.             for i, batch in enumerate(dataloader):
    44.  
    45.                 # Step
    46.                 batch = [item.to(self.device) for item in batch]  # Set device
    47.                 inputs, y_true = batch[:-1], batch[-1]
    48.                 z = self.model(inputs)  # Forward pass
    49.                 J = self.loss_fn(z, y_true).item()
    50.  
    51.                 # Cumulative Metrics
    52.                 loss += (J - loss) / (i + 1)
    53.  
    54.                 # Store outputs
    55.                 y_prob = F.softmax(z).cpu().numpy()
    56.                 y_probs.extend(y_prob)
    57.                 y_trues.extend(y_true.cpu().numpy())
    58.  
    59.         return loss, np.vstack(y_trues), np.vstack(y_probs)
    60.  
    61.     def predict_step(self, dataloader):
    62.         """Prediction step."""
    63.         # Set model to eval mode
    64.         self.model.eval()
    65.         y_probs = []
    66.  
    67.         # Iterate over val batches
    68.         with torch.inference_mode():
    69.             for i, batch in enumerate(dataloader):
    70.  
    71.                 # Forward pass w/ inputs
    72.                 inputs, targets = batch[:-1], batch[-1]
    73.                 z = self.model(inputs)
    74.  
    75.                 # Store outputs
    76.                 y_prob = F.softmax(z).cpu().numpy()
    77.                 y_probs.extend(y_prob)
    78.  
    79.         return np.vstack(y_probs)
    80.  
    81.     def train(self, num_epochs, patience, train_dataloader, val_dataloader):
    82.         best_val_loss = np.inf
    83.         for epoch in range(num_epochs):
    84.             # Steps
    85.             train_loss = self.train_step(dataloader=train_dataloader)
    86.             val_loss, _, _ = self.eval_step(dataloader=val_dataloader)
    87.             self.scheduler.step(val_loss)
    88.  
    89.             # Early stopping
    90.             if val_loss < best_val_loss:
    91.                 best_val_loss = val_loss
    92.                 best_model = self.model
    93.                 _patience = patience  # reset _patience
    94.             else:
    95.                 _patience -= 1
    96.             if not _patience:  # 0
    97.                 print("Stopping early!")
    98.                 break
    99.  
    100.             # Logging
    101.             print(
    102.                 f"Epoch: {epoch+1} | "
    103.                 f"train_loss: {train_loss:.5f}, "
    104.                 f"val_loss: {val_loss:.5f}, "
    105.                 f"lr: {self.optimizer.param_groups[0]['lr']:.2E}, "
    106.                 f"_patience: {_patience}"
    107.             )
    108.         return best_model
    1.  
    2. # Trainer module
    3. trainer = Trainer(
    4.     model=model, device=device, loss_fn=loss_fn,
    5.     optimizer=optimizer, scheduler=scheduler)
    1. # Train
    2. best_model = trainer.train(
    3.     NUM_EPOCHS, PATIENCE, train_dataloader, val_dataloader)
    1. Epoch: 1 | train_loss: 0.73999, val_loss: 0.58441, lr: 1.00E-02, _patience: 3
    2. Epoch: 2 | train_loss: 0.52631, val_loss: 0.41542, lr: 1.00E-02, _patience: 3
    3. Epoch: 3 | train_loss: 0.40919, val_loss: 0.30673, lr: 1.00E-02, _patience: 3
    4. Epoch: 4 | train_loss: 0.31421, val_loss: 0.22428, lr: 1.00E-02, _patience: 3
    5. ...
    6. Epoch: 48 | train_loss: 0.04100, val_loss: 0.02100, lr: 1.00E-02, _patience: 2
    7. Epoch: 49 | train_loss: 0.04155, val_loss: 0.02008, lr: 1.00E-02, _patience: 3
    8. Epoch: 50 | train_loss: 0.05295, val_loss: 0.02094, lr: 1.00E-02, _patience: 2
    9. Epoch: 51 | train_loss: 0.04619, val_loss: 0.02179, lr: 1.00E-02, _patience: 1
    10. Stopping early!

    评估

    1. import json
    2. from sklearn.metrics import precision_recall_fscore_support
    1. def get_metrics(y_true, y_pred, classes):
    2.     """Per-class performance metrics."""
    3.     # Performance
    4.     performance = {"overall": {}, "class": {}}
    5.  
    6.     # Overall performance
    7.     metrics = precision_recall_fscore_support(y_true, y_pred, average="weighted")
    8.     performance["overall"]["precision"] = metrics[0]
    9.     performance["overall"]["recall"] = metrics[1]
    10.     performance["overall"]["f1"] = metrics[2]
    11.     performance["overall"]["num_samples"] = np.float64(len(y_true))
    12.  
    13.     # Per-class performance
    14.     metrics = precision_recall_fscore_support(y_true, y_pred, average=None)
    15.     for i in range(len(classes)):
    16.         performance["class"][classes[i]] = {
    17.             "precision": metrics[0][i],
    18.             "recall": metrics[1][i],
    19.             "f1": metrics[2][i],
    20.             "num_samples": np.float64(metrics[3][i]),
    21.         }
    22.  
    23.     return performance
    1.  
    2. # Get predictions
    3. test_loss, y_true, y_prob = trainer.eval_step(dataloader=test_dataloader)
    4. y_pred = np.argmax(y_prob, axis=1)
    1.  
    2. # Determine performance
    3. performance = get_metrics(
    4.     y_true=y_test, y_pred=y_pred, classes=label_encoder.classes)
    5. print (json.dumps(performance["overall"], indent=2))
    1. {
    2.   “精度”：0.9956140350877193，
    3.   “召回”：0.9955555555555556，
    4.   “f1”：0.9955553580159119，
    5.   “num_samples”：225.0
    6. }

    保存和加载

    许多教程从未向您展示如何保存您创建的组件，以便您可以加载它们进行推理。

    from pathlib import Path
    1. # Save artifacts
    2. dir = Path("mlp")
    3. dir.mkdir(parents=True, exist_ok=True)
    4. label_encoder.save(fp=Path(dir, "label_encoder.json"))
    5. X_scaler.save(fp=Path(dir, "X_scaler.json"))
    6. torch.save(best_model.state_dict(), Path(dir, "model.pt"))
    7. with open(Path(dir, 'performance.json'), "w") as fp:
    8.     json.dump(performance, indent=2, sort_keys=False, fp=fp)
    1. # Load artifacts
    2. device = torch.device("cpu")
    3. label_encoder = LabelEncoder.load(fp=Path(dir, "label_encoder.json"))
    4. X_scaler = StandardScaler.load(fp=Path(dir, "X_scaler.json"))
    5. model = MLP(
    6.     input_dim=INPUT_DIM, hidden_dim=HIDDEN_DIM,
    7.     dropout_p=DROPOUT_P, num_classes=NUM_CLASSES)
    8. model.load_state_dict(torch.load(Path(dir, "model.pt"), map_location=device))
    9. model.to(device)
    1. MLP(
    2.   (fc1): Linear(in_features=2, out_features=100, bias=True)
    3.   (dropout): Dropout(p=0.1, inplace=False)
    4.   (fc2): Linear(in_features=100, out_features=3, bias=True)
    5. )
    1. # Initialize trainer
    2. trainer = Trainer(model=model, device=device)
    1. # Dataloader
    2. sample = [[0.106737, 0.114197]] # c1
    3. X = X_scaler.scale(sample)
    4. y_filler = label_encoder.encode([label_encoder.classes[0]]*len(X))
    5. dataset = Dataset(X=X, y=y_filler)
    6. dataloader = dataset.create_dataloader(batch_size=batch_size)
    1. # Inference
    2. y_prob = trainer.predict_step(dataloader)
    3. y_pred = np.argmax(y_prob, axis=1)
    4. label_encoder.decode(y_pred)
    [“c1”]

  • 相关阅读:
    flutter（学习日记篇-1）
    关于quartus 13.1出现的问题的一些总结
    无痕 PS、读得懂文字，OpenAI 的二代 DALL·E 惊艳亮相
    MIT6.S081 2021 mmap
    全网最细，真实企业性能测试落地实施，一文带你快速打通...
    《持续交付：发布可靠软件的系统方法》- 读书笔记（六）
    代码提交规范
    分布式与集群区别
    微信聚合聊天，自动回复
    Qt判断一个点在多边形内还是外（支持凸边形和凹变形）
  • 原文地址：https://blog.csdn.net/sikh_0529/article/details/126804898