神经网络与深度学习：案例与实践——第三章(1)

神经网络与深度学习：案例与实践——第三章(1)

3.1 基于Logistic回归的二分类任务

主要任务：实现一个Logistic回归模型，并对一个简单的数据集进行二分类实验。

3.1.1 数据集构建

先构建一个简单的分类任务，并构建训练集、验证集和测试集。
本任务的数据来自带噪音的两个弯月形状函数，每个弯月对一个类别。我们采集1000条样本，每个样本包含2个特征。
①数据集构建函数：具体讲解在代码内部.

#first.py
import copy
import paddle
import math
# 数据集构建：1.输入参数
def make_moons(n_samples = 1000,shuffle = True,noise = None):'''生成弯月状数据集：输入：n_samples: 总样本量,默认是1000shuffle: 表示是否打乱顺序,当值为True时,表示打乱顺序。noise: 数据类型为None或者float,当数据类型为None,表示没有噪声值输出：X: 特征数据,shape = [n_samples,2] ,表示生成是n_samples行,2列Y: 标签数据,shape = [n_samples],表示生成是 n_samples维'''#2.数据分配——将总样本数大致平分给两个半圆。n_sample_out = n_samples // 2 # 外半圆数据量n_sample_in = n_samples - n_sample_out # 内半圆数据量#上述代码运算符解释： '/': 返回浮点数, '//': 返回整数（向下取整，3.1 取 3）#3.生成外半圆数据outer_circ_x = paddle.cos(paddle.linspace(0,math.pi,n_sample_out))outer_circ_y = paddle.sin(paddle.linspace(0,math.pi,n_sample_out))#上述代码解释，linspace 方法，在0到Π 之间生成 n_samples_out个等距数据，对这些点进行cos,sin计算取得x和y坐标，生成上半圆#4.生成内半圆数据inner_circ_x = 1 - paddle.cos(paddle.linspace(0, math.pi, n_sample_in))inner_circ_y = 0.5 - paddle.sin(paddle.linspace(0, math.pi, n_sample_in))#内半圆通过变换外半圆得到: x坐标: 1 - cos(θ),使内半圆向右偏移, y坐标: 0.5 - sin(θ),使内半圆向下偏移并缩小print('outer_circ_x.shape:', outer_circ_x.shape, 'outer_circ_y.shape:', outer_circ_y.shape)print('inner_circ_x.shape:', inner_circ_x.shape, 'inner_circ_y.shape:', inner_circ_y.shape)#5.合并特征数据X = paddle.stack([paddle.concat([outer_circ_x, inner_circ_x]),paddle.concat([outer_circ_y, inner_circ_y])],axis=1)#paddle.concat: 沿第0维拼接内外半圆的x/y坐标, paddle.stack: 将x和y坐标沿第1维堆叠，形成(n_samples, 2)的特征矩阵#paddle.concat:沿指定维度拼接张量,paddle.concat(tensors, axis=0)#paddle.stack: 沿新维度堆叠张量, paddle.stack(tensors, axis=0)#6.生成标签数据y = paddle.concat([paddle.zeros(shape=[n_sample_out]), paddle.ones(shape=[n_sample_in])])#外半圆标签为0, 内半圆标签为1print('y shape:', y.shape)#7.打乱数据if shuffle:idx = paddle.randperm(X.shape[0])X = X[idx]Y = y[idx]#paddle.randperm(X.shape[0]) 的作用, 生成一个 [0, 1, 2, ..., n-1] 的随机排列的索引数组。#例如：若 X.shape[0] = 4，可能生成 [2, 0, 3, 1]。#用 idx 的每个元素作为位置索引，从 X 中提取对应行的数据。#原数据 X = [[a], [b], [c], [d]], idx = [2, 0, 3, 1], 结果 X[idx] = [[c], [a], [d], [b]]#8.添加噪声值if noise is not None:X += paddle.normal(mean=0.0, std=noise, shape=X.shape)# 使用'paddle.normal'生成符合正态分布的随机Tensor作为噪声，并加到原始特征上,# mean=0.0: 噪声均值为0, std=noise: 噪声标准差由参数指定# paddle.normal: 生成正态分布随机数, paddle.normal(mean, std, shape)return X,y

②随机采集1000个样本，之后进行可视化。

