使用tensorflow实现了简单的rnn网络用来学习加法运算。

tensorflow 版本：1.1

import tensorflow as tf

from tensorflow.contrib import rnn

class RNN():

    def __init__(self, input_dim , hidden_dim , step_num , class_num,learning_rate):

        # # tf Graph input

        self.x = tf.placeholder("float", [None, step_num, input_dim])

        self.y = tf.placeholder("float", [None, class_num])

        # Define weights

        weights = {

            'out': tf.Variable(tf.random_normal([hidden_dim, hidden_dim])),

            'sigout':tf.Variable(tf.random_normal([hidden_dim , class_num]))

        }

        biases = {

            'out': tf.Variable(tf.random_normal([hidden_dim])),

            'sigout':tf.Variable(tf.random_normal([class_num]))

        }

        # Unstack to get a list of 'step_num' tensors of shape (batch_size, input_dim)

        x_unstack = tf.unstack(self.x, step_num, 1)

        # Define a lstm cell with tensorflow

        lstm_cell = rnn.BasicLSTMCell(hidden_dim, forget_bias=1.0)

        # Get lstm cell output

        outputs, states = rnn.static_rnn(lstm_cell, x_unstack, dtype=tf.float32)

        # Linear activation, using rnn inner loop last output

        pred = tf.matmul(outputs[-1], weights['out']) + biases['out']

        sigmodout = tf.matmul(tf.nn.sigmoid(pred , 'relu') , weights['sigout']) + biases['sigout']

        self.predict = tf.round(sigmodout)

        correct_pred = tf.equal(self.predict , self.y)

        self.accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32))

        cost = tf.reduce_sum(tf.abs(tf.subtract(sigmodout , self.y)))

        self.loss = cost

        tf.summary.scalar('acc' , self.accuracy)

        tf.summary.scalar('cost' , cost)

        self.merge_all = tf.summary.merge_all()

import random

import numpy as np

class DataProcess():

    def __init__(self):

        self.point = 0

        self.max_integer = 10000

        self.max_len = len(self.filter(bin(self.max_integer))) + 1

    def nextBatch(self,batch_size , fake = False ,is_test = False):

        x_batch = []

        y_batch = []

        t_batch = []

        for i in range( 0 , batch_size):

            a = random.randint(0,self.max_integer)

            b = random.randint(0,self.max_integer)

            if fake:

                a = 1

                b = 1

            c = a + b

            abin = self.process(a)

            bbin = self.process(b)

            cbin = self.process(c)

            xa = np.array(abin)

            xb = np.array(bbin)

            x_ = np.concatenate((xa , xb),axis= 0)

            y_ = np.array(cbin)

            x_batch.append(x_.reshape(2,self.max_len))

            y_batch.append(cbin)

            if is_test:

                temp = []

                temp.append(a)

                temp.append(b)

                t_batch.append(temp)

        # if not is_test:

        return np.array(x_batch) , np.array(y_batch),np.array(t_batch)

        # else:

        #     return np.array(x_batch) , np.array(y_batch)

    def process(self , str):

        bstr = bin(str)

        bstr = self.filter(bstr)

        bstr = self.completion(bstr)

        return bstr[::-1]

    def filter(self , bstr):

        return bstr.replace('0b' , '')

    def completion(self , bstr):

        lst = []

        for num in list(bstr):

            lst.append(int(num))

        length = len(bstr)

        for i in range(length , self.max_len):

            lst.insert(0,0)

        return[ (float)(i) for i in lst]

import tensorflow as tf

import time

import os

from src.bitplus.Process import DataProcess

from src.bitplus.rnn import RNN

dp = DataProcess()

tf.app.flags.DEFINE_float("learning_rate",0.01,'learning rate')

tf.app.flags.DEFINE_integer("training_iters",500000,'')

tf.app.flags.DEFINE_integer("batch_size",100,'')

tf.app.flags.DEFINE_integer("display_step",10,'')

tf.app.flags.DEFINE_integer("input_dim" , 2,'')

tf.app.flags.DEFINE_integer("steps_num" , dp.max_len,'')

tf.app.flags.DEFINE_integer("hidden_dim", dp.max_len * 2 ,'')

tf.app.flags.DEFINE_integer("class_num" , dp.max_len,'')

tf.app.flags.DEFINE_string("rnn_model",'rnn.model','')

FLAGS = tf.app.flags.FLAGS

logfolder = time.strftime("%Y%m%d_%H%M%S", time.localtime())

def train():

    rnn = RNN(FLAGS.input_dim , FLAGS.hidden_dim ,FLAGS.steps_num , FLAGS.class_num , FLAGS.learning_rate)

    optimizer = tf.train.AdamOptimizer(learning_rate=FLAGS.learning_rate).minimize(rnn.loss)

    init = tf.global_variables_initializer()

    sess = tf.Session()

    sess.run(init)

    train_writer = tf.summary.FileWriter('../LOGS/'+logfolder,sess.graph)

    step = 1

    while step  * FLAGS.batch_size < FLAGS.training_iters:

        batch_x, batch_y ,_= dp.nextBatch(FLAGS.batch_size , False)

        batch_x = batch_x.transpose((0, 2, 1))

        # Run optimization op (backprop)

        sess.run(optimizer, feed_dict={rnn.x: batch_x, rnn.y: batch_y})

        if step % FLAGS.display_step == 0:

            # Calculate batch loss

            merge_all , loss , acc,predict= sess.run([rnn.merge_all ,rnn.loss ,rnn.accuracy,rnn.predict],feed_dict={rnn.x: batch_x, rnn.y: batch_y})

            print("Iter " + str(step*FLAGS.batch_size) + ", Minibatch Loss= " + \

                  "{:.6f}".format(loss) + ", Training Accuracy= " + \

                  "{:.5f}".format(acc))

            train_writer.add_summary(merge_all , step)

        step += 1

    print("Optimization Finished!")

    train_writer.close()

    saver = tf.train.Saver(tf.all_variables())

    if not os.path.exists('model'):

        os.mkdir('model')

    saver.save(sess, './model/' + FLAGS.rnn_model)

def tModel(model):

    rnn = RNN(FLAGS.input_dim , FLAGS.hidden_dim ,FLAGS.steps_num , FLAGS.class_num , FLAGS.learning_rate)

    saver = tf.train.Saver()

    init = tf.global_variables_initializer()

    sess = tf.Session()

    sess.run(init)

    saver.restore(sess,model)

    batch_x , batch_y , batch_test = dp.nextBatch(100 , False,True)

    batch_x = batch_x.transpose((0, 2, 1))

    predict = sess.run([rnn.predict],feed_dict={rnn.x: batch_x, rnn.y: batch_y})

    for ndx in range(len(predict[0])):

        print('%s + %s = %d(%d)'%(batch_test[ndx][0] , batch_test[ndx][1] , batch_test[ndx][0] + batch_test[ndx][1],bin2Ten(predict[0][ndx])))

    print('pause!')

def bin2Ten(bin):

    lst = []

    for i in bin[::-1]:

        lst.append(str(int(i)))

    return int(''.join(lst) , base=2)

def main(_):

    print('start...')

    print('max len:' , dp.max_len)

    # train()

    tModel('./model/' + FLAGS.rnn_model)

if __name__ == '__main__':

    tf.app.run()