PyTorch基础篇:
Hook函数机制:不改变主体,实现额外功能,像一个挂件一样将功能挂到函数主体上。Hook函数与PyTorch中的动态图运算机制有关,因为在动态图计算,在运算结束后,中间变量是会被释放掉的,例如:非叶子节点的梯度。但是,我们往往想要提取这些中间变量,这时,我们就可以采用Hook函数在前向传播与反向传播主体上挂上一些额外的功能(函数),通过这些函数获取中间的梯度,甚至是改变中间的梯度。PyTorch一共提供了四种Hook函数:
torch.Tensor.register_hook(hook)torch.nn.Module.register_forward_hooktorch.nn.Module.register_forward_pre_hooktorch.nn.Module.register_backward_hook一种是针对Tensor,其余三种是针对网络的
Tensor.register_hook
def register_hook(self, hook):
"""
接受一个hook函数
"""
...
功能:注册一个反向传播hook函数,这是因为张量在反向传播的时候,如果不是叶子节点,它的梯度就会消失。由于反向传播过程中存在数据的释放,所以就有了反向传播的hook函数
hook函数仅一个输入参数,为张量的梯度
hook(grad) -> Tensor or None
下面,我们通过计算图流程来观察张量梯度的获取以及熟悉Hook函数。
y
=
(
x
+
w
)
∗
(
w
+
1
)
y=(x+w)*(w+1)
y=(x+w)∗(w+1)

import torch
import torch.nn as nn
w = torch.tensor([1.], requires_grad=True)
x = torch.tensor([2.], requires_grad=True)
a = torch.add(w, x)
b = torch.add(w, 1)
y = torch.mul(a, b)
# 存储张量的梯度
a_grad = list()
def grad_hook(grad):
"""
定义一个hook函数,将梯度存储到列表中
:param grad: 梯度
:return:
"""
a_grad.append(grad)
# 注册一个反向传播的hook函数,功能是将梯度存储到a_grad列表中
handle = a.register_hook(grad_hook)
# 反向传播
y.backward()
# 查看梯度
print("gradient:", w.grad, x.grad, a.grad, b.grad, y.grad)
print("a_grad[0]: ", a_grad[0])
handle.remove()
tensor([5.]) tensor([2.]) None None None
a_grad[0]: tensor([2.])
如果对叶子节点的张量使用hook函数,那么会怎么样呢?
import torch
import torch.nn as nn
w = torch.tensor([1.], requires_grad=True)
x = torch.tensor([2.], requires_grad=True)
a = torch.add(w, x)
b = torch.add(w, 1)
y = torch.mul(a, b)
a_grad = list()
def grad_hook(grad):
grad *= 2
return grad*3
handle = w.register_hook(grad_hook)
y.backward()
# 查看梯度
print("gradient:", w.grad, x.grad, a.grad, b.grad, y.grad)
print("w.grad: ", w.grad)
handle.remove()
gradient: tensor([30.]) tensor([2.]) None None None
w.grad: tensor([30.])
与上面比较,发现hook函数相当于对已有张量进行原地操作
Module.register_forward_hook
def register_forward_hook(self, hook):
...
功能:注册module的前向传播hook函数
参数:
module: 当前网络层input:当前网络层输入数据output:当前网络层输出数据Module.register_forward_pre_hook
功能:注册module前向传播前的hook函数
参数:
module: 当前网络层input:当前网络层输入数据Module.register_backward_hook
功能:注册module反向传播的hook函数
参数:
module: 当前网络层grad_input:当前网络层输入梯度数据grad_output:当前网络层输出梯度数据下面例子展示这三个hook函数

import torch
import torch.nn as nn
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.conv1 = nn.Conv2d(1, 2, 3)
self.pool1 = nn.MaxPool2d(2, 2)
def forward(self, x):
x = self.conv1(x)
x = self.pool1(x)
return x
def forward_hook(module, data_input, data_output):
"""
定义前向传播hook函数
:param module:网络
:param data_input:输入数据
:param data_output:输出数据
"""
fmap_block.append(data_output)
input_block.append(data_input)
def forward_pre_hook(module, data_input):
"""
定义前向传播前的hook函数
:param module: 网络
:param data_input: 输入数据
:return:
"""
print("forward_pre_hook input:{}".format(data_input))
def backward_hook(module, grad_input, grad_output):
"""
定义反向传播的hook函数
:param module: 网络
:param grad_input: 输入梯度
:param grad_output: 输出梯度
:return:
"""
print("backward hook input:{}".format(grad_input))
print("backward hook output:{}".format(grad_output))
# 初始化网络
net = Net()
# 第一个卷积核全设置为1
net.conv1.weight[0].detach().fill_(1)
# 第二个卷积核全设置为2
net.conv1.weight[1].detach().fill_(2)
# bias不考虑
net.conv1.bias.data.detach().zero_()
# 注册hook
fmap_block = list()
input_block = list()
# 给卷积层注册前向传播hook函数
net.conv1.register_forward_hook(forward_hook)
# 给卷积层注册前向传播前的hook函数
net.conv1.register_forward_pre_hook(forward_pre_hook)
# 给卷积层注册反向传播的hook函数
net.conv1.register_backward_hook(backward_hook)
# inference
fake_img = torch.ones((1, 1, 4, 4)) # batch size * channel * H * W
output = net(fake_img)
loss_fnc = nn.L1Loss()
target = torch.randn_like(output)
loss = loss_fnc(target, output)
loss.backward()
# 观察
print("output shape: {}\noutput value: {}\n".format(output.shape, output))
print("feature maps shape: {}\noutput value: {}\n".format(fmap_block[0].shape, fmap_block[0]))
print("input shape: {}\ninput value: {}".format(input_block[0][0].shape, input_block[0]))
forward_pre_hook input:(tensor([[[[1., 1., 1., 1.],
[1., 1., 1., 1.],
[1., 1., 1., 1.],
[1., 1., 1., 1.]]]]),)
backward hook input:(None, tensor([[[[0.5000, 0.5000, 0.5000],
[0.5000, 0.5000, 0.5000],
[0.5000, 0.5000, 0.5000]]],
[[[0.5000, 0.5000, 0.5000],
[0.5000, 0.5000, 0.5000],
[0.5000, 0.5000, 0.5000]]]]), tensor([0.5000, 0.5000]))
backward hook output:(tensor([[[[0.5000, 0.0000],
[0.0000, 0.0000]],
[[0.5000, 0.0000],
[0.0000, 0.0000]]]]),)
output shape: torch.Size([1, 2, 1, 1])
output value: tensor([[[[ 9.]],
[[18.]]]], grad_fn=<MaxPool2DWithIndicesBackward>)
feature maps shape: torch.Size([1, 2, 2, 2])
output value: tensor([[[[ 9., 9.],
[ 9., 9.]],
[[18., 18.],
[18., 18.]]]], grad_fn=<MkldnnConvolutionBackward>)
input shape: torch.Size([1, 1, 4, 4])
input value: (tensor([[[[1., 1., 1., 1.],
[1., 1., 1., 1.],
[1., 1., 1., 1.],
[1., 1., 1., 1.]]]]),)
正则化方法是机器学习(深度学习)中重要的方法,它目的在于减小方差。下面借助周志华老师西瓜书中的对于方差、偏差的定义来进行理解。
泛化误差可分解为:偏差、方差与噪声之和。

