【深度学习实验】卷积神经网络（六）：自定义卷积神经网络模型（VGG）实现图片多分类任务

Qomolangma

发布于 2024-07-30 08:35:04

1640

发布于 2024-07-30 08:35:04

文章被收录于专栏：深度学习

一、实验介绍

本实验实现了一个简化版VGG网络，并基于此完成图像分类任务。

VGG网络是深度卷积神经网络中的经典模型之一，由牛津大学计算机视觉组（Visual Geometry Group）提出。它在2014年的ImageNet图像分类挑战中取得了优异的成绩（分类任务第二，定位任务第一），被广泛应用于图像分类、目标检测和图像生成等任务。 VGG网络的主要特点是使用了非常小的卷积核尺寸（通常为3x3）和更深的网络结构。该网络通过多个卷积层和池化层堆叠在一起，逐渐增加网络的深度，从而提取图像的多层次特征表示。VGG网络的基本构建块是由连续的卷积层组成，每个卷积层后面跟着一个ReLU激活函数。在每个卷积块的末尾，都会添加一个最大池化层来减小特征图的尺寸。VGG网络的这种简单而有效的结构使得它易于理解和实现，并且在不同的任务上具有很好的泛化性能。 VGG网络有几个不同的变体，如VGG11、VGG13、VGG16和VGG19，它们的数字代表网络的层数。这些变体在网络深度和参数数量上有所区别，较深的网络通常具有更强大的表示能力，但也更加复杂。

二、实验环境

本系列实验使用了PyTorch深度学习框架，相关操作如下：

1. 配置虚拟环境

conda create -n DL python=3.7

conda activate DL

pip install torch==1.8.1+cu102 torchvision==0.9.1+cu102 torchaudio==0.8.1 -f https://download.pytorch.org/whl/torch_stable.html

conda install matplotlib

 conda install scikit-learn

2. 库版本介绍

软件包	本实验版本	目前最新版
matplotlib	3.5.3	3.8.0
numpy	1.21.6	1.26.0
python	3.7.16
scikit-learn	0.22.1	1.3.0
torch	1.8.1+cu102	2.0.1
torchaudio	0.8.1	2.0.2
torchvision	0.9.1+cu102	0.15.2

三、实验内容

ChatGPT：

卷积神经网络（Convolutional Neural Network，简称CNN）是一种深度学习模型，广泛应用于图像识别、计算机视觉和模式识别等领域。它的设计灵感来自于生物学中视觉皮层的工作原理。卷积神经网络通过多个卷积层、池化层和全连接层组成。

卷积层主要用于提取图像的局部特征，通过卷积操作和激活函数的处理，可以学习到图像的特征表示。
池化层则用于降低特征图的维度，减少参数数量，同时保留主要的特征信息。
全连接层则用于将提取到的特征映射到不同类别的概率上，进行分类或回归任务。

卷积神经网络在图像处理方面具有很强的优势，它能够自动学习到具有层次结构的特征表示，并且对平移、缩放和旋转等图像变换具有一定的不变性。这些特点使得卷积神经网络成为图像分类、目标检测、语义分割等任务的首选模型。除了图像处理，卷积神经网络也可以应用于其他领域，如自然语言处理和时间序列分析。通过将文本或时间序列数据转换成二维形式，可以利用卷积神经网络进行相关任务的处理。

0. 导入必要的工具包

import torch 
from torch import nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from torchvision.io import read_image
import matplotlib.pyplot as plt
import os

1. 构建数据集（CIFAR10Dataset）

a. read_csv_labels（）

从CSV文件中读取标签信息并返回一个标签字典。

def read_csv_labels(fname):
    """读取fname来给标签字典返回一个文件名"""
    with open(fname, 'r') as f:
        # 跳过文件头行(列名)
        lines = f.readlines()[1:]
    tokens = [l.rstrip().split(',') for l in lines]
    return dict(((name, label) for name, label in tokens))

