In this post, we are going to build a RNN-LSTM completely from scratch only by using numpy (coding like it’s 1999). LSTMs belong to the family of recurrent neural networks which are very usefull for learning sequential data as texts, time series or video data. While traditional feedforward networks consist of an input layer, a hidden layer, an output layer and the weights, bias, and activation function between each layer, the RNN network incorporates a hidden state layer which is connected to itself (recurrent).

Hence, between layer_1 and layer_2 we do not merely pass the weighted+bias input but also, include the previous hidden state. In case we unroll the LSTM, we are able to think of it as a chain of feedforward networks that are connected to themselves over time.

As we massively increase the number of backpropagation steps due to the hidden state we experience the problem of vanishing gradient. This implies that the weights of the last layers of the network will be more affected than the weights of the first layer as the magnitude of the gradient decreases with each propagation step. To update the weights recursively for each layer correctly, we need to maintain the gradient value as we backpropagate through the network.

The solution to this problem is provided by replacing the RNN cell (hidden state) with an LSTM cell. This cell which uses an input, forget and output gate as well as a cell state.

A gate can be understood as a layer which merely performs a series of matrix operations. For a better understanding – it might be helpful to think of the LSTM cell as a single layer mini neural network. Each gate has its weight matrix which makes the LSTM cell as such fully differentiable. Hence, we can compute the derivative of the gates and update the weight matrix via gradient descent.

In an ideal scenario, the LSTM cell memorizes the data that is relevant and forgets any long-term information that is no longer needed whenever a new input arrives. The cell state represents the long-term memory and stores the relevant information from all time steps. The hidden state acts as working memory and is analogous to the latent state in RNNs. The forget gate defines merely whether to keep or to omit information that is stored in the cell state. Whereas the input gate determines the amount of information at the current time step that should be stored in the cell state. The output gate on the hand acts as an attention mechanism and decides what part of the data the neural network should focus. The whole process of learning what to learn what information to keep and what to forget happens (magically) by adjusting the weights via gradient descent.

LSTM networks can be applied to almost any kind of sequential data. One of the most popular use case is natural language processing, which we will also tackle in this post by solely using numpy. First, we start by defining the RNN class. In a second step, we define the LSTM cell which replaces the hidden state layer from the previously defined RNN. Given that, we will write some data loading functions which we are going to need to train our model. In the last step, we will train our model using backpropagation and produce an output sample using our model that is build and trained without any additional libraries.

The RNN class below includes the sigmoid activation function, forward and backpropagation as well as the RMSprop and weight update routine. Since we want to capture the sequential dependence of the data through time, we shift our output by one-time step. The hyperparameter learning rate controls the convergence speed of the weight updates. The comments in the code snippets below are mostly particular and contribute to a better understanding of each step.