L1正则化与L2正则化,通常就是损失函数加上正则项。关于L1正则化与L2正则化面试中常涉及到如下问题:百面机器学习—13.L1正则化与稀疏性
加入L2正则项后,目标函数:
O
b
j
=
L
o
s
s
+
λ
2
∗
∑
i
N
w
i
2
w
i
+
1
=
w
i
−
∂
O
b
j
∂
w
i
=
w
i
(
1
−
λ
)
−
∂
L
o
s
s
∂
w
i
Obj=Loss + \frac{\lambda}{2}*\sum_i^Nw_i^2\\w_{i+1}=w_i-\frac{\partial{Obj}}{\partial{w_i}}=w_i(1-\lambda)-\frac{\partial{Loss}}{\partial{w_i}}
Obj=Loss+2λ∗i∑Nwi2wi+1=wi−∂wi∂Obj=wi(1−λ)−∂wi∂Loss
0
<
λ
<
1
0<\lambda<1
0<λ<1,L2正则项又称为权值衰减。
下面通过一个小例子来试验weight decay。
import torch
import torch.nn as nn
import matplotlib.pyplot as plt
from torch.utils.tensorboard import SummaryWriter
n_hidden = 200
max_iter = 2000
disp_interval = 200
lr_init = 0.01
# ============================ step 1/5 数据 ============================
def gen_data(num_data=10, x_range=(-1, 1)):
w = 1.5
train_x = torch.linspace(*x_range, num_data).unsqueeze_(1)
train_y = w*train_x + torch.normal(0, 0.5, size=train_x.size())
test_x = torch.linspace(*x_range, num_data).unsqueeze_(1)
test_y = w*test_x + torch.normal(0, 0.3, size=test_x.size())
return train_x, train_y, test_x, test_y
train_x, train_y, test_x, test_y = gen_data(x_range=(-1, 1))
# ============================ step 2/5 模型 ============================
class MLP(nn.Module):
def __init__(self, neural_num):
super(MLP, self).__init__()
# 三层的全连接网络
self.linears = nn.Sequential(
nn.Linear(1, neural_num),
nn.ReLU(inplace=True),
nn.Linear(neural_num, neural_num),
nn.ReLU(inplace=True),
nn.Linear(neural_num, neural_num),
nn.ReLU(inplace=True),
nn.Linear(neural_num, 1),
)
# 前向传播
def forward(self, x):
return self.linears(x)
# 网络实例化
net_normal = MLP(neural_num=n_hidden)
net_weight_decay = MLP(neural_num=n_hidden)
# ============================ step 3/5 优化器 ============================
optim_normal = torch.optim.SGD(net_normal.parameters(), lr=lr_init, momentum=0.9)
optim_wdecay = torch.optim.SGD(net_weight_decay.parameters(), lr=lr_init, momentum=0.9, weight_decay=1e-2)
# ============================ step 4/5 损失函数 ============================
loss_func = torch.nn.MSELoss()
# ============================ step 5/5 迭代训练 ============================
writer = SummaryWriter(comment='_test_tensorboard', filename_suffix="12345678")
for epoch in range(max_iter):
# forward
pred_normal, pred_wdecay = net_normal(train_x), net_weight_decay(train_x)
loss_normal, loss_wdecay = loss_func(pred_normal, train_y), loss_func(pred_wdecay, train_y)
# 梯度清零
optim_normal.zero_grad()
optim_wdecay.zero_grad()
# 反向传播
loss_normal.backward()
loss_wdecay.backward()
# 模型逐步更新
optim_normal.step()
optim_wdecay.step()
if (epoch+1) % disp_interval == 0:
# 可视化-统计直方图分布
for name, layer in net_normal.named_parameters():
writer.add_histogram(name + '_grad_normal', layer.grad, epoch)
writer.add_histogram(name + '_data_normal', layer, epoch)
for name, layer in net_weight_decay.named_parameters():
writer.add_histogram(name + '_grad_weight_decay', layer.grad, epoch)
writer.add_histogram(name + '_data_weight_decay', layer, epoch)
# 测试
test_pred_normal, test_pred_wdecay = net_normal(test_x), net_weight_decay(test_x)
# 绘图
plt.scatter(train_x.data.numpy(), train_y.data.numpy(), c='blue', s=50, alpha=0.3, label='train')
plt.scatter(test_x.data.numpy(), test_y.data.numpy(), c='red', s=50, alpha=0.3, label='test')
plt.plot(test_x.data.numpy(), test_pred_normal.data.numpy(), 'r-', lw=3, label='no weight decay')
plt.plot(test_x.data.numpy(), test_pred_wdecay.data.numpy(), 'b--', lw=3, label='weight decay')
plt.text(-0.25, -1.5, 'no weight decay loss={:.6f}'.format(loss_normal.item()), fontdict={'size': 15, 'color': 'red'})
plt.text(-0.25, -2, 'weight decay loss={:.6f}'.format(loss_wdecay.item()), fontdict={'size': 15, 'color': 'red'})
plt.ylim((-2.5, 2.5))
plt.legend(loc='upper left')
plt.title("Epoch: {}".format(epoch+1))
plt.show()
plt.close()



