mlp

mlp_cro/net.py

@@ -0,0 +1,29 @@
+from torch import nn
+import torch
+
+class FullyConnectedNet(nn.Module):
+    def __init__(self):
+        super().__init__()
+        # nn.Sequential 会按顺序执行每一层。
+        self.layer = nn.Sequential(
+            nn.Flatten(),             # [batch, 1, 28, 28] -> [batch, 784]
+            nn.Linear(28 * 28, 512),  # 784 个像素点映射到 512 个隐藏特征
+            nn.ReLU(),                # 增加非线性表达能力
+            nn.Linear(512, 256),      # 继续提取更紧凑的隐藏特征
+            nn.ReLU(),
+            nn.Linear(256, 128),
+            nn.ReLU(),
+            nn.Linear(128, 10),       # 输出 10 个数字类别的分数
+        )
+
+    def forward(self, x):
+        # forward 定义“数据如何从输入流到输出”。
+        # 输入 x 是一批图片，输出是每张图片对应 10 个数字类别的 logits。
+        return self.layer(x)
+
+
+if __name__ == '__main__':
+    data = torch.randn(1,1,28,28)
+    net = FullyConnectedNet()
+    output = net(data)
+    print(output)

mlp_cro/test.py

@@ -0,0 +1,61 @@
+import torch  
+from net import FullyConnectedNet  
+from PIL import Image  
+import numpy as np  
+from torchvision import transforms  
+import torch.nn.functional as F 
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+model = FullyConnectedNet()  
+model_path = "D:/code/xwdev/mlp_cro/model/mnist_net_best.pth"  
+
+# 加载训练好的权重到模型中：
+# 1. torch.load(model_path, map_location=device) 从磁盘读取权重字典（state_dict），
+#    map_location=device 保证权重被映射到当前使用的设备上（防止在 GPU 上训练的模型在纯 CPU 机器上加载失败）
+# 2. model.load_state_dict(...) 把读取到的权重参数逐个复制进模型的对应层中
+model.load_state_dict(torch.load(model_path, map_location=device))
+
+model.to(device)  # 把模型的所有参数和缓冲区移动到指定设备上（GPU 或 CPU），之后前向计算就在该设备上进行
+model.eval()  
+# 读取图片并进行预处理
+image_path = "D:/code/xwdev/mlp_cro/mnist_pytorch_samples/image_0000_label_7.png"  
+image = Image.open(image_path)  # 用 Pillow 打开图片文件，得到一个 PIL Image 对象
+
+# 定义图像预处理流水线，transforms.Compose 会把多个变换按顺序串联起来依次执行
+transform = transforms.Compose([
+        transforms.Grayscale(num_output_channels=1),  # 转为灰度图：无论原图是 RGB 还是 RGBA，都强制变成单通道灰度图（MNIST 是灰度图，通道数必须为 1）
+        transforms.Resize((28, 28)),                  # 调整大小为 28x28：MNIST 模型训练时输入就是 28x28，推理时尺寸必须一致
+        transforms.ToTensor(),                        # 转为 tensor：把 PIL 图片转成 PyTorch 张量，形状为 [C, H, W]（这里是 [1, 28, 28]），
+                                                      # 同时把像素值从 0~255 的整数缩放到 0.0~1.0 的浮点数
+    ])
+
+input_tensor = transform(image)  # 对图片执行上面定义的预处理流水线，得到形状为 [1, 28, 28] 的张量（通道、高、宽）
+input_tensor = input_tensor.unsqueeze(0)  # 添加 batch 维度：在第 0 维插入一个大小为 1 的维度，
+                                          # 形状从 [1, 28, 28] 变为 [1, 1, 28, 28]，即 NCHW 格式（批量数、通道、高、宽），
+                                          # 因为模型要求输入是一“批”图片，即使只有一张也要包成 batch
+input_tensor = input_tensor.to(device)  # 把输入张量移动到与模型相同的设备上，设备和模型必须一致，否则计算时会报错
+
+# 使用模型进行预测
+with torch.no_grad():  # 上下文管理器：在其内部所有操作都不会记录梯度信息。
+                       # 推理阶段不需要反向传播，关闭梯度可以节省内存、加快计算
+    output = model(input_tensor)  # 前向传播：把图片张量送入模型，得到输出。
+    output = F.softmax(output, dim=1)  #  Softmax(dim=1)，output 是形状 [1, 10] 的概率分布，
+                                  # 每一维表示该图片属于数字 0~9 的概率，10 个概率之和为 1
+    
+
+    predicted_class = torch.argmax(output, dim=1).item()  # 取概率最大的那一维的索引作为预测类别：
+                                                          # argmax(output, dim=1) 在第 1 维（类别维）上找最大值的下标，返回形状 [1] 的张量
+                                                          # .item() 把只含一个元素的张量转成普通的 Python 整数
+    print(f"Predicted class: {predicted_class}")  # 打印预测结果，例如 "Predicted class: 0"
+
+    # 下面三行是把送入模型的图片显示出来，方便人工核对预处理结果
+    img_display = input_tensor.squeeze().cpu().numpy()  # 把张量变回 NumPy 数组以便转成图片：
+                                                        # squeeze() 去掉所有大小为 1 的维度，[1,1,28,28] -> [28,28]
+                                                        # .cpu() 把张量从 GPU 移回 CPU（NumPy 只能处理 CPU 上的数据）
+                                                        # .numpy() 把 PyTorch 张量转成 NumPy 数组，此时像素值仍是 0.0~1.0 的浮点数
+    image = Image.fromarray((img_display * 255).astype(np.uint8))  # 把 NumPy 数组转回 PIL 图片：
+                                                                   # *255 把像素值从 0~1 还原回 0~255
+                                                                   # astype(np.uint8) 转成 8 位无符号整数（图片标准格式）
+                                                                   # Image.fromarray 根据数组创建灰度图对象
+    image.show()  # 调用系统默认的图片查看器显示这张图片