#second.py
from first import make_moons
import matplotlib.pyplot as plt
# %matplotlib inline# 设置样本量，调用函数
n_samples = 1000
X, y = make_moons(n_samples=n_samples, shuffle=True, noise=0.5)#创建图形
plt.figure(figsize=(5,5))
#plt.figure()：创建一个新的图形窗口, figsize=(5,5)：设置图形大小为5×5英寸# 绘制散点图
plt.scatter(x=X[:, 0].tolist(), y=X[:, 1].tolist(), marker='*', c=y.tolist())
#plt.scatter()：绘制散点图
# x=X[:, 0].tolist()：X的第一列作为x坐标（转换为Python列表）
# y=X[:, 1].tolist()：X的第二列作为y坐标
# marker='*'：使用星号作为点的标记
# c=y.tolist()：使用y值作为颜色编码（不同类别不同颜色）# 设置坐标轴范围
plt.xlim(-3,4)
plt.ylim(-3,4)
#plt.xlim()/plt.ylim()：设置x/y轴显示范围, 这里设置为(-3,4)以完整显示数据分布# 保存图片，显示图片，但是现在vscode还没办法显示pdf，所以先不保存
#plt.savefig('linear-dataset-vis.pdf')
plt.show()

③将1000条样本数据拆分成训练集、验证集和测试集，其中训练集640条、验证集160条、测试集200条。

#third.py
from first import make_moons
# 设置样本量，调用函数
n_samples = 1000
X, y = make_moons(n_samples=n_samples, shuffle=True, noise=0.5)#1. 定义数据集
num_train = 640    # 训练集样本数
num_dev = 160      # 开发集(验证集)样本数
num_test = 200     # 测试集样本数#2.数据分割
# 训练集：前640个样本
X_train, y_train = X[:num_train], y[:num_train]# 开发集：接下来的160个样本 (640到800)
X_dev, y_dev = X[num_train:num_train + num_dev], y[num_train:num_train + num_dev]# 测试集：剩余样本 (800到1000)
X_test, y_test = X[num_train + num_dev:], y[num_train + num_dev:]#3.标签维度调整
# 将一维标签 [n_samples] 转为二维 [n_samples, 1]
y_train = y_train.reshape([-1,1])  # -1表示自动计算该维度大小
y_dev = y_dev.reshape([-1,1])
y_test = y_test.reshape([-1,1])# 打印X_train和y_train的维度
print("X_train shape: ", X_train.shape, "y_train shape: ", y_train.shape)
# 打印一下前5个数据的标签
print (y_train[:5])

现在，完成了数据集的构建

3.1.2 模型构建

Logistic回归是一种常用的处理二分类问题的线性模型。

特性	线性回归	Logistic回归
输出范围	任意实数	概率值（0，1）区间
目标	预测连续值	预测类别概率
函数转换	无	使用Sigmoid函数（非线性转换）

经过上面的对比，得知Logistic回归与线性回归的相同之处为：都会将输入特征与权重做线性叠加。但是，Logistic回归还引入了非线性函数（Sigmoid函数），取预测类别标签的厚颜概率p(y = 1|X)，从而解决了连续的线性函数不适合进行分类的这一问题。


Logistic函数是Logistic回归的核心。其特点与优势：
①输出范围：将任意实数映射到 (0,1) 区间，适合表示概率。
②平滑性：连续可导，便于优化（如梯度下降）。
③与阶跃函数的对比：解决了阶跃函数在阈值点不连续的问题，更适合分类。
Logistic函数代码实现如下：

结果分析：输入在0附近时，Logistic函数近似为线性函数；当输入值非常大或者非常小时，函数会对输入进行抑制.输入越小，越接近0；输入越大，越接近1。正是因为这样的性质，才使得其输出可以直接看作是概率分布。

Logistic回归算子
Logistic回归模型其实就是线性层与Logistic函数的组合，通常会讲Logistic回归模型中的权重和偏置初始化为0.并且，为例提高预测样本效率，将N个样本归为一组进行成批预测。

实现代码如下：