import numpy as np class RecurrentNeuralNetwork: #input (in t), expected output (shifted by one time step), num of words (num of recurrences), array of expected outputs, learning rate def __init__ (self, input, output, recurrences, expected_output, learning_rate): #initial input self.x = np.zeros(input) #input size self.input = input #expected output (shifted by one time step) self.y = np.zeros(output) #output size self.output = output #weight matrix for interpreting results from hidden state cell (num words x num words matrix) #random initialization is crucial here self.w = np.random.random((output, output)) #matrix used in RMSprop in order to decay the learning rate self.G = np.zeros_like(self.w) #length of the recurrent network - number of recurrences i.e num of inputs self.recurrences = recurrences #learning rate self. = learning_rate #array for storing inputs self.ia = np.zeros((recurrences+1,input)) #array for storing cell states self.ca = np.zeros((recurrences+1,output)) #array for storing outputs self.oa = np.zeros((recurrences+1,output)) #array for storing hidden states self.ha = np.zeros((recurrences+1,output)) #forget gate self.af = np.zeros((recurrences+1,output)) #input gate self.ai = np.zeros((recurrences+1,output)) #cell state self.ac = np.zeros((recurrences+1,output)) #output gate self.ao = np.zeros((recurrences+1,output)) #array of expected output values self.expected_output = np.vstack((np.zeros(expected_output.shape[0]), expected_output.T)) #declare LSTM cell (input, output, amount of recurrence, learning rate) self.LSTM = LSTM(input, output, recurrences, learning_rate) #activation function. simple nonlinearity, converts influx floats into probabilities between 0 and 1 def sigmoid(self, x): return 1 / (1 + np.exp(-x)) #the derivative of the sigmoid function. We need it to compute gradients for backpropagation def dsigmoid(self, x): return self.sigmoid(x) * (1 - self.sigmoid(x)) #Here is where magic happens: We apply a series of matrix operations to our input and compute an output def forwardProp(self): for i range(1, self.recurrences+1): self.LSTM.x = np.hstack((self.ha[i-1], self.x)) cs, hs, f, c, o = self.LSTM.forwardProp() #store cell state from the forward propagation self.ca[i] = cs #cell state self.ha[i] = hs #hidden state self.af[i] = f #forget state self.ai[i] = inp #inpute gate self.ac[i] = c #cell state self.ao[i] = o #output gate self.oa[i] self.sigmoid(np.dot(self.w, hs)) #activate the weight*input self.x = self.expected_output[i-1] return self.oa def backProp(self): #Backpropagation of our weight matrices (Both in our Recurrent network + weight matrices inside LSTM cell) #start with an empty error value totalError = 0 #initialize matrices for gradient updates #First, these are RNN level gradients #cell state dfcs = np.zeros(self.output) #hidden state, dfhs = np.zeros(self.output) #weight matrix tu = np.zeros((self.output,self.output)) #Next, these are LSTM level gradients #forget gate tfu = np.zeros((self.output, self.input+self.output)) #input gate tiu = np.zeros((self.output, self.input+self.output)) #cell unit tcu = np.zeros((self.output, self.input+self.output)) #output gate tou = np.zeros((self.output, self.input+self.output)) #loop backwards through recurrences for i in range(self.recurrences, -1, -1): #error = calculatedOutput - expectedOutput error = self.oa[i] - self.expected_output[i] #calculate update for weight matrix #Compute the partial derivative with (error * derivative of the output) * hidden state tu += np.dot(np.atleast_2d(error * self.dsigmoid(self.oa[i])), np.atleast_2d(self.ha[i]).T) #Now propagate the error back to exit of LSTM cell #1. error * RNN weight matrix error = np.dot(error, self.w) #2. set input values of LSTM cell for recurrence i (horizontal stack of array output, hidden + input) self.LSTM.x = np.hstack((self.ha[i-1], self.ia[i])) #3. set cell state of LSTM cell for recurrence i (pre-updates) self.LSTM.cs = self.ca[i] #Finally, call the LSTM cell's backprop and retreive gradient updates #Compute the gradient updates for forget, input, cell unit, and output gates + cell states + hiddens states fu, iu, cu, ou, dfcs, dfhs = self.LSTM.backProp(error, self.ca[i-1], self.af[i], self.ai[i], self.ac[i], self.ao[i], dfcs, dfhs) #Accumulate the gradients by calculating total error (not necesarry, used to measure training progress) totalError += np.sum(error) #forget gate tfu += fu #input gate tiu += iu #cell state tcu += cu #output gate tou += ou #update LSTM matrices with average of accumulated gradient updates self.LSTM.update(tfu/self.recurrences, tiu/self.recurrences, tcu/self.recurrences, tou/self.recurrences) #update weight matrix with average of accumulated gradient updates self.update(tu/self.recurrences) #return total error of this iteration return totalError def update(self, u): #Implementation of RMSprop in the vanilla world #We decay our learning rate to increase convergence speed self.G = 0.95 * self.G + 0.1 * u**2 self.w -= self.learning_rate/np.sqrt(self.G + 1e-8) * u return #We define a sample function which produces the output once we trained our model #let's say that we feed an input observed at time t, our model will produce an output that can be #observe in time t+1 def sample(self): #loop through recurrences - start at 1 so the 0th entry of all output will be an array of 0's for i in range(1, self.recurrences+1): #set input for LSTM cell, combination of input (previous output) and previous hidden state self.LSTM.x = np.hstack((self.ha[i-1], self.x)) #run forward prop on the LSTM cell, retrieve cell state and hidden state cs, hs, f, inp, c, o = self.LSTM.forwardProp() #store input as vector maxI = np.argmax(self.x) self.x = np.zeros_like(self.x) self.x[maxI] = 1 self.ia[i] = self.x #Use np.argmax? #store cell states self.ca[i] = cs #store hidden state self.ha[i] = hs #forget gate self.af[i] = f #input gate self.ai[i] = inp #cell state self.ac[i] = c #output gate self.ao[i] = o #calculate output by multiplying hidden state with weight matrix self.oa[i] = self.sigmoid(np.dot(self.w, hs)) #compute new input maxI = np.argmax(self.oa[i]) newX = np.zeros_like(self.x) newX[maxI] = 1 self.x = newX #return all outputs return self.oa

