LeNet实战

m0_62318828

160人浏览 · 2026-06-27 10:51:29

m0_62318828 · 2026-06-27 10:51:29 发布

1.源码

#main
import torch
from torch import nn
from torchsummary import summary

class LeNet(nn.Module):
    def __init__(self):
        super(LeNet,self).__init__()

        self.c1 = nn.Conv2d(in_channels=1,out_channels=6,kernel_size=5,padding=2)
        self.sig = nn.Sigmoid()
        self.s2 = nn.AvgPool2d(kernel_size=2,stride=2)
        self.c3 = nn.Conv2d(in_channels=6,out_channels=16,kernel_size=5)
        self.s4 = nn.AvgPool2d(kernel_size=2,stride=2)

        self.flatten = nn.Flatten()
        self.f5 = nn.Linear(400,120)
        self.f6 = nn.Linear(120, 84)
        self.f7 = nn.Linear(84, 10)
    def forward(self,x):
        x = self.sig(self.c1(x))
        x = self.s2(x)
        x = self.sig(self.c3(x))
        x = self.s4(x)
        x = self.flatten(x)
        x = self.f5(x)
        x = self.f6(x)
        x = self.f7(x)
        return x

if __name__ == "__main__":
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    model = LeNet().to(device)
    print(summary(model,(1, 28, 28)))

#plot
from torchvision.datasets import FashionMNIST
from torchvision import transforms
import torch.utils.data as Data
import numpy as np
import matplotlib.pyplot as plt


train_data = FashionMNIST(root='./data',
                          train = True,
                          transform=transforms.Compose([transforms.Resize(size=224),transforms.ToTensor()]),
                          download= True)

train_loader = Data.DataLoader(dataset=train_data,
                               batch_size=64,
                               shuffle=True,
                               num_workers=0)

for step, (b_x, b_y) in enumerate(train_loader):
    if step > 0:
        break
batch_x = b_x.squeeze().numpy()  # 将四维张量移除第1维，并转换成Numpy数组
batch_y = b_y.numpy()  # 将张量转换成Numpy数组
class_label = train_data.classes  # 训练集的标签
print(class_label)

plt.figure(figsize=(12, 5))
for ii in np.arange(len(batch_y)):
    plt.subplot(4, 16, ii + 1)
    plt.imshow(batch_x[ii, :, :], cmap=plt.cm.gray)
    plt.title(class_label[batch_y[ii]], size=10)
    plt.axis("off")
    plt.subplots_adjust(wspace=0.05)
plt.show()

#model-train
import copy
import time
import torch
import pandas as pd
from model import LeNet
from torchvision.datasets import FashionMNIST
from torchvision import transforms
import torch.utils.data as Data
import numpy as np
import matplotlib.pyplot as plt

import torch.nn as nn



def train_val_data_process():
    train_data = FashionMNIST(root='./data',
                              train=True,
                              transform=transforms.Compose([
                                  transforms.ToTensor()
                                  ]),
                              download=True)

    train_data,val_data = Data.random_split(train_data,[round(0.8*len(train_data)),round(0.2*len(train_data))])

    train_dataloader = Data.DataLoader(dataset=train_data,
                                       batch_size=32,
                                       shuffle=True,
                                       num_workers=2)

    val_dataloader = Data.DataLoader(dataset=val_data,
                                       batch_size=32,
                                       shuffle=True,
                                       num_workers=2)


    return train_dataloader,val_dataloader