可以发现:随着训练轮数的增加,有权值衰减的模型的泛化能力越强。
未权值衰减的权值变化为:

权值衰减的权值变化为:

权值衰减控制权值逐渐减小,以至于模型不复杂,从而不会产生非常严重的过拟合。
下面,我们看一下pytorch中的权值衰减是如何实现的?
def step(self, closure=None):
"""Performs a single optimization step.
Arguments:
closure (callable, optional): A closure that reevaluates the model
and returns the loss.
"""
loss = None
if closure is not None:
with torch.enable_grad():
loss = closure()
for group in self.param_groups:
weight_decay = group['weight_decay']
momentum = group['momentum']
dampening = group['dampening']
nesterov = group['nesterov']
for p in group['params']:
if p.grad is None:
continue
d_p = p.grad
if weight_decay != 0:
d_p = d_p.add(p, alpha=weight_decay)
if momentum != 0:
param_state = self.state[p]
if 'momentum_buffer' not in param_state:
buf = param_state['momentum_buffer'] = torch.clone(d_p).detach()
else:
buf = param_state['momentum_buffer']
buf.mul_(momentum).add_(d_p, alpha=1 - dampening)
if nesterov:
d_p = d_p.add(buf, alpha=momentum)
else:
d_p = buf
p.add_(d_p, alpha=-group['lr'])
return loss
核心代码为:
for group in self.param_groups:
weight_decay = group['weight_decay']
momentum = group['momentum']
dampening = group['dampening']
nesterov = group['nesterov']
for p in group['params']:
if p.grad is None:
continue
d_p = p.grad
if weight_decay != 0:
d_p = d_p.add(p, alpha=weight_decay)
if momentum != 0:
param_state = self.state[p]
即
d
p
=
d
p
+
p
∗
w
e
i
g
h
t
_
d
e
c
a
y
d_p=d_p+p*weight\_decay
dp=dp+p∗weight_decay
Dropout可以总结为四个字:随机失活。失活可以理解为权值为0。从对特征的依赖性角度来讲,假设某个神经元接受上一层的5个神经元,如果其特别依赖第一个神经元,比如权重是5,剩余4个权重都是0.0001,那么,这个神经元就非常依赖权重为5的神经元。如果第一个神经元出现,则这个神经元激活,否则,可能不被激活。如果加入Dropout后,当前的神经元就不会知道上一层的神经元哪些神经元会出现,所以就不会对某些神经元进行过度依赖。减轻神经元对某些特征的过度依赖性,提高了特征的鲁棒性,实现了正则化。
Dropout是指在深度网络的训练中,以一定的概率
p
p
p随机地“临时丢弃”一部分神经元节点(即以一定概率使部分神经元节点随机失活)。具体来讲,Dropout作用于每份小批量训练数据,由于其随机丢弃部分神经元的机制,相当于每次迭代都在训练不同结构的神经网络。Dropout在小批量级别上的操作,提供了一种轻量级的 Bagging 集成近似,能够实现指数级数量神经网络的训练与评测。
Dropout可被认为是一种实用的大规模深度神经网络的模型集成算法。这是由于传统意义上的Bagging涉及多个模型的同时训练与测试评估,当网络与参数规模庞大时,这种集成方式需要消耗大量的运算时间与空间。
Dropout的具体实现中,要求某个神经元节点激活值以一定的概率p被“丢弃”,即该神经元暂时停止工作。

因此,对于包含
N
N
N个神经元节点的网络,在 Dropout的作用下可看作为
2
N
2^N
2N个模型的集成。这
2
N
2^N
2N个模型可认为是原始网络的子网络,它们共享部分权值,并且具有相同的网络层数,而模型整体的参数数目不变,这就大大简化了运算。对于任意神经元,每次训练中都与一组随机挑选的不同的神经元集合共同进行优化,这个过程会减弱全体神经元之间的联合适应性,减少过拟合的风险,增强泛化能力。
标准网络与Dropout网络对比如下:

原始网络对应的前向传播公式为:

Dropout网络对应的前向传播公式为:


上面的 Bernoulli 函数的作用是以概率系数p随机生成一个取值为0或1的向量,代表每个神经元是否需要被丢弃。如果取值为0,则该神经元将不会计算梯度或参与后面的误差传播。
深度学习中常见的正则化方法—Dropout,Dropout是简洁高效的正则化方法,但需要注意其在实现过程中的权值数据尺度问题。在测试时,所有权重乘以
1
−
d
r
o
p
_
p
r
o
b
1-drop\_prob
1−drop_prob,这个在《Dropout: A simple way to prevent neural networks from overfitting》论文中被提到。在写代码时要注意到将训练时的数据尺度与测试时的数据尺度保持一致。比如:
d
r
o
p
_
p
r
o
b
=
0.3
drop\_prob=0.3
drop_prob=0.3,那么训练时的数据尺度为测试时的数据尺度的0.7,所以需要在测试时,所有权重乘以
1
−
d
r
o
p
_
p
r
o
b
1-drop\_prob
1−drop_prob。
torch.nn.Dropout(p=0.5, inplace=False)
功能:Dropout层
参数:
p:被舍弃概率,失活概率 下面我们通过代码来观察Dropout的影响
import torch
import torch.nn as nn
import matplotlib.pyplot as plt
from torch.utils.tensorboard import SummaryWriter
n_hidden = 200
max_iter = 2000
disp_interval = 400
lr_init = 0.01
# ============================ step 1/5 数据 ============================
def gen_data(num_data=10, x_range=(-1, 1)):
w = 1.5
train_x = torch.linspace(*x_range, num_data).unsqueeze_(1)
train_y = w*train_x + torch.normal(0, 0.5, size=train_x.size())
test_x = torch.linspace(*x_range, num_data).unsqueeze_(1)
test_y = w*test_x + torch.normal(0, 0.3, size=test_x.size())
return train_x, train_y, test_x, test_y
train_x, train_y, test_x, test_y = gen_data(x_range=(-1, 1))
# ============================ step 2/5 模型 ============================
class MLP(nn.Module):
def __init__(self, neural_num, d_prob=0.5):
super(MLP, self).__init__()
self.linears = nn.Sequential(
nn.Linear(1, neural_num),
nn.ReLU(inplace=True),
nn.Dropout(d_prob),
nn.Linear(neural_num, neural_num),
nn.ReLU(inplace=True),
nn.Dropout(d_prob),
nn.Linear(neural_num, neural_num),
nn.ReLU(inplace=True),
nn.Dropout(d_prob),
nn.Linear(neural_num, 1),
)
def forward(self, x):
# 采用Sequential可以一步实现forward
return self.linears(x)
# 模型实例化
net_prob_0 = MLP(neural_num=n_hidden, d_prob=0.)
net_prob_05 = MLP(neural_num=n_hidden, d_prob=0.5)
# ============================ step 3/5 优化器 ============================
optim_normal = torch.optim.SGD(net_prob_0.parameters(), lr=lr_init, momentum=0.9)
optim_reglar = torch.optim.SGD(net_prob_05.parameters(), lr=lr_init, momentum=0.9)
# ============================ step 4/5 损失函数 ============================
loss_func = torch.nn.MSELoss()
# ============================ step 5/5 迭代训练 ============================
writer = SummaryWriter(comment='_test_tensorboard', filename_suffix="12345678")
for epoch in range(max_iter):
pred_normal, pred_wdecay = net_prob_0(train_x), net_prob_05(train_x)
loss_normal, loss_wdecay = loss_func(pred_normal, train_y), loss_func(pred_wdecay, train_y)
optim_normal.zero_grad()
optim_reglar.zero_grad()
loss_normal.backward()
loss_wdecay.backward()
optim_normal.step()
optim_reglar.step()
if (epoch+1) % disp_interval == 0:
# 由于dropout的训练与测试是不一样的,eval()表示设置当前网络采用测试的状态,而不是训练的状态
net_prob_0.eval()
net_prob_05.eval()
# 可视化
for name, layer in net_prob_0.named_parameters():
writer.add_histogram(name + '_grad_normal', layer.grad, epoch)
writer.add_histogram(name + '_data_normal', layer, epoch)
for name, layer in net_prob_05.named_parameters():
writer.add_histogram(name + '_grad_regularization', layer.grad, epoch)
writer.add_histogram(name + '_data_regularization', layer, epoch)
# 测试,在测试前,需要将模型设置为测试状态
test_pred_prob_0, test_pred_prob_05 = net_prob_0(test_x), net_prob_05(test_x)
# 绘图
plt.clf()
plt.scatter(train_x.data.numpy(), train_y.data.numpy(), c='blue', s=50, alpha=0.3, label='train')
plt.scatter(test_x.data.numpy(), test_y.data.numpy(), c='red', s=50, alpha=0.3, label='test')
plt.plot(test_x.data.numpy(), test_pred_prob_0.data.numpy(), 'r-', lw=3, label='d_prob_0')
plt.plot(test_x.data.numpy(), test_pred_prob_05.data.numpy(), 'b--', lw=3, label='d_prob_05')
plt.text(-0.25, -1.5, 'd_prob_0 loss={:.8f}'.format(loss_normal.item()), fontdict={'size': 15, 'color': 'red'})
plt.text(-0.25, -2, 'd_prob_05 loss={:.6f}'.format(loss_wdecay.item()), fontdict={'size': 15, 'color': 'red'})
plt.ylim((-2.5, 2.5))
plt.legend(loc='upper left')
plt.title("Epoch: {}".format(epoch+1))
plt.show()
plt.close()
# 在测试完后,需要将网络重新恢复到训练状态
net_prob_0.train()
net_prob_05.train()

Dropout与weight_decay一样,同样使得权重尺度发生变化。下面通过TensorBoard来观察线性层的权值区别。
未dropout的权值:

dropout的权值:

