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])
