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

目录

一、实验介绍

二、实验环境

1. 配置虚拟环境

2. 库版本介绍

三、实验内容

0. 导入必要的工具包

1. __init__(初始化)

2. train(训练)

3. evaluate(评估)

4. predict(预测)

5. save_model(保存模型参数)

6. load_model(加载模型参数)

7. 代码整合

---

 

 

 

 

一、实验介绍

      本实验实现了Runner类，用于模型的训练、评估和预测。通过该类，可以更方便地进行模型的训练和评估，并获取训练过程中的损失变化和评价指标的变化情况。

 

二、实验环境

    本系列实验使用了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. __init__(初始化)

    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

• 五个参数：
  • model（模型）：用于进行训练、评估和预测；
  • optimizer（优化器）：用于更新模型的参数；
  • loss_fn（损失函数）：用于计算模型的损失；
  • metric（评价指标）：用于评估模型的性能；
  • **kwargs（其他可选参数）
• 该类还定义了一些用于记录训练过程中的指标变化和全局最优指标的属性：
  • self.dev_scores（记录验证集评价指标的变化）
  • self.train_epoch_losses（记录训练集损失的变化）
  • self.dev_losses（记录验证集损失的变化）
  • self.best_score（记录全局最优评价指标）

 

2. train(训练)

       遍历训练数据集进行模型训练，并在一定轮次上进行验证。

• train_loader：训练数据集的数据加载器。
• dev_loader：验证数据集的数据加载器（可选）。
• **kwargs：可选的其他参数，如训练轮数、日志输出频率等。

 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')

 

• 通过self.model.train()将模型设置为训练模式，这意味着在训练过程中模型的参数会被更新。

• 从kwargs参数中获取一些配置信息，如训练的轮数num_epochs、日志输出步数log_steps、模型保存路径save_path和评估步数eval_steps等。

• 初始化全局步数global_step为0。

• 如果设置了评估步数eval_steps，则需要进行模型评估。

  • 如果没有提供验证数据集dev_loader，会抛出一个RuntimeError异常。

  • 如果没有设置评估指标self.metric，也会抛出一个RuntimeError异常。

• 使用循环遍历训练的轮数num_epochs。

  • 在每一轮训练开始前，初始化总损失total_loss为0。

  • 使用enumerate(train_loader)遍历训练数据集，获取数据data和对应的索引step。

    • 从data中获取输入数据x和目标数据y。

    • 将输入数据x传入模型self.model，得到模型的预测结果logits。

    • 使用损失函数self.loss_fn计算预测结果logits和目标数据y之间的损失loss。

    • 累加每个样本的损失到总损失total_loss中。

    • 如果设置了日志输出步数log_steps，并且当前步数global_step是日志输出步数的倍数，则打印当前步数的损失值。

    • 对损失值进行反向传播计算梯度，并更新模型的参数。

    • 清空梯度，为下一轮训练做准备。

  • 如果当前轮次是需要进行验证的轮次（根据eval_steps参数控制）

    • 调用self.evaluate函数对模型进行评估，并获取评估结果dev_score和验证损失dev_loss。

    • 打印验证结果。

    • 如果当前的验证结果dev_score优于之前的最佳结果self.best_score，则保存模型，并更新最佳结果self.best_score。

    • 在完成验证后，将模型调回训练模式，以便继续训练。

  • 更新全局步数global_step。

  • 计算当前轮次的平均训练损失train_loss，并将其添加到self.train_epoch_losses列表中。

• 完成所有训练轮次后，打印训练完成的提示信息。

 

3. evaluate(评估)

        使用验证数据集评估模型性能，并计算评价指标和损失值。（设置模型为验证模式，可以保持模型参数不变，以便在验证过程中进行准确的评估）

    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

 

• 使用assert self.metric is not None断言确保评估指标self.metric不为空。

• 将模型设置为验证模式，即通过self.model.eval()将模型的状态设置为验证模式。在验证模式下，模型的参数不会被更新。

• 从kwargs参数中获取全局步数global_step，默认为-1。

• 初始化总损失total_loss为0。

• 调用评估指标的reset()方法，用于重置评估指标的内部状态。

• 使用循环遍历验证数据集dev_loader。

  • 对于每个批次的数据，获取输入数据x和目标数据y。

  • 将输入数据x传入模型self.model，得到模型的预测结果logits。

  • 使用损失函数self.loss_fn计算预测结果logits和目标数据y之间的损失loss。使用.item()方法将损失值转换为标量。

  • 将每个批次的损失值累加到总损失total_loss中。

  • 调用评估指标的update()方法，传入预测结果logits和目标数据y，用于更新评估指标的状态。

• 计算验证集的平均损失dev_loss，即总损失除以验证数据集的长度。

• 将当前全局步数global_step和验证损失dev_loss添加到self.dev_losses列表中，用于后续的记录和分析。

• 调用评估指标的accumulate()方法，计算并返回累积的评估指标值。

• 将当前的评估指标值dev_score添加到self.dev_scores列表中，用于后续的记录和分析。

• 返回评估指标值dev_score和验证损失dev_loss。

 

4. predict(预测)

    用于对输入数据进行预测。将模型设置为评估模式（self.model.eval()），然后通过self.model进行前向传播，得到预测结果logits。

 def predict(self, x, **kwargs):
        self.model.eval()
        logits = self.model(x)
        return logits

 

5. save_model(保存模型参数)

 

 def save_model(self, save_path):
        torch.save(self.model.state_dict(),save_path)

 

6. load_model(加载模型参数)

  def load_model(self, model_path):
        self.model.load_state_dict(torch.load(model_path, map_location=torch.device('cpu')))

 

7. 代码整合

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')))