小白劝退预告

  • 仅简单介绍思路，没有用作教程的打算，如果读者没有机器学习基础、循环神经网络基础或Pytorch基础 —— 会很不友好的(

Pytorch实现神经网络

Pytorch 准备正弦函数数据集

其中超参数的含义如下:
1. input_size: 输入预测的特征值数目
2. output_size: 最终输出的预测数值数目
3. hidden_size: 循环神经元的特征数目
4. batch_size: 一次训练批量数目
5. num_hidden: 循环层神经元的层数
6. length_num: 数据集的数据数目

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
input_size = 1
output_size = 1
hidden_size = 16
batch_size = 49
num_hidden = 1
length_num = 50

X = torch.linspace(0, 10, length_num)
Y = torch.sin(X)
plt.scatter(X, Y)
plt.show()

data0_48 = Y[0:49].reshape(num_hidden, batch_size, input_size)
data1_49 = Y[1:50].reshape(num_hidden, batch_size, input_size)
hidden = torch.zeros(num_hidden, input_size, hidden_size)
data0_48.shape, data1_49.shape, hidden.shape

训练数据集展示如下：

raw

Pytorch 实现循环神经网路

用 torch.nn.Module 来自定义模型，模型不必太复杂，一个循环节点和一个多层感知机就足够了
一定要留意各神经元间张量格式的衔接！！！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import torch
from torch import nn, optim
import torch.nn.functional as F
import matplotlib.pyplot as plt
from time import perf_counter

class RNN(nn.Module):
    def __init__(self):
        super(RNN, self).__init__()
        self.rnn = nn.RNN(input_size, hidden_size, num_hidden, batch_first=True)
        self.linear = nn.Linear(hidden_size, output_size)
        
    def forward(self, inputs, hidden):
        # inputs.shape [1, 49, 1]
        # hidden.shape [1,  1, 16]
        #    out.shape [1, 49, 1]
        #    hid.shape [1,  1, 16]
        out, hid = self.rnn(inputs, hidden)
        out = self.linear(out)
        return out, hid

定义超参数

超参数就是全局变量，我们需要定义以下几个超参数：

  1. model: 要训练的模型，实例化上文我们定义的类即可
  2. Step: 训练步数，共 6000 次
  3. learning_rate: 学习率，控制着模型训练速率的常数
  4. criterion: 损失函数，本质为 torch.nn 对象，对于多分类的模型，我们采用平方损失函数(即 MSELoss )
  5. optimizer: 模型优化器，本质为 torch.optim 对象，在本次训练中，我们选择 Adam 作为我们模型的优化方法
1
2
3
4
5
6
net = RNN()
learning_rate = 1e-3
criterion = nn.MSELoss()
optimizer = optim.Adam(net.parameters(), lr=learning_rate)
Step = 6000
Test_step = 1000

实现检测

这个函数用于可视化我们的模型预测结果如何
我们先输入一个初始值，我们的模型会预测出第二个值
我们再把第二个值交付给模型，模型会给出第三个值，以此类推直到 49 个预测值全部给出
最后通过 plt 将数据可视化

1
2
3
4
5
6
7
8
9
10
11
12
def test(hidden_back):
    prediction = []
    input = data0_48[:, 0, :].reshape(1, 1, 1)
    for idx in range(batch_size):
        out, hidden_back = net(input, hidden_back)
        input = out
        prediction.append(out.item())
    plt.scatter(X[1:50], data1_49.reshape(49), label="raw")
    plt.scatter(X[1:50], prediction, label="RNN-pre")
    plt.legend()
    plt.show()
    return

Pytorch 实现训练过程

我们将模型的训练过程整合成为一个函数，这个函数接受三个输入
我们拆分这个训练过程，它主要由这几步构成：

  1. 每次训练，将训练的函数数据和隐藏层数据放进模型
  2. 通过模型计算得到结果，存进 outputs 里，并通过损失函数得到损失值 loss
  3. 先清空模型的参数梯度设置，调用 loss 的反向传播重新计算梯度，并依据梯度对模型参数赋值
  4. 每 1000 次训练，计算这轮计算的时间、损失值、正确率，用作训练时的日志输出
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def train(Step, test_step, hidden):
    start = perf_counter()
    Loss = 0
    for step in range(1, Step + 1):
        outputs, hidden_temp = net(data0_48, hidden)
        hidden = hidden_temp.detach()
        
        optimizer.zero_grad()
        loss = criterion(outputs, data1_49)
        loss.backward()
        optimizer.step()
        
        Loss += loss.item()
        
        if step % test_step == 0:
            time = perf_counter() - start
            start = perf_counter()
            Loss /= test_step
            print(f"[INFO] Step:{step} Loss:{Loss} Time:{time:.2f}s")
            test(hidden)
            
    print("Finished!")
    return

代码总览

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
72
73
74
75
76
77
78
79
import torch
from torch import nn, optim
import torch.nn.functional as F
import matplotlib.pyplot as plt
from time import perf_counter

input_size = 1
output_size = 1
hidden_size = 16
batch_size = 49
num_hidden = 1
length_num = 50

class RNN(nn.Module):
    def __init__(self):
        super(RNN, self).__init__()
        self.rnn = nn.RNN(input_size, hidden_size, num_hidden, batch_first=True)
        self.linear = nn.Linear(hidden_size, output_size)
        
    def forward(self, inputs, hidden):
        out, hid = self.rnn(inputs, hidden)
        out = self.linear(out)
        return out, hid

X = torch.linspace(0, 10, length_num)
Y = torch.sin(X)
plt.scatter(X, Y)
plt.show()

data0_48 = Y[0:49].reshape(num_hidden, batch_size, input_size)
data1_49 = Y[1:50].reshape(num_hidden, batch_size, input_size)
hidden = torch.zeros(num_hidden, input_size, hidden_size)
print(data0_48.shape, data1_49.shape, hidden.shape)

net = RNN()
learning_rate = 1e-3
criterion = nn.MSELoss()
optimizer = optim.Adam(net.parameters(), lr=learning_rate)
Step = 6000
Test_step = 1000

def test(hidden_back):
    prediction = []
    input = data0_48[:, 0, :].reshape(1, 1, 1)
    for idx in range(batch_size):
        out, hidden_back = net(input, hidden_back)
        input = out
        prediction.append(out.item())
    plt.scatter(X[1:50], data1_49.reshape(49), label="raw")
    plt.scatter(X[1:50], prediction, label="RNN-pre")
    plt.legend()
    plt.show()
    return

def train(Step, test_step, hidden):
    start = perf_counter()
    Loss = 0
    for step in range(1, Step + 1):
        outputs, hidden_temp = net(data0_48, hidden)
        hidden = hidden_temp.detach()
        
        optimizer.zero_grad()
        loss = criterion(outputs, data1_49)
        loss.backward()
        optimizer.step()
        
        Loss += loss.item()
        
        if step % test_step == 0:
            time = perf_counter() - start
            start = perf_counter()
            Loss /= test_step
            print(f"[INFO] Step:{step} Loss:{Loss} Time:{time:.2f}s")
            test(hidden)
            
    print("Finished!")
    return

train(Step, Test_step, hidden)

最终，这个模型在训练 6,000 次训练后的预测结果如下
pre
最终检查测试的可视化