首先将数据集放入和底下代码同一目录中,然后导入一些相关函数的文件cnn_utils.py:
import math
import numpy as np
import h5py
import matplotlib.pyplot as plt
import tensorflow as tf
from tensorflow.python.framework import ops
def load_dataset():
train_dataset = h5py.File('datasets/train_signs.h5', "r")
train_set_x_orig = np.array(train_dataset["train_set_x"][:]) # your train set features
train_set_y_orig = np.array(train_dataset["train_set_y"][:]) # your train set labels
test_dataset = h5py.File('datasets/test_signs.h5', "r")
test_set_x_orig = np.array(test_dataset["test_set_x"][:]) # your test set features
test_set_y_orig = np.array(test_dataset["test_set_y"][:]) # your test set labels
classes = np.array(test_dataset["list_classes"][:]) # the list of classes
train_set_y_orig = train_set_y_orig.reshape((1, train_set_y_orig.shape[0]))
test_set_y_orig = test_set_y_orig.reshape((1, test_set_y_orig.shape[0]))
return train_set_x_orig, train_set_y_orig, test_set_x_orig, test_set_y_orig, classes
def random_mini_batches(X, Y, mini_batch_size = 64, seed = 0):
"""
Creates a list of random minibatches from (X, Y)
Arguments:
X -- input data, of shape (input size, number of examples) (m, Hi, Wi, Ci)
Y -- true "label" vector (containing 0 if cat, 1 if non-cat), of shape (1, number of examples) (m, n_y)
mini_batch_size - size of the mini-batches, integer
seed -- this is only for the purpose of grading, so that you're "random minibatches are the same as ours.
Returns:
mini_batches -- list of synchronous (mini_batch_X, mini_batch_Y)
"""
m = X.shape[0] # number of training examples
mini_batches = []
np.random.seed(seed)
# Step 1: Shuffle (X, Y)
permutation = list(np.random.permutation(m))
shuffled_X = X[permutation,:,:,:]
shuffled_Y = Y[permutation,:]
# Step 2: Partition (shuffled_X, shuffled_Y). Minus the end case.
num_complete_minibatches = math.floor(m/mini_batch_size) # number of mini batches of size mini_batch_size in your partitionning
for k in range(0, num_complete_minibatches):
mini_batch_X = shuffled_X[k * mini_batch_size : k * mini_batch_size + mini_batch_size,:,:,:]
mini_batch_Y = shuffled_Y[k * mini_batch_size : k * mini_batch_size + mini_batch_size,:]
mini_batch = (mini_batch_X, mini_batch_Y)
mini_batches.append(mini_batch)
# Handling the end case (last mini-batch < mini_batch_size)
if m % mini_batch_size != 0:
mini_batch_X = shuffled_X[num_complete_minibatches * mini_batch_size : m,:,:,:]
mini_batch_Y = shuffled_Y[num_complete_minibatches * mini_batch_size : m,:]
mini_batch = (mini_batch_X, mini_batch_Y)
mini_batches.append(mini_batch)
return mini_batches
def convert_to_one_hot(Y, C):
Y = np.eye(C)[Y.reshape(-1)].T
return Y
def forward_propagation_for_predict(X, parameters):
"""
Implements the forward propagation for the model: LINEAR -> RELU -> LINEAR -> RELU -> LINEAR -> SOFTMAX
Arguments:
X -- input dataset placeholder, of shape (input size, number of examples)
parameters -- python dictionary containing your parameters "W1", "b1", "W2", "b2", "W3", "b3"
the shapes are given in initialize_parameters
Returns:
Z3 -- the output of the last LINEAR unit
"""
# Retrieve the parameters from the dictionary "parameters"
W1 = parameters['W1']
b1 = parameters['b1']
W2 = parameters['W2']
b2 = parameters['b2']
W3 = parameters['W3']
b3 = parameters['b3']
# Numpy Equivalents:
Z1 = tf.add(tf.matmul(W1, X), b1) # Z1 = np.dot(W1, X) + b1
A1 = tf.nn.relu(Z1) # A1 = relu(Z1)
Z2 = tf.add(tf.matmul(W2, A1), b2) # Z2 = np.dot(W2, a1) + b2
A2 = tf.nn.relu(Z2) # A2 = relu(Z2)
Z3 = tf.add(tf.matmul(W3, A2), b3) # Z3 = np.dot(W3,Z2) + b3
return Z3
def predict(X, parameters):
W1 = tf.convert_to_tensor(parameters["W1"])
b1 = tf.convert_to_tensor(parameters["b1"])
W2 = tf.convert_to_tensor(parameters["W2"])
b2 = tf.convert_to_tensor(parameters["b2"])
W3 = tf.convert_to_tensor(parameters["W3"])
b3 = tf.convert_to_tensor(parameters["b3"])
params = {"W1": W1,
"b1": b1,
"W2": W2,
"b2": b2,
"W3": W3,
"b3": b3}
x = tf.placeholder("float", [12288, 1])
z3 = forward_propagation_for_predict(x, params)
p = tf.argmax(z3)
sess = tf.Session()
prediction = sess.run(p, feed_dict = {x: X})
return prediction
padding
卷积网络通常会出现两大问题:
- 图像缩小问题。正常卷积后会缩小高度和宽度,特别如果是深层的CNN,这样的问题会影响网络性能。
- 边缘信息丢失,因为边缘的值被卷积核卷积的次数较。
而padding可以很好的解决这两个问题。
首先实现padding填充的代码:
def zero_pad(X, pad):
"""
把数据集X的图像边界全部使用0来扩充pad个宽度和高度。
参数:
X - 图像数据集,维度为(样本数,图像高度,图像宽度,图像通道数)
pad - 整数,每个图像在垂直和水平维度上的填充量
返回:
X_paded - 扩充后的图像数据集,维度为(样本数,图像高度 + 2*pad,图像宽度 + 2*pad,图像通道数)
"""
X_paded = np.pad(X, (
(0, 0), # 样本数,不填充
(pad, pad), # 图像高度,你可以视为上面填充x个,下面填充y个(x,y)
(pad, pad), # 图像宽度,你可以视为左边填充x个,右边填充y个(x,y)
(0, 0)), # 通道数,不填充
'constant', constant_values=0) # 连续一样的值填充
return X_paded
可以测试代码看padding后的效果:
np.random.seed(1)
x = np.random.randn(4, 3, 3, 2)
x_paded = zero_pad(x, 2)
# 查看信息
print("x.shape =", x.shape)
print("x_paded.shape =", x_paded.shape)
print("x[1, 1] =", x[1, 1])
print("x_paded[1, 1] =", x_paded[1, 1])
# 绘制图
fig, axarr = plt.subplots(1, 2) # 一行两列
axarr[0].set_title('x')
axarr[0].imshow(x[0, :, :, 0])
axarr[1].set_title('x_paded')
axarr[1].imshow(x_paded[0, :, :, 0])
plt.show()
输出:
x.shape = (4, 3, 3, 2)
x_paded.shape = (4, 7, 7, 2)
x[1, 1] = [[ 0.90085595 -0.68372786]
[-0.12289023 -0.93576943]
[-0.26788808 0.53035547]]
x_paded[1, 1] = [[0. 0.]
[0. 0.]
[0. 0.]
[0. 0.]
[0. 0.]
[0. 0.]
[0. 0.]]
Conv
前向传播
单个卷积核的卷积过程如图(2D),3D则是相同通道的卷积核的每个通道与输入对应通道的进行2D的卷积然后相加,3D卷积的输出的通道数=卷积核的个数。
n
×
n
×
n
c
−
−
C
o
n
v
(
f
×
f
×
n
c
,
n
1
c
个
,
s
t
r
i
d
e
,
p
a
d
d
i
n
g
)
−
−
>
m
×
m
×
n
1
c
n×n×n_c--Conv(f×f×n_c,n1_c个, stride, padding)-->m×m×n1_c
n×n×nc−−Conv(f×f×nc,n1c个,stride,padding)−−>m×m×n1c
m
=
⌊
n
−
f
+
2
×
p
a
d
d
i
n
g
s
t
r
i
d
e
⌋
+
1
m = ⌊ \frac{n − f + 2 × p a dding }{ st r i d e }⌋ + 1
m=⌊striden−f+2×padding⌋+1
若除以步幅不为整数时要向下取整。
因为卷积核的个数决定输出的通道数,第l层的卷积核通道数=第l-1层的输出(即l层的输入)的通道数,第l层的卷积核的个数=第l+1层的输入(即第l层的输出)的通道数,所以l层的卷积核(大小×通道数×个数)为:
f
×
f
×
n
c
l
−
1
×
n
c
l
f×f×n^{l-1}_c×n^{l}_c
f×f×ncl−1×ncl
n
c
l
−
1
为
l
−
1
层的卷积核个数,
n
c
l
为
l
层的卷积核个数
n^{l-1}_c为l-1层的卷积核个数,n^{l}_c为l层的卷积核个数
ncl−1为l−1层的卷积核个数,ncl为l层的卷积核个数
# 单个卷积操作
def conv_single_step(a_slice_prev, W, b):
"""
在前一层的激活输出的一个片段上应用一个由参数W定义的过滤器。
这里切片大小和过滤器大小相同
参数:
a_slice_prev - 输入数据的一个片段,维度为(过滤器大小,过滤器大小,上一通道数)
W - 权重参数,包含在了一个矩阵中,维度为(过滤器大小,过滤器大小,上一通道数)
b - 偏置参数,包含在了一个矩阵中,维度为(1,1,1)
返回:
Z - 在输入数据的片X上卷积滑动窗口(w,b)的结果。
"""
s = np.multiply(a_slice_prev, W) + b
Z = np.sum(s)
return Z
def conv_forward(A_prev, W, b, hparameters):
"""
实现卷积函数的前向传播
参数:
A_prev - 上一层的激活输出矩阵,维度为(m, n_H_prev, n_W_prev, n_C_prev),(样本数量,上一层图像的高度,上一层图像的宽度,上一层过滤器数量)
W - 权重矩阵,维度为(f, f, n_C_prev, n_C),(过滤器大小,过滤器大小,上一层的过滤器数量,这一层的过滤器数量)
b - 偏置矩阵,维度为(1, 1, 1, n_C),(1,1,1,这一层的过滤器数量)
hparameters - 包含了"stride"与 "pad"的超参数字典。
返回:
Z - 卷积输出,维度为(m, n_H, n_W, n_C),(样本数,图像的高度,图像的宽度,过滤器数量)
cache - 缓存了一些反向传播函数conv_backward()需要的一些数据
"""
# 获取来自上一层数据的基本信息
(m, n_H_prev, n_W_prev, n_C_prev) = A_prev.shape
# 获取权重矩阵的基本信息
(f, f, n_C_prev, n_C) = W.shape
# 获取超参数hparameters的值
stride = hparameters["stride"]
pad = hparameters["pad"]
# 计算卷积后的图像的宽度高度,参考上面的公式,使用int()来进行板除
n_H = int((n_H_prev - f + 2 * pad) / stride) + 1
n_W = int((n_W_prev - f + 2 * pad) / stride) + 1
# 使用0来初始化卷积输出Z
Z = np.zeros((m, n_H, n_W, n_C))
# 通过A_prev创建填充过了的A_prev_pad
A_prev_pad = zero_pad(A_prev, pad)
for i in range(m): # 遍历样本
a_prev_pad = A_prev_pad[i] # 选择第i个样本的扩充后的激活矩阵
for h in range(n_H): # 在输出的垂直轴上循环
for w in range(n_W): # 在输出的水平轴上循环
for c in range(n_C): # 循环遍历输出的通道
# 定位当前的切片位置
vert_start = h * stride # 竖向,开始的位置
vert_end = vert_start + f # 竖向,结束的位置
horiz_start = w * stride # 横向,开始的位置
horiz_end = horiz_start + f # 横向,结束的位置
# 切片位置定位好了我们就把它取出来,需要注意的是我们是“穿透”取每层通道的(3D层面)
# 自行脑补一下吸管插入一层层的橡皮泥就明白了
a_slice_prev = a_prev_pad[vert_start:vert_end, horiz_start:horiz_end, :]
# 执行单步卷积,遍历到的切片与第c个卷积核进行单卷集操作,得到输出的第c层的对应的一个值
Z[i, h, w, c] = conv_single_step(a_slice_prev, W[:, :, :, c], b[0, 0, 0, c])
# 数据处理完毕,验证数据格式是否正确
assert (Z.shape == (m, n_H, n_W, n_C))
# 存储一些缓存值,以便于反向传播使用
cache = (A_prev, W, b, hparameters)
return (Z, cache)
'''
卷积层的最后应该要进行激活的,activation可以是sigmoid、relu、tanh等,但这里我们就不这么做了,
只是用无框架代码解释其底层逻辑
A[i, h, w, c] = activation(Z[i, h, w, c])
'''
这里遍历切片的位置定义可参考下图:
测试代码:
np.random.seed(1)
A_prev = np.random.randn(10, 4, 4, 3)
W = np.random.randn(2, 2, 3, 8)
b = np.random.randn(1, 1, 1, 8)
hparameters = {"pad": 2, "stride": 1}
Z, cache_conv = conv_forward(A_prev, W, b, hparameters)
print("np.mean(Z) = ", np.mean(Z))
print("A_prev.shape =", cache_conv[0].shape)
print("W.shape =", cache_conv[1].shape)
print("b.shape =", cache_conv[2].shape)
输出:
np.mean(Z) = 0.15585932488906465
A_prev.shape = (10, 4, 4, 3)
W.shape = (2, 2, 3, 8)
b.shape = (1, 1, 1, 8)
可以看到输入、卷积核及偏差的shape与测试代码中的一致。
反向传播
若是使用DL框架,只需要写出前向传播框架就会自动计算出反向传播的梯度值。
首先列出反向传播所需要的计算公式:
- 计算dA的:
d A + = ∑ h = 0 n H ∑ w = 0 n W W c × d Z h w (1) dA += \sum ^{n_H} _{h=0} \sum ^{n_W} _{w=0} W_{c} \times dZ_{hw} \tag{1} dA+=h=0∑nHw=0∑nWWc×dZhw(1)
其中 W c W_c Wc是过滤器, Z h w Z_{hw} Zhw是一个标量,是卷积层第h行第w列的使用点乘计算后的输出Z的梯度(即输出Z的每个切片对应的梯度)。
需要注意的是在每次更新dA的时候,都会用第c个过滤器 W c W_c Wc乘以对应的第c层通道的dZ,因为在前向传播的时候,每个过滤器都与a_slice进行了点乘相加得到某一层通道的一个值(即输入的每一块切片都是用不同的卷积核卷积过,第c个卷积核与其卷积得到第c层通道的一个值),所以在计算dA的每一块的切片的梯度的时候,我们是计算每个卷积核与对应层的dZ的反向操作的结果之和。公式可以如下表示:
da_perv_pad[vert_start:vert_end,horiz_start:horiz_end,:] += W[:,:,:,c] * dZ[i,h,w,c]
如图(网图加上自己的画笔)是一个单个卷积操作的正向过程,可以看到是一个卷积核对一个切片(3D)进行卷积是得到了某一层的一个值(2D)(图中的红蓝框);而反向计算我们也按这个顺序反向,每个卷积核与dZ对应层(通道)的一个值(2D)相乘然后每个结果相加,就得到对应切片的范围的dA(3D)。大概更直观的过程就是:
正向过程:
3
D
∗
卷积核
−
−
>
2
D
正向过程:3D * 卷积核 --> 2D
正向过程:3D∗卷积核−−>2D
反向过程:
2
D
∗
卷积核
−
−
>
3
D
反向过程:2D * 卷积核 --> 3D
反向过程:2D∗卷积核−−>3D
- 计算dW的:
d
W
c
+
=
∑
h
=
0
n
H
∑
w
=
0
n
W
a
s
l
i
c
e
×
d
Z
h
w
(2)
dW_c += \sum^{n_H}_{h=0} \sum^{n_W}_{w=0}a_{slice} \times dZ_{hw} \tag{2}
dWc+=h=0∑nHw=0∑nWaslice×dZhw(2)
其中,
a
s
l
i
c
e
a_{slice}
aslice 对应着
Z
i
j
Z_{ij}
Zij的激活值。由此,我们就可以推导W的梯度,因为我们使用了卷积核来对数据进行窗口滑动,在这里,我们实际上是切出了和过滤器一样大小的切片,切了多少次就产生了多少个梯度,所以我们需要把它们加起来得到这个数据集的整体dW。用公式概括:
dW[:,:,:, c] += a_slice * dZ[i , h , w , c]
同样可以参考上面的dA求解过程即配图。单步卷积的正向是一个切片与一个卷积核卷积得到一层的一个值,而输入的每个切片都是用到这个卷积核(但一个卷积层有一个或多个卷积核),所以反向操作是每个卷积核的梯度是所有切片与某一层通道(这个卷积核卷积后得到的对应层)的所有值一一对应相乘后相加。大概更直观的过程就是:
正向过程:卷积核
1
−
−
>
输出结果的第
1
层通道,卷积核
2
−
−
>
输出结果的第
2
层通道
正向过程:卷积核1-->输出结果的第1层通道,卷积核2-->输出结果的第2层通道
正向过程:卷积核1−−>输出结果的第1层通道,卷积核2−−>输出结果的第2层通道
反向过程:输出结果的第
1
层通道
−
−
>
卷积核
1
,输出结果的第
2
层通道
−
−
>
卷积核
2
反向过程:输出结果的第1层通道-->卷积核1,输出结果的第2层通道-->卷积核2
反向过程:输出结果的第1层通道−−>卷积核1,输出结果的第2层通道−−>卷积核2
- 计算db的:
每个卷积核只对应一个b的值,所以b的梯度比较好计算:
d b = ∑ h ∑ w d Z h w (3) db = \sum^{}_{h} \sum^{}_{w}dZ_{hw} \tag{3} db=h∑w∑dZhw(3)
可以用一行代码进行概括:
db[:,:,:,c] += dZ[ i, h, w, c]
因为一个卷积核对应一个b,输出结果的每一层对应的是一个卷积核,所以输出结果的每一层对应的是一个b。
接下来用这3个公式求卷积的反向过程:
def conv_backward(dZ, cache):
"""
实现卷积层的反向传播
参数:
dZ - 卷积层的输出Z的 梯度,维度为(m, n_H, n_W, n_C) 取决于使用的是什么激活函数
cache - 反向传播所需要的参数,conv_forward()的输出之一
返回:
dA_prev - 卷积层的输入(A_prev)的梯度值,维度为(m, n_H_prev, n_W_prev, n_C_prev)
dW - 卷积层的权值的梯度,维度为(f,f,n_C_prev,n_C)
db - 卷积层的偏置的梯度,维度为(1,1,1,n_C)
"""
# 获取cache的值
(A_prev, W, b, hparameters) = cache
# 获取A_prev的基本信息
(m, n_H_prev, n_W_prev, n_C_prev) = A_prev.shape
# 获取dZ的基本信息
(m, n_H, n_W, n_C) = dZ.shape
# 获取权值的基本信息
(f, f, n_C_prev, n_C) = W.shape
# 获取hparaeters的值
pad = hparameters["pad"]
stride = hparameters["stride"]
# 初始化各个梯度的结构
dA_prev = np.zeros((m, n_H_prev, n_W_prev, n_C_prev))
dW = np.zeros((f, f, n_C_prev, n_C))
db = np.zeros((1, 1, 1, n_C))
# 前向传播中我们使用了pad,反向传播也需要使用,这是为了保证数据结构一致
A_prev_pad = zero_pad(A_prev, pad)
dA_prev_pad = zero_pad(dA_prev, pad)
# 现在处理数据
for i in range(m):
# 选择第i个扩充了的数据的样本,降了一维。
a_prev_pad = A_prev_pad[i]
da_prev_pad = dA_prev_pad[i]
for h in range(n_H):
for w in range(n_W):
for c in range(n_C):
# 定位切片位置
vert_start = h
vert_end = vert_start + f
horiz_start = w
horiz_end = horiz_start + f
# 定位完毕,开始切片
a_slice = a_prev_pad[vert_start:vert_end, horiz_start:horiz_end, :]
# 切片完毕,使用上面的公式计算梯度,第c个卷积核与dZ的第c层通道反向计算得到对应的3D切片
da_prev_pad[vert_start:vert_end, horiz_start:horiz_end, :] += W[:, :, :, c] * dZ[i, h, w, c]
dW[:, :, :, c] += a_slice * dZ[i, h, w, c]
db[:, :, :, c] += dZ[i, h, w, c]
# 设置第i个样本最终的dA_prev,即把非填充的数据取出来。dA_prev的shape要和A_prev的保持一致才是正确的,下面assert验证
dA_prev[i, :, :, :] = da_prev_pad[pad:-pad, pad:-pad, :]
# 数据处理完毕,验证数据格式是否正确
assert (dA_prev.shape == (m, n_H_prev, n_W_prev, n_C_prev))
return (dA_prev, dW, db)
测试上面的代码:
np.random.seed(1)
# 初始化参数
A_prev = np.random.randn(10, 4, 4, 3)
W = np.random.randn(2, 2, 3, 8)
b = np.random.randn(1, 1, 1, 8)
hparameters = {"pad": 2, "stride": 1}
# 前向传播
Z, cache_conv = conv_forward(A_prev, W, b, hparameters)
# 反向传播
dA, dW, db = conv_backward(Z, cache_conv)
print("dA.shape =", dA.shape)
print("dW.shape =", dW.shape)
print("db.shape =", db.shape)
输出:
dA.shape = (10, 4, 4, 3)
dW.shape = (2, 2, 3, 8)
db.shape = (1, 1, 1, 8)
可以看到输入的dA、卷积核的dW及偏差的db的shape都与之相对应,从正向传播的输出输入到反向传播过程中而产生的结果无误。
pooling
正向传播
池化层可以减小图像的宽高(通道数不变),池化核对每层分别进行池化操作。池化层是CNN的静态属性,没有参数进行学习更新,只需要设置超参即可(size、stride、pooling等)。
池化层可以缩减模型大小,提高计算速度;提高提取特征的鲁棒性。
池化分为最大值池化和均值池化:
- max pooling:取池化核区域内的最大值
- 项目2取池化核区域内的平均值
池化操作后的高宽的计算跟卷积一样,不同的是输出的通道数不变:
n
c
l
=
n
c
l
−
1
n^l_c = n^{l-1}_c
ncl=ncl−1
def pool_forward(A_prev, hparameters, mode="max"):
"""
实现池化层的前向传播
参数:
A_prev - 输入数据,维度为(m, n_H_prev, n_W_prev, n_C_prev)
hparameters - 包含了 "f" 和 "stride"的超参数字典
mode - 模式选择【"max" | "average"】
返回:
A - 池化层的输出,维度为 (m, n_H, n_W, n_C)
cache - 存储了一些反向传播需要用到的值,包含了输入和超参数的字典。
"""
# 获取输入数据的基本信息
(m, n_H_prev, n_W_prev, n_C_prev) = A_prev.shape
# 获取超参数的信息
f = hparameters["f"]
stride = hparameters["stride"]
# 计算输出维度
n_H = int((n_H_prev - f) / stride) + 1
n_W = int((n_W_prev - f) / stride) + 1
n_C = n_C_prev
# 初始化输出矩阵
A = np.zeros((m, n_H, n_W, n_C))
for i in range(m): # 遍历样本
for h in range(n_H): # 在输出的垂直轴上循环
for w in range(n_W): # 在输出的水平轴上循环
for c in range(n_C): # 循环遍历输出的通道
# 定位当前的切片位置
vert_start = h * stride # 竖向,开始的位置
vert_end = vert_start + f # 竖向,结束的位置
horiz_start = w * stride # 横向,开始的位置
horiz_end = horiz_start + f # 横向,结束的位置
# 遍历到的区域切片,这里是对c层通道进行切片(2D层面)
a_slice_prev = A_prev[i, vert_start:vert_end, horiz_start:horiz_end, c]
# 对切片进行池化操作,对遍历到的第c层的区域进行池化操作
if mode == "max":
A[i, h, w, c] = np.max(a_slice_prev)
elif mode == "average":
A[i, h, w, c] = np.mean(a_slice_prev)
# 池化完毕,校验数据格式
assert (A.shape == (m, n_H, n_W, n_C))
# 校验完毕,开始存储用于反向传播的值
cache = (A_prev, hparameters)
return A, cache
测试代码:
np.random.seed(1)
A_prev = np.random.randn(2, 4, 4, 3)
hparameters = {"f": 4, "stride": 1}
A, cache = pool_forward(A_prev, hparameters, mode="max")
print("mode = max")
print("A =", A)
print("----------------------------")
A, cache = pool_forward(A_prev, hparameters, mode="average")
print("mode = average")
print("A =", A)
输出结果:
mode = max
A = [[[[1.74481176 1.6924546 2.10025514]]]
[[[1.19891788 1.51981682 2.18557541]]]]
----------------------------
mode = average
A = [[[[-0.09498456 0.11180064 -0.14263511]]]
[[[-0.09525108 0.28325018 0.33035185]]]]
4×4×3的输入经过stride为1的4×4的池化之后输出为1×1×3,样本数为2,输出的shape为(2, 1, 1, 3),没有错误。
反向传播
即使池化层没有反向传播过程中要更新的参数,我们仍然需要通过池化层反向传播梯度,以便为在池化层之前的层(比如卷积层)计算梯度。
- max pooling的反向传播:
创建一个create_mask_from_window()
的函数,用来记池化核区域的最大值的位置。因为正向的操作是取输入的池化核区域内的最大值一个值,而反向操作是还是取这个输入,在对应的池化核区域内只保留最大值的值其余置0,所以我们先记录最大值位置的矩阵通过相乘来达到我们的目的。
X = [ 1 3 4 2 ] → M = [ 0 0 1 0 ] X = \begin{bmatrix} 1 && 3 \\ 4 && 2 \end {bmatrix} \quad \rightarrow \quad M = \begin{bmatrix} 0 && 0 \\ 1 && 0 \end {bmatrix} X=[1432]→M=[0100]
正向传播首先是经过卷积层,然后滑动地取卷积层最大值构成了池化层,要记录最大值的位置,才能反向传播到卷积层。
def create_mask_from_window(x):
"""
从输入矩阵中创建掩码,以保存最大值的矩阵的位置。
参数:
x - 一个维度为(f,f)的矩阵
返回:
mask - 包含x的最大值的位置的矩阵
"""
mask = x == np.max(x)
return mask
测试一下代码,看输出结果:
np.random.seed(1)
x = np.random.randn(2, 3)
mask = create_mask_from_window(x)
print("x = " + str(x))
print("mask = " + str(mask))
输出:
x = [[ 1.62434536 -0.61175641 -0.52817175]
[-1.07296862 0.86540763 -2.3015387 ]]
mask = [[ True False False]
[False False False]]
是以布尔值为矩阵来记录最大值的位置。
- average pooling的反向传播:
而均值池化的正向操作是取池化核内的平均值得到一个值,所以反向操作就根据后一层的梯度的其中一个值来得到对应区域内的所有平均值。
d Z = 1 → d Z = [ 1 4 1 4 1 4 1 4 ] dZ=1 \quad \rightarrow \quad dZ = \begin{bmatrix} \frac{1}{4} & \frac{1}{4} \\ \frac{1}{4} & \frac{1}{4} \\ \end{bmatrix} dZ=1→dZ=[41414141]
def distribute_value(dz, shape):
"""
给定一个值,为按矩阵大小平均分配到每一个矩阵位置中。
参数:
dz - 输入的实数
shape - 元组,两个值,分别为n_H , n_W
返回:
a - 已经分配好了值的矩阵,里面的值全部一样。
"""
# 获取矩阵的大小
(n_H, n_W) = shape
# 计算平均值
average = dz / (n_H * n_W)
# 填充入矩阵
a = np.ones(shape) * average
return a
测试:
dz = 2
shape = (2, 2)
a = distribute_value(dz, shape)
print("a = " + str(a))
输出:
a = [[0.5 0.5]
[0.5 0.5]]
可以看到2平均分为2×2=4份平均值为0.5验正确。
池化层的反向操作的实现:
def pool_backward(dA, cache, mode="max"):
"""
实现池化层的反向传播
参数:
dA - 池化层的输出的梯度,和池化层的输出的维度一样
cache - 池化层前向传播时所存储的参数。
mode - 模式选择,【"max" | "average"】
返回:
dA_prev - 池化层的输入的梯度,和A_prev的维度相同
"""
# 获取cache中的值
(A_prev, hparaeters) = cache
# 获取hparaeters的值
f = hparaeters["f"]
stride = hparaeters["stride"]
# 获取A_prev和dA的基本信息
(m, n_H_prev, n_W_prev, n_C_prev) = A_prev.shape
(m, n_H, n_W, n_C) = dA.shape
# 初始化输出的结构
dA_prev = np.zeros_like(A_prev)
# 开始处理数据
for i in range(m):
a_prev = A_prev[i]
for h in range(n_H):
for w in range(n_W):
for c in range(n_C):
# 定位切片位置
vert_start = h
vert_end = vert_start + f
horiz_start = w
horiz_end = horiz_start + f
# 选择反向传播的计算方式
if mode == "max":
# 开始切片
a_prev_slice = a_prev[vert_start:vert_end, horiz_start:horiz_end, c]
# 创建掩码
mask = create_mask_from_window(a_prev_slice)
# 计算dA_prev,一层一层对应的反操作
dA_prev[i, vert_start:vert_end, horiz_start:horiz_end, c] += np.multiply(mask, dA[i, h, w, c])
elif mode == "average":
# 获取dA的值
da = dA[i, h, w, c]
# 定义过滤器大小
shape = (f, f)
# 平均分配
dA_prev[i, vert_start:vert_end, horiz_start:horiz_end, c] += distribute_value(da, shape)
# 数据处理完毕,开始验证格式
assert (dA_prev.shape == A_prev.shape)
return dA_prev
测试一下代码:
np.random.seed(1)
A_prev = np.random.randn(5, 5, 3, 2)
hparameters = {"stride": 1, "f": 2}
A, cache = pool_forward(A_prev, hparameters)
dA = np.random.randn(5, 4, 2, 2)
dA_prev = pool_backward(dA, cache, mode="max")
print("mode = max")
print('mean of dA = ', np.mean(dA))
print('dA_prev[1,1] = ', dA_prev[1, 1])
print()
dA_prev = pool_backward(dA, cache, mode="average")
print("mode = average")
print('mean of dA = ', np.mean(dA))
print('dA_prev[1,1] = ', dA_prev[1, 1])
输出:
mode = max
mean of dA = 0.14571390272918056
dA_prev[1,1] = [[ 0. 0. ]
[ 5.05844394 -1.68282702]
[ 0. 0. ]]
mode = average
mean of dA = 0.14571390272918056
dA_prev[1,1] = [[ 0.08485462 0.2787552 ]
[ 1.26461098 -0.25749373]
[ 1.17975636 -0.53624893]]
无报错证明通过了assert验证,dA_prev[1, 1]输出的是第二个样本的第二行的切片,因为宽度为3,通道数为2,所以输出了个shape为(3, 2)的矩阵也没问题。
因为池化是对每一层分别池化进行池化操作(不改变通道数的操作),所以反操作也是一层一层对应的反操作。