【深度学习实验】前馈神经网络（final）：自定义鸢尾花分类前馈神经网络模型并进行训练及评价

目录

一、实验介绍

 二、实验环境

1. 配置虚拟环境

2. 库版本介绍

三、实验内容

0. 导入必要的工具包

1. 构建数据集（IrisDataset）

2. 构建模型（FeedForward）

a. __init__(初始化)

b. forward(前向传播)

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

4. 模型评价（Accuracy）

5. __main__

6. 代码整合

---

 

 

 

一、实验介绍

        鸢尾花分类（通过输入鸢尾花的特征信息，预测鸢尾花的种类)是一个常见的机器学习问题。本次实验旨在使用PyTorch构建一个简单的鸢尾花分类前馈神经网络模型，并进行训练和评价，以了解PyTorch的基本使用方法和模型训练过程。

 

 二、实验环境

    本系列实验使用了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. 库版本介绍

 

三、实验内容

ChatGPT：

        前馈神经网络（Feedforward Neural Network）是一种常见的人工神经网络模型，也被称为多层感知器（Multilayer Perceptron，MLP）。它是一种基于前向传播的模型，主要用于解决分类和回归问题。

        前馈神经网络由多个层组成，包括输入层、隐藏层和输出层。它的名称"前馈"源于信号在网络中只能向前流动，即从输入层经过隐藏层最终到达输出层，没有反馈连接。

以下是前馈神经网络的一般工作原理：

1. 输入层：接收原始数据或特征向量作为网络的输入，每个输入被表示为网络的一个神经元。每个神经元将输入加权并通过激活函数进行转换，产生一个输出信号。

2. 隐藏层：前馈神经网络可以包含一个或多个隐藏层，每个隐藏层由多个神经元组成。隐藏层的神经元接收来自上一层的输入，并将加权和经过激活函数转换后的信号传递给下一层。

3. 输出层：最后一个隐藏层的输出被传递到输出层，输出层通常由一个或多个神经元组成。输出层的神经元根据要解决的问题类型（分类或回归）使用适当的激活函数（如Sigmoid、Softmax等）将最终结果输出。

4. 前向传播：信号从输入层通过隐藏层传递到输出层的过程称为前向传播。在前向传播过程中，每个神经元将前一层的输出乘以相应的权重，并将结果传递给下一层。这样的计算通过网络中的每一层逐层进行，直到产生最终的输出。

5. 损失函数和训练：前馈神经网络的训练过程通常涉及定义一个损失函数，用于衡量模型预测输出与真实标签之间的差异。常见的损失函数包括均方误差（Mean Squared Error）和交叉熵（Cross-Entropy）。通过使用反向传播算法（Backpropagation）和优化算法（如梯度下降），网络根据损失函数的梯度进行参数调整，以最小化损失函数的值。

        前馈神经网络的优点包括能够处理复杂的非线性关系，适用于各种问题类型，并且能够通过训练来自动学习特征表示。然而，它也存在一些挑战，如容易过拟合、对大规模数据和高维数据的处理较困难等。为了应对这些挑战，一些改进的网络结构和训练技术被提出，如卷积神经网络（Convolutional Neural Networks）和循环神经网络（Recurrent Neural Networks）等。

本系列为实验内容，对理论知识不进行详细阐释

（咳咳，其实是没时间整理，待有缘之时，回来填坑）

 

0. 导入必要的工具包

import torch
from torch import nn
import torch.nn.functional as F
# 绘画时使用的工具包
import matplotlib.pyplot as plt
# 导入鸢尾花数据集
from sklearn.datasets import load_iris
# 构建自己的数据集,继承自Dataset类
from torch.utils.data import Dataset, DataLoader

 

1. 构建数据集（IrisDataset）

• 本实验使用了鸢尾花数据集，该数据集包含150个样本，每个样本有4个特征和1个标签。
• load_iris函数加载数据集，并进行数据归一化处理；
• 自定义的IrisDataset类用于数据加载，构建训练集、验证集和测试集。

【深度学习实验】前馈神经网络（七）：批量加载数据（直接加载数据→定义类封装数据）_QomolangmaH的博客-CSDN博客https://blog.csdn.net/m0_63834988/article/details/133181882?spm=1001.2014.3001.5501

 

2. 构建模型（FeedForward）

       本实验构建了一个简单的两层前馈神经网络。这个前馈神经网络和前文实现的MLP类最大的区别在于，我们实现类中使用了自己写的激活函数，该激活函数不能通过反向传播更新参数，但深度学习框架已经帮我们完成了这个功能。（其实通过简单的改动，我们的激活函数也可以反传梯度）

【深度学习实验】前馈神经网络（三）：自定义多层感知机（激活函数logistic、线性层算Linear）_QomolangmaH的博客-CSDN博客https://blog.csdn.net/m0_63834988/article/details/133097102?spm=1001.2014.3001.5501