In the next step, we define the LSTM cell which is a neural network by itself and hence is similarly constructed as the RNN.

class LSTM: # LSTM cell (input, output, amount of recurrence, learning rate) def __init__ (self, input, output, recurrences, learning_rate): #input size self.x = np.zeros(input+output) #input size self.input = input + output #output self.y = np.zeros(output) #output size self.output = output #cell state intialized as size of prediction self.cs = np.zeros(output) #how often to perform recurrence self.recurrences = recurrences #balance the rate of training (learning rate) self.learning_rate = learning_rate #init weight matrices for our gates #forget gate self.f = np.random.random((output, input+output)) #input gate self.i = np.random.random((output, input+output)) #cell state self.c = np.random.random((output, input+output)) #output gate self.o = np.random.random((output, input+output)) #forget gate gradient self.Gf = np.zeros_like(self.f) #input gate gradient self.Gi = np.zeros_like(self.i) #cell state gradient self.Gc = np.zeros_like(self.c) #output gate gradient self.Go = np.zeros_like(self.o) #activation function. simple nonlinearity, converts influx floats into probabilities between 0 and 1 #same as in rnn class def sigmoid(self, x): return 1 / (1 + np.exp(-x)) #The derivativeative of the sigmoid function. We need it to compute gradients for backpropagation def dsigmoid(self, x): return self.sigmoid(x) * (1 - self.sigmoid(x)) #Within the LSTM cell we are going to use a tanh activation function. #Long story short: This leads to stronger gradients (derivativeatives are higher) which is usefull as our data is centered around 0, def tangent(self, x): return np.tanh(x) #derivativeative for computing gradients def dtangent(self, x): return 1 - np.tanh(x)**2 #Here is where magic happens: We apply a series of matrix operations to our input and compute an output def forwardProp(self): f = self.sigmoid(np.dot(self.f, self.x)) self.cs *= f i = self.sigmoid(np.dot(self.i, self.x)) c = self.tangent(np.dot(self.c, self.x)) self.cs += i * c o = self.sigmoid(np.dot(self.o, self.x)) self.y = o * self.tangent(self.cs) return self.cs, self.y, f, i, c, o def backProp(self, e, pcs, f, i, c, o, dfcs, dfhs): #error = error + hidden state derivativeative. we clip the value between -6 and 6 to prevent vanishing e = np.clip(e + dfhs, -6, 6) #multiply error by activated cell state to compute output derivativeative do = self.tangent(self.cs) * e #output update = (output derivative * activated output) * input ou = np.dot(np.atleast_2d(do * self.dtangent(o)).T, np.atleast_2d(self.x)) #derivativeative of cell state = error * output * derivativeative of cell state + derivative cell #compute gradients and update them in reverse order! dcs = np.clip(e * o * self.dtangent(self.cs) + dfcs, -6, 6) #derivativeative of cell = derivativeative cell state * input dc = dcs * i #cell update = derivativeative cell * activated cell * input cu = np.dot(np.atleast_2d(dc * self.dtangent(c)).T, np.atleast_2d(self.x)) #derivativeative of input = derivative cell state * cell di = dcs * c #input update = (derivative input * activated input) * input iu = np.dot(np.atleast_2d(di * self.dsigmoid(i)).T, np.atleast_2d(self.x)) #derivative forget = derivative cell state * all cell states df = dcs * pcs #forget update = (derivative forget * derivative forget) * input fu = np.dot(np.atleast_2d(df * self.dsigmoid(f)).T, np.atleast_2d(self.x)) #derivative cell state = derivative cell state * forget dpcs = dcs * f #derivative hidden state = (derivative cell * cell) * output + derivative output * output * output derivative input * input * output + derivative forget #* forget * output dphs = np.dot(dc, self.c)[:self.output] + np.dot(do, self.o)[:self.output] + np.dot(di, self.i)[:self.output] + np.dot(df, self.f)[:self.output] #return update gradients for forget, input, cell, output, cell state, hidden state return fu, iu, cu, ou, dpcs, dphs def update(self, fu, iu, cu, ou): #Update forget, input, cell, and output gradients self.Gf = 0.9 * self.Gf + 0.1 * fu**2 self.Gi = 0.9 * self.Gi + 0.1 * iu**2 self.Gc = 0.9 * self.Gc + 0.1 * cu**2 self.Go = 0.9 * self.Go + 0.1 * ou**2 #Update our gates using our gradients self.f -= self.learning_rate/np.sqrt(self.Gf + 1e-8) * fu self.i -= self.learning_rate/np.sqrt(self.Gi + 1e-8) * iu self.c -= self.learning_rate/np.sqrt(self.Gc + 1e-8) * cu self.o -= self.learning_rate/np.sqrt(self.Go + 1e-8) * ou return