下面分析pytorch中的eval()与train()实现了什么功能?
def train(self: T, mode: bool = True) -> T:
r"""Sets the module in training mode.
This has any effect only on certain modules. See documentations of
particular modules for details of their behaviors in training/evaluation
mode, if they are affected, e.g. :class:`Dropout`, :class:`BatchNorm`,
etc.
Args:
mode (bool): whether to set training mode (``True``) or evaluation
mode (``False``). Default: ``True``.
Returns:
Module: self
"""
self.training = mode
for module in self.children():
module.train(mode)
return self
def eval(self: T) -> T:
r"""Sets the module in evaluation mode.
This has any effect only on certain modules. See documentations of
particular modules for details of their behaviors in training/evaluation
mode, if they are affected, e.g. :class:`Dropout`, :class:`BatchNorm`,
etc.
This is equivalent with :meth:`self.train(False) `.
Returns:
Module: self
"""
return self.train(False)
self.training是False,则表示模型是测试状态;self.training是True,则表示模型是训练状态。
为了在测试时,测试更简单,训练时权重均乘以 1 1 − 𝑝 \frac{1}{1−𝑝} 1−p1 ,即除以 1 − p 1-p 1−p;此时,测试时,所有权重就不需要乘以 1 − d r o p _ p r o b 1-drop\_prob 1−drop_prob,两者抵消。
import torch
import torch.nn as nn
class Net(nn.Module):
def __init__(self, neural_num, d_prob=0.5):
super(Net, self).__init__()
self.linears = nn.Sequential(
nn.Dropout(d_prob),
nn.Linear(neural_num, 1, bias=False),
nn.ReLU(inplace=True)
)
def forward(self, x):
return self.linears(x)
# 输入张量
input_num = 10000
x = torch.ones((input_num, ), dtype=torch.float32)
net = Net(input_num, d_prob=0.5)
# 权重设置为1
net.linears[1].weight.detach().fill_(1.)
# train模式
net.train()
y = net(x)
print("output in training mode", y)
# eval模式
net.eval()
y = net(x)
print("output in eval mode", y)
output in training mode tensor([9956.], grad_fn=<ReluBackward1>)
output in eval mode tensor([10000.], grad_fn=<ReluBackward1>)
Batch Normalization即“批标准化”,批指的是一批数据,通常为mini-batch。标准化指的是mean=0,std=1。
Batch Normalization有以下优点:
Batch Normalization时,如果学习率过大,很容易导致梯度激增,从而使得模型无法训练。详细的可以了解《 Batch Normalization: Accelerating Deep Network Training by Reducing Internal Covariate Shift》
神经网络训练过程的本质是学习数据分布,如果训练数据与测试数据的分布不同将大大降低网络的泛化能力,因此我们需要在训练开始前对所有输入数据进行归一化处理。然而随着网络训练的进行,每个隐层的参数变化使得后一层的输入发生变化,从而每一批训练数据的分布也随之改变,致使网络在每次迭代中都需要拟合不同的数据分布,增大训练的复杂度以及过拟合的风险。批量归一化可以看做是在每一层输入和上一层输出之间加入一个计算层,这个计算层的作用就是归一化处理,将所有批数据强制在统一的数据分布下,从而增强模型的泛化能力。
批量归一化,虽然增强了模型的泛化能力,但同时降低了模型的拟合能力。因此,在批量归一化的具体实现中引入了变量重构以及可学习参数 γ \gamma γ和 β \beta β, γ \gamma γ和 β \beta β变成了该层的学习参数,仅用两个参数就可以恢复最优的输入数据分布。
完整的批量归一化网络层前向传播公式:
在原始论文中Batch Normalization的提出是为了解决Internal Covariate Shift问题,即数据分布(尺度)的变化,导致训练困难。在学习权值初始化 时,我们分析过网络方差的变化,Batch Normalization就是为了解决这个问题。在解决了这个问题后带来了上述一系列的优点。下面,我们通过代码来观察这些优点。
在未权值初始化,未bn情况下,发现数据尺度发生了巨大变化。
import torch
import numpy as np
import torch.nn as nn
class MLP(nn.Module):
def __init__(self, neural_num, layers=100):
super(MLP, self).__init__()
self.linears = nn.ModuleList([nn.Linear(neural_num, neural_num, bias=False) for i in range(layers)])
self.bns = nn.ModuleList([nn.BatchNorm1d(neural_num) for i in range(layers)])
self.neural_num = neural_num
def forward(self, x):
# 前向传播
for (i, linear), bn in zip(enumerate(self.linears), self.bns):
x = linear(x)
# x = bn(x)
x = torch.relu(x)
if torch.isnan(x.std()):
print("output is nan in {} layers".format(i))
break
print("layers:{}, std:{}".format(i, x.std().item()))
return x
# 权值初始化
def initialize(self):
for m in self.modules():
if isinstance(m, nn.Linear):
# method 1
# nn.init.normal_(m.weight.materials, std=1) # normal: mean=0, std=1
# method 2 kaiming
nn.init.kaiming_normal_(m.weight.data)
neural_nums = 256
layer_nums = 100
batch_size = 16
net = MLP(neural_nums, layer_nums)
# net.initialize()
inputs = torch.randn((batch_size, neural_nums)) # normal: mean=0, std=1
output = net(inputs)
print(output)
layers:0, std:0.32943010330200195
layers:1, std:0.1318356990814209
layers:2, std:0.052585702389478683
layers:3, std:0.02193078212440014
layers:4, std:0.00893945898860693
layers:5, std:0.0036938649136573076
layers:6, std:0.0015579452738165855
layers:7, std:0.0006277725333347917
layers:8, std:0.0002646839711815119
...
layers:47, std:7.859776476425103e-20
layers:48, std:2.9688882202417704e-20
layers:49, std:1.1333666053890026e-20
layers:50, std:4.123510701654615e-21
layers:51, std:1.6957295453597266e-21
layers:52, std:6.230306804239323e-22
layers:53, std:2.4648755417600425e-22
...
layers:90, std:5.762514160668824e-37
layers:91, std:2.4922294974995734e-37
layers:92, std:9.623848803677693e-38
layers:93, std:4.248455843360902e-38
layers:94, std:1.7813144929173677e-38
layers:95, std:6.768123045051648e-39
layers:96, std:2.81580556797436e-39
layers:97, std:1.1762345166711439e-39
layers:98, std:4.812521354984649e-40
layers:99, std:1.925075804320147e-40
如果我们设计标准正态初始化,则数据变化仍然很大
import torch
import numpy as np
import torch.nn as nn
class MLP(nn.Module):
def __init__(self, neural_num, layers=100):
super(MLP, self).__init__()
self.linears = nn.ModuleList([nn.Linear(neural_num, neural_num, bias=False) for i in range(layers)])
self.bns = nn.ModuleList([nn.BatchNorm1d(neural_num) for i in range(layers)])
self.neural_num = neural_num
def forward(self, x):
# 前向传播
for (i, linear), bn in zip(enumerate(self.linears), self.bns):
x = linear(x)
# x = bn(x)
x = torch.relu(x)
if torch.isnan(x.std()):
print("output is nan in {} layers".format(i))
break
print("layers:{}, std:{}".format(i, x.std().item()))
return x
# 权值初始化
def initialize(self):
for m in self.modules():
if isinstance(m, nn.Linear):
# method 1
nn.init.normal_(m.weight.data, std=1) # normal: mean=0, std=1
# method 2 kaiming
# nn.init.kaiming_normal_(m.weight.data)
neural_nums = 256
layer_nums = 100
batch_size = 16
net = MLP(neural_nums, layer_nums)
net.initialize()
inputs = torch.randn((batch_size, neural_nums)) # normal: mean=0, std=1
output = net(inputs)
print(output)
layers:0, std:9.278536796569824
layers:1, std:110.19374084472656
layers:2, std:1147.586181640625
layers:3, std:12954.3427734375
layers:4, std:140433.40625
layers:5, std:1587572.125
layers:6, std:17144180.0
layers:7, std:193045600.0
...
layers:29, std:2.299191217901132e+31
layers:30, std:2.5098634669304556e+32
layers:31, std:2.8851340804932224e+33
layers:32, std:3.580790586058518e+34
layers:33, std:3.951448749544361e+35
layers:34, std:4.6563579532070865e+36
output is nan in 35 layers
如果我们使用kaiming初始化,则数据变化合理
import torch
import numpy as np
import torch.nn as nn
class MLP(nn.Module):
def __init__(self, neural_num, layers=100):
super(MLP, self).__init__()
self.linears = nn.ModuleList([nn.Linear(neural_num, neural_num, bias=False) for i in range(layers)])
self.bns = nn.ModuleList([nn.BatchNorm1d(neural_num) for i in range(layers)])
self.neural_num = neural_num
def forward(self, x):
# 前向传播
for (i, linear), bn in zip(enumerate(self.linears), self.bns):
x = linear(x)
# x = bn(x)
x = torch.relu(x)
if torch.isnan(x.std()):
print("output is nan in {} layers".format(i))
break
print("layers:{}, std:{}".format(i, x.std().item()))
return x
# 权值初始化
def initialize(self):
for m in self.modules():
if isinstance(m, nn.Linear):
# method 1
# nn.init.normal_(m.weight.data, std=1) # normal: mean=0, std=1
# method 2 kaiming
nn.init.kaiming_normal_(m.weight.data)
neural_nums = 256
layer_nums = 100
batch_size = 16
net = MLP(neural_nums, layer_nums)
net.initialize()
inputs = torch.randn((batch_size, neural_nums)) # normal: mean=0, std=1
output = net(inputs)
print(output)
layers:0, std:0.8312186002731323
layers:1, std:0.8596137166023254
layers:2, std:0.775924026966095
layers:3, std:0.8062796592712402
layers:4, std:0.821081280708313
layers:5, std:0.8172966241836548
layers:6, std:0.7615244388580322
layers:7, std:0.8058425784111023
layers:8, std:0.7926238775253296
layers:9, std:0.7935766577720642
...
layers:50, std:0.6049355268478394
layers:51, std:0.6208720803260803
layers:52, std:0.6958529353141785
layers:53, std:0.6533211469650269
layers:54, std:0.6593946218490601
layers:55, std:0.6131288409233093
layers:56, std:0.6278454065322876
layers:57, std:0.6850351691246033
layers:58, std:0.754036009311676
layers:59, std:0.6656206250190735
...
layers:90, std:0.3476549983024597
layers:91, std:0.3065834045410156
layers:92, std:0.3101688623428345
layers:93, std:0.3357504904270172
layers:94, std:0.36929771304130554
layers:95, std:0.364469975233078
layers:96, std:0.32304060459136963
layers:97, std:0.3217012584209442
layers:98, std:0.34530118107795715
layers:99, std:0.3646430969238281
如果我们使用Batch Normalization初始化,则数据变化更加合理。
import torch
import numpy as np
import torch.nn as nn
class MLP(nn.Module):
def __init__(self, neural_num, layers=100):
super(MLP, self).__init__()
self.linears = nn.ModuleList([nn.Linear(neural_num, neural_num, bias=False) for i in range(layers)])
self.bns = nn.ModuleList([nn.BatchNorm1d(neural_num) for i in range(layers)])
self.neural_num = neural_num
def forward(self, x):
# 前向传播
for (i, linear), bn in zip(enumerate(self.linears), self.bns):
x = linear(x)
x = bn(x)
x = torch.relu(x)
if torch.isnan(x.std()):
print("output is nan in {} layers".format(i))
break
print("layers:{}, std:{}".format(i, x.std().item()))
return x
# 权值初始化
def initialize(self):
for m in self.modules():
if isinstance(m, nn.Linear):
# method 1
# nn.init.normal_(m.weight.data, std=1) # normal: mean=0, std=1
# method 2 kaiming
nn.init.kaiming_normal_(m.weight.data)
neural_nums = 256
layer_nums = 100
batch_size = 16
net = MLP(neural_nums, layer_nums)
# net.initialize()
inputs = torch.randn((batch_size, neural_nums)) # normal: mean=0, std=1
output = net(inputs)
print(output)
layers:0, std:0.5896828174591064
layers:1, std:0.5799940824508667
layers:2, std:0.5818725824356079
layers:3, std:0.5713619589805603
layers:4, std:0.5826764106750488
layers:5, std:0.5809409022331238
layers:6, std:0.5783893465995789
layers:7, std:0.5778737664222717
layers:8, std:0.5741480588912964
layers:9, std:0.5806161165237427
layers:10, std:0.5795959830284119
...
layers:40, std:0.5803288817405701
layers:41, std:0.5799254775047302
layers:42, std:0.5809890627861023
layers:43, std:0.5694544911384583
layers:44, std:0.5809769630432129
layers:45, std:0.5793504118919373
layers:46, std:0.5785672664642334
layers:47, std:0.5776182413101196
layers:48, std:0.5859651565551758
layers:49, std:0.5858767032623291
...
layers:90, std:0.5870027542114258
layers:91, std:0.5834522247314453
layers:92, std:0.5806059837341309
layers:93, std:0.5809077024459839
layers:94, std:0.57407146692276
layers:95, std:0.5835222601890564
layers:96, std:0.577896773815155
layers:97, std:0.5759225487709045
layers:98, std:0.5789481401443481
layers:99, std:0.5877866148948669
有了bn层,我们可以不用精心设计网络初始化,甚至不用初始化。bn层一定要在激活函数之前来使用。
PyTorch中的Batch Normalization都继承基类_BatchNorm
__init__(self, num_features,
eps=1e-5,
momentum=0.1,
affine=True,
track_running_stats=True)
参数:
num_features:一个样本特征数量(最重要)eps:分母修正项,防止除以0导致的计算错误momentum:指数加权平均估计当前mean/varaffine:是否需要affine transform(恢复最优的输入数据分布)track_running_stats:是训练状态,还是测试状态基本方法:
nn.BatchNorm1d:输入数据大小为
(
b
a
t
c
h
∗
特征数
(
神经元数目
)
∗
1
d
特征
)
(batch*特征数(神经元数目)*1d特征)
(batch∗特征数(神经元数目)∗1d特征)
4个参数需要在batch维度上进行计算,在每个维度上都有这四个参数
import torch
import numpy as np
import torch.nn as nn
batch_size = 3
num_features = 5
momentum = 0.3
features_shape = (1)
feature_map = torch.ones(features_shape) # 1D
feature_maps = torch.stack([feature_map*(i+1) for i in range(num_features)], dim=0) # 2D
feature_maps_bs = torch.stack([feature_maps for i in range(batch_size)], dim=0) # 3D
print("input materials:\n{} shape is {}".format(feature_maps_bs, feature_maps_bs.shape))
bn = nn.BatchNorm1d(num_features=num_features, momentum=momentum)
running_mean, running_var = 0, 1
for i in range(2):
outputs = bn(feature_maps_bs)
print("\niteration:{}, running mean: {} ".format(i, bn.running_mean))
print("iteration:{}, running var:{} ".format(i, bn.running_var))
mean_t, var_t = 2, 0
running_mean = (1 - momentum) * running_mean + momentum * mean_t
running_var = (1 - momentum) * running_var + momentum * var_t
print("iteration:{}, 第二个特征的running mean: {} ".format(i, running_mean))
print("iteration:{}, 第二个特征的running var:{}".format(i, running_var))
input materials:
tensor([[[1.],
[2.],
[3.],
[4.],
[5.]],
[[1.],
[2.],
[3.],
[4.],
[5.]],
[[1.],
[2.],
[3.],
[4.],
[5.]]]) shape is torch.Size([3, 5, 1])
iteration:0, running mean: tensor([0.3000, 0.6000, 0.9000, 1.2000, 1.5000])
iteration:0, running var:tensor([0.7000, 0.7000, 0.7000, 0.7000, 0.7000])
iteration:0, 第二个特征的running mean: 0.6
iteration:0, 第二个特征的running var:0.7
iteration:1, running mean: tensor([0.5100, 1.0200, 1.5300, 2.0400, 2.5500])
iteration:1, running var:tensor([0.4900, 0.4900, 0.4900, 0.4900, 0.4900])
iteration:1, 第二个特征的running mean: 1.02
iteration:1, 第二个特征的running var:0.48999999999999994
nn.BatchNorm2d:输入数据大小为
(
b
a
t
c
h
∗
特征数
∗
2
d
特征
)
(batch*特征数*2d特征)
(batch∗特征数∗2d特征)
4个参数需要在batch维度上进行计算,在每个维度上都有这四个参数
import torch
import numpy as np
import torch.nn as nn
batch_size = 3
num_features = 6
momentum = 0.3
features_shape = (2, 2)
feature_map = torch.ones(features_shape) # 2D
feature_maps = torch.stack([feature_map*(i+1) for i in range(num_features)], dim=0) # 3D
feature_maps_bs = torch.stack([feature_maps for i in range(batch_size)], dim=0) # 4D
print("input materials:\n{} shape is {}".format(feature_maps_bs, feature_maps_bs.shape))
bn = nn.BatchNorm2d(num_features=num_features, momentum=momentum)
running_mean, running_var = 0, 1
for i in range(2):
outputs = bn(feature_maps_bs)
print("\niter:{}, running_mean.shape: {}".format(i, bn.running_mean.shape))
print("iter:{}, running_var.shape: {}".format(i, bn.running_var.shape))
print("iter:{}, weight.shape: {}".format(i, bn.weight.shape))
print("iter:{}, bias.shape: {}".format(i, bn.bias.shape))
input materials:
tensor([[[[1., 1.],
[1., 1.]],
[[2., 2.],
[2., 2.]],
[[3., 3.],
[3., 3.]],
[[4., 4.],
[4., 4.]],
[[5., 5.],
[5., 5.]],
[[6., 6.],
[6., 6.]]],
[[[1., 1.],
[1., 1.]],
[[2., 2.],
[2., 2.]],
[[3., 3.],
[3., 3.]],
[[4., 4.],
[4., 4.]],
[[5., 5.],
[5., 5.]],
[[6., 6.],
[6., 6.]]],
[[[1., 1.],
[1., 1.]],
[[2., 2.],
[2., 2.]],
[[3., 3.],
[3., 3.]],
[[4., 4.],
[4., 4.]],
[[5., 5.],
[5., 5.]],
[[6., 6.],
[6., 6.]]]]) shape is torch.Size([3, 6, 2, 2])
iter:0, running_mean.shape: torch.Size([6])
iter:0, running_var.shape: torch.Size([6])
iter:0, weight.shape: torch.Size([6])
iter:0, bias.shape: torch.Size([6])
iter:1, running_mean.shape: torch.Size([6])
iter:1, running_var.shape: torch.Size([6])
iter:1, weight.shape: torch.Size([6])
iter:1, bias.shape: torch.Size([6])
nn.BatchNorm3d:输入数据大小为
(
b
a
t
c
h
∗
特征数
∗
3
d
特征
)
(batch*特征数*3d特征)
(batch∗特征数∗3d特征)
4个参数需要在batch维度上进行计算,在每个维度上都有这四个参数
import torch
import numpy as np
import torch.nn as nn
batch_size = 3
num_features = 4
momentum = 0.3
features_shape = (2, 2, 3)
feature = torch.ones(features_shape) # 3D
feature_map = torch.stack([feature * (i + 1) for i in range(num_features)], dim=0) # 4D
feature_maps = torch.stack([feature_map for i in range(batch_size)], dim=0) # 5D
print("input materials:\n{} shape is {}".format(feature_maps, feature_maps.shape))
bn = nn.BatchNorm3d(num_features=num_features, momentum=momentum)
running_mean, running_var = 0, 1
for i in range(2):
outputs = bn(feature_maps)
print("\niter:{}, running_mean.shape: {}".format(i, bn.running_mean.shape))
print("iter:{}, running_var.shape: {}".format(i, bn.running_var.shape))
print("iter:{}, weight.shape: {}".format(i, bn.weight.shape))
print("iter:{}, bias.shape: {}".format(i, bn.bias.shape))
input materials:
tensor([[[[[1., 1., 1.],
[1., 1., 1.]],
[[1., 1., 1.],
[1., 1., 1.]]],
[[[2., 2., 2.],
[2., 2., 2.]],
[[2., 2., 2.],
[2., 2., 2.]]],
[[[3., 3., 3.],
[3., 3., 3.]],
[[3., 3., 3.],
[3., 3., 3.]]],
[[[4., 4., 4.],
[4., 4., 4.]],
[[4., 4., 4.],
[4., 4., 4.]]]],
[[[[1., 1., 1.],
[1., 1., 1.]],
[[1., 1., 1.],
[1., 1., 1.]]],
[[[2., 2., 2.],
[2., 2., 2.]],
[[2., 2., 2.],
[2., 2., 2.]]],
[[[3., 3., 3.],
[3., 3., 3.]],
[[3., 3., 3.],
[3., 3., 3.]]],
[[[4., 4., 4.],
[4., 4., 4.]],
[[4., 4., 4.],
[4., 4., 4.]]]],
[[[[1., 1., 1.],
[1., 1., 1.]],
[[1., 1., 1.],
[1., 1., 1.]]],
[[[2., 2., 2.],
[2., 2., 2.]],
[[2., 2., 2.],
[2., 2., 2.]]],
[[[3., 3., 3.],
[3., 3., 3.]],
[[3., 3., 3.],
[3., 3., 3.]]],
[[[4., 4., 4.],
[4., 4., 4.]],
[[4., 4., 4.],
[4., 4., 4.]]]]]) shape is torch.Size([3, 4, 2, 2, 3])
iter:0, running_mean.shape: torch.Size([4])
iter:0, running_var.shape: torch.Size([4])
iter:0, weight.shape: torch.Size([4])
iter:0, bias.shape: torch.Size([4])
iter:1, running_mean.shape: torch.Size([4])
iter:1, running_var.shape: torch.Size([4])
iter:1, weight.shape: torch.Size([4])
iter:1, bias.shape: torch.Size([4])
主要属性:
训练:均值和方差采用指数加权平均计算
测试:当前统计值
running_mean:均值running_var:方差weight:affine transform中的gammabias:affine transform中的beta 这几个Normalization方法的区别在于均值与方差的求取方式。
Batch Normalization(BN)
在一个batch中求均值与方差
Layer Normalizatoin(LN)
在一个网络层中求均值与方差
起因:BN不适用于变长的网络,如RNN
思路:逐层计算均值和方差

