rnn实现不使用tf.nn.rnn_cell

参考这篇文章写的

1.导入模块

import tensorflow as tf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

2.获取数据

def get_data(size = 40000):
X = np.array(np.random.choice(2, size=(size)))
Y = []
for i in range(size):
threshold = 0.5
if X[i-3] == 1:
threshold += 0.5
if X[i-8] == 1:
threshold -= 0.25
if np.random.rand() > threshold:
Y.append(0)
else:
Y.append(1)
return X,np.array(Y)

x值分别为0，1的概率都是0.5
当x的3个时刻之前的值为1时，y为1的概率增加0.5
当x的8个时刻之前的值为1时，y为1的概率减少0.25
由此，输出y便于输入x在时间上建立了关系

3.获取每一批的数据

def get_batch(x,y,batch_size,seq_length):
lens = len(x)
batches = lens//(batch_size*seq_length)
x = x[:batches*batch_size*seq_length]
y = y[:batches*batch_size*seq_length]
x = x.reshape(batch_size, -1)
print(x.shape)
y = y.reshape(batch_size,-1)
for i in range(0,(batches-1)*seq_length,seq_length):
raw_x = x[:,i:i+seq_length]
raw_y = y[:,i:i+seq_length]
yield (raw_x, raw_y)

将数据进行，把整除之后多余的部分去掉，然后对列表进行reshape处理，变成shape为[batch_size,-1]的数据，方便迭代

4.获取每一轮的数据

def get_epochs(n, num_steps):
x,y=get_data()
for i in range(n):
yield get_batch(x,y, batch_size, num_steps)

当get_batch函数执行结束后，又会重新执行一遍get_batch函数，直至执行完轮数

5.参数设置

batch_size = 3
num_classes = 2
state_size = 4
num_steps = 10
learning_rate = 0.2

6.定义输入输出

"""
tf.unstack
>>> import numpy as np
>>> import tensorflow as tf
>>> sess = tf.Session()
>>> b = tf.constant(np.array([[0,1,2],[3,4,5]]))
>>> c = tf.one_hot(b,6)
>>> sess.run(c)
array([[[1., 0., 0., 0., 0., 0.],
[0., 1., 0., 0., 0., 0.],
[0., 0., 1., 0., 0., 0.]],

[[0., 0., 0., 1., 0., 0.],
[0., 0., 0., 0., 1., 0.],
[0., 0., 0., 0., 0., 1.]]], dtype=float32)
>>> d = tf.unstack(c,axis=1)
>>> sess.run(d)
[array([[1., 0., 0., 0., 0., 0.],
[0., 0., 0., 1., 0., 0.]], dtype=float32), array([[0., 1., 0., 0., 0., 0.],
[0., 0., 0., 0., 1., 0.]], dtype=float32), array([[0., 0., 1., 0., 0., 0.],
[0., 0., 0., 0., 0., 1.]], dtype=float32)]
>>> e = tf.unstack(c,axis=0)
>>> sess.run(e)
[array([[1., 0., 0., 0., 0., 0.],
[0., 1., 0., 0., 0., 0.],
[0., 0., 1., 0., 0., 0.]], dtype=float32), array([[0., 0., 0., 1., 0., 0.],
[0., 0., 0., 0., 1., 0.],
[0., 0., 0., 0., 0., 1.]], dtype=float32)]
>>> f = tf.unstack(c,axis=2)
>>> sess.run(f)
[array([[1., 0., 0.],
[0., 0., 0.]], dtype=float32), array([[0., 1., 0.],
[0., 0., 0.]], dtype=float32), array([[0., 0., 1.],
[0., 0., 0.]], dtype=float32), array([[0., 0., 0.],
[1., 0., 0.]], dtype=float32), array([[0., 0., 0.],
[0., 1., 0.]], dtype=float32), array([[0., 0., 0.],
[0., 0., 1.]], dtype=float32)]
"""

x = tf.placeholder(tf.int32, [batch_size, num_steps], name='inputs')  #[batch_size, num_steps]
y = tf.placeholder(tf.int32, [batch_size, num_steps], name='targets')
init_state = tf.zeros([batch_size, state_size])
x_onehot = tf.one_hot(x, num_classes) #[batch_size, num_steps, num_classes]
rnn_inputs = tf.unstack(x_onehot, axis=1) #对数据进行分解 [num_steps, batch_size, num_classes]

其中注释部分对tf.unstack函数做了一下解释

7.定义rnn_cell

#计算state
def rnn_cell(rnn_input, state):
with tf.variable_scope('rnn_cell', reuse=tf.AUTO_REUSE):
W = tf.get_variable('W', [num_classes + state_size, state_size])
b = tf.get_variable('b', [state_size], initializer=tf.constant_initializer(0.0))
return tf.tanh(tf.matmul(tf.concat([rnn_input, state], 1), W) + b)

8.计算每一个时刻的state

state = init_state
rnn_outputs = []
#循环num_steps次，即将一个序列输入RNN模型
for rnn_input in rnn_inputs:
state = rnn_cell(rnn_input, state)
rnn_outputs.append(state)
final_state = rnn_outputs[-1]

对输入进行遍历，然后带入rnn_cell中进行计算就可

ps:此时的输入数据已经被处理为了shape=[num_steps, batch_size, num_classes]的数据，每次遍历即是取出各个时刻对应的数据

9.构建输出层

with tf.variable_scope('softmax'):
W = tf.get_variable('W', [state_size, num_classes])
b = tf.get_variable('b', [num_classes], initializer=tf.constant_initializer(0.0))
#注意，这里要将num_steps个输出全部分别进行计算其输出，然后使用softmax预测
logits = [tf.matmul(rnn_output, W) + b for rnn_output in rnn_outputs]
predictions = [tf.nn.softmax(logit) for logit in logits]

10.计算loss并进行反向传播

# Turn our y placeholder into a list of labels
y_as_list = tf.unstack(y, num=num_steps, axis=1)
#losses and train_step
losses = [tf.nn.sparse_softmax_cross_entropy_with_logits(labels=label, logits=logit) for \
logit, label in zip(logits, y_as_list)]
total_loss = tf.reduce_mean(losses)
train_step = tf.train.GradientDescentOptimizer(learning_rate).minimize(total_loss)

11.模型训练

def train_network(num_epochs, num_steps, state_size=4, verbose=True):
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
training_losses = []
#得到数据，因为num_epochs==5，所以外循环只执行五次
for idx, epoch in enumerate(get_epochs(num_epochs, num_steps)):
training_loss = 0
#保存每次执行后的最后状态，然后赋给下一次执行
training_state = np.zeros((batch_size, state_size))
if verbose:
print("\nEPOCH", idx)
#这是具体获得数据的部分
for step, (X, Y) in enumerate(epoch):
tr_losses, training_loss_, training_state, _ = \
sess.run([losses,
total_loss,
final_state,
train_step],
feed_dict={x:X, y:Y, init_state:training_state})
training_loss += training_loss_
if step % 100 == 0 and step > 0:
if verbose:
print("Average loss at step", step,
"for last 100 steps:", training_loss/100)
training_losses.append(training_loss/100)
training_loss = 0

return training_losses
training_losses = train_network(5,num_steps)
plt.plot(training_losses)
plt.show()

运行结果如下：

• 点赞
• 收藏
• 分享
• 文章举报

jiang199912 发布了18 篇原创文章 · 获赞 0 · 访问量 533 私信 关注