keras 使用RNN实现2进制加法

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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
#coding=utf-8
'''
Created on 2018年8月28日

'''

import copy, numpy as np                                       
np.random.seed(0)

# compute sigmoid nonlinearity                         #定义sigmoid函数
def sigmoid(x):
    output = 1/(1+np.exp(-x))
    return output

# convert output of sigmoid function to its derivative   #计算sigmoid函数的倒数
def sigmoid_output_to_derivative(output):
    return output*(1-output)

# training dataset generation                            
int2binary = {}                                    #用于将输入的整数转为计算机可运行的二进制数用
binary_dim = 8                                     #定义了二进制数的长度=8

largest_number = pow(2,binary_dim)                         #二进制数最大能取的数就=256喽
binary = np.unpackbits(
    np.array([range(largest_number)],dtype=np.uint8).T,axis=1)            
for i in range(largest_number):                                #将二进制数与十进制数做个一一对应关系
    int2binary[i] = binary[i]

# input variables
alpha = 0.1              #反向传播时参数w更新的速度
input_dim = 2            #输入数据的维度，程序是实现两个数相加的
hidden_dim = 16          #隐藏层神经元个数=16
output_dim = 1           #输出结果值是1维的


# initialize neural network weights                          #初始化神经网络的权重参数
synapse_0 = 2*np.random.random((input_dim,hidden_dim)) - 1   #输入至神经元的w0，维度为2X16，取值约束在[-1,1]间
synapse_1 = 2*np.random.random((hidden_dim,output_dim)) - 1  #神经元至输出层的权重w1,维度为16X1，取值约束在[-1,1]间
synapse_h = 2*np.random.random((hidden_dim,hidden_dim)) - 1  #神经元前一时刻状态至当前状态权重wh,维度为16X16，取值约束在[-1,1]间                                   

print(synapse_0.shape)
print(synapse_h.shape)
print(synapse_1.shape)

synapse_0_update = np.zeros_like(synapse_0)          #构造与w0相同维度的矩阵，并初始化为全0；
synapse_1_update = np.zeros_like(synapse_1)
synapse_h_update = np.zeros_like(synapse_h)

