『迷你教程』Keras六步实现简单循环神经网络_keras simplernn overide

Keras 算是神经网络框架中封装比较高级的网络，虽然很高级但是实现起来也是非常的简单，如果作为算法工程师入门级别的技能来说简单的 6 步就可以实现一个简单的神经网络了。

这个往往也是面试算法工程师一个比较基础的面试题，考量的是求职者的基本功是否扎实，来看看如何在 6 步就能实现一个简单的回归网络吧。

了解循环神经网络 (RNN) 的工作原理以及如何通过 Keras 深度学习库使用。就先需要了解RNN的结构、给定输入时，RNN 如何计算输出、如何在 Keras 中为 SimpleRNN 准备数据、如何训练 SimpleRNN 模型。

RNN 的依赖项

from pandas import read_csv
import numpy as np
from keras.models import Sequential
from keras.layers import Dense, SimpleRNN
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error
import math
import matplotlib.pyplot as plt
1
2
3
4
5
6
7
8

Keras SimpleRNN

下面的函数返回一个模型，该模型包括一个SimpleRNN层和一个Dense用于学习顺序数据的层。在input_shape指定的参数(time_steps x features)。我们将简化一切并使用单变量数据，即只有一个特征；下面讨论 time_steps。

def create_RNN(hidden_units, dense_units, input_shape, activation):
    model = Sequential()
    model.add(SimpleRNN(hidden_units, input_shape=input_shape, 
                        activation=activation[0]))
    model.add(Dense(units=dense_units, activation=activation[1]))
    model.compile(loss='mean_squared_error', optimizer='adam')
    return model

demo_model = create_RNN(2, 1, (3,1), activation=['linear', 'linear'])
1
2
3
4
5
6
7
8
9

demo_model 返回的对象带有 2 个通过SimpleRNN层创建的隐藏单元和 1 个通过Dense层创建的密集单元。在input_shape被设定在3×1和linear激活功能是在两个层为简单起见使用。只是回忆一下线性激活函数不改变输入。网络如下所示：

如果我们有 隐藏单位（ 在上述情况下），则：

输入：$x\in R$
隐藏单位：$h，R^m$
输入单位的权重：$w_x，以R^m$为单位
隐藏单位的权重：$w_h(R^{MXM}$)
隐藏单位的偏差：$b_h(R^m$)
密集层的权重：$w_y(R^m$)
致密层的偏压：$b_y\in R$

让我们看看上面的权重。注意：由于权重是随机初始化的，所以这里粘贴的结果会和你的不一样。重要的是要了解正在使用的每个对象的结构是什么样的，以及它如何与其他对象交互以产生最终输出。

wx = demo_model.get_weights()[0]
wh = demo_model.get_weights()[1]
bh = demo_model.get_weights()[2]
wy = demo_model.get_weights()[3]
by = demo_model.get_weights()[4]


print('wx = ', wx, ' wh = ', wh, ' bh = ', bh, ' wy =', wy, 'by = ', by)
1
2
3
4
5
6
7
8

输出结果为

wx =  [[ 0.18662322 -1.2369459 ]]  wh =  [[ 0.86981213 -0.49338293]
 [ 0.49338293  0.8698122 ]]  bh =  [0. 0.]  wy = [[-0.4635998]
 [ 0.6538409]] by =  [0.]
1
2
3

现在让我们做一个简单的实验，看看来自 SimpleRNN 和 Dense 层的层是如何产生输出的。记住这个数字。

我们将输入x三个时间步长，然后让网络生成一个输出。将计算时间步长 1、2 和 3 中隐藏单元的值。初始化为零向量。输出 是从 和 . 由于我们使用的是线性单元，因此不需要激活函数。

x = np.array([1, 2, 3])
# 将输入重塑为所需的sample_size x time_steps x特性  
x_input = np.reshape(x,(1, 3, 1))
y_pred_model = demo_model.predict(x_input)


m = 2
h0 = np.zeros(m)
h1 = np.dot(x[0], wx) + h0 + bh
h2 = np.dot(x[1], wx) + np.dot(h1,wh) + bh
h3 = np.dot(x[2], wx) + np.dot(h2,wh) + bh
o3 = np.dot(h3, wy) + by