class model_LR(op.Op):#继承自定义的算子基类def __init__(self, input_dim): # 输入特征维度super(model_LR, self).__init__()self.params = {}# 将线性层的权重参数全部初始化为0self.params['w'] = paddle.zeros(shape=[input_dim, 1])# self.params['w'] = paddle.normal(mean=0, std=0.01, shape=[input_dim, 1])#这个是替代初始化方案，小随机数，实践中常用# 将线性层的偏置参数初始化为0self.params['b'] = paddle.zeros(shape=[1])def __call__(self, inputs): # 辅助方法#使类实例可像函数一样调用，使Python的特殊方法，实现对象可调用性return self.forward(inputs)#前向计算，def forward(self, inputs):"""输入：- inputs: shape=[N,D], N是样本数量，D为特征维度输出：- outputs：预测标签为1（类别为1）的概率，shape=[N,1]"""# 线性计算 ，matmul实现矩阵乘法 Xw，再加上偏置bscore = paddle.matmul(inputs, self.params['w']) + self.params['b']# Logistic 函数，非线性激活，将分数映射到【0，1】中outputs = logistic(score)return outputs# 固定随机种子，保持每次运行结果一致
paddle.seed(0)
# 随机生成3条长度为4的数据
inputs = paddle.randn(shape=[3,4])
print('Input is:', inputs)
# 实例化模型
model = model_LR(4)
outputs = model(inputs)
print('Output is:', outputs)

输出结果：

Input is: Tensor(shape=[3, 4], dtype=float32, place=CPUPlace, stop_gradient=True,
[[-0.75711036, -0.38059190, 0.10946669, 1.34467661],
[-0.84002435, -1.27341712, 2.47224617, 0.14070207],
[ 0.60608417, 0.23396523, 1.35604191, 0.10350471]])
Output is: Tensor(shape=[3, 1], dtype=float32, place=CPUPlace, stop_gradient=True,
[[0.50000000],
[0.50000000],
[0.50000000]])

从输出结果看，模型最终的输出g(⋅)恒为0.5。这是由于采用全0初始化后，不论输入值的大小为多少，Logistic函数的输入值恒为0，因此输出恒为0.5。

损失函数

这段内容主要阐述了在训练Logistic回归模型时，如何通过**交叉熵损失函数（Cross-Entropy Loss）**量化预测概率与真实标签之间的差异，并给出了数学形式和直观意义。
使用损失函数量化预测性hi和真实值之间的差异。现有一个分类任务，y表示样本x的标签的概率分布。
向量
表示预测的标签概率分布，训练目标是使得预测概率分布尽可能接近真实概率分布，通常使用交叉熵损失函数。
在给定y的情况下，若预测与真实越接近，则交叉熵越小，反之，越大。

对于二分类任务：

Logistic回归的风险函数计算为：


向量化形式为：
二分类任务的交叉熵损失函数的代码实现如下：

import paddle
# 实现交叉熵损失函数
class BinaryCrossEntropyLoss(op.Op):def __init__(self):self.predicts = None  # 存储预测值self.labels = None   # 存储真实标签self.num = None    #存储样本数量# 等同于直接调用前向计算def __call__(self, predicts, labels):return self.forward(predicts, labels)#前向计算：完成了从模型输出到损失值的计算过程def forward(self, predicts, labels):"""输入：- predicts：预测值，shape=[N, 1]，N为样本数量- labels：真实标签，shape=[N, 1]输出：- 损失值：shape=[1]"""self.predicts = predictsself.labels = labelsself.num = self.predicts.shape[0]#label.t() ：将列向量转为行向量，形状[1,N],方便矩阵乘法#paddle,log ：逐元素计算预测值的对数#损失计算：正类部分：label.t,log(perdicts),负类部分：1 - label.t这些loss = -1. / self.num * (paddle.matmul(self.labels.t(), paddle.log(self.predicts)) + paddle.matmul((1-self.labels.t()), paddle.log(1-self.predicts)))#压缩维度，去除多余维度。loss = paddle.squeeze(loss, axis=1)return loss# 测试一下
# 生成一组长度为3，值为1的标签数据
labels = paddle.ones(shape=[3,1])
# 计算风险函数
bce_loss = BinaryCrossEntropyLoss()
# 有了call辅助方法，使得类实例可以想函数一样直接调用，如下。
print(bce_loss(outputs, labels))#paddle.matmul(a, b)	    矩阵乘法	                 a、b 需满足矩阵乘法维度规则
#paddle.log(x)	            逐元素自然对数	              输入需为正数（用clip避免0）
#paddle.squeeze(x, axis)	压缩指定维度（长度为1的维度）	axis：需压缩的维度
#tensor.t()	                矩阵转置	                 将 [N,1] 转为 [1,N]