mlp_cro/train.py

@@ -0,0 +1,96 @@
+
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader
+from torch.utils.tensorboard import SummaryWriter
+from torchvision import datasets, transforms
+from net import FullyConnectedNet
+import torch.nn.functional as F
+import tqdm #让循环在运行时，自动在控制台显示一个动态更新的进度条
+
+# 指定日志目录
+writer = SummaryWriter(log_dir='D:/code/xwdev/mlp_cro/logs')
+
+# 1. 判断是否使用CUDA
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+print(f'Using device: {device}')
+# 2. 准备数据
+train_set = datasets.MNIST(root='D:/code/xwdev/mlp_cro/data', 
+                           train=True, download=True, transform=transforms.ToTensor())
+test_set = datasets.MNIST(root='D:/code/xwdev/mlp_cro/data', 
+                          train=False, download=True, transform=transforms.ToTensor())
+train_loader = DataLoader(dataset=train_set, batch_size=100, shuffle=True)
+test_loader = DataLoader(dataset=test_set, batch_size=100, shuffle=False)
+# 3. 创建模型
+model = FullyConnectedNet()
+model = model.to(device)
+# 4. 确定损失函数
+loss_fn = nn.CrossEntropyLoss()
+# 5. 创建优化器 使用梯度下降算法 参数的更新
+opt = torch.optim.Adam(model.parameters())
+
+# 早停法参数：当测试集损失连续 patience 轮不再变好时，提前终止训练
+patience = 3
+best_test_loss = float('inf')  # 记录到目前为止最好的测试损失
+no_improve_count = 0           # 记录测试损失连续没有变好的轮数
+
+max_epochs = 1000
+for epoch in range(max_epochs):
+    # 6. 训练模型
+    model.train()
+    train_total_loss = 0
+    for images, labels in tqdm.tqdm(train_loader, desc="train", total=len(train_loader)):
+        # 将数据移动到设备
+        images, labels = images.to(device), labels.to(device)
+        outputs = model(images) # 前向传播
+        loss = loss_fn(outputs, labels) # 计算损失
+        #loss = loss_fn(labels, outputs)
+
+        opt.zero_grad() # 清空梯度
+        loss.backward() # 反向传播 计算梯度
+        opt.step() # 更新参数
+        train_total_loss += loss.item()
+    train_avg_loss = train_total_loss / len(train_loader)
+    print(f'Epoch {epoch+1}/{max_epochs}, Loss: {train_avg_loss:.4f}')
+    # 7. 测试模型
+    model.eval()
+    test_total_loss = 0
+    test_total_acc = 0
+    #禁用梯度计算
+    #在推理时，我们不需要反向传播，因此不需要计算损失函数对参数的梯度
+    with torch.inference_mode(): 
+        for images, labels in tqdm.tqdm(test_loader, desc="test", total=len(test_loader)):
+            images, labels = images.to(device), labels.to(device)
+            outputs = model(images)
+            # 计算损失
+            loss = loss_fn(outputs, labels)
+            test_total_loss += loss.item()
+
+            pred = torch.argmax(outputs, dim=1)
+            acc = torch.eq(pred, labels).float().mean()
+            test_total_acc += acc.item()
+
+    test_avg_acc = test_total_acc / len(test_loader)
+    print(f'epoch:{epoch+1}, Test Accuracy: {test_avg_acc:.4f}')
+
+    test_avg_loss = test_total_loss / len(test_loader)
+    print(f'epoch:{epoch+1},Test Loss: {test_avg_loss:.4f}')
+    # 记录数据
+    writer.add_scalars('Loss/train', {'train_avg_loss': train_avg_loss,
+    'test_avg_loss': test_avg_loss}, epoch)
+    writer.add_scalar('Accuracy/test', test_avg_acc, epoch)
+    # 8. 早停判断与模型保存
+    if test_avg_loss < best_test_loss:
+        # 测试损失变好了：更新最好成绩，重置计数器，保存当前最优模型
+        best_test_loss = test_avg_loss
+        no_improve_count = 0
+        torch.save(model.state_dict(), 'D:/code/xwdev/mlp_cro/model/mnist_net_best.pth')
+        print(f'测试损失改善，保存最优模型 (best loss: {best_test_loss:.4f})')
+    else:
+        # 测试损失没有变好：计数器加一
+        no_improve_count += 1
+        print(f'测试损失未改善 ({no_improve_count}/{patience})')
+        if no_improve_count >= patience:
+            # 连续 patience 轮没有变好，提前终止训练
+            print(f'早停：测试损失连续 {patience} 轮未改善，在第 {epoch+1} 轮终止训练')
+            break

