1
$\begingroup$

I have recently started studying Deep Learning and have become quite confident in my understanding of the theory of how NNs work. I have written a couple of simple NNs from scratch in Python, to ensure that I know how they work without any libraries. Now, in a hope to become more familiar with implementing my NNs using TensorFlow, I have decided to write an LSTM using it.

The LSTM I am trying to implement is the well-known example of learning from the entire works of Shakespeare, and then producing its own text in the same style.

Here is my code that I thought would work fine, up until a few minutes ago where I have become very confused.

from __future__ import print_function
import tensorflow as tf
from tensorflow.contrib import rnn
import numpy as np
import random

text = open("/home/kev/Documents/NeuralNetworks/CharacterPredicter/input.txt", 'r').read()
#text = "abcdefghi"
vocab = list(set(text))
textSize = len(text)
vocabSize = len(vocab)
testingStartChar = int(textSize * 0.8)

charToInt = {char : i for i, char in enumerate(vocab)}
intToChar = {i : char for i, char in enumerate(vocab)}

sequenceLength = 25  # Must be strictly < (textSize / 5)
trainingIterations = 5000
batchSize = 100
numTestingSequences = int((textSize - testingStartChar - 1) / sequenceLength)

hiddenDimension = 500
learningRate = 0.005
forgetRate = 1.0
printStep = 100

# Returns two 3D arrays of shape (batchSize x seqLength x vocabSize)
def generateData(batchSize, seqLength, isTraining):
    inputs = []                                                 # e.g.  a, b = generateData(2, 3, True)
    labels = []
    charPointer = testingStartChar                              #       a -> [[[0, 1, ..., 0], [1, 0, ..., 0], [0, 0, ..., 1]]
    for _ in range(batchSize):                                  #             [[1, 0, ..., 0], [0, 0, ..., 1], [0, 1, ..., 0]]]
        inputSequence = []
        labelSequence = []
        if isTraining:
            charPointer = random.randint(0, testingStartChar - seqLength)
        for _ in range(seqLength):
            oneHotInput = charToOneHot(text[charPointer])
            oneHotLabel = charToOneHot(text[charPointer + 1])
            inputSequence.append(oneHotInput)
            labelSequence.append(oneHotLabel)
            charPointer += 1
        inputs.append(inputSequence)
        labels.append(labelSequence)
    return inputs, labels

# Takes in a single character and returns its corresponding
# one-hot vector of length vocabSize
def charToOneHot(char):
    oneHot = [0] * vocabSize
    oneHot[charToInt[char]] = 1
    return oneHot

# Takes in a one-hot vector and returns its corresponding char
def oneHotToChar(oneHot):
    charIndex = np.argmax(oneHot)
    return intToChar[charIndex]

# Takes in a 3D array (last axis is one-hot vectors) and returns
# an array of the corresponding strings per row (batchSize)
# (Used for printing outputs)
def oneHotsToChars(array):                                          # e.g. [[[0, 1, ..., 0], [1, 0, ..., 0], [0, 0, ..., 1]],
    chars = []                                                      #       [[1, 0, ..., 0], [0, 0, ..., 1], [0, 1, ..., 0]]]
    for batchIndex in range(len(array)):
        batchChars = ""                                             # ->   ['hpb',
        for oneHotIndex in range(len(array[batchIndex])):           #      'pbh']
            batchChars += oneHotToChar(array[batchIndex][oneHotIndex])
        chars.append(batchChars)
    return chars

# Takes a 3D array of real numbers (batchSize x sequenceLength x vocabSize)
# and returns a 3D array with the last axis being one-hot vectors
# corresponding to the real number of the highest value
def predictionToOneHots(prediction):
    oneHots = []
    for batchIndex in range(len(prediction)):
        batchOneHots = []
        for charArrayIndex in range(len(prediction[batchIndex])):
            charArray = prediction[batchIndex][charArrayIndex]
            charOneHot = [0] * len(charArray)
            charIndex = np.argmax(charArray)
            charOneHot[charIndex] = 1
            batchOneHots.append(charOneHot)
        oneHots.append(batchOneHots)
    return oneHots