使用open函数打开指定文件名的CSV文件，并将文件对象赋值给变量f。这里使用'r'参数以只读模式打开文件。
使用文件对象的readlines()方法读取文件的所有行，并将结果存储在名为lines的列表中。通过切片操作[1:]，跳过了文件的第一行（列名），将剩余的行存储在lines列表中。
列表推导式（list comprehension）：对lines列表中的每一行进行处理。对于每一行，使用rstrip()方法去除行末尾的换行符，并使用split(',')方法将行按逗号分割为多个标记。最终，将所有行的标记组成的子列表存储在tokens列表中。
使用字典推导式（dictionary comprehension）将tokens列表中的子列表转换为字典。对于tokens中的每个子列表，将子列表的第一个元素作为键（name），第二个元素作为值（label），最终返回一个包含这些键值对的字典。

b. CIFAR10Dataset

class CIFAR10Dataset(Dataset):
    def __init__(self, folder_path, fname):
        self.labels = read_csv_labels(os.path.join(folder_path, fname))
        self.folder_path = os.path.join(folder_path, 'train')

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, idx):
        img = read_image(self.folder_path + '/' + str(idx + 1) + '.png')
        label = self.labels[str(idx + 1)]

        return img, torch.tensor(int(label))

构造函数：
- 接受两个参数
  - folder_path表示数据集所在的文件夹路径
  - fname表示包含标签信息的文件名。
- 调用read_csv_labels函数，传递folder_path和fname作为参数，以读取CSV文件中的标签信息，并将返回的标签字典存储在self.labels变量中。
- 通过拼接folder_path和字符串'train'来构建数据集的文件夹路径，将结果存储在self.folder_path变量中。
def __len__(self)
- 这是CIFAR10Dataset类的方法，用于返回数据集的长度，即样本的数量。
def __getitem__(self, idx): 这是CIFAR10Dataset类的方法，用于根据给定的索引idx获取数据集中的一个样本。它首先根据索引idx构建图像文件的路径，并调用read_image函数来读取图像数据，将结果存储在img变量中。然后，它通过将索引转换为字符串，并使用该字符串作为键来从self.labels字典中获取相应的标签，将结果存储在label变量中。最后，它返回一个元组，包含图像数据和经过torch.tensor转换的标签。

2. 构建模型（FeedForward）

参考前文：

【深度学习实验】卷积神经网络（五）：深度卷积神经网络经典模型——VGG网络（卷积层、池化层、全连接层）_QomolangmaH的博客-CSDN博客

https://blog.csdn.net/m0_63834988/article/details/133350927?spm=1001.2014.3001.5501

#  每个卷积块由Conv2d卷积 + BatchNorm2d(批量标准化处理) + ReLU激活层组成
def conv_layer(chann_in, chann_out, k_size, p_size):
    layer = nn.Sequential(
        nn.Conv2d(chann_in, chann_out, kernel_size=k_size, padding=p_size),
        nn.BatchNorm2d(chann_out),
        nn.ReLU()
    )
    return layer
    
# vgg卷积模块是由几个相同的卷积块以及最大池化组成
def vgg_conv_block(in_list, out_list, k_list, p_list, pooling_k, pooling_s):

    layers = [conv_layer(in_list[i], out_list[i], k_list[i], p_list[i]) for i in range(len(in_list)) ]
    layers += [nn.MaxPool2d(kernel_size = pooling_k, stride = pooling_s)]
    return nn.Sequential(*layers)

# vgg全连接层由Linear + BatchNorm1d + ReLU组成
def vgg_fc_layer(size_in, size_out):
    layer = nn.Sequential(
        nn.Linear(size_in, size_out),
        nn.BatchNorm1d(size_out),
        nn.ReLU()
    )
    return layer