mlp_mse/net.py

@@ -0,0 +1,32 @@
+from torch import nn
+import torch
+
+class FullyConnectedNet(nn.Module):
+    def __init__(self):
+        super().__init__()
+        # nn.Sequential 会按顺序执行每一层。
+        self.layer = nn.Sequential(
+            nn.Flatten(),             # [batch, 1, 28, 28] -> [batch, 784]
+            nn.Linear(28 * 28, 512),  # 784 个像素点映射到 512 个隐藏特征
+            nn.ReLU(),                # 增加非线性表达能力
+            nn.Linear(512, 256),      # 继续提取更紧凑的隐藏特征
+            nn.ReLU(),
+            nn.Linear(256, 128),
+            nn.ReLU(),
+            nn.Linear(128, 10),       # 输出 10 个数字类别的分数
+            
+            #输出函数
+            nn.Softmax(dim=1)  # NV结构，激活V
+        )
+
+    def forward(self, x):
+        # forward 定义“数据如何从输入流到输出”。
+        # 输入 x 是一批图片，输出是每张图片对应 10 个数字类别的 logits。
+        return self.layer(x)
+
+
+if __name__ == '__main__':
+    data = torch.randn(1,1,28,28)
+    net = FullyConnectedNet()
+    output = net(data)
+    print(output)

mlp_mse/test.py