注意事项:
running_mean和running_var下面学习PyTorch中的Layer Normalizatoin
nn.LayerNorm(
normalized_shape,
eps=1e-05,
elementwise_affine=True)
主要参数:
normalized_shape:该层特征形状(重要参数)eps:分母修正项elementwise_affine:是否需要affine transform(逐元素进行)import torch
import numpy as np
import torch.nn as nn
batch_size = 3
num_features = 5
features_shape = (2, 2)
feature_map = torch.ones(features_shape) # 2D
feature_maps = torch.stack([feature_map * (i + 1) for i in range(num_features)], dim=0) # 3D
feature_maps_bs = torch.stack([feature_maps for i in range(batch_size)], dim=0) # 4D
# feature_maps_bs shape is [3, 5, 2, 2], B * C * H * W
ln = nn.LayerNorm(feature_maps_bs.size()[1:], elementwise_affine=True)
# 当设置为elementwise_affine=False时,不会生成权重
# ln = nn.LayerNorm(feature_maps_bs.size()[1:], elementwise_affine=False)
output = ln(feature_maps_bs)
print("Layer Normalization")
print(ln.weight.shape)
print(feature_maps_bs[0, ...])
print(output[0, ...])
Layer Normalization
torch.Size([5, 2, 2])
tensor([[[1., 1.],
[1., 1.]],
[[2., 2.],
[2., 2.]],
[[3., 3.],
[3., 3.]],
[[4., 4.],
[4., 4.]],
[[5., 5.],
[5., 5.]]])
tensor([[[-1.4142, -1.4142],
[-1.4142, -1.4142]],
[[-0.7071, -0.7071],
[-0.7071, -0.7071]],
[[ 0.0000, 0.0000],
[ 0.0000, 0.0000]],
[[ 0.7071, 0.7071],
[ 0.7071, 0.7071]],
[[ 1.4142, 1.4142],
[ 1.4142, 1.4142]]], grad_fn=<SelectBackward>)
Instance Normalizatoin(IN)
针对图像生成过程中的Normalizatoin方法,因为在图像生成当中,每个channel是一个风格,我们不能将风格不同的图像混为一谈,所以,不能加起来计算均值方差,因此,我们要逐通道(channel)的计算均值方差。
起因:BN在图像生成(Image Generation)中不适用
思路:逐Instance(channel)计算均值和方差
PyTorch中提供的Instance Normalizatoin的使用
nn.InstanceNorm2d(
num_features,
eps=1e-05,
momentum=0.1,
affine=False,
track_running_stats=False)
主要参数:
num_features:一个样本特征数量(最重要)eps:分母修正项momentum:指数加权平均估计当前mean/varaffine:是否需要affine transformtrack_running_stats:是训练状态,还是测试状态Batch Normalization一样的,也分一维二维与三维。这里不做详细介绍Group Normalizatoin(GN)
Group Normalizatoin是因为模型过大,GPU吃不了太多的数据,只能获取少量的batch size的数据,如果采用BN,就会导致均值方差估计不准确。因此,研究者提出数据不够,通道(分组)来凑。
起因:小batch样本中,BN估计的值不准
思路:数据不够,通道来凑
注意事项:
running_mean和running_vargamma和beta为逐通道(channel)的应用场景:大模型(小batch size)任务
nn.GroupNorm(
num_groups,
num_channels,
eps=1e-05,
affine=True)
主要参数:
num_groups:分组数(根据特征图的数量计算分组数)num_channels:通道数(特征数)eps:分母修正项affine:是否需要affine transformimport torch
import numpy as np
import torch.nn as nn
batch_size = 2
num_features = 4
num_groups = 2 # 3 Expected number of channels in input to be divisible by num_groups
features_shape = (2, 2)
feature_map = torch.ones(features_shape) # 2D
feature_maps = torch.stack([feature_map * (i + 1) for i in range(num_features)], dim=0) # 3D
feature_maps_bs = torch.stack([feature_maps * (i + 1) for i in range(batch_size)], dim=0) # 4D
gn = nn.GroupNorm(num_groups, num_features)
outputs = gn(feature_maps_bs)
print("Group Normalization")
print(gn.weight.shape)
print(outputs[0])
Group Normalization
torch.Size([4])
tensor([[[-1.0000, -1.0000],
[-1.0000, -1.0000]],
[[ 1.0000, 1.0000],
[ 1.0000, 1.0000]],
[[-1.0000, -1.0000],
[-1.0000, -1.0000]],
[[ 1.0000, 1.0000],
[ 1.0000, 1.0000]]], grad_fn=<SelectBackward>)
BN、LN、IN和GN都是为了克服Internal Covariate Shift (内部协变量偏移)问题。