# 为了简化，我们少使用了几层卷积层，方便大家使用
class VGG_S(nn.Module):
    def __init__ (self, num_classes):
        super().__init__()
        
        self.layer1 = vgg_conv_block([3,64], [64,64], [3,3], [1,1], 2, 2)   
        self.layer2 = vgg_conv_block([64,128], [128,128], [3,3], [1,1], 2, 2)
        self.layer3 = vgg_conv_block([128,256,256], [256,256,256], [3,3,3], [1,1,1], 2, 2)

        # 全连接层
        self.layer4 = vgg_fc_layer(4096, 1024)
        # Final layer
        self.layer5 = nn.Linear(1024, num_classes)
        
    def forward(self, x):
        out = self.layer1(x)
        out = self.layer2(out)
        vgg16_features = self.layer3(out)
        out = vgg16_features.view(out.size(0), -1)
        out = self.layer4(out)
        out = self.layer5(out)

        return out

3.整合训练、评估、预测过程（Runner）

参考前文：

【深度学习实验】前馈神经网络（九）：整合训练、评估、预测过程（Runner）_QomolangmaH的博客-CSDN博客

https://blog.csdn.net/m0_63834988/article/details/133219448?spm=1001.2014.3001.5501

（略有改动：）

class Runner(object):
    def __init__(self, model, optimizer, loss_fn, metric=None):
        self.model = model
        self.optimizer = optimizer
        self.loss_fn = loss_fn
        # 用于计算评价指标
        self.metric = metric
        
        # 记录训练过程中的评价指标变化
        self.dev_scores = []
        # 记录训练过程中的损失变化
        self.train_epoch_losses = []
        self.dev_losses = []
        # 记录全局最优评价指标
        self.best_score = 0
   
 
# 模型训练阶段
    def train(self, train_loader, dev_loader=None, **kwargs):
        # 将模型设置为训练模式，此时模型的参数会被更新
        self.model.train()
        
        num_epochs = kwargs.get('num_epochs', 0)
        log_steps = kwargs.get('log_steps', 100)
        save_path = kwargs.get('save_path','best_model.pth')
        eval_steps = kwargs.get('eval_steps', 0)
        # 运行的step数，不等于epoch数
        global_step = 0
        
        if eval_steps:
            if dev_loader is None:
                raise RuntimeError('Error: dev_loader can not be None!')
            if self.metric is None:
                raise RuntimeError('Error: Metric can not be None')
                
        # 遍历训练的轮数
        for epoch in range(num_epochs):
            total_loss = 0
            # 遍历数据集
            for step, data in enumerate(train_loader):
                x, y = data
                logits = self.model(x.float())
                loss = self.loss_fn(logits, y.long())
                total_loss += loss
                if step%log_steps == 0:
                    print(f'loss:{loss.item():.5f}')
                    
                loss.backward()
                self.optimizer.step()
                self.optimizer.zero_grad()
            # 每隔一定轮次进行一次验证，由eval_steps参数控制，可以采用不同的验证判断条件
            if eval_steps != 0 :
                if (epoch+1) % eval_steps ==  0:

                    dev_score, dev_loss = self.evaluate(dev_loader, global_step=global_step)
                    print(f'[Evalute] dev score:{dev_score:.5f}, dev loss:{dev_loss:.5f}')
                
                    if dev_score > self.best_score:
                        self.save_model(f'model_{epoch+1}.pth')
                    
                        print(f'[Evaluate]best accuracy performance has been updated: {self.best_score:.5f}-->{dev_score:.5f}')
                        self.best_score = dev_score
                    
                # 验证过程结束后，请记住将模型调回训练模式   
                    self.model.train()
            
            global_step += 1
            # 保存当前轮次训练损失的累计值
            train_loss = (total_loss/len(train_loader)).item()
            self.train_epoch_losses.append((global_step,train_loss))
        self.save_model(f'{save_path}.pth')   
        print('[Train] Train done')
        
    # 模型评价阶段
    def evaluate(self, dev_loader, **kwargs):
        assert self.metric is not None
        # 将模型设置为验证模式，此模式下，模型的参数不会更新
        self.model.eval()
        global_step = kwargs.get('global_step',-1)
        total_loss = 0
        self.metric.reset()
        
        for batch_id, data in enumerate(dev_loader):
            x, y = data
            logits = self.model(x.float())
            loss = self.loss_fn(logits, y.long()).item()
            total_loss += loss 
            self.metric.update(logits, y)
            
        dev_loss = (total_loss/len(dev_loader))
        self.dev_losses.append((global_step, dev_loss))
        dev_score = self.metric.accumulate()
        self.dev_scores.append(dev_score)
        return dev_score, dev_loss
    
    # 模型预测阶段，
    def predict(self, x, **kwargs):
        self.model.eval()
        logits = self.model(x)
        return logits
    
    # 保存模型的参数
    def save_model(self, save_path):
        torch.save(self.model.state_dict(), save_path)
        
    # 读取模型的参数
    def load_model(self, model_path):
        self.model.load_state_dict(torch.load(model_path, map_location=torch.device('cpu')))

