前言
本篇文章是深度学习第七周的实验内容，实践基于前馈神经网络完成鸢尾花分类任务，我们一起来学习吧(? ?_?)?
导入需要用到的库：
import numpy as np
import matplotlib.pyplot as plt
from sklearn import svm, datasets
import copy
import torch.nn as nn
import numpy as np
import torch
from sklearn.datasets import load_iris
import torch.optim as opt
import torch.nn.functional as F
深入研究鸢尾花数据集
画出数据集中150个数据的前两个特征的散点分布图：
iris = datasets.load_iris()
X = iris.data[:, :2] # 只取前两维特征
y = iris.target
h = .02 # 网格中的步长
C = 1.0 # SVM正则化参数
svc = svm.SVC(kernel='linear', C=C).fit(X, y) # 线性核
rbf_svc = svm.SVC(kernel='rbf', gamma=0.7, C=C).fit(X, y) # 径向基核
poly_svc = svm.SVC(kernel='poly', degree=3, C=C).fit(X, y) # 多项式核
lin_svc = svm.LinearSVC(C=C).fit(X, y) # 线性核
x_min, x_max = X[:, 0].min() - 1, X[:, 0].max() + 1
y_min, y_max = X[:, 1].min() - 1, X[:, 1].max() + 1
xx, yy = np.meshgrid(np.arange(x_min, x_max, h),
np.arange(y_min, y_max, h))
titles = ['SVC with linear kernel',
'LinearSVC (linear kernel)',
'SVC with RBF kernel',
'SVC with polynomial (degree 3) kernel']
for i, clf in enumerate((svc, lin_svc, rbf_svc, poly_svc)):
# 绘出决策边界，不同的区域分配不同的颜色
plt.subplot(2, 2, i + 1) # 创建一个2行2列的图，并以第i个图为当前图
plt.subplots_adjust(wspace=0.4, hspace=0.4) # 设置子图间隔
Z = clf.predict(np.c_[xx.ravel(), yy.ravel()]) # 将xx和yy中的元素组成一对对坐标，作为支持向量机的输入，返回一个array
# 把分类结果绘制出来
Z = Z.reshape(xx.shape) # (220, 280)
plt.contourf(xx, yy, Z)
# 将训练数据以离散点的形式绘制出来
plt.scatter(X[:, 0], X[:, 1], c=y, cmap="brg")
plt.xlabel('Sepal length')
plt.ylabel('Sepal width')
plt.xlim(xx.min(), xx.max())
plt.ylim(yy.min(), yy.max())
plt.xticks(())
plt.yticks(())
plt.title(titles[i])
plt.show()
运行结果：
4.5 实践：基于前馈神经网络完成鸢尾花分类
继续使用第三章中的鸢尾花分类任务，将Softmax分类器替换为前馈神经网络。
损失函数：交叉熵损失；
优化器：随机梯度下降法；
评价指标：准确率。
4.5.1 小批量梯度下降法
在梯度下降法中，目标函数是整个训练集上的风险函数，这种方式称为批量梯度下降法（Batch Gradient Descent，BGD）。 批量梯度下降法在每次迭代时需要计算每个样本上损失函数的梯度并求和。当训练集中的样本数量N很大时，空间复杂度比较高，每次迭代的计算开销也很大。
为了减少每次迭代的计算复杂度，我们可以在每次迭代时只采集一小部分样本，计算在这组样本上损失函数的梯度并更新参数，这种优化方式称为小批量梯度下降法（Mini-Batch Gradient Descent，Mini-Batch GD）。
第t次迭代时，随机选取一个包含K个样本的子集Bt，计算这个子集上每个样本损失函数的梯度并进行平均，然后再进行参数更新。
其中K为批量大小(Batch Size)。K通常不会设置很大，一般在1～100之间。在实际应用中为了提高计算效率，通常设置为2的幂2n。
在实际应用中，小批量随机梯度下降法有收敛快、计算开销小的优点，因此逐渐成为大规模的机器学习中的主要优化算法。
此外，随机梯度下降相当于在批量梯度下降的梯度上引入了随机噪声。在非凸优化问题中，随机梯度下降更容易逃离局部最优点。
小批量随机梯度下降法的训练过程如下:
为了小批量梯度下降法，我们需要对数据进行随机分组。
目前，机器学习中通常做法是构建一个数据迭代器，每个迭代过程中从全部数据集中获取一批指定数量的数据。
数据迭代器的实现原理如下图所示：
1.首先，将数据集封装为Dataset类，传入一组索引值，根据索引从数据集合中获取数据；
2.其次，构建DataLoader类，需要指定数据批量的大小和是否需要对数据进行乱序，通过该类即可批量获取数据
4.5.2 数据处理
#加载数据集
def load_data(shuffle=True):
#加载原始数据
X = np.array(load_iris().data, dtype=np.float32)
y = np.array(load_iris().target, dtype=np.int64)
X = torch.as_tensor(X)
y = torch.as_tensor(y)
#数据归一化
X_min = torch.min(X, dim=0)
X_max = torch.max(X, dim=0)
X = (X-X_min.values) / (X_max.values-X_min.values)
#如果shuffle为True，随机打乱数据
if shuffle:
idx = torch.randperm(X.shape[0])
X_new = copy.deepcopy(X)
y_new = copy.deepcopy(y)
for i in range(X.shape[0]):
X_new[i] = X[idx[i]]
y_new[i] = y[idx[i]]
X = X_new
y = y_new
return X, y
class IrisDataset(torch.utils.data.Dataset):
def __init__(self, mode='train', num_train=120, num_dev=15):
super(IrisDataset, self).__init__()
# 调用第三章中的数据读取函数，其中不需要将标签转成one-hot类型
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.y)
train_dataset = IrisDataset(mode='train')
dev_dataset = IrisDataset(mode='dev')
test_dataset = IrisDataset(mode='test')
4.5.2.2 用DataLoader进行封装
# 批量大小
batch_size = 16
# 加载数据
train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
dev_loader = torch.utils.data.DataLoader(dev_dataset, batch_size=batch_size)
test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=batch_size)
4.5.3 模型构建
输入层神经元个数为4，输出层神经元个数为3，隐含层神经元个数为6。
# 实现一个两层前馈神经网络
class Model_MLP_L2_V3(torch.nn.Module):
def __init__(self, input_size, hidden_size, output_size):
super(Model_MLP_L2_V3, self).__init__()
self.fc1 = torch.nn.Linear(input_size, hidden_size)
w_ = torch.normal(0, 0.01, size=(hidden_size, input_size), requires_grad=True)
self.fc1.weight = torch.nn.Parameter(w_)
self.fc1.bias = torch.nn.init.constant_(self.fc1.bias, val=1.0)
self.fc2 = torch.nn.Linear(hidden_size, output_size )
w2 = torch.normal(0, 0.01, size=(output_size, hidden_size), requires_grad=True)
self.fc2.weight = nn.Parameter(w2)
self.fc2.bias = torch.nn.init.constant_(self.fc2.bias, val=1.0)
self.act = torch.sigmoid
def forward(self, inputs):
outputs = self.fc1(inputs)
outputs = self.act(outputs)
outputs = self.fc2(outputs)
return outputs
fnn_model =Model_MLP_L2_V3(input_size=4, hidden_size=6,output_size=3)
4.5.4 完善Runner类
class RunnerV3(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 = [] # 一个epoch记录一次loss
self.train_step_losses = [] # 一个step记录一次loss
self.dev_losses = []
# 记录全局最优指标
self.best_score = 0
def train(self, train_loader, dev_loader=None, **kwargs):
# 将模型切换为训练模式
self.model.train()
# 传入训练轮数，如果没有传入值则默认为0
num_epochs = kwargs.get("num_epochs", 0)
# 传入log打印频率，如果没有传入值则默认为100
log_steps = kwargs.get("log_steps", 100)
# 评价频率
eval_steps = kwargs.get("eval_steps", 0)
# 传入模型保存路径，如果没有传入值则默认为"best_model.pdparams"
save_path = kwargs.get("save_path", "best_model.pdparams")
custom_print_log = kwargs.get("custom_print_log", None)
# 训练总的步数
num_training_steps = num_epochs * len(train_loader)
if eval_steps:
if self.metric is None:
raise RuntimeError('Error: Metric can not be None!')
if dev_loader is None:
raise RuntimeError('Error: dev_loader can not be None!')
# 运行的step数目
global_step = 0
# 进行num_epochs轮训练
for epoch in range(num_epochs):
# 用于统计训练集的损失
total_loss = 0
for step, data in enumerate(train_loader):
X, y = data
# 获取模型预测
logits = self.model(X)
loss = self.loss_fn(logits, y) # 默认求mean
total_loss += loss
# 训练过程中，每个step的loss进行保存
self.train_step_losses.append((global_step, loss.item()))
if log_steps and global_step % log_steps == 0:
print(
f"[Train] epoch: {epoch}/{num_epochs}, step: {global_step}/{num_training_steps}, loss: {loss.item():.5f}")
# 梯度反向传播，计算每个参数的梯度值
loss.backward()
if custom_print_log:
custom_print_log(self)
# 小批量梯度下降进行参数更新
self.optimizer.step()
# 梯度归零
self.optimizer.zero_grad()
# 判断是否需要评价
if eval_steps > 0 and global_step > 0 and \
(global_step % eval_steps == 0 or global_step == (num_training_steps - 1)):
dev_score, dev_loss = self.evaluate(dev_loader, global_step=global_step)
print(f"[Evaluate] dev score: {dev_score:.5f}, dev loss: {dev_loss:.5f}")
# 将模型切换为训练模式
self.model.train()
# 如果当前指标为最优指标，保存该模型
if dev_score > self.best_score:
self.save_model(save_path)
print(
f"[Evaluate] best accuracy performence has been updated: {self.best_score:.5f} --> {dev_score:.5f}")
self.best_score = dev_score
global_step += 1
# 当前epoch 训练loss累计值
trn_loss = (total_loss / len(train_loader)).item()
# epoch粒度的训练loss保存
self.train_epoch_losses.append(trn_loss)
print("[Train] Training done!")
# 模型评估阶段，使用'torch.no_grad()'控制不计算和存储梯度
@torch.no_grad()
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)
# 计算损失函数
loss = self.loss_fn(logits, y).item()
# 累积损失
total_loss += loss
# 累积评价
self.metric.update(logits, y)
dev_loss = (total_loss / len(dev_loader))
dev_score = self.metric.accumulate()
# 记录验证集loss
if global_step != -1:
self.dev_losses.append((global_step, dev_loss))
self.dev_scores.append(dev_score)
return dev_score, dev_loss
# 模型评估阶段，使用'torch.no_grad()'控制不计算和存储梯度
@torch.no_grad()
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):
model_state_dict = torch.load(model_path)
self.model.load_state_dict(model_state_dict)
4.5.5 模型训练
import torch.optim as opt
lr = 0.2
# 定义网络
model = fnn_model
# 定义优化器
optimizer = opt.SGD(model.parameters(),lr=lr)
# 定义损失函数。softmax+交叉熵
loss_fn = F.cross_entropy
# 定义评价指标
metric = Accuracy(is_logist=True)
runner = RunnerV3(model, optimizer, loss_fn, metric)
# 启动训练
log_steps = 100
eval_steps = 50
runner.train(train_loader, dev_loader,
num_epochs=150, log_steps=log_steps, eval_steps = eval_steps,
save_path="best_model.pdparams")
运行结果;
可视化观察训练集损失和训练集loss变化情况:
import matplotlib.pyplot as plt
# 绘制训练集和验证集的损失变化以及验证集上的准确率变化曲线
def plot_training_loss_acc(runner, fig_name,
fig_size=(16, 6),
sample_step=20,
loss_legend_loc="upper right",
acc_legend_loc="lower right",
train_color="#e4007f",
dev_color='#f19ec2',
fontsize='large',
train_linestyle="-",
dev_linestyle='--'):
plt.figure(figsize=fig_size)
plt.subplot(1, 2, 1)
train_items = runner.train_step_losses[::sample_step]
train_steps = [x[0] for x in train_items]
train_losses = [x[1] for x in train_items]
plt.plot(train_steps, train_losses, color=train_color, linestyle=train_linestyle, label="Train loss")
if len(runner.dev_losses) > 0:
dev_steps = [x[0] for x in runner.dev_losses]
dev_losses = [x[1] for x in runner.dev_losses]
plt.plot(dev_steps, dev_losses, color=dev_color, linestyle=dev_linestyle, label="Dev loss")
# 绘制坐标轴和图例
plt.ylabel("loss", fontsize=fontsize)
plt.xlabel("step", fontsize=fontsize)
plt.legend(loc=loss_legend_loc, fontsize='x-large')
# 绘制评价准确率变化曲线
if len(runner.dev_scores) > 0:
plt.subplot(1, 2, 2)
plt.plot(dev_steps, runner.dev_scores,
color=dev_color, linestyle=dev_linestyle, label="Dev accuracy")
# 绘制坐标轴和图例
plt.ylabel("score", fontsize=fontsize)
plt.xlabel("step", fontsize=fontsize)
plt.legend(loc=acc_legend_loc, fontsize='x-large')
plt.savefig(fig_name)
plt.show()
plot_training_loss_acc(runner, 'fw-loss.pdf')
运行结果：
从输出结果可以看出准确率随着迭代次数增加逐渐上升，损失函数下降。
4.5.6 模型评价
使用测试数据对在训练过程中保存的最佳模型进行评价，观察模型在测试集上的准确率以及Loss情况。代码实现如下：
# 加载最优模型
runner.load_model('best_model.pdparams')
# 模型评价
score, loss = runner.evaluate(test_loader)
print("[Test] accuracy/loss: {:.4f}/{:.4f}".format(score, loss))
运行结果：
4.5.7 模型预测
同样地，也可以使用保存好的模型，对测试集中的某一个数据进行模型预测，观察模型效果。代码实现如下：
# 获取测试集中第一条数据
X, label = train_dataset[0]
logits = runner.predict(X)
pred_class = torch.argmax(logits[0]).numpy()
label = label.numpy()
# 输出真实类别与预测类别
print("The true category is {} and the predicted category is {}".format(label, pred_class))
运行结果：
思考题
1. 对比Softmax分类和前馈神经网络分类。（必做）
softmax实现对鸢尾花进行分类：
softmax分类结果：
前馈神经网络的分类结果：
对比两种结果可知对于鸢尾花分类，前馈神经网络的准确率要高于Softmax分类。
softmax损失函数：
其中，被称为softmax函数。
softmax经常在神经网络中代替sigmoid用于输出层上，通过结果得到的0-1区间的值代表概率来判断谁更可能是符合的输出
softmax直观理解：max是a>b一定取a，没有比别的选择。而在神经网络中有时候我们并不想这样，因为这样会造成分小的那个值的饥饿，我们希望分小的那个值在小概率的情况下仍然会被取到，这个时候我们就用到了softmax。在softmax中值大的会大概率被取到，也经常被取到，取到的次数便多，而值小的也会大概率被取到。而这个概率与值本身和该层各个值有关。
而使用指数的原因：，第一个原因是要模拟max的行为，所以要让大的更大。第二个原因是需要一个可导的函数。让大的更大的原因是让错的更错，这样学习效率更高
2. 自定义隐藏层层数和每个隐藏层中的神经元个数，尝试找到最优超参数完成多分类。（选做）
lr=0.1， 隐藏层神经元个数=4
lr=0.1， 隐藏层神经元个数=6
lr=0.1， 隐藏层神经元个数=8
lr=0.2， 隐藏层神经元个数=4
lr=0.2， 隐藏层神经元个数=6
lr=0.2， 隐藏层神经元个数=8
经过多次测试后，在学习率为0.1的时候，隐藏层神经元的个数为6和8的时候，准确率可以达到0.93.
在学习率为0.2的时候拟合效果最好，在验证集上准确率达到了1.0。
3. 对比SVM与FNN分类效果，谈谈自己看法。（选做）
前馈神经网络：
分类效果图可参考：
对于特征维度少、数据量小的数据集SVM非常适用，对于特征维度多、数据量大的数据集神经网络适用。
SVM：就是寻找最大分类间隔的过程，使得数据点到分类超平面之间的距离最大化。
SVM分类适合二分类问题，在文本分类尤其是针对二分类任务性能卓越，也可用于多分类
优点：如果新增一类，不需要重新训练所有的 SVM，只需要训练和新增这一类样本的分类器。而且这种方式在训练单个 SVM 模型的时候，训练速度快。
缺点：分类器的个数与 K 的平方成正比，所以当 K 较大时，训练和测试的时间会比较慢。
FNN：结合了神经网络系统和模糊系统的长处，它在处理非线性、模糊性等问题上有很大的优越性，在 智能信息处理方面存在巨大的潜力。 前馈神经网络是设计的第一种也是最简单的人工神经网络。在该网络中，信息仅在一个方向上从输入节点向前移动，通过隐藏节点（如果有）并到达输出节点。网络中没有循环或环路。
神经网络有很强的非线性拟合能力，可映射任意复杂的非线性关系，而且学习规则简单，便于计算机实现。具有很强的鲁棒性、记忆能力、非线性映射能力以及强大的自学习能力，因此有很大的应用市场。
缺点：
（1）最严重的问题是没能力来解释自己的推理过程和推理依据。
（2）不能向用户提出必要的询问，而且当数据不充分的时候，神经网络就无法进行工作。
（3）把一切问题的特征都变为数字，把一切推理都变为数值计算，其结果势必是丢失信息。
（4）理论和学习算法还有待于进一步完善和提高。
4. 尝试基于MNIST手写数字识别数据集，设计合适的前馈神经网络进行实验，并取得95%以上的准确率。（选做）
import torch
from torch import nn
from torch.autograd import Variable
from torch.utils.data import DataLoader
import torchvision.datasets as dataset
import torchvision.transforms as transforms
batch_size = 100
# 加载数据集
train_dataset = dataset.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True)
test_dataset = dataset.MNIST(root='./data', train=False, transform=transforms.ToTensor(), download=True)
# 封装数据集
train_loader = DataLoader(dataset=train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(dataset=test_dataset, batch_size=batch_size, shuffle=True)
# 初始化超参数
input_size = 784
hidden_size = 500
num_classes = 10 # 类别数目
# 定义神经网络模型
class module(nn.Module):
def __init__(self, input_size, hidden_size, output_num):
super(module, self).__init__()
self.fc1 = nn.Linear(input_size, hidden_size)
self.fc2 = nn.Linear(hidden_size, output_num)
# 前向传播
def forward(self, x):
out = self.fc1(x)
out = torch.relu(out)
out = self.fc2(out)
return out
module = module(input_size, hidden_size, num_classes)
# 超参数初始化
lr = 1e-1
epochs = 5
# 定义损失函数及优化器
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(module.parameters(), lr=lr)
# 模型开始训练
for epoch in range(epochs + 1):
print("==============第 {} 轮 训练开始==============".format(epoch + 1))
for i, (images, labels) in enumerate(train_loader):
images = Variable(images.view(-1, 28 * 28))
labels = Variable(labels)
outputs = module(images)
# 计算损失
loss = criterion(outputs, labels)
# 梯度清0
optimizer.zero_grad()
# 反向传播
loss.backward()
# 参数更新
optimizer.step() # update parameters
#
if i % 100 == 0:
print("交叉熵损失为: %.5f" % loss.item())
# 利用训练好的模型进行预测
T = 0 # 总共测试集样本个数
CC = 0 # 测试集正确识别的个数
for images, labels in test_loader:
images = Variable(images.view(-1, 28 * 28))
labels = Variable(labels)
outputs = module(images)
_, predicts = torch.max(outputs.data, 1)
T += labels.size(0)
CC += (predicts == labels).sum()
print("识别准确率为：%.2f %%" % (100 * CC / T))
运行结果：
总结
1. 总结本次实验；
本次实验实践了基于前馈神经网络完成鸢尾花分类任务，在深入了解前馈神经网络的基本概念并且实现了鸢尾花数据集分类的任务，还对softmax，SVM，FNN进行多方面的比较。
2. 全面总结前馈神经网络，梳理知识点，建议画思维导图。