# training logic
for j in range(10000):          #模型迭代次数，可自行更改

    # generate a simple addition problem (a + b = c)
    a_int = np.random.randint(largest_number/2) # int version      #约束初始化的输入加数a的值不超过128
    a = int2binary[a_int] # binary encoding                        #将加数a的转为对应二进制数
    b_int = np.random.randint(largest_number/2) # int version
    b = int2binary[b_int] # binary encoding

    # true answer
    c_int = a_int + b_int    #真实和
    c = int2binary[c_int]    

    # where we'll store our best guess (binary encoded)
    d = np.zeros_like(c)                                        #用于存储预测的和

    overallError = 0                                           #打印显示误差

    layer_2_deltas = list()                                  #反向求导用
    layer_1_values = list()
    layer_1_values.append(np.zeros(hidden_dim))             #先对隐藏层前一时刻状态初始化为0

    # moving along the positions in the binary encoding
    for position in range(binary_dim):                         #前向传播；二进制求和，低位在右，高位在左

        # generate input and output
        X = np.array([[a[binary_dim - position - 1],b[binary_dim - position - 1]]]) #输入的a与b（二进制形式）
        y = np.array([[c[binary_dim - position - 1]]]).T                            #真实label值

        # hidden layer (input ~+ prev_hidden)
        layer_1 = sigmoid(np.dot(X,synapse_0) + np.dot(layer_1_values[-1],synapse_h))  # X*w0+RNN前一时刻状态值*wh

        # output layer (new binary representation)
        layer_2 = sigmoid(np.dot(layer_1,synapse_1))          #layer_1*w1

        # did we miss?... if so, by how much?
        layer_2_error = y - layer_2                                            #求误差
        layer_2_deltas.append((layer_2_error)*sigmoid_output_to_derivative(layer_2))  #代价函数
        overallError += np.abs(layer_2_error[0])                             #误差，打印显示用

        # decode estimate so we can print it out
        d[binary_dim - position - 1] = np.round(layer_2[0][0])                    #预测的和

        # store hidden layer so we can use it in the next timestep
        layer_1_values.append(copy.deepcopy(layer_1))                     #深拷贝，将RNN模块状态值存储，用于反向传播

    future_layer_1_delta = np.zeros(hidden_dim)

    for position in range(binary_dim):   #反向传播，计算从左到右，即二进制高位到低位

        X = np.array([[a[position],b[position]]])
        layer_1 = layer_1_values[-position-1]
        prev_layer_1 = layer_1_values[-position-2]

        # error at output layer
        layer_2_delta = layer_2_deltas[-position-1]
        # error at hidden layer
        layer_1_delta = (future_layer_1_delta.dot(synapse_h.T) + layer_2_delta.dot(synapse_1.T)) * sigmoid_output_to_derivative(layer_1)

        # let's update all our weights so we can try again
        synapse_1_update += np.atleast_2d(layer_1).T.dot(layer_2_delta)       #对w1进行更新
        synapse_h_update += np.atleast_2d(prev_layer_1).T.dot(layer_1_delta)  #对wh进行更新
        synapse_0_update += X.T.dot(layer_1_delta)                            #对w0进行更新

        future_layer_1_delta = layer_1_delta


    synapse_0 += synapse_0_update * alpha
    synapse_1 += synapse_1_update * alpha
    synapse_h += synapse_h_update * alpha    

    synapse_0_update *= 0
    synapse_1_update *= 0
    synapse_h_update *= 0

    # print out progress
    if(j % 1000 == 0):  #每1000次打印结果
        print ("Error:" + str(overallError))
        print ("Pred:" + str(d))
        print ("True:" + str(c))
        out = 0
        for index,x in enumerate(reversed(d)):
            out += x*pow(2,index)
        print (str(a_int) + " + " + str(b_int) + " = " + str(out))
        print ("------------")

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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
#coding=utf-8
'''
Created on 2018年8月27日

第一种输入数据:
77 + 11 = 88
3个整数的二进制分别为(低位->高危表示):
[0 1 0 0 1 1 0 1] [0 0 0 0 1 0 1 1] [0 1 0 1 1 0 0 0]
1-4列分别表示 00，10,01,11
X:
[[ 1.  0.  0.  0.]
 [ 0.  1.  0.  0.]
 [ 1.  0.  0.  0.]
 [ 1.  0.  0.  0.]
 [ 0.  0.  0.  1.]
 [ 0.  1.  0.  0.]
 [ 0.  0.  1.  0.]
 [ 0.  0.  0.  1.]]
Y:
[[0]
 [1]
 [0]
 [1]
 [1]
 [0]
 [0]
 [0]]

第二种输入数据:
60 + 72 = 132
3个整数的二进制分别为(低位->高危表示):
[0, 0, 1, 1, 1, 1, 0, 0] [0, 0, 0, 1, 0, 0, 1, 0] [0, 0, 1, 0, 0, 0, 0, 1]
1,2列表示60这个[0, 0, 1, 1, 1, 1, 0, 0]进行了one-hot编码
3,4列表示72这个[0, 0, 0, 1, 0, 0, 1, 0]进行了one-hot编码
X:
[[ 1.  0.  1.  0.]
 [ 1.  0.  1.  0.]
 [ 0.  1.  1.  0.]
 [ 0.  1.  0.  1.]
 [ 0.  1.  1.  0.]
 [ 0.  1.  1.  0.]
 [ 1.  0.  0.  1.]
 [ 1.  0.  1.  0.]]
Y:
[[0]
 [0]
 [1]
 [0]
 [0]
 [0]
 [0]
 [1]]

第三种输入数据:
3 + 23 = 26
3个整数的二进制分别为(低位->高危表示):
[1, 1, 0, 0, 0, 0, 0, 0] [1, 1, 1, 0, 1, 0, 0, 0] [0, 1, 0, 1, 1, 0, 0, 0]
1,2列分别表示3,23这两个数字的二进制
X:
[[1 1]
 [1 1]
 [0 1]
 [0 0]
 [0 1]
 [0 0]
 [0 0]
 [0 0]]
Y:
[[0]
 [1]
 [0]
 [1]
 [1]
 [0]
 [0]
 [0]]

第4种可以参考:https://blog.csdn.net/whai362/article/details/52523439
"789+123"这样的序列
'''
from keras.models import Sequential
from keras.layers import Activation, TimeDistributed, Dense,SimpleRNN
import numpy as np
from keras.utils import to_categorical
from keras.optimizers import SGD
from sklearn.model_selection import train_test_split