batchSz = tf.placeholder(tf.int32, shape=())
x = tf.placeholder(tf.float32, [None, sequenceLength, vocabSize])
y = tf.placeholder(tf.float32, [None, sequenceLength, vocabSize])
xFlat = tf.contrib.layers.flatten(x)                                                # [batchSize, sequenceLength*vocabSize]

W = tf.Variable(tf.random_normal([hiddenDimension, sequenceLength, vocabSize]))
b = tf.Variable(tf.random_normal([1, sequenceLength, vocabSize]))
WFlat = tf.contrib.layers.flatten(W)                                                # [hiddenDimension, sequenceLength*vocabSize]
bFlat = tf.contrib.layers.flatten(b)                                                # [1, sequenceLength*vocabSize]

cell = rnn.BasicLSTMCell(hiddenDimension, forget_bias=forgetRate)
outputs, states = tf.nn.static_rnn(cell, [xFlat], dtype=tf.float32)                 # outputs    = [[batchSize, hiddenDimension]]

predictionFlat = tf.add(tf.matmul(outputs[0], WFlat), bFlat)                        # outputs[0] = [batchSize, hiddenDimension]
prediction = tf.reshape(predictionFlat, [batchSz, sequenceLength, vocabSize])

# 2D array corresponding to whether character per sequence per batch was predicted correctly
# A correct prediction is when the highest predicted value is the same index as the 1 of the one-hot label
correctPrediction = tf.equal(tf.argmax(prediction, axis=2), tf.argmax(y, axis=2))   # [batchSize, sequenceLength]
accuracy = tf.reduce_mean(tf.cast(correctPrediction, tf.float32))

loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(logits=prediction, labels=y))
optimiser = tf.train.AdamOptimizer(learning_rate=learningRate).minimize(loss)

with tf.Session() as session:
    session.run(tf.global_variables_initializer())

    ############################################################################
    #   TRAINING
    ############################################################################
    for iteration in range(trainingIterations):
        batchX, batchY = generateData(batchSize, sequenceLength, isTraining=True)
        dict = {batchSz: batchSize, x: batchX, y: batchY}
        session.run(optimiser, dict)
        if (iteration + 1) % printStep == 0 or iteration in (0, trainingIterations - 1):
            batchAccuracy = session.run(accuracy, dict)
            batchLoss  = session.run(loss, dict)
            inputOneHots = session.run(x, dict)
            labelOneHots = session.run(y, dict)
            predictions = session.run(prediction, dict)
            correctPredictions = session.run(correctPrediction, dict)
            print("Iteration:\t" + str(iteration + 1))
            print("Accuracy:\t" + str("%.2f" % (batchAccuracy * 100) + "%"))
            print("Loss:\t\t" + str(batchLoss) + "\n")
            print("Inputs:\n" + str(oneHotsToChars(inputOneHots)) + "\n")
            print("Labels:\n" + str(oneHotsToChars(labelOneHots)) + "\n")
            print("Prediction:\n" + str(oneHotsToChars(predictionToOneHots(predictions))) + "\n")
            print("Correct:\n" + str(correctPredictions) + "\n")

    ############################################################################
    #   TESTING
    ############################################################################
    testX, testY = generateData(numTestingSequences, sequenceLength, isTraining=False)
    testAccuracy = session.run(accuracy, feed_dict={batchSz: numTestingSequences, x: testX, y: testY})
    print("Testing Accuracy: " + str("%.2f" % (testAccuracy * 100) + "%"))

    ############################################################################
    #   GENERATING
    ############################################################################
    def stringToOneHots(string):
        oneHots = []
        for char in string:
            oneHots.append(charToOneHot(char))
        return oneHots

    randIndex = random.randint(0, textSize - sequenceLength)
    seedSequence = text[randIndex : randIndex + sequenceLength]
    inputOneHots = stringToOneHots(seedSequence)

    with open("/home/kev/Documents/NeuralNetworks/CharacterPredicter/output.txt", "w+") as output:
        generatedText = ""
        lineLength = 0
        for _ in range(1000):
            dict = {x: [inputOneHots], batchSz: 1}
            pred = session.run(prediction, dict)
            predString = oneHotsToChars(predictionToOneHots(pred))[0]
            inputOneHots = stringToOneHots(predString)
            if predString[-2] == " " and lineLength >= 70:
                output.write("\n")
                lineLength = 0
            if predString[-1] == "\n":
                lineLength = 0
            lineLength += 1
            output.write(predString[-1])