模型优化

不同于线性回归中直接使用最小二乘法即可进行模型参数的求解，Logistic回归需要使用优化算法对模型参数进行有限次地迭代来获取更优的模型，从而尽可能地降低风险函数的值。 在机器学习任务中，最简单、常用的优化算法是梯度下降法。
使用梯度下降法进行模型优化，首先需要初始化参数W和 b，然后不断地计算它们的梯度，并沿梯度的反方向更新参数。

通常将偏导数的计算过程定义在Logistic回归算子的backward函数中.代码实现如下：

class model_LR(op.Op):def __init__(self, input_dim):super(model_LR, self).__init__()# 存放线性层参数self.params = {}# 将线性层的权重参数全部初始化为0self.params['w'] = paddle.zeros(shape=[input_dim, 1])# self.params['w'] = paddle.normal(mean=0, std=0.01, shape=[input_dim, 1])# 将线性层的偏置参数初始化为0self.params['b'] = paddle.zeros(shape=[1])# 存放参数的梯度self.grads = {}self.X = Noneself.outputs = Nonedef __call__(self, inputs):return self.forward(inputs)def forward(self, inputs):self.X = inputs# 线性计算score = paddle.matmul(inputs, self.params['w']) + self.params['b']# Logistic 函数self.outputs = logistic(score)return self.outputsdef backward(self, labels):"""输入：- labels：真实标签，shape=[N, 1]"""N = labels.shape[0]# 计算偏导数self.grads['w'] = -1 / N * paddle.matmul(self.X.t(), (labels - self.outputs))self.grads['b'] = -1 / N * paddle.sum(labels - self.outputs)

参数更新：计算参数梯度之后，按照下面的公式更新参数

把上面参数更新过程称为优化器。首先定义一个优化器基类Optimizer，方便后续的调用。在该类中，会初始化优化器的初始学习率init_ir，以及指定优化器需要优化的参数。代码实现如下：

from abc import abstractmethod# 优化器基类
class Optimizer(object):def __init__(self, init_lr, model):"""优化器类初始化"""# 初始化学习率，用于参数更新的计算self.init_lr = init_lr# 指定优化器需要优化的模型self.model = model#表示这是一个抽象方法，子类必须重写该方法，否则抛出TypeError@abstractmethoddef step(self):"""定义每次迭代如何更新参数"""pass

然后实现一个梯度下降法的优化器函数SimpleBatchGD来执行参数更新过程。其中step函数从模型的grads属性取出参数的梯度并更新。代码实现如下：

#实现了简单的批量梯度下降优化器Sim，继承自Optimizer基类
class SimpleBatchGD(Optimizer):def __init__(self, init_lr, model):super(SimpleBatchGD, self).__init__(init_lr=init_lr, model=model)#调用父类Optimizer的·初始化方法，传入两个参数def step(self):# 参数更新# 遍历所有参数，按照公式(3.8)和(3.9)更新参数if isinstance(self.model.params, dict):for key in self.model.params.keys():self.model.params[key] = self.model.params[key] - self.init_lr * self.model.grads[key]

评价指标

在分类任务中，通常使用准确率（Accuracy）作为评价指标。如果模型预测的类别与真实类别一致，则说明模型预测正确。准确率即正确预测的数量与总的预测数量的比值：

其中I(⋅)是指示函数。
求准确率的代码实现如下：