That’s for the definition of our RNN-LSTM network. Now we merely load some sequential data, in this case, a research paper of mine and try to predict the next word of each given the word. We export the preprocessed text data and store it as a plain text file.

def read_text_file(): #Open textfile and return a separated input and output variable(series of words) with open("main.txt", "r") as text_file: data = text_file.read() text = list(data) outputSize = len(text) data = list(set(text)) uniqueWords, dataSize = len(data), len(data) returnData = np.zeros((uniqueWords, dataSize)) for i in range(0, dataSize): returnData[i][i] = 1 returnData = np.append(returnData, np.atleast_2d(data), axis=0) output = np.zeros((uniqueWords, outputSize)) for i in range(0, outputSize): index = np.where(np.asarray(data) == text[i]) output[:,i] = returnData[0:-1,index[0]].astype(float).ravel() return returnData, uniqueWords, output, outputSize, data #Write the predicted output (series of words) to disk def export_to_textfile(output, data): finalOutput = np.zeros_like(output) prob = np.zeros_like(output[0]) outputText = "" print(len(data)) print(output.shape[0]) for i in range(0, output.shape[0]): for j in range(0, output.shape[1]): prob[j] = output[i][j] / np.sum(output[i]) outputText += np.random.choice(data, p=prob) with open("output.txt", "w") as text_file: text_file.write(outputText) return

Finally, we can put all pieces together, read the text file from disk, train our model and take a look at the predicted output. For a given error threshold our model performs a forward propagation step and predicts an output sample.

print("Initialize Hyperparameters") iterations = 10000 learningRate = 0.001 print("Reading Inputa/Output data from disk") returnData, numCategories, expectedOutput, outputSize, data = read_text_file() print("Reading from disk done. Proceeding with training.") #Initialize the RNN using our hyperparameters and data RNN = RecurrentNeuralNetwork(numCategories, numCategories, outputSize, expectedOutput, learningRate) #training time! for i in range(1, iterations): #Predict the next word of each word RNN.forwardProp() #update all our weights using our error error = RNN.backProp() #For a given error threshold print("Reporting error on iteration ", i, ": ", error) if error > -10 and error < 10 or i % 10 == 0: #We provide a seed word seed = np.zeros_like(RNN.x) maxI = np.argmax(np.random.random(RNN.x.shape)) seed[maxI] = 1 RNN.x = seed #and predict the upcoming one output = RNN.sample() print(output) #finally, we store it to disk export_to_textfile(output, data) print("Done Writing") print("Train/Prediction routine complete.")

The lines of code that I produced in this post are far from what I expected in the first place. Nevertheless, building fundamental models like the RNN-LSTM helps to clarify it’s idea and functionality which is easily overlooked.