My network works by taking in sequences of 25 characters, and predicting the character that will follow each of these (25 outputs).

In the testing loop, I have added a whole load of prints just to see the progression of the NN as it trains. It is clearer to see what is going on if sequenceLength and batchSize are reduced to single digit number.

The main confusions I am having are:

The model learns to predict the first 24 output characters perfectly, extremely quickly (it seems spot on after 100 training iterations of 100 batches). For the rest of the training, it is getting the first 24 characters perfect, and then the only errors are in the final output character. I am lost as to how this is happening so quickly. Is it because the first 24 output characters are already in the network as the last 24 input characters? I'm sure it is something along the lines of this, but I am failing to see exactly what is happening and am in need of an explanation.

This same problem is also leading to my accuracy and loss calculations to be very high and low respectively. The network very quickly reaches 95%+ accuracy since it is constantly getting the first 24 characters correct, and the last one sometimes wrong, sometimes right. The same reasoning can be said for why the loss is calculated to a small amount. Should I somehow only be focusing on the prediction of the 25th character - the one that isn't contained within any of the input time steps? Should my prediction only be this final character, or still the 25 outputs?

No matter if I train it over 1,000 iterations or 100,000, the prediction of this final character never seems to get very accurate and so the text generated isn't terribly impressive. I just don't know if I am going about this problem in the wrong way entirely.

EDIT

I think I have figured it all out:

from __future__ import print_function
import tensorflow as tf
import numpy as np
import random

text = open("/home/kev/Documents/NeuralNetworks/CharacterPredicter/input.txt", 'r').read()
#text = "abcdefghij"
vocab = list(set(text))
textSize = len(text)
vocabSize = len(vocab)
testingStartChar = int(textSize * 0.8)

charToInt = {char : i for i, char in enumerate(vocab)}
intToChar = {i : char for i, char in enumerate(vocab)}

sequenceLength = 25  # Must be strictly < (textSize / 5)
batchSize = 100

numValidationSequences = (textSize - testingStartChar - 1) // sequenceLength
numEpochs = textSize // (sequenceLength * batchSize)
trainingIterations = numEpochs * 100
#trainingIterations = 1000

hiddenDimension = 500
learningRate = 0.01
forgetRate = 2.0
printStep = 100

# [batchSize, seqLength, vocabSize]
def generateData(batchSize, seqLength, isTraining, charPointer):
    inputs = []                                                 # e.g.  a, b = generateData(2, 3, True)
    labels = []                                                 #       a -> [[[0, 1, ..., 0], [1, 0, ..., 0], [0, 0, ..., 1]]
    for _ in range(batchSize):                                  #             [[1, 0, ..., 0], [0, 0, ..., 1], [0, 1, ..., 0]]]
        inputSequence = []
        labelSequence = []
        #if isTraining:
        #    charPointer = random.randint(0, testingStartChar - seqLength)  # Uncomment for random sequences
        for _ in range(seqLength):
            oneHotInput = charToOneHot(text[charPointer])
            oneHotLabel = charToOneHot(text[charPointer + 1])
            inputSequence.append(oneHotInput)
            labelSequence.append(oneHotLabel)
            if charPointer + sequenceLength >= textSize - 1:
                charPointer = 0
            else:
                charPointer += 1
        inputs.append(inputSequence)
        labels.append(labelSequence)
    return inputs, labels, charPointer

# Converts a single  character into a
# one-hot vector of length vocabSize
def charToOneHot(char):
    oneHot = [0] * vocabSize
    oneHot[charToInt[char]] = 1
    return oneHot