# training dataset generation                            
int2binary = {}                                    #用于将输入的整数转为计算机可运行的二进制数用
binary_dim = 8                                     #定义了二进制数的长度=8

largest_number = pow(2,binary_dim)                         #二进制数最大能取的数就=256喽
binary = np.unpackbits(
    np.array([range(largest_number)],dtype=np.uint8).T,axis=1)            
for i in range(largest_number):                                #将二进制数与十进制数做个一一对应关系
    int2binary[i] = binary[i]

x_train = []  
y_train = []  
for j in range(1):          #模型迭代次数，可自行更改
    # generate a simple addition problem (a + b = c)
    a_int = np.random.randint(largest_number/2) # int version      #约束初始化的输入加数a的值不超过128
    a = list(int2binary[a_int]) # binary encoding #将加数a的转为对应二进制数
    a.reverse()

    b_int = np.random.randint(largest_number/2) # int version
    b = list(int2binary[b_int]) # binary encoding
    b.reverse()

    c_int = a_int + b_int    #真实和
    c = list(int2binary[c_int])
    c.reverse()  

    #第2种输入数据，记得修改最后的keras model的输入维度
    #tempx = np.hstack((to_categorical(np.array([a]).T,num_classes=2),to_categorical(np.array([b]).T,num_classes=2)))

    #第3种输入数据，记得修改最后的keras model的输入维度
    #tempx = np.hstack((np.array([a]).T,np.array([b]).T))

    #第一种输入数据
    tempx = np.hstack((np.array([a]).T,np.array([b]).T))
    tempx_2 = []
    for i in range(8):
        if tempx[i][0] == 0 and tempx[i][1]==0:
            tempx_2.append(0)
        if tempx[i][0] == 1 and tempx[i][1]==0:
            tempx_2.append(1)
        if tempx[i][0] == 0 and tempx[i][1]==1:
            tempx_2.append(2)
        if tempx[i][0] == 1 and tempx[i][1]==1:
            tempx_2.append(3)
    tempx = to_categorical(np.array([tempx_2]).T,num_classes=4)

    tempy = np.array([c]).reshape((8,1))

    x_train.append(tempx)
    y_train.append(tempy)

x_train = np.array(x_train)
y_train = np.array(y_train)
x_train, x_val, y_train, y_val = train_test_split(x_train, y_train, test_size=0.3, random_state=0)
print(x_train.shape)  
print(y_train.shape)




model = Sequential()
model.add(SimpleRNN(128, input_shape=(8, 4),return_sequences=True,activation="sigmoid"))
model.add(TimeDistributed(Dense(1)))
model.add(Activation('sigmoid'))
#经过测试最好不要用sgd,训练速度慢，建议使用adam
#sgd = SGD(lr=0.001, decay=0.0002, momentum=0.9, nesterov=True)
model.compile(loss='binary_crossentropy',
              optimizer='adam',
              metrics=['accuracy'])
model.summary() 

model.fit(x_train, y_train, 32, 1000,validation_data=(x_val, y_val))