print('h1 = ', h1,'h2 = ', h2,'h3 = ', h3)

print("Prediction from network ", y_pred_model)
print("Prediction from our computation ", o3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

输出结果为

h1 =  [[ 0.18662322 -1.23694587]] h2 =  [[-0.07471441 -3.64187904]] h3 =  [[-1.30195881 -6.84172557]]
Prediction from network  [[-3.8698118]]
Prediction from our computation  [[-3.86981216]]
1
2
3

在太阳黑子数据集上运行 RNN

现在我们了解了 SimpleRNN 和 Dense 层是如何组合在一起的。让我们在一个简单的时间序列数据集上运行一个完整的 RNN。我们需要按照以下步骤操作

1. 从给定的 URL 读取数据集
2. 将数据拆分为训练集和测试集
3. 准备输入所需的 Keras 格式
4. 创建一个 RNN 模型并训练它
5. 对训练集和测试集进行预测，并在两个集上打印均方根误差
6. 查看结果

步骤 1、2：读取数据并拆分为训练和测试

以下函数从给定的 URL 读取训练和测试数据，并将其拆分为给定百分比的训练和测试数据。在使用MinMaxScaler来自 scikit-learn的数据在 0 和 1 之间缩放后，它返回训练和测试数据的一维数组。

# 参数split_percent定义了训练示例的比例  
def get_train_test(url, split_percent=0.8):
    df = read_csv(url, usecols=[1], engine='python')
    data = np.array(df.values.astype('float32'))
    scaler = MinMaxScaler(feature_range=(0, 1))
    data = scaler.fit_transform(data).flatten()
    n = len(data)
    # 将数据分解为训练和测试
    split = int(n*split_percent)
    train_data = data[range(split)]
    test_data = data[split:]
    return train_data, test_data, data

sunspots_url = 'https://raw.githubusercontent.com/jbrownlee/Datasets/master/monthly-sunspots.csv'
train_data, test_data, data = get_train_test(sunspots_url)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

第 3 步：为 Keras 重塑数据

下一步是为 Keras 模型训练准备数据。输入数组应当成形为：total_samples x time_steps x features。

有很多方法可以为训练准备时间序列数据。我们将创建具有非重叠时间步长的输入行。下图显示了 time_steps = 2 的示例。这里 time_steps 表示用于预测时间序列数据的下一个值的先前时间步数。

以下函数get_XY()将一维数组作为输入并将其转换为所需的输入X 和目标Y 数组。我们将使用 12time_steps作为太阳黑子数据集，因为太阳黑子通常有 12 个月的周期。您可以尝试使用 的其他值time_steps。

# 准备输入X和目标Y
def get_XY(dat, time_steps):
    # Indices of target array
    Y_ind = np.arange(time_steps, len(dat), time_steps)
    Y = dat[Y_ind]
    # Prepare X
    rows_x = len(Y)
    X = dat[range(time_steps*rows_x)]
    X = np.reshape(X, (rows_x, time_steps, 1))    
    return X, Y

time_steps = 12
trainX, trainY = get_XY(train_data, time_steps)
testX, testY = get_XY(test_data, time_steps)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

第 4 步：创建 RNN 模型并训练

对于这一步，我们可以重用create_RNN()上面定义的函数。

model = create_RNN(hidden_units=3, dense_units=1, input_shape=(time_steps,1), 
                   activation=['tanh', 'tanh'])
model.fit(trainX, trainY, epochs=20, batch_size=1, verbose=2)
1
2
3

第 5 步：计算并打印均方根误差

该函数print_error()计算实际值和预测值之间的均方误差。

def print_error(trainY, testY, train_predict, test_predict):    
    # 错误的预测
    train_rmse = math.sqrt(mean_squared_error(trainY, train_predict))
    test_rmse = math.sqrt(mean_squared_error(testY, test_predict))
    # 打印RMSE
    print('Train RMSE: %.3f RMSE' % (train_rmse))
    print('Test RMSE: %.3f RMSE' % (test_rmse))    

# 做出预测
train_predict = model.predict(trainX)
test_predict = model.predict(testX)
# 均方误差
print_error(trainY, testY, train_predict, test_predict)
1
2
3
4
5
6
7
8
9
10
11
12
13

输出结果

Train RMSE: 0.058 RMSE
Test RMSE: 0.077 RMSE
1
2

第六步：查看结果

以下函数绘制实际目标值和预测值。红线将训练和测试数据点分开。

# 绘图
def plot_result(trainY, testY, train_predict, test_predict):
    actual = np.append(trainY, testY)
    predictions = np.append(train_predict, test_predict)
    rows = len(actual)
    plt.figure(figsize=(15, 6), dpi=80)
    plt.plot(range(rows), actual)
    plt.plot(range(rows), predictions)
    plt.axvline(x=len(trainY), color='r')
    plt.legend(['Actual', 'Predictions'])
    plt.xlabel('Observation number after given time steps')
    plt.ylabel('Sunspots scaled')
    plt.title('Actual and Predicted Values. The Red Line Separates The Training And Test Examples')
plot_result(trainY, testY, train_predict, test_predict)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

生成以下图：

合并代码
下面给出的是本教程的完整代码。一定要在最后尝试一下，并尝试不同的隐藏单元和时间步长。您可以SimpleRNN向网络添加一秒钟，看看它的行为。您还可以使用该scaler对象将数据重新缩放回其正常范围。

# 参数split_percent定义了训练示例的比例  
def get_train_test(url, split_percent=0.8):
    df = read_csv(url, usecols=[1], engine='python')
    data = np.array(df.values.astype('float32'))
    scaler = MinMaxScaler(feature_range=(0, 1))
    data = scaler.fit_transform(data).flatten()
    n = len(data)
    # 将数据分解为训练和测试
    split = int(n*split_percent)
    train_data = data[range(split)]
    test_data = data[split:]
    return train_data, test_data, data

# 准备输入X和目标Y
def get_XY(dat, time_steps):
    Y_ind = np.arange(time_steps, len(dat), time_steps)
    Y = dat[Y_ind]
    rows_x = len(Y)
    X = dat[range(time_steps*rows_x)]
    X = np.reshape(X, (rows_x, time_steps, 1))    
    return X, Y

def create_RNN(hidden_units, dense_units, input_shape, activation):
    model = Sequential()
    model.add(SimpleRNN(hidden_units, input_shape=input_shape, activation=activation[0]))
    model.add(Dense(units=dense_units, activation=activation[1]))
    model.compile(loss='mean_squared_error', optimizer='adam')
    return model

def print_error(trainY, testY, train_predict, test_predict):    
    # 错误的预测
    train_rmse = math.sqrt(mean_squared_error(trainY, train_predict))
    test_rmse = math.sqrt(mean_squared_error(testY, test_predict))
    # 打印RMSE
    print('Train RMSE: %.3f RMSE' % (train_rmse))
    print('Test RMSE: %.3f RMSE' % (test_rmse))    

# 绘图
def plot_result(trainY, testY, train_predict, test_predict):
    actual = np.append(trainY, testY)
    predictions = np.append(train_predict, test_predict)
    rows = len(actual)
    plt.figure(figsize=(15, 6), dpi=80)
    plt.plot(range(rows), actual)
    plt.plot(range(rows), predictions)
    plt.axvline(x=len(trainY), color='r')
    plt.legend(['Actual', 'Predictions'])
    plt.xlabel('Observation number after given time steps')
    plt.ylabel('Sunspots scaled')
    plt.title('Actual and Predicted Values. The Red Line Separates The Training And Test Examples')

sunspots_url = 'https://raw.githubusercontent.com/jbrownlee/Datasets/master/monthly-sunspots.csv'
time_steps = 12
train_data, test_data, data = get_train_test(sunspots_url)
trainX, trainY = get_XY(train_data, time_steps)
testX, testY = get_XY(test_data, time_steps)

# 创建模型并训练
model = create_RNN(hidden_units=3, dense_units=1, input_shape=(time_steps,1), 
                   activation=['tanh', 'tanh'])
model.fit(trainX, trainY, epochs=20, batch_size=1, verbose=2)

# 做出预测
train_predict = model.predict(trainX)
test_predict = model.predict(testX)

# 打印误差
print_error(trainY, testY, train_predict, test_predict)

# 绘图结果
plot_result(trainY, testY, train_predict, test_predict)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71