@@ -0,0 +1,61 @@
+import torch  
+from net import FullyConnectedNet  
+from PIL import Image  
+import numpy as np  
+from torchvision import transforms  
+import torch.nn.functional as F 
+
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+model = FullyConnectedNet()  #模型结构，参数是随机初始化的，和训练好的模型权重没有关系
+model_path = "./mlp_mse/model/mnist_net_best.pth"  
+
+# 加载训练好的权重到模型中：
+# 1. torch.load(model_path, map_location=device) 从磁盘读取权重字典（state_dict），
+#    map_location=device 保证权重被映射到当前使用的设备上（防止在 GPU 上训练的模型在纯 CPU 机器上加载失败）
+# 2. model.load_state_dict(...) 把读取到的权重参数逐个复制进模型的对应层中
+model.load_state_dict(torch.load(model_path, map_location=device))
+
+model.to(device)  # 把模型的所有参数和缓冲区移动到指定设备上（GPU 或 CPU），之后前向计算就在该设备上进行
+model.eval()  
+# 读取图片并进行预处理
+image_path = "D:/code/xwdev/mlp_mse/mnist_pytorch_samples/image_0003_label_0.png"  
+image = Image.open(image_path)  # 用 Pillow 打开图片文件，得到一个 PIL Image 对象
+
+# 定义图像预处理流水线，transforms.Compose 会把多个变换按顺序串联起来依次执行
+transform = transforms.Compose([
+        transforms.Grayscale(num_output_channels=1),  # 转为灰度图：无论原图是 RGB 还是 RGBA，都强制变成单通道灰度图（MNIST 是灰度图，通道数必须为 1）
+        transforms.Resize((28, 28)),                  # 调整大小为 28x28：MNIST 模型训练时输入就是 28x28，推理时尺寸必须一致
+        transforms.ToTensor(),                        # 转为 tensor：把 PIL 图片转成 PyTorch 张量，形状为 [C, H, W]（这里是 [1, 28, 28]），
+                                                      # 同时把像素值从 0~255 的整数缩放到 0.0~1.0 的浮点数
+    ])
+
+input_tensor = transform(image)  # 对图片执行上面定义的预处理流水线，得到形状为 [1, 28, 28] 的张量（通道、高、宽）
+input_tensor = input_tensor.unsqueeze(0)  # 添加 batch 维度：在第 0 维插入一个大小为 1 的维度，
+                                          # 形状从 [1, 28, 28] 变为 [1, 1, 28, 28]，即 NCHW 格式（批量数、通道、高、宽），
+                                          # 因为模型要求输入是一“批”图片，即使只有一张也要包成 batch
+input_tensor = input_tensor.to(device)  # 把输入张量移动到与模型相同的设备上，设备和模型必须一致，否则计算时会报错
+
+# 使用模型进行预测
+with torch.no_grad():  # 上下文管理器：在其内部所有操作都不会记录梯度信息。
+                       # 推理阶段不需要反向传播，关闭梯度可以节省内存、加快计算
+    output = model(input_tensor)  # 前向传播：把图片张量送入模型，得到输出。
+                                  # 由于网络最后一层是 Softmax(dim=1)，output 是形状 [1, 10] 的概率分布，
+                                  # 每一维表示该图片属于数字 0~9 的概率，10 个概率之和为 1
+
+    predicted_class = torch.argmax(output, dim=1).item()  # 取概率最大的那一维的索引作为预测类别：
+                                                          # argmax(output, dim=1) 在第 1 维（类别维）上找最大值的下标，返回形状 [1] 的张量
+                                                          # .item() 把只含一个元素的张量转成普通的 Python 整数
+    print(f"Predicted class: {predicted_class}")  # 打印预测结果，例如 "Predicted class: 0"
+
+    # 下面三行是把送入模型的图片显示出来，方便人工核对预处理结果
+    img_display = input_tensor.squeeze().cpu().numpy()  # 把张量变回 NumPy 数组以便转成图片：
+                                                        # squeeze() 去掉所有大小为 1 的维度，[1,1,28,28] -> [28,28]
+                                                        # .cpu() 把张量从 GPU 移回 CPU（NumPy 只能处理 CPU 上的数据）
+                                                        # .numpy() 把 PyTorch 张量转成 NumPy 数组，此时像素值仍是 0.0~1.0 的浮点数
+    image = Image.fromarray((img_display * 255).astype(np.uint8))  # 把 NumPy 数组转回 PIL 图片：
+                                                                   # *255 把像素值从 0~1 还原回 0~255
+                                                                   # astype(np.uint8) 转成 8 位无符号整数（图片标准格式）
+                                                                   # Image.fromarray 根据数组创建灰度图对象
+    image.show()  # 调用系统默认的图片查看器显示这张图片

mlp_mse/train.py