4. main

if __name__ == '__main__':
    batch_size = 20
    # 构建训练集
    train_data = CIFAR10Dataset('cifar10_tiny', 'trainLabels.csv')
    train_iter = DataLoader(train_data, batch_size=batch_size)
    # 构建测试集
    test_data = CIFAR10Dataset('cifar10_tiny', 'trainLabels.csv')
    test_iter = DataLoader(test_data, batch_size=batch_size)

    # 模型训练
    num_classes = 10
    # 定义模型
    model = VGG_S(num_classes)
    # 定义损失函数
    loss_fn = F.cross_entropy
    # 定义优化器
    optimizer = torch.optim.SGD(model.parameters(), lr=0.1)

    runner = Runner(model, optimizer, loss_fn, metric=None)
    runner.train(train_iter, num_epochs=10, save_path='chapter_5')

    # 模型预测
    runner.load_model('chapter_5.pth')
    x, label = next(iter(test_iter))
    predict = torch.argmax(runner.predict(x.float()), dim=1)
    print('predict:', predict)
    print('  label:', label)

预测结果

predict: tensor([6, 1, 9, 6, 1, 1, 6, 7, 0, 3, 4, 7, 7, 1, 9, 0, 9, 5, 3, 6])
  label: tensor([6, 9, 9, 4, 1, 1, 2, 7, 8, 3, 4, 7, 7, 2, 9, 9, 9, 3, 2, 6])

5. 代码整合

# 导入必要的工具包
import torch
from torch import nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from torchvision.io import read_image
import matplotlib.pyplot as plt
import os


def read_csv_labels(fname):
    """读取fname来给标签字典返回一个文件名"""
    with open(fname, 'r') as f:
        # 跳过文件头行(列名)
        lines = f.readlines()[1:]
    tokens = [l.rstrip().split(',') for l in lines]
    return dict(((name, label) for name, label in tokens))


class CIFAR10Dataset(Dataset):
    def __init__(self, folder_path, fname):
        self.labels = read_csv_labels(os.path.join(folder_path, fname))
        self.folder_path = os.path.join(folder_path, 'train')

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, idx):
        img = read_image(self.folder_path + '/' + str(idx + 1) + '.png')
        label = self.labels[str(idx + 1)]

        return img, torch.tensor(int(label))


#  每个卷积块由Conv2d卷积 + BatchNorm2d(批量标准化处理) + ReLU激活层组成
def conv_layer(chann_in, chann_out, k_size, p_size):
    layer = nn.Sequential(
        nn.Conv2d(chann_in, chann_out, kernel_size=k_size, padding=p_size),
        nn.BatchNorm2d(chann_out),
        nn.ReLU()
    )
    return layer


