diff --git a/14/2.py b/14/2.py new file mode 100644 index 0000000..9268ae8 --- /dev/null +++ b/14/2.py @@ -0,0 +1,170 @@ +import numpy as np +import torch +import torch.nn.functional as F +from matplotlib import pyplot as plt +from torch.utils.data import DataLoader +from torchvision import datasets, transforms + +# Super parameters ------------------------------------------------------------------------------------ +batch_size = 64 +learning_rate = 0.01 +momentum = 0.5 +EPOCH = 10 + +# Prepare dataset ------------------------------------------------------------------------------------ +transform = transforms.Compose( + [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))] +) + +train_dataset = datasets.MNIST( + root="./data/mnist", train=True, download=True, transform=transform +) +test_dataset = datasets.MNIST( + root="./data/mnist", train=False, download=True, transform=transform +) +train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) +test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False) + + +# Design model using class ------------------------------------------------------------------------------ +class Net(torch.nn.Module): + def __init__(self): + super(Net, self).__init__() + self.fc1 = torch.nn.Linear( + 28 * 28, 512 + ) # Flatten 28x28 input to a vector of 512 units + self.fc2 = torch.nn.Linear(512, 128) # First hidden layer + self.fc3 = torch.nn.Linear( + 128, 10 + ) # Output layer with 10 units (one for each digit) + + def forward(self, x): + x = x.view(-1, 28 * 28) # Flatten the image + x = self.fc1(x) # the first linear layer + x = F.relu(x) # Activation + x = self.fc2(x) # the second linear layer + x = F.relu(x) # Activation + x = self.fc3(x) # No activation on the final output layer (for classification) + return x + + +model = Net() + +# Construct loss and optimizer ---------------------------------------------------------------------- +criterion = ( + torch.nn.CrossEntropyLoss() +) # CrossEntropy loss for multi-class classification +optimizer = torch.optim.SGD( + model.parameters(), lr=learning_rate, momentum=momentum +) # SGD optimizer + + +# Train and Test CLASS ----------------------------------------------------------------------------------- +def train(epoch): + running_loss = 0.0 + running_total = 0 + running_correct = 0 + for batch_idx, data in enumerate(train_loader): + inputs, target = data + optimizer.zero_grad() + + # forward + backward + update + outputs = model(inputs) + loss = criterion(outputs, target) + + loss.backward() + optimizer.step() + + # Accumulate loss and accuracy + running_loss += loss.item() + _, predicted = torch.max(outputs.data, dim=1) + running_total += target.shape[0] + running_correct += (predicted == target).sum().item() + + # Print after each epoch + avg_loss = running_loss / len(train_loader) + avg_acc = 100 * running_correct / running_total + print( + "[%d / %d]: Training Loss: %.3f, Training Accuracy: %.2f %%" + % (epoch + 1, EPOCH, avg_loss, avg_acc) + ) + + +# ... existing code ... + + +def add_noise(images, noise_level=0.5): + """ + 添加噪声函数 + 参数: + images: 输入数据 batch (torch.Tensor) + noise_level: 噪声强度 + 返回: + 添加噪声后的数据 + """ + noise = noise_level * torch.randn_like(images) # 生成随机噪声 + noisy_images = images + noise # 将噪声添加到数据中 + noisy_images = torch.clamp(noisy_images, 0.0, 1.0) # 限制值范围在 [0, 1] + return noisy_images + + +def test(epoch, noise_level=0.2): + correct = 0 + total = 0 + with torch.no_grad(): + for data in test_loader: + images, labels = data + # 添加噪声到测试数据 + noisy_images = add_noise(images, noise_level=noise_level) + outputs = model(noisy_images) + _, predicted = torch.max(outputs.data, dim=1) + total += labels.size(0) + correct += (predicted == labels).sum().item() + acc = correct / total + print( + "[%d / %d]: Accuracy on test set with noise level %.2f after epoch %d: %.1f %%" + % (epoch + 1, EPOCH, noise_level, epoch + 1, 100 * acc) + ) + return 100 * acc + + +# Add EarlyStopping class after imports +class EarlyStopping: + def __init__(self, patience=3, min_delta=0): + self.patience = patience + self.min_delta = min_delta + self.counter = 0 + self.best_acc = None + self.should_stop = False + + def __call__(self, acc): + if self.best_acc is None: + self.best_acc = acc + elif acc <= self.best_acc + self.min_delta: + self.counter += 1 + if self.counter >= self.patience: + self.should_stop = True + else: + self.best_acc = acc + self.counter = 0 + + +if __name__ == "__main__": + acc_list_test_noisy = [] + early_stopping = EarlyStopping(patience=3) + + for epoch in range(EPOCH): + train(epoch) + acc_test_noisy = test(epoch, noise_level=0.8) + acc_list_test_noisy.append(acc_test_noisy) + + # Early stopping check + early_stopping(acc_test_noisy) + if early_stopping.should_stop: + print(f"Early stopping triggered at epoch {epoch + 1}") + break + + plt.plot(range(1, len(acc_list_test_noisy) + 1), acc_list_test_noisy) + plt.xlabel("Epoch") + plt.ylabel("Accuracy On Noisy Test Set") + plt.show()