@@ -0,0 +1,101 @@
+
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader
+from torch.utils.tensorboard import SummaryWriter
+from torchvision import datasets, transforms
+from net import FullyConnectedNet
+import torch.nn.functional as F
+import tqdm #让循环在运行时，自动在控制台显示一个动态更新的进度条
+
+# 指定日志目录
+writer = SummaryWriter(log_dir='./mlp/logs')
+
+# 1. 判断是否使用CUDA
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+print(f'Using device: {device}')
+# 2. 准备数据,读数据，数据预处理
+train_set = datasets.MNIST(root='./mlp/data', 
+                           train=True, download=True, transform=transforms.ToTensor())
+test_set = datasets.MNIST(root='./mlp/data', 
+                          train=False, download=True, transform=transforms.ToTensor())
+
+#显存，CUDA out of memory 错误，可能是因为 batch_size 太大了，导致显存不足。可以尝试减小 batch_size。
+train_loader = DataLoader(dataset=train_set, batch_size=100, shuffle=True)
+test_loader = DataLoader(dataset=test_set, batch_size=100, shuffle=False)
+# 3. 创建模型
+model = FullyConnectedNet()
+model = model.to(device)
+# 4. 确定损失函数
+loss_fn = nn.MSELoss()
+# 5. 创建优化器 使用梯度下降算法 参数的更新
+opt = torch.optim.Adam(model.parameters())
+
+# 早停法参数：当测试集损失连续 patience 轮不再变好时，提前终止训练
+patience = 5
+best_test_loss = float('inf')  # 记录到目前为止最好的测试损失
+no_improve_count = 0           # 记录测试损失连续没有变好的轮数
+
+max_epochs = 1000
+#轮次， 训练集 60000 张图片，batch_size=100, 60000/100=600
+for epoch in range(max_epochs):
+    # 6. 训练模型
+    model.train()
+    train_total_loss = 0
+    for images, labels in tqdm.tqdm(train_loader, desc="train", total=len(train_loader)):
+        # 将数据移动到设备
+        images, labels = images.to(device), labels.to(device)
+        labels = F.one_hot(labels, num_classes=10).float()
+        outputs = model(images) # 前向传播
+        loss = loss_fn(outputs, labels) # 计算损失
+
+        opt.zero_grad() # 清空梯度
+        loss.backward() # 反向传播 计算梯度
+        opt.step() # 更新参数
+        train_total_loss += loss.item()
+    train_avg_loss = train_total_loss / len(train_loader)
+    print(f'Epoch {epoch+1}/{max_epochs}, Loss: {train_avg_loss:.4f}')
+    # 7. 测试模型
+    model.eval()
+    test_total_loss = 0 #一轮测试的总损失
+    test_total_acc = 0 #一轮测试的总得分
+    #禁用梯度计算
+    #在推理时，我们不需要反向传播，因此不需要计算损失函数对参数的梯度
+    with torch.inference_mode(): 
+        for images, labels in tqdm.tqdm(test_loader, desc="test", total=len(test_loader)):
+            images, labels = images.to(device), labels.to(device)
+            labels = F.one_hot(labels, num_classes=10).float()
+            outputs = model(images)
+            # 计算损失
+            loss = loss_fn(outputs, labels)
+            test_total_loss += loss.item()
+
+            pred = torch.argmax(outputs, dim=1)
+            target = torch.argmax(labels, dim=1)
+            acc = torch.eq(pred, target).float().mean()
+            test_total_acc += acc.item()
+
+    test_avg_acc = test_total_acc / len(test_loader)
+    print(f'epoch:{epoch+1}, Test Accuracy: {test_avg_acc:.4f}')
+
+    test_avg_loss = test_total_loss / len(test_loader)
+    print(f'epoch:{epoch+1},Test Loss: {test_avg_loss:.4f}')
+    # 记录数据
+    writer.add_scalars('Loss/train', {'train_avg_loss': train_avg_loss,
+    'test_avg_loss': test_avg_loss}, epoch)
+    writer.add_scalar('Accuracy/test', test_avg_acc, epoch)
+    # 8. 早停判断与模型保存
+    if test_avg_loss < best_test_loss:
+        # 测试损失变好了：更新最好成绩，重置计数器，保存当前最优模型
+        best_test_loss = test_avg_loss
+        no_improve_count = 0
+        torch.save(model.state_dict(), './mlp/model/mnist_net_best.pth')
+        print(f'测试损失改善，保存最优模型 (best loss: {best_test_loss:.4f})')
+    else:
+        # 测试损失没有变好：计数器加一
+        no_improve_count += 1
+        print(f'测试损失未改善 ({no_improve_count}/{patience})')
+        if no_improve_count >= patience:
+            # 连续 patience 轮没有变好，提前终止训练
+            print(f'早停：测试损失连续 {patience} 轮未改善，在第 {epoch+1} 轮终止训练')
+            break