# Converts a list of sequenceLength characters
# into a list of corresponding one-hot vectors
def charsToOneHots(chars):
    oneHots = []
    for char in chars:
        oneHots.append(charToOneHot(char))
    return oneHots

# Converts a single one-hot vector into
# a list of corresponding characters
def oneHotToChar(oneHot):
    charIndex = np.argmax(oneHot)
    return intToChar[charIndex]

# Converts a list of sequenceLength one-hot vectors
# into a list of corresponding characters
def oneHotsToChars(oneHots):
    chars = []
    for batchIndex in range(len(oneHots)):
        batchChars = []
        for charIndex in range(len(oneHots[batchIndex])):
            batchChars.append(oneHotToChar(oneHots[batchIndex][charIndex]))
        chars.append(batchChars)
    return chars

# Converts a 3D array of predictions of shape [batchSize, sequenceLength, vocabSize]
# into a list of one-hot vectors where the 1 corresponds to the prediction with the
# highest score. These are then converted into batchSize lists of sequenceLength
# predicted characters
def predictionsToChars(predictions):
    chars = []
    for batchIndex in range(len(predictions)):
        batchChars = []
        for charIndex in range(len(predictions[batchIndex])):
            predictionOneHot = np.zeros_like(predictions[batchIndex][charIndex])
            maxChar = np.argmax(predictions[batchIndex][charIndex])
            predictionOneHot[maxChar] = 1
            batchChars.append(oneHotToChar(predictionOneHot))
        chars.append(batchChars)
    return chars

x = tf.placeholder(tf.float32, [None, sequenceLength, vocabSize])
y = tf.placeholder(tf.float32, [None, sequenceLength, vocabSize])
xTensors = tf.unstack(x, axis=1)                                                                # sequenceLength tensors of shape [batchSize, vocabSize]

W = tf.Variable(tf.random_normal([hiddenDimension, vocabSize]))
b = tf.Variable(tf.random_normal([vocabSize]))

cell = tf.contrib.rnn.BasicLSTMCell(hiddenDimension, forget_bias=forgetRate)
outputs, states = tf.nn.static_rnn(cell, xTensors, dtype=tf.float32)                            # sequenceLength list of tensors of shape [batchSize, hiddenDimension]
predictions = [tf.add(tf.matmul(output, W), b) for output in outputs]                           # sequenceLength list of tensors of shape [batchSize, vocabSize]
predictions = tf.transpose(predictions, [1, 0, 2])                                              # tensor of shape [batchSize, sequenceLength, vocabSize]

correctPredictions = tf.equal(tf.argmax(predictions, axis=2), tf.argmax(y, axis=2))             # tensor of shape [batchSize, sequenceLength], true if one-hot vector of predicted char = label
accuracy = tf.reduce_mean(tf.cast(correctPredictions, tf.float32))                              # real number in [0, 1] giving the total accuracy of predictions for this batch
loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(logits=predictions, labels=y))    # real number giving the total error in predictions for this batch
optimiser = tf.train.AdamOptimizer(learningRate).minimize(loss)                                 # adjusts the paramaters based on the loss