class FeedForward(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(FeedForward,self).__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.fc2 = nn.Linear(hidden_size, output_size)
        self.act = nn.Sigmoid()
    
    def forward(self, inputs):
        outputs = self.fc1(inputs)
        outputs = self.act(outputs)
        outputs = self.fc2(outputs)
        return outputs

a. __init__(初始化)

• 三个参数：
  • input_size（输入大小）
  • hidden_size（隐藏层大小）
  • output_size（输出大小）
• 调用父类nn.Module的初始化方法super(FeedForward, self).init()，确保正确地初始化该类作为一个nn.Module。
• 两个线性层self.fc1和self.fc2：
  • self.fc1的输入大小为input_size，输出大小为hidden_size；
  • self.fc2的输入大小为hidden_size，输出大小为output_size。
• 一个激活函数self.act，这里使用的是nn.Sigmoid()，即Sigmoid激活函数。

b. forward(前向传播)

• 接受一个输入张量inputs。在前向传播过程中，
  • 输入经过self.fc1线性层，
  • 然后通过self.act激活函数进行非线性变换，
  • 再经过self.fc2线性层得到最终的输出张量outputs

 

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

        Runner类封装了模型训练和评价的过程。

• 初始化函数接收模型、优化器、损失函数和评价指标等参数，并定义了一些成员变量用于记录训练过程中的损失和评价指标变化。
• train函数进行模型的训练
• evaluate函数进行模型的评价
• predict函数进行模型的预测
• save_model和load_model函数用于保存和加载模型的参数

【深度学习实验】前馈神经网络（九）：整合训练、评估、预测过程（Runner）_QomolangmaH的博客-CSDN博客https://blog.csdn.net/m0_63834988/article/details/133219448?spm=1001.2014.3001.5501

4. 模型评价（Accuracy）

 

【深度学习实验】前馈神经网络（八）：模型评价（自定义支持分批进行评价的Accuracy类）_QomolangmaH的博客-CSDN博客https://blog.csdn.net/m0_63834988/article/details/133186305?spm=1001.2014.3001.5501

 

5. __main__

if __name__ == '__main__':
    batch_size = 16
 
    # 分别构建训练集、验证集和测试集
    train_dataset = IrisDataset(mode='train')
    dev_dataset = IrisDataset(mode='dev')
    test_dataset = IrisDataset(mode='test')
 
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    dev_loader = DataLoader(dev_dataset, batch_size=batch_size)
    test_loader = DataLoader(test_dataset, batch_size=1, shuffle=True)
 
    input_size = 4
    output_size = 3
    hidden_size = 6
    # 定义模型
    model = FeedForward(input_size, hidden_size, output_size)
    # 定义损失函数
    loss_fn = F.cross_entropy
    # 定义优化器
    optimizer = torch.optim.SGD(model.parameters(), lr=0.2)
    # 定义评价方法
    metric = Accuracy(is_logist=True)
    # 实例化辅助runner类
    runner = Runner(model, optimizer, loss_fn, metric)
    # 模型训练
    runner.train(train_loader, dev_loader, num_epochs=50, log_steps=10, eval_steps=5)
    # 训练结束后，网络的参数会自动保存为.pth结尾的文件，且与训练文件在同一目录下
    model_path = 'model_25.pth'
    # 首先读入经过训练后的网络的参数
    runner.load_model(model_path)
    x, label = next(iter(test_loader))
    print(runner.predict(x.float()))
    print(label)

 

• batch_size = 16 设置批大小，用于数据加载器；

• 构建训练集、验证集和测试集的数据集对象；

• 创建数据加载器；

• 设置模型的输入大小、输出大小和隐藏层大小：

  • input_size = 4 输入大小为4，对应于鸢尾花数据集的特征数量。
  • output_size = 3 输出大小为3，对应于鸢尾花数据集的类别数量。
  • hidden_size = 6 隐藏层大小为6，作为前馈神经网络模型的隐藏层的单元数。

• 定义模型、损失函数、优化器和评价指标：

  • 前馈神经网络模型：使用FeedForward类，设置输入、隐藏层和输出大小。
  • 损失函数：使用交叉熵损失函数F.cross_entropy。
  • 优化器：使用随机梯度下降法（SGD），学习率为0.2。
  • 评价指标：使用Accuracy类。

• 创建辅助的Runner类对象用于训练和评价模型。

• 进行模型训练：

  • 设置训练数据加载器为train_loader，验证数据加载器为dev_loader，训练的轮次为50，打印日志的频率为每10步打印一次日志，评价的频率为每5个epoch进行一次评价。

• 训练结束后，通过调用runner.load_model(model_path)方法加载模型的参数

• 构造一个测试样本 x 和标签 label，调用 runner.predict(x.float()) 方法对样本进行预测，并输出预测结果。随后输出真实标签 label。

 

6. 代码整合

# 导入必要的工具包
import torch
from torch import nn
import torch.nn.functional as F
# 绘画时使用的工具包
import matplotlib.pyplot as plt
# 导入鸢尾花数据集
from sklearn.datasets import load_iris
# 构建自己的数据集,继承自Dataset类
from torch.utils.data import Dataset, DataLoader
 
 
# 此函数用于加载鸢尾花数据集
def load_data(shuffle=True):
    x = torch.tensor(load_iris().data)
    y = torch.tensor(load_iris().target)
 
    # 数据归一化
    x_min = torch.min(x, dim=0).values
    x_max = torch.max(x, dim=0).values
    x = (x - x_min) / (x_max - x_min)
 
    if shuffle:
        idx = torch.randperm(x.shape[0])
        x = x[idx]
        y = y[idx]
    return x, y
 
 
class IrisDataset(Dataset):
    def __init__(self, mode='train', num_train=120, num_dev=15):
        super(IrisDataset, self).__init__()
        x, y = load_data(shuffle=True)
        if mode == 'train':
            self.x, self.y = x[:num_train], y[:num_train]
        elif mode == 'dev':
            self.x, self.y = x[num_train:num_train + num_dev], y[num_train:num_train + num_dev]
        else:
            self.x, self.y = x[num_train + num_dev:], y[num_train + num_dev:]
 
    def __getitem__(self, idx):
        return self.x[idx], self.y[idx]
 
    def __len__(self):
        return len(self.x)
 
 
class FeedForward(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(FeedForward, self).__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.fc2 = nn.Linear(hidden_size, output_size)
        self.act = nn.Sigmoid()
 
    def forward(self, inputs):
        outputs = self.fc1(inputs)
        outputs = self.act(outputs)
        outputs = self.fc2(outputs)
        return outputs
 
 
# 支持分批进行模型评价的 Accuracy 类
class Accuracy:
    def __init__(self, is_logist=True):
        # 正确样本个数
        self.num_correct = 0
        # 样本总数
        self.num_count = 0
        self.is_logist = is_logist
 
    def update(self, outputs, labels):
        # 判断是否为二分类任务
        if outputs.shape[1] == 1:
            outputs = outputs.squeeze(-1)
            # 判断是否是logit形式的预测值
            if self.is_logist:
                preds = (outputs >= 0).long()
            else:
                preds = (outputs >= 0.5).long()
        else:
            # 多分类任务时，计算最大元素索引作为类别
            preds = torch.argmax(outputs, dim=1).long()
 
        # 获取本批数据中预测正确的样本个数
        labels = labels.squeeze(-1)
        batch_correct = (preds == labels).float().sum()
        batch_count = len(labels)
        # 更新
        self.num_correct += batch_correct
        self.num_count += batch_count
 
    def accumulate(self):
        # 使用累计的数据，计算总的评价指标
        if self.num_count == 0:
            return 0
        return self.num_correct / self.num_count
 
    def reset(self):
        self.num_correct = 0
        self.num_count = 0
 
 
class Runner(object):
    def __init__(self, model, optimizer, loss_fn, metric, **kwargs):
        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_mode.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 log_steps and global_step % log_steps == 0:
                    print(f'loss:{loss.item():.5f}')
 
                loss.backward()
                self.optimizer.step()
                self.optimizer.zero_grad()
            # 每隔一定轮次进行一次验证，由eval_steps参数控制，可以采用不同的验证判断条件
            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))
 
        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 = 16
 
    # 分别构建训练集、验证集和测试集
    train_dataset = IrisDataset(mode='train')
    dev_dataset = IrisDataset(mode='dev')
    test_dataset = IrisDataset(mode='test')
 
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    dev_loader = DataLoader(dev_dataset, batch_size=batch_size)
    test_loader = DataLoader(test_dataset, batch_size=1, shuffle=True)
 
    input_size = 4
    output_size = 3
    hidden_size = 6
    # 定义模型
    model = FeedForward(input_size, hidden_size, output_size)
    # 定义损失函数
    loss_fn = F.cross_entropy
    # 定义优化器
    optimizer = torch.optim.SGD(model.parameters(), lr=0.2)
    # 定义评价方法
    metric = Accuracy(is_logist=True)
    # 实例化辅助runner类
    runner = Runner(model, optimizer, loss_fn, metric)
    # 模型训练
    runner.train(train_loader, dev_loader, num_epochs=50, log_steps=10, eval_steps=5)
    # 训练结束后，网络的参数会自动保存为.pth结尾的文件，且与训练文件在同一目录下
    model_path = 'model_25.pth'
    # 首先读入经过训练后的网络的参数
    runner.load_model(model_path)
    x, label = next(iter(test_loader))
    print(runner.predict(x.float()))
    print(label)