In Part 1 and Part 2, we covered the mathematical foundations of forward-backward propagation and single-word sequence modeling. However, real-world language processing cannot rely solely on the immediate previous word; human dialogue requires temporal context and sequential memory.
A standard feedforward network treats every input independently. To give an AI short-term context awareness, we introduce Recurrent Neural Networks (RNN). An RNN maintains an internal Hidden State that acts as continuous working memory, passing information from previous tokens down the execution timeline.
1. The Recurrent Memory Equation
In an RNN cell, the hidden state at timestep t is computed using both the current input vector and the hidden state from the previous step:
ht = tanh(Wxh · xt + Whh · ht-1 + bh)
- Wxh: Weight matrix connecting the input to the hidden state.
- Whh: Recurrent weight matrix connecting the previous hidden state to the current hidden state.
- tanh: Activation function squashing memory values strictly between -1 and 1.
2. Complete Python Script: Scratch RNN Chatbot
The script below builds a recurrent neural memory loop, trains on conversational prompt-response pairs, and provides an interactive terminal chatbot interface:
import numpy as np
import re
class RecurrentChatbotEngine:
def __init__(self, hidden_dim=48, learning_rate=0.02):
self.hidden_dim = hidden_dim
self.lr = learning_rate
self.word_to_idx = {}
self.idx_to_word = {}
self.vocab_size = 0
def build_vocab(self, dialogue_corpus):
clean_text = re.sub(r'[^a-zA-Z0-9\s]', '', dialogue_corpus.lower())
tokens = clean_text.split()
unique_words = sorted(list(set(tokens)))
if "<pad>" not in unique_words: unique_words.insert(0, "<pad>")
if "<eos>" not in unique_words: unique_words.append("<eos>")
self.vocab_size = len(unique_words)
for idx, word in enumerate(unique_words):
self.word_to_idx[word] = idx
self.idx_to_word[idx] = word
self.init_weights()
def init_weights(self):
np.random.seed(42)
self.W_xh = np.random.randn(self.hidden_dim, self.vocab_size) * 0.01
self.W_hh = np.random.randn(self.hidden_dim, self.hidden_dim) * 0.01
self.W_hy = np.random.randn(self.vocab_size, self.hidden_dim) * 0.01
self.b_h = np.zeros((self.hidden_dim, 1))
self.b_y = np.zeros((self.vocab_size, 1))
def softmax(self, x):
e_x = np.exp(x - np.max(x))
return e_x / np.sum(e_x, axis=0)
def forward(self, inputs):
xs, hs, ps = {}, {}, {}
hs[-1] = np.zeros((self.hidden_dim, 1))
for t in range(len(inputs)):
xs[t] = np.zeros((self.vocab_size, 1))
xs[t][inputs[t]] = 1
hs[t] = np.tanh(np.dot(self.W_xh, xs[t]) + np.dot(self.W_hh, hs[t-1]) + self.b_h)
ys = np.dot(self.W_hy, hs[t]) + self.b_y
ps[t] = self.softmax(ys)
return xs, hs, ps
def backward(self, xs, hs, ps, targets):
dW_xh = np.zeros_like(self.W_xh)
dW_hh = np.zeros_like(self.W_hh)
dW_hy = np.zeros_like(self.W_hy)
db_h = np.zeros_like(self.b_h)
db_y = np.zeros_like(self.b_y)
dh_next = np.zeros_like(hs[0])
loss = 0
for t in reversed(range(len(targets))):
dy = np.copy(ps[t])
dy[targets[t]] -= 1
loss += -np.log(ps[t][targets[t], 0] + 1e-9)
dW_hy += np.dot(dy, hs[t].T)
db_y += dy
dh = np.dot(self.W_hy.T, dy) + dh_next
dh_raw = (1 - hs[t] * hs[t]) * dh
db_h += dh_raw
dW_xh += np.dot(dh_raw, xs[t].T)
dW_hh += np.dot(dh_raw, hs[t-1].T)
dh_next = np.dot(self.W_hh.T, dh_raw)
for dparam in [dW_xh, dW_hh, dW_hy, db_h, db_y]:
np.clip(dparam, -5, 5, out=dparam)
self.W_xh -= self.lr * dW_xh
self.W_hh -= self.lr * dW_hh
self.W_hy -= self.lr * dW_hy
self.b_h -= self.lr * db_h
self.b_y -= self.lr * db_y
return loss
def train(self, conversation_data, epochs=800):
for epoch in range(1, epochs + 1):
total_loss = 0
for prompt, response in conversation_data:
full_seq = (prompt + " " + response + " <eos>").lower().split()
indices = [self.word_to_idx.get(w, 0) for w in full_seq]
xs, hs, ps = self.forward(indices[:-1])
total_loss += self.backward(xs, hs, ps, indices[1:])
if epoch % 200 == 0 or epoch == 1:
print(f"Epoch {epoch:4d} | Loss: {total_loss:.4f}")
def reply(self, user_input, max_len=8):
clean_input = re.sub(r'[^a-zA-Z0-9\s]', '', user_input.lower()).split()
input_indices = [self.word_to_idx.get(w, 0) for w in clean_input]
if not input_indices: return "..."
xs, hs, ps = self.forward(input_indices)
h = hs[len(input_indices) - 1]
current_word_idx = np.argmax(ps[len(input_indices) - 1])
response_tokens = []
for _ in range(max_len):
x = np.zeros((self.vocab_size, 1))
x[current_word_idx] = 1
h = np.tanh(np.dot(self.W_xh, x) + np.dot(self.W_hh, h) + self.b_h)
p = self.softmax(np.dot(self.W_hy, h) + self.b_y)
current_word_idx = np.argmax(p)
word = self.idx_to_word[current_word_idx]
if word in ["<eos>", "<pad>"]: break
response_tokens.append(word)
return " ".join(response_tokens)
if __name__ == "__main__":
dialogue_pairs = [
("hello", "hi there human"),
("what is ai", "ai is math and gradient descent")
]
bot = RecurrentChatbotEngine(hidden_dim=32, learning_rate=0.03)
bot.build_vocab(" ".join([p + " " + r for p, r in dialogue_pairs]))
bot.train(dialogue_pairs, epochs=800)
print("User: hello -> AI:", bot.reply("hello"))
3. Key Architectural Mechanisms
- State Persistence: The internal state acts as dynamic RAM, holding the mathematical footprint of previous words to generate coherent multi-word replies.
- Gradient Clipping: Recurrent networks suffer from multiplying gradients over extended timelines; clipping derivative vectors prevents numerical overflow.