with tf.Session() as session:
    session.run(tf.global_variables_initializer())

    ############################################################################
    #   TRAINING
    ############################################################################
    pointer = 0
    for iteration in range(trainingIterations):
        batchX, batchY, pointer = generateData(batchSize, sequenceLength, isTraining=True, charPointer=pointer)
        dict = {x: batchX, y: batchY}
        session.run(optimiser, dict)
        if (iteration + 1) % printStep == 0 or iteration in (0, trainingIterations - 1):
            inputOneHots = session.run(x, dict)
            labelOneHots = session.run(y, dict)
            preds = session.run(predictions, dict)
            correct = session.run(correctPredictions, dict)
            iteractionAccuracy = session.run(accuracy, dict)
            iterationLoss = session.run(loss, dict)
            print("Iteration:\t" + str(iteration + 1) + " / " + str(trainingIterations))
            print("Accuracy:\t" + str("%.2f" % (iteractionAccuracy * 100) + "%"))
            print("Loss:\t\t" + str(iterationLoss))
            print("Inputs:\n" + str(oneHotsToChars(inputOneHots)))
            print("Labels:\n" + str(oneHotsToChars(labelOneHots)))
            print("Prediction:\n" + str(predictionsToChars(preds)))
            print("Correct:\n" + str(correct) + "\n")

    ############################################################################
    #   VALIDATING
    ############################################################################
    testX, testY, _ = generateData(numValidationSequences, sequenceLength, isTraining=False, charPointer=testingStartChar)
    testAccuracy = session.run(accuracy, {x: testX, y: testY})
    print("Testing Accuracy: " + str("%.2f" % (testAccuracy * 100) + "%"))

    ############################################################################
    #   GENERATING
    ############################################################################
    randIndex = random.randint(0, textSize - sequenceLength)
    seedSequence = text[randIndex : randIndex + sequenceLength]
    inputOneHots = charsToOneHots(seedSequence)

    with open("/home/kev/Documents/NeuralNetworks/CharacterPredicter/output.txt", "w+") as output:
        generatedText = ""
        lineLength = 0
        for _ in range(100):
            dict = {x: [inputOneHots]}
            pred = session.run(predictions, dict)
            predChars = predictionsToChars(pred)[0]
            inputOneHots = charsToOneHots(predChars)
            if predChars[-1] == "\n":
                lineLength = 0
            elif predChars[-2] == " " and lineLength >= 70:
                output.write("\n")
                lineLength = 0
            lineLength += 1
            output.write(predChars[-1])
$\endgroup$
2
$\begingroup$

If you read the documentation in tensorflow for static_rnn https://github.com/tensorflow/tensorflow/blob/master/tensorflow/python/ops/rnn.py#L1076

You will see that the input should be a list of tensors with size [batch_size, input_size].

But you have flattened your input, previously shaped like [None, sequenceLength, vocabSize] into the shape [None, sequenceLength*vocabSize]

When you write outputs, states = tf.nn.static_rnn(cell, [xFlat], dtype=tf.float32) you are passing into the RNN a sequence of length 1 (because there is only one item in the list), but you are giving the RNN the first 24 characters, all in the first time step! Clearly, if you give the RNN all 24 characters in the first time step, it will be able to give you the first 24 characters back.

To be clear, since you only passed in one-time step, this is not even an RNN anymore. It is effectively a feed-forward network with one hidden layer.

What you should do is not flatten the input x. I'm not sure why you did that. Instead, pass in a list with 24 tensors, each with size [None, vocabSize].

$\endgroup$
  • $\begingroup$ Wow, that is so obvious now that you have pointed it out to me, thank you. In terms of passing in a list of sequenceLength tensors into the RNN, can it be done as simply as x = tf.placeholder(tf.float32, [None, vocabSize]) and then outputs, states = tf.nn.static_rnn(cell, [x] * sequenceLength, dtype=tf.float32), or is this a very "un-TensorFlow-like" way of doing that? $\endgroup$ – KOB Aug 4 '17 at 9:16
  • $\begingroup$ I am continuously confusing myself with this as I struggle to get it working. If each time step in my RNN is a single char in the 25 character sequence, and each char is represented by a vocabSize one-hot vector, then I am failing to understand why the x placeholder is not 25 one-hot vectors? Does the x placeholder represent one whole sequence being input into the RNN (i.e. 25 time-steps) or does it just represent a single character at each time step? $\endgroup$ – KOB Aug 4 '17 at 11:38
  • $\begingroup$ in python multiplying lists is a dangerous thing to do if you don't know exactly what you're doing. [x] * sequenceLength really means to create a list of references, each one pointing to the same placeholder x. Instead, you want to make 24 distinct placeholders. $\endgroup$ – shimao Aug 4 '17 at 17:24
  • $\begingroup$ I think I figured it out earlier on. I have edited my post with my final code. I now need to train it and see how it does! If you have time, I'd appreciate it greatly if you could let me know what you think of my code. $\endgroup$ – KOB Aug 4 '17 at 18:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.