def accuracy(preds, labels):"""输入：- preds：预测值，二分类时，shape=[N, 1]，N为样本数量，多分类时，shape=[N, C]，C为类别数量--二分类任务时，每个样本通常通过sigmoid激活--多分类任务·时，每个样本通常通过softmax激活- labels：真实标签，shape=[N, 1]输出：- 准确率：shape=[1]"""# 判断是二分类任务还是多分类任务，preds.shape[1]=1时为二分类任务，preds.shape[1]>1时为多分类任务if preds.shape[1] == 1:# 二分类时，判断每个概率值是否大于0.5，当大于0.5时，类别为1，否则类别为0# 使用'paddle.cast'将preds的数据类型转换为float32类型preds = paddle.cast((preds>=0.5),dtype='float32')else:# 多分类时，使用'paddle.argmax'，沿着类别维度（axis = 1）找到 每个样本预测分数最大的类别索引，返回的索引值作为预测类别preds = paddle.argmax(preds,axis=1, dtype='int32')#paddle.mean(): 计算平均值，也就是准确率#paddle.equal():逐元素比较预测类别和真实标签return paddle.mean(paddle.cast(paddle.equal(preds, labels),dtype='float32'))# 假设模型的预测值为[[0.],[1.],[1.],[0.]]，真实类别为[[1.],[1.],[0.],[0.]]，计算准确率
preds = paddle.to_tensor([[0.],[1.],[1.],[0.]])
labels = paddle.to_tensor([[1.],[1.],[0.],[0.]])
print("accuracy is:", accuracy(preds, labels))

完善Runner类

基于RunnerV1，本章的RunnerV2类在训练过程中使用梯度下降法进行网络优化，模型训练过程中计算在训练集和验证集上的损失及评估指标并打印，训练过程中保存最优模型。代码实现如下：

import paddle
# 用RunnerV2类封装整个训练过程
# 负责管理模型训练，验证，评估和保存的全过程
# RunnerV2 的主要目标时：①提供一个统一的训练过程框架；②自动化训练过程中的常见操作（梯度计算，参数更新等）
# ③记录训练指标和损失变化；④实现模型保存和加载功能；⑤支持训练过程中的监控和日志输出；
class RunnerV2(object):def __init__(self, model, optimizer, metric, loss_fn):'''-model: 包含前向计算和反向传播方法的模型-optimizer: 负责参数更新的优化器-metric: 评估模型的性能函数（准确率）-loss_fn: 计算模型损失函数（交叉熵）'''self.model = modelself.optimizer = optimizerself.loss_fn = loss_fnself.metric = metric# 记录训练过程（训练集和验证集）中的评价指标变化情况self.train_scores = []self.dev_scores = []# 记录训练过程（训练集和验证集）中的损失函数变化情况self.train_loss = []self.dev_loss = []def train(self, train_set, dev_set, **kwargs):# 传入训练轮数，如果没有传入值则默认为0num_epochs = kwargs.get("num_epochs", 0)# 传入log打印频率，如果没有传入值则默认为100log_epochs = kwargs.get("log_epochs", 100)# 传入模型保存路径，如果没有传入值则默认为"best_model.pdparams"save_path = kwargs.get("save_path", "best_model.pdparams")# 梯度打印函数，如果没有传入则默认为"None"print_grads = kwargs.get("print_grads", None)# 记录全局最优指标best_score = 0# 进行num_epochs轮训练for epoch in range(num_epochs):X, y = train_set# 前向计算，获取模型预测logits = self.model(X)# 计算交叉熵损失trn_loss = self.loss_fn(logits, y).item()self.train_loss.append(trn_loss)# 计算评价指标trn_score = self.metric(logits, y).item()self.train_scores.append(trn_score)# 反向传播与优化：计算参数梯度self.model.backward(y)if print_grads is not None:# 打印每一层的梯度print_grads(self.model)# 更新模型参数self.optimizer.step() # 参数更新#验证步骤dev_score, dev_loss = self.evaluate(dev_set)# 如果当前指标为最优指标，保存该模型if dev_score > best_score:self.save_model(save_path)print(f"best accuracy performence has been updated: {best_score:.5f} --> {dev_score:.5f}")best_score = dev_score#日志输出if epoch % log_epochs == 0:print(f"[Train] epoch: {epoch}, loss: {trn_loss}, score: {trn_score}")print(f"[Dev] epoch: {epoch}, loss: {dev_loss}, score: {dev_score}")#再给定数据集上评估模型性能，返回评估指标和损失值；记录历史评估结果         def evaluate(self, data_set):X, y = data_set# 计算模型输出logits = self.model(X)# 计算损失函数loss = self.loss_fn(logits, y).item()self.dev_loss.append(loss)# 计算评价指标score = self.metric(logits, y).item()self.dev_scores.append(score)return score, loss#模型预测#对输入数据进行预测，返回模型输出def predict(self, X):return self.model(X)# 模型保存def save_model(self, save_path):paddle.save(self.model.paramloadel_models, save_path)# 模型加载（loadel_model）def load_model(self, model_path):self.model.params = paddle.load(model_path)