def train_model_process(model,train_dataloader,val_dataloader,num_epochs):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    #使用Adam优化器，学习率为0.01
    optimizer = torch.optim.Adam(model.parameters(),lr=0.01)

    #损失函数为交叉熵函数
    criterion = nn.CrossEntropyLoss()
    #模型放入训练设备
    model = model.to(device)
    #复制模型参数
    best_model_wts = copy.deepcopy(model.state_dict())


    #初始化参数，提高准确度
    best_acc = 0.0

    train_loss_all = []

    val_loss_all = []

    train_acc_all = []

    val_acc_all = []

    since = time.time()

    for epoch in range(num_epochs):
        print("Epoch {}/{}".format(epoch,num_epochs-1))
        print("-"*10)



        #初始化参数
        train_loss =0.0

        train_corrects = 0

        val_loss = 0.0

        val_corrects = 0

        train_num = 0

        val_num = 0
        #对每一个mini_batch训练和计算
        for step,(b_x, b_y) in enumerate(train_dataloader):
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.train()

            #前向传播过程，输入为一个batch，输出为一个batch中对应的预测
            output = model(b_x)

            #最大值作为索引
            pre_lab = torch.argmax(output,dim=1)

            loss = criterion(output,b_y)
            #梯度初始化为0
            optimizer.zero_grad()
            #反向传播计算
            loss.backward()
            #梯度下降法更新网络参数
            optimizer.step()

            train_loss+= loss.item()*b_x.size(0)

            train_corrects+= torch.sum(pre_lab == b_y.data)

            train_num += b_x.size(0)

        for step,(b_x, b_y) in enumerate(val_dataloader):
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.eval()
            # 前向传播过程，输入为一个batch，输出为一个batch中对应的预测
            output = model(b_x)

            # 最大值作为索引
            pre_lab = torch.argmax(output, dim=1)

            loss = criterion(output, b_y)

            val_loss += loss.item() * b_x.size(0)

            val_corrects += torch.sum(pre_lab == b_y.data)

            val_num += b_x.size(0)

        #计算并保存每一次迭代的loss值和准确率
        #计算并保存训练集的loss值
        train_loss_all.append(train_loss/train_num)
        #计算并保存训练集准确率
        train_acc_all.append(train_corrects.double().item() / train_num)

        val_loss_all.append(val_loss/val_num)
        val_acc_all.append(val_corrects.double().item() / val_num)

        print("{} train loss:{:.4f} train acc: {:.4f}".format(epoch, train_loss_all[-1], train_acc_all[-1]))
        print("{} val loss:{:.4f} val acc: {:.4f}".format(epoch, val_loss_all[-1], val_acc_all[-1]))

        if val_acc_all[-1] > best_acc:
            # 保存当前最高准确度
            best_acc = val_acc_all[-1]
            # 保存当前最高准确度的模型参数
            best_model_wts = copy.deepcopy(model.state_dict())

            # 计算训练和验证的耗时
        time_use = time.time() - since
        print("训练和验证耗费的时间{:.0f}m{:.0f}s".format(time_use // 60, time_use % 60))

        #选择最优参数
        model.load_state_dict(best_model_wts)
        torch.save(model.state_dict(best_model_wts),"C:/Users/12072/Desktop/LeNet/best_model.pth")

    train_process = pd.DataFrame(data={"epoch": range(num_epochs),
                                        "train_loss_all": train_loss_all,
                                        "val_loss_all": val_loss_all,
                                        "train_acc_all": train_acc_all,
                                        "val_acc_all": val_acc_all, })

    return train_process

def matplot_acc_loss(train_process):
        # 显示每一次迭代后的训练集和验证集的损失函数和准确率
        plt.figure(figsize=(12, 4))
        plt.subplot(1, 2, 1)
        plt.plot(train_process['epoch'], train_process.train_loss_all, "ro-", label="Train loss")
        plt.plot(train_process['epoch'], train_process.val_loss_all, "bs-", label="Val loss")
        plt.legend()
        plt.xlabel("epoch")
        plt.ylabel("Loss")
        plt.subplot(1, 2, 2)
        plt.plot(train_process['epoch'], train_process.train_acc_all, "ro-", label="Train acc")
        plt.plot(train_process['epoch'], train_process.val_acc_all, "bs-", label="Val acc")
        plt.xlabel("epoch")
        plt.ylabel("acc")
        plt.legend()
        plt.show()

if __name__ == '__main__':
        # 加载需要的模型
        LeNetModel = LeNet()
        # 加载数据集
        train_data, val_data = train_val_data_process()
        # 利用现有的模型进行模型的训练
        train_process = train_model_process(LeNetModel, train_data, val_data, num_epochs=20)
        matplot_acc_loss(train_process)

#model-test
import torch
import torch.utils.data as Data
from torchvision import transforms
from torchvision.datasets import FashionMNIST
from model import LeNet



def test_data_process():
    test_data = FashionMNIST(root='./data',
                              train=False,
                              transform=transforms.Compose([transforms.Resize(size=28), transforms.ToTensor()]),
                              download=True)

    test_dataloader = Data.DataLoader(dataset=test_data,
                                       batch_size=1,
                                       shuffle=True,
                                       num_workers=0)
    return test_dataloader


def test_model_process(model, test_dataloader):
    # 设定测试所用到的设备，有GPU用GPU没有GPU用CPU
    device = "cuda" if torch.cuda.is_available() else 'cpu'

    model = model.to(device)

    # 初始化参数
    test_corrects = 0.0
    test_num = 0

    # 只进行前向传播计算，不计算梯度，从而节省内存，加快运行速度
    with torch.no_grad():
        for test_data_x, test_data_y in test_dataloader:
            test_data_x = test_data_x.to(device)
            test_data_y = test_data_y.to(device)
            # 设置模型为评估模式
            model.eval()
            # 前向传播过程，输入为测试数据集，输出为对每个样本的预测值
            output= model(test_data_x)
            # 查找每一行中最大值对应的行标
            pre_lab = torch.argmax(output, dim=1)
            # 如果预测正确，则准确度test_corrects加1
            test_corrects += torch.sum(pre_lab == test_data_y.data)
            # 将所有的测试样本进行累加
            test_num += test_data_x.size(0)

    # 计算测试准确率
    test_acc = test_corrects.double().item() / test_num
    print("测试的准确率为：", test_acc)


if __name__=="__main__":
    # 加载模型
    model = LeNet()
    model.load_state_dict(torch.load('best_model.pth'))
    # 加载测试数据
    test_dataloader = test_data_process()
    # 加载模型测试的函数
    test_model_process(model, test_dataloader)

    # device = "cuda" if torch.cuda.is_available() else 'cpu'
    # model = model.to(device)
    #
    # #classes = FashionMNIST.classes
    # classes = ['T-shirt/top','Trouser','Pullover','Dress','Coat','Sandal','Shirt','Sneaker','Bag','Ankle boot']
    # with torch.no_grad():
    #     for b_x , b_y in test_dataloader:
    #         b_x = b_x.to(device)
    #         b_y = b_y.to(device)
    #
    #         model.eval()
    #         output = model(b_x)
    #         pre_lab = torch.argmax(output,dim=1)
    #         result = pre_lab.item()
    #         label = b_y.item()
    #
    #         print("预测值",classes[result],"--------","真实值",classes[label])

2.结果展示

model-train展示

model-test展示

3.代码分析

3.1model.py

3.1.1 模型建立

model的python文件阐述的就是简单的层次构建，通过Conv2d，AvgPool2d这些函数直接构建而成，在全连接层通过Linear函数（即矩阵乘法 + 偏置加法），直接构成了整个LeNet结构。

class LeNet(nn.Module):
    def __init__(self):
        super(LeNet,self).__init__()

        self.c1 = nn.Conv2d(in_channels=1,out_channels=6,kernel_size=5,padding=2)
        self.sig = nn.Sigmoid()
        self.s2 = nn.AvgPool2d(kernel_size=2,stride=2)
        self.c3 = nn.Conv2d(in_channels=6,out_channels=16,kernel_size=5)
        self.s4 = nn.AvgPool2d(kernel_size=2,stride=2)

        self.flatten = nn.Flatten()
        self.f5 = nn.Linear(400,120)
        self.f6 = nn.Linear(120, 84)
        self.f7 = nn.Linear(84, 10)

这里有个疑问我也询问了AI，全连接层为什么不用sigmoid函数？以下是AI解答仅作参考：

在经典的LeNet-5论文中，全连接层确实使用了激活函数（当时用的是tanh，不是sigmoid，但性质类似）。但不使用激活函数在现代深度学习中是完全正确的。

1.分类任务需要。LeNet用于手写数字识别（MNIST），是多分类问题。最后一层输出10个类别的得分（logits），需要经过Softmax转换为概率分布。如果在最后一层使用Sigmoid：

输出会被压缩到(0,1)区间，破坏了Softmax所需的未归一化得分
实际上最后一层不用激活函数，直接将Softmax放在后面

2.梯度消失问题。Sigmoid函数在两端梯度接近0，如果全连接层也使用Sigmoid：多层Sigmoid堆叠会导致梯度连乘，很快趋近于0，使得浅层网络几乎无法更新。

总结来说，全连接层不加sigmoid是为了保留模型的表达能力，避免梯度消失，并让最后一层能正常使用Softmax。

3.1.2 前向传播（forward）

    def forward(self,x):
        x = self.sig(self.c1(x))
        x = self.s2(x)
        x = self.sig(self.c3(x))
        x = self.s4(x)
        x = self.flatten(x)
        x = self.f5(x)
        x = self.f6(x)
        x = self.f7(x)
        return x

3.2 plot.py

plot的python文件比较公式化，导入数据，加载数据处理器，数据转化可视化等

from torchvision.datasets import FashionMNIST
from torchvision import transforms
import torch.utils.data as Data
import numpy as np
import matplotlib.pyplot as plt


train_data = FashionMNIST(root='./data',
                          train = True,
                          transform=transforms.Compose([transforms.Resize(size=224),transforms.ToTensor()]),
                          download= True)

train_loader = Data.DataLoader(dataset=train_data,
                               batch_size=64,
                               shuffle=True,
                               num_workers=0)

for step, (b_x, b_y) in enumerate(train_loader):
    if step > 0:
        break
batch_x = b_x.squeeze().numpy()  # 将四维张量移除第1维，并转换成Numpy数组
batch_y = b_y.numpy()  # 将张量转换成Numpy数组
class_label = train_data.classes  # 训练集的标签
print(class_label)

plt.figure(figsize=(12, 5))
for ii in np.arange(len(batch_y)):
    plt.subplot(4, 16, ii + 1)
    plt.imshow(batch_x[ii, :, :], cmap=plt.cm.gray)
    plt.title(class_label[batch_y[ii]], size=10)
    plt.axis("off")
    plt.subplots_adjust(wspace=0.05)
plt.show()

3.3 model-train.py

3.3.1 数据加载和划分

def train_val_data_process():
    train_data = FashionMNIST(root='./data',
                              train=True,
                              transform=transforms.Compose([
                                  transforms.ToTensor()
                                  ]),
                              download=True)

    train_data,val_data = Data.random_split(train_data,[round(0.8*len(train_data)),round(0.2*len(train_data))])

    train_dataloader = Data.DataLoader(dataset=train_data,
                                       batch_size=32,
                                       shuffle=True,
                                       num_workers=2)

    val_dataloader = Data.DataLoader(dataset=val_data,
                                       batch_size=32,
                                       shuffle=True,
                                       num_workers=2)


    return train_dataloader,val_dataloader

3.3.2 模型训练

def train_model_process(model,train_dataloader,val_dataloader,num_epochs):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    #使用Adam优化器，学习率为0.01
    optimizer = torch.optim.Adam(model.parameters(),lr=0.01)

    #损失函数为交叉熵函数
    criterion = nn.CrossEntropyLoss()
    #模型放入训练设备
    model = model.to(device)
    #复制模型参数
    best_model_wts = copy.deepcopy(model.state_dict())


    #初始化参数，提高准确度
    best_acc = 0.0

    train_loss_all = []

    val_loss_all = []

    train_acc_all = []

    val_acc_all = []

    since = time.time()

    for epoch in range(num_epochs):
        print("Epoch {}/{}".format(epoch,num_epochs-1))
        print("-"*10)



        #初始化参数
        train_loss =0.0

        train_corrects = 0

        val_loss = 0.0

        val_corrects = 0

        train_num = 0

        val_num = 0
        #对每一个mini_batch训练和计算
        for step,(b_x, b_y) in enumerate(train_dataloader):
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.train()

            #前向传播过程，输入为一个batch，输出为一个batch中对应的预测
            output = model(b_x)

            #最大值作为索引
            pre_lab = torch.argmax(output,dim=1)

            loss = criterion(output,b_y)
            #梯度初始化为0
            optimizer.zero_grad()
            #反向传播计算
            loss.backward()
            #梯度下降法更新网络参数
            optimizer.step()

            train_loss+= loss.item()*b_x.size(0)

            train_corrects+= torch.sum(pre_lab == b_y.data)

            train_num += b_x.size(0)

        for step,(b_x, b_y) in enumerate(val_dataloader):
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.eval()
            # 前向传播过程，输入为一个batch，输出为一个batch中对应的预测
            output = model(b_x)

            # 最大值作为索引
            pre_lab = torch.argmax(output, dim=1)

            loss = criterion(output, b_y)

            val_loss += loss.item() * b_x.size(0)

            val_corrects += torch.sum(pre_lab == b_y.data)

            val_num += b_x.size(0)

        #计算并保存每一次迭代的loss值和准确率
        #计算并保存训练集的loss值
        train_loss_all.append(train_loss/train_num)
        #计算并保存训练集准确率
        train_acc_all.append(train_corrects.double().item() / train_num)

        val_loss_all.append(val_loss/val_num)
        val_acc_all.append(val_corrects.double().item() / val_num)

        print("{} train loss:{:.4f} train acc: {:.4f}".format(epoch, train_loss_all[-1], train_acc_all[-1]))
        print("{} val loss:{:.4f} val acc: {:.4f}".format(epoch, val_loss_all[-1], val_acc_all[-1]))

        if val_acc_all[-1] > best_acc:
            # 保存当前最高准确度
            best_acc = val_acc_all[-1]
            # 保存当前最高准确度的模型参数
            best_model_wts = copy.deepcopy(model.state_dict())

            # 计算训练和验证的耗时
        time_use = time.time() - since
        print("训练和验证耗费的时间{:.0f}m{:.0f}s".format(time_use // 60, time_use % 60))

        #选择最优参数
        model.load_state_dict(best_model_wts)
        torch.save(model.state_dict(best_model_wts),"C:/Users/12072/Desktop/LeNet/best_model.pth")

    train_process = pd.DataFrame(data={"epoch": range(num_epochs),
                                        "train_loss_all": train_loss_all,
                                        "val_loss_all": val_loss_all,
                                        "train_acc_all": train_acc_all,
                                        "val_acc_all": val_acc_all, })

    return train_process

整体其实没什么问题，先对数据进行初始化，分为训练集和验证集，测试集开启训练模式，验证集开启评估模式，反向传播，保存最佳模型，训练过程记录。

但在跟着炮哥敲还是遇到了几个问题，1.缩进问题。由于train-process被错误缩进，导致传给 pd.DataFrame 的各个列表长度不一致，一直报错。2.数据集问题。具体就是全连接层维度不匹配的同时，内存爆炸。修改如下：

原代码：

transform=transforms.Compose([transforms.Resize(size=224), transforms.ToTensor()])

修改后代码：

transform=transforms.Compose([transforms.ToTensor()])

3.这是个优化问题。保存最佳模型我代码中是每一轮循环都进行复制比较，其实可以缩进到外面减少判断，减少时空损耗。

3.4 model-test.py

简单进行判断：

import torch
import torch.utils.data as Data
from torchvision import transforms
from torchvision.datasets import FashionMNIST
from model import LeNet



def test_data_process():
    test_data = FashionMNIST(root='./data',
                              train=False,
                              transform=transforms.Compose([transforms.Resize(size=28), transforms.ToTensor()]),
                              download=True)

    test_dataloader = Data.DataLoader(dataset=test_data,
                                       batch_size=1,
                                       shuffle=True,
                                       num_workers=0)
    return test_dataloader


def test_model_process(model, test_dataloader):
    # 设定测试所用到的设备，有GPU用GPU没有GPU用CPU
    device = "cuda" if torch.cuda.is_available() else 'cpu'

    model = model.to(device)

    # 初始化参数
    test_corrects = 0.0
    test_num = 0

    # 只进行前向传播计算，不计算梯度，从而节省内存，加快运行速度
    with torch.no_grad():
        for test_data_x, test_data_y in test_dataloader:
            test_data_x = test_data_x.to(device)
            test_data_y = test_data_y.to(device)
            # 设置模型为评估模式
            model.eval()
            # 前向传播过程，输入为测试数据集，输出为对每个样本的预测值
            output= model(test_data_x)
            # 查找每一行中最大值对应的行标
            pre_lab = torch.argmax(output, dim=1)
            # 如果预测正确，则准确度test_corrects加1
            test_corrects += torch.sum(pre_lab == test_data_y.data)
            # 将所有的测试样本进行累加
            test_num += test_data_x.size(0)

    # 计算测试准确率
    test_acc = test_corrects.double().item() / test_num
    print("测试的准确率为：", test_acc)


if __name__=="__main__":
    # 加载模型
    model = LeNet()
    model.load_state_dict(torch.load('best_model.pth'))
    # 加载测试数据
    test_dataloader = test_data_process()
    # 加载模型测试的函数
    test_model_process(model, test_dataloader)

具体化显示修改主函数：

if __name__=="__main__":
    # 加载模型
    model = LeNet()
    model.load_state_dict(torch.load('best_model.pth'))
    # 加载测试数据
    test_dataloader = test_data_process()
    # 加载模型测试的函数
    #test_model_process(model, test_dataloader)

    device = "cuda" if torch.cuda.is_available() else 'cpu'
    model = model.to(device)

    #classes = FashionMNIST.classes
    classes = ['T-shirt/top','Trouser','Pullover','Dress','Coat','Sandal','Shirt','Sneaker','Bag','Ankle boot']
    with torch.no_grad():
        for b_x , b_y in test_dataloader:
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.eval()
            output = model(b_x)
            pre_lab = torch.argmax(output,dim=1)
            result = pre_lab.item()
            label = b_y.item()

            print("预测值",classes[result],"--------","真实值",classes[label])