# vgg卷积模块是由几个相同的卷积块以及最大池化组成
def vgg_conv_block(in_list, out_list, k_list, p_list, pooling_k, pooling_s):
    layers = [conv_layer(in_list[i], out_list[i], k_list[i], p_list[i]) for i in range(len(in_list))]
    layers += [nn.MaxPool2d(kernel_size=pooling_k, stride=pooling_s)]
    return nn.Sequential(*layers)


# vgg全连接层由Linear + BatchNorm1d + ReLU组成
def vgg_fc_layer(size_in, size_out):
    layer = nn.Sequential(
        nn.Linear(size_in, size_out),
        nn.BatchNorm1d(size_out),
        nn.ReLU()
    )
    return layer


# 为了简化，我们少使用了几层卷积层，方便大家使用
class VGG_S(nn.Module):
    def __init__(self, num_classes):
        super().__init__()

        self.layer1 = vgg_conv_block([3, 64], [64, 64], [3, 3], [1, 1], 2, 2)
        self.layer2 = vgg_conv_block([64, 128], [128, 128], [3, 3], [1, 1], 2, 2)
        self.layer3 = vgg_conv_block([128, 256, 256], [256, 256, 256], [3, 3, 3], [1, 1, 1], 2, 2)

        # 全连接层
        self.layer4 = vgg_fc_layer(4096, 1024)
        # Final layer
        self.layer5 = nn.Linear(1024, num_classes)

    def forward(self, x):
        out = self.layer1(x)
        out = self.layer2(out)
        vgg16_features = self.layer3(out)
        out = vgg16_features.view(out.size(0), -1)
        out = self.layer4(out)
        out = self.layer5(out)

        return out



class Runner(object):
    def __init__(self, model, optimizer, loss_fn, metric=None):
        self.model = model
        self.optimizer = optimizer
        self.loss_fn = loss_fn
        # 用于计算评价指标
        self.metric = metric

        # 记录训练过程中的评价指标变化
        self.dev_scores = []
        # 记录训练过程中的损失变化
        self.train_epoch_losses = []
        self.dev_losses = []
        # 记录全局最优评价指标
        self.best_score = 0

    # 模型训练阶段
    def train(self, train_loader, dev_loader=None, **kwargs):
        # 将模型设置为训练模式，此时模型的参数会被更新
        self.model.train()

        num_epochs = kwargs.get('num_epochs', 0)
        log_steps = kwargs.get('log_steps', 100)
        save_path = kwargs.get('save_path', 'best_model.pth')
        eval_steps = kwargs.get('eval_steps', 0)
        # 运行的step数，不等于epoch数
        global_step = 0

        if eval_steps:
            if dev_loader is None:
                raise RuntimeError('Error: dev_loader can not be None!')
            if self.metric is None:
                raise RuntimeError('Error: Metric can not be None')

        # 遍历训练的轮数
        for epoch in range(num_epochs):
            total_loss = 0
            # 遍历数据集
            for step, data in enumerate(train_loader):
                x, y = data
                logits = self.model(x.float())
                loss = self.loss_fn(logits, y.long())
                total_loss += loss
                if step % log_steps == 0:
                    print(f'loss:{loss.item():.5f}')

                loss.backward()
                self.optimizer.step()
                self.optimizer.zero_grad()
            # 每隔一定轮次进行一次验证，由eval_steps参数控制，可以采用不同的验证判断条件
            if eval_steps != 0:
                if (epoch + 1) % eval_steps == 0:

                    dev_score, dev_loss = self.evaluate(dev_loader, global_step=global_step)
                    print(f'[Evalute] dev score:{dev_score:.5f}, dev loss:{dev_loss:.5f}')

                    if dev_score > self.best_score:
                        self.save_model(f'model_{epoch + 1}.pth')

                        print(
                            f'[Evaluate]best accuracy performance has been updated: {self.best_score:.5f}-->{dev_score:.5f}')
                        self.best_score = dev_score

                    # 验证过程结束后，请记住将模型调回训练模式
                    self.model.train()

            global_step += 1
            # 保存当前轮次训练损失的累计值
            train_loss = (total_loss / len(train_loader)).item()
            self.train_epoch_losses.append((global_step, train_loss))
        self.save_model(f'{save_path}.pth')
        print('[Train] Train done')

    # 模型评价阶段
    def evaluate(self, dev_loader, **kwargs):
        assert self.metric is not None
        # 将模型设置为验证模式，此模式下，模型的参数不会更新
        self.model.eval()
        global_step = kwargs.get('global_step', -1)
        total_loss = 0
        self.metric.reset()

        for batch_id, data in enumerate(dev_loader):
            x, y = data
            logits = self.model(x.float())
            loss = self.loss_fn(logits, y.long()).item()
            total_loss += loss
            self.metric.update(logits, y)

        dev_loss = (total_loss / len(dev_loader))
        self.dev_losses.append((global_step, dev_loss))
        dev_score = self.metric.accumulate()
        self.dev_scores.append(dev_score)
        return dev_score, dev_loss

    # 模型预测阶段，
    def predict(self, x, **kwargs):
        self.model.eval()
        logits = self.model(x)
        return logits

    # 保存模型的参数
    def save_model(self, save_path):
        torch.save(self.model.state_dict(), save_path)

    # 读取模型的参数
    def load_model(self, model_path):
        self.model.load_state_dict(torch.load(model_path, map_location=torch.device('cpu')))


if __name__ == '__main__':
    batch_size = 20
    # 构建训练集
    train_data = CIFAR10Dataset('cifar10_tiny', 'trainLabels.csv')
    train_iter = DataLoader(train_data, batch_size=batch_size)
    # 构建测试集
    test_data = CIFAR10Dataset('cifar10_tiny', 'trainLabels.csv')
    test_iter = DataLoader(test_data, batch_size=batch_size)

    # 模型训练
    num_classes = 10
    # 定义模型
    model = VGG_S(num_classes)
    # 定义损失函数
    loss_fn = F.cross_entropy
    # 定义优化器
    optimizer = torch.optim.SGD(model.parameters(), lr=0.1)

    runner = Runner(model, optimizer, loss_fn, metric=None)
    runner.train(train_iter, num_epochs=10, save_path='chapter_5')

    # 模型预测
    runner.load_model('chapter_5.pth')
    x, label = next(iter(test_iter))
    predict = torch.argmax(runner.predict(x.float()), dim=1)
    print('predict:', predict)
    print('  label:', label)

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2023-09-30，如有侵权请联系 cloudcommunity@tencent.com 删除

卷积神经网络

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度

【深度学习实验】卷积神经网络（六）：自定义卷积神经网络模型（VGG）实现图片多分类任务

【深度学习实验】卷积神经网络（六）：自定义卷积神经网络模型（VGG）实现图片多分类任务

一、实验介绍

二、实验环境

1. 配置虚拟环境

2. 库版本介绍

三、实验内容

0. 导入必要的工具包

1. 构建数据集（CIFAR10Dataset）

a. read_csv_labels（）

b. CIFAR10Dataset

2. 构建模型（FeedForward）

3.整合训练、评估、预测过程（Runner）

4. main

预测结果

5. 代码整合

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

【深度学习实验】卷积神经网络（六）：自定义卷积神经网络模型（VGG）实现图片多分类任务

【深度学习实验】卷积神经网络（六）：自定义卷积神经网络模型（VGG）实现图片多分类任务

一、实验介绍

二、实验环境

1. 配置虚拟环境

2. 库版本介绍

三、实验内容

0. 导入必要的工具包

1. 构建数据集（CIFAR10Dataset）

a. read_csv_labels（）

b. CIFAR10Dataset

2. 构建模型（FeedForward）

3.整合训练、评估、预测过程（Runner）

4. __main__

预测结果

5. 代码整合

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

4. main