模型训练

下面进行Logistic回归模型的训练，使用交叉熵损失函数和梯度下降法进行优化。 使用训练集和验证集进行模型训练，共训练 500个epoch，每隔50个epoch打印出训练集上的指标。 代码实现如下：

# 固定随机种子，保持每次运行结果一致
paddle.seed(102)# 特征维度
input_dim = 2
# 设置梯度下降的学习率，这是一个相对较大的学习率，适合Logistic回归这种简单模型
lr = 0.1# 实例化模型
model = model_LR(input_dim=input_dim)
# 指定优化器，更新参数
optimizer = SimpleBatchGD(init_lr=lr, model=model)
# 指定损失函数
loss_fn = BinaryCrossEntropyLoss()
# 指定评价方式
metric = accuracy# 实例化RunnerV2类(训练流程管理类),并传入训练配置
runner = RunnerV2(model, optimizer, metric, loss_fn)
# 训练执行
#-X_train: 训练数据，特征
#-y_train: 训练数据，标签
#-X_dev: 验证数据，特征
#-y_dev: 验证数据，输出值，评估模型泛化能力
#-num_epochs: 训练轮数
#-log_epochs:   每隔 log_epochs  轮打印一次日志
#-save_path:  最佳模型保存路径
runner.train([X_train, y_train], [X_dev, y_dev], num_epochs=500, log_epochs=50, save_path="best_model.pdparams")

可视化观察训练集与验证集的准确率和损失的变化情况。

# 可视化观察训练集与验证集的指标变化情况
def plot(runner,fig_name):plt.figure(figsize=(10,5))plt.subplot(1,2,1)epochs = [i for i in range(len(runner.train_scores))]# 绘制训练损失变化曲线plt.plot(epochs, runner.train_loss, color='#8E004D', label="Train loss")# 绘制评价损失变化曲线plt.plot(epochs, runner.dev_loss, color='#E20079', linestyle='--', label="Dev loss")# 绘制坐标轴和图例plt.ylabel("loss")plt.xlabel("epoch")plt.legend(loc='upper right')plt.subplot(1,2,2)# 绘制训练准确率变化曲线plt.plot(epochs, runner.train_scores, color='#8E004D', label="Train accuracy")# 绘制评价准确率变化曲线plt.plot(epochs, runner.dev_scores, color='#E20079', linestyle='--', label="Dev accuracy")# 绘制坐标轴和图例plt.ylabel("score")plt.xlabel("epoch")plt.legend(loc='lower right')plt.tight_layout()plt.savefig(fig_name)plt.show()plot(runner,fig_name='linear-acc.pdf')

从输出结果可以看到，在训练集与验证集上，loss得到了收敛，同时准确率指标都达到了较高的水平，训练比较充分。

模型评价

使用测试集对训练完成后的最终模型进行评价，观察模型在测试集上的准确率和loss数据。代码实现如下：

score, loss = runner.evaluate([X_test, y_test])
print("[Test] score/loss: {:.4f}/{:.4f}".format(score, loss))

输出指标：

score：评估指标得分（此处为准确率）

loss：损失函数值（此处为交叉熵损失）

输出结果：[Test] score/loss: 0.8350/0.3925
可视化观察拟合的决策边界 Xw+b=0。

def decision_boundary(w, b, x1):w1, w2 = wx2 = (- w1 * x1 - b) / w2return x2plt.figure(figsize=(5,5))
# 绘制原始数据
plt.scatter(X[:, 0].tolist(), X[:, 1].tolist(), marker='*', c=y.tolist())
#X[:, 0]：第一个特征作为x轴，X[:, 1]：第二个特征作为y轴，c=y.tolist()：用颜色区分不同类别
#获取模型参数：
w = model.params['w']
b = model.params['b']
#生成决策边界：
x1 = paddle.linspace(-2, 3, 1000)
x2 = decision_boundary(w, b, x1)
# 绘制决策边界
plt.plot(x1.tolist(), x2.tolist(), color="red")
plt.show()