A hidden state is the compact internal record a sequence model uses to carry information forward. Instead of predicting from an isolated token, the model conditions what comes next on the state built from what is happening now. section 2.2.3 section 2.2.7
The same idea appears at increasing levels of model complexity:
- Markov chains store probabilistic state and use the current state to model future token behavior, such as making “happy” more likely after “am.” section 2.2.3
- Recurrent neural networks update an internal state or memory as each token arrives, allowing them to process variable-length sequences. section 2.2.7
- LSTMs extend recurrent state with memory cells and gating mechanisms to manage information across sequences. section 2.2.7
Why sequence length matters
For sentiment analysis, an RNN may rate “I loved the movie last night” more positively than a longer semantically similar sentence because information changes as it passes through the recurrent state. Long sequences can also produce vanishing or exploding gradients. section 2.2.7
Listing 2.3 Generative hidden Markov language model implementation
import re
import random
from nltk.tokenize import word_tokenize
from collections import defaultdict, deque
class MarkovChain:
def __init__(self):
self.lookup_dict = defaultdict(list)
self._seeded = False
self.__seed_me()
def __seed_me(self, rand_seed=None):
if self._seeded is not True:
try:
if rand_seed is not None:
random.seed(rand_seed)
else:
random.seed()
self._seeded = True
except NotImplementedError:
self._seeded = False
def add_document(self, str):
preprocessed_list = self._preprocess(str)
pairs = self.__generate_tuple_keys(preprocessed_list)
for pair in pairs:
self.lookup_dict[pair[0]].append(pair[1])
def _preprocess(self, str):
cleaned = re.sub(r"\W+", " ", str).lower()
tokenized = word_tokenize(cleaned)
return tokenized
def __generate_tuple_keys(self, data):
if len(data) < 1:
return
for i in range(len(data) - 1):
yield [data[i], data[i + 1]]
def generate_text(self, max_length=50):
context = deque()
output = []
if len(self.lookup_dict) > 0:
self.__seed_me(rand_seed=len(self.lookup_dict))
chain_head = [list(self.lookup_dict)[0]]
context.extend(chain_head)
while len(output) < (max_length - 1):
next_choices = self.lookup_dict[context[-1]]
if len(next_choices) > 0:
next_word = random.choice(next_choices)
context.append(next_word)
output.append(context.popleft())
else:
break
output.extend(list(context))
return " ".join(output)
if __name__ == "__main__":
with open("hamlet.txt", "r", encoding="utf-8") as f:
text = f.read()
HMM = MarkovChain()
HMM.add_document(text)
print(HMM.generate_text(max_length=25))
Inspect a minimal generative hidden Markov model that tokenizes documents and records possible next tokens for each observed state. section 2.2.3
Listing 2.7 RNN and LSTM PyTorch class implementations
import torch
import pandas as pd
import numpy as np
from gensim.models import Word2Vec
from sklearn.model_selection import train_test_split
import nltk
import spacy
tokenizer = nltk.tokenize.RegexpTokenizer("\w+'?\w+|\w+'")
tokenizer.tokenize("This is a test")
stop_words = nltk.corpus.stopwords.words("english")
nlp = spacy.load("en_core_web_lg", disable=["parser", "tagger", "ner"])
dataset = pd.read_csv("./data/twitter.csv") #1
text_data = list(
map(lambda x: tokenizer.tokenize(x.lower()), dataset["text"])
)
text_data = [
[token.lemma_ for word in text for token in nlp(word)]
for text in text_data
]
label_data = list(map(lambda x: x, dataset["feeling"]))
assert len(text_data) == len(
label_data
), f"{len(text_data)} does not equal {len(label_data)}"
EMBEDDING_DIM = 100
model = Word2Vec(
text_data, vector_size=EMBEDDING_DIM, window=5, min_count=1, workers=4
)
word_vectors = model.wv
print(f"Vocabulary Length: {len(model.wv)}")
del model
padding_value = len(word_vectors.index_to_key)
embedding_weights = torch.Tensor(word_vectors.vectors) #2
class RNN(torch.nn.Module):
def __init__(
self,
input_dim,
embedding_dim,
hidden_dim,
output_dim,
embedding_weights,
):
super().__init__()
self.embedding = torch.nn.Embedding.from_pretrained(
embedding_weights
)
self.rnn = torch.nn.RNN(embedding_dim, hidden_dim)
self.fc = torch.nn.Linear(hidden_dim, output_dim)
def forward(self, x, text_lengths):
embedded = self.embedding(x)
packed_embedded = torch.nn.utils.rnn.pack_padded_sequence(
embedded, text_lengths
)
packed_output, hidden = self.rnn(packed_embedded)
output, output_lengths = torch.nn.utils.rnn.pad_packed_sequence(
packed_output
)
return self.fc(hidden.squeeze(0))
INPUT_DIM = padding_value
EMBEDDING_DIM = 100
HIDDEN_DIM = 256
OUTPUT_DIM = 1
rnn_model = RNN(
INPUT_DIM, EMBEDDING_DIM, HIDDEN_DIM, OUTPUT_DIM, embedding_weights
)
rnn_optimizer = torch.optim.SGD(rnn_model.parameters(), lr=1e-3)
rnn_criterion = torch.nn.BCEWithLogitsLoss()
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
class LSTM(torch.nn.Module):
def __init__(
self,
input_dim,
embedding_dim,
hidden_dim,
output_dim,
n_layers,
bidirectional,
dropout,
embedding_weights,
):
super().__init__()
self.embedding = torch.nn.Embedding.from_pretrained(
embedding_weights
)
self.rnn = torch.nn.LSTM(
embedding_dim,
hidden_dim,
num_layers=n_layers,
bidirectional=bidirectional,
dropout=dropout,
)
self.fc = torch.nn.Linear(hidden_dim * 2, output_dim)
self.dropout = torch.nn.Dropout(dropout)
def forward(self, x, text_lengths):
embedded = self.embedding(x)
packed_embedded = torch.nn.utils.rnn.pack_padded_sequence(
embedded, text_lengths
)
packed_output, (hidden, cell) = self.rnn(packed_embedded)
hidden = self.dropout(
torch.cat((hidden[-2, :, :], hidden[-1, :, :]), dim=1)
)
return self.fc(hidden.squeeze(0))
INPUT_DIM = padding_value
EMBEDDING_DIM = 100
HIDDEN_DIM = 256
OUTPUT_DIM = 1
N_LAYERS = 2
BIDIRECTIONAL = True
DROPOUT = 0.5
lstm_model = LSTM(
INPUT_DIM,
EMBEDDING_DIM,
HIDDEN_DIM,
OUTPUT_DIM,
N_LAYERS,
BIDIRECTIONAL,
DROPOUT,
embedding_weights,
)
lstm_optimizer = torch.optim.Adam(lstm_model.parameters())
lstm_criterion = torch.nn.BCEWithLogitsLoss()
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
def binary_accuracy(preds, y):
rounded_preds = torch.round(torch.sigmoid(preds))
correct = (rounded_preds == y).float()
acc = correct.sum() / len(correct)
return acc
def train(model, iterator, optimizer, criterion):
epoch_loss = 0
epoch_acc = 0
model.train()
for batch in iterator:
optimizer.zero_grad()
predictions = model(batch["text"], batch["length"]).squeeze(1)
loss = criterion(predictions, batch["label"])
acc = binary_accuracy(predictions, batch["label"])
loss.backward()
optimizer.step()
epoch_loss += loss.item()
epoch_acc += acc.item()
return epoch_loss / len(iterator), epoch_acc / len(iterator)
def evaluate(model, iterator, criterion):
epoch_loss = 0
epoch_acc = 0
model.eval()
with torch.no_grad():
for batch in iterator:
predictions = model(batch["text"], batch["length"]).squeeze(1)
loss = criterion(predictions, batch["label"])
acc = binary_accuracy(predictions, batch["label"])
epoch_loss += loss.item()
epoch_acc += acc.item()
return epoch_loss / len(iterator), epoch_acc / len(iterator)
batch_size = 2 #3
def iterator(X, y):
size = len(X)
permutation = np.random.permutation(size)
iterate = []
for i in range(0, size, batch_size):
indices = permutation[i : i + batch_size]
batch = {}
batch["text"] = [X[i] for i in indices]
batch["label"] = [y[i] for i in indices]
batch["text"], batch["label"] = zip(
*sorted(
zip(batch["text"], batch["label"]),
key=lambda x: len(x[0]),
reverse=True,
)
)
batch["length"] = [len(utt) for utt in batch["text"]]
batch["length"] = torch.IntTensor(batch["length"])
batch["text"] = torch.nn.utils.rnn.pad_sequence(
batch["text"], batch_first=True
).t()
batch["label"] = torch.Tensor(batch["label"])
batch["label"] = batch["label"].to(device)
batch["length"] = batch["length"].to(device)
batch["text"] = batch["text"].to(device)
iterate.append(batch)
return iterate
index_utt = [
torch.tensor([word_vectors.key_to_index.get(word, 0) for word in text])
for text in text_data
]
X_train, X_test, y_train, y_test = train_test_split(
index_utt, label_data, test_size=0.2
) #4
X_train, X_val, y_train, y_val = train_test_split(
X_train, y_train, test_size=0.2
)
train_iterator = iterator(X_train, y_train)
validate_iterator = iterator(X_val, y_val)
test_iterator = iterator(X_test, y_test)
print(len(train_iterator), len(validate_iterator), len(test_iterator))
N_EPOCHS = 25
for model in [rnn_model, lstm_model]:
print(
"|-----------------------------------------------------------------------------------------|"
)
print(f"Training with {model.__class__.__name__}")
if "RNN" in model.__class__.__name__:
for epoch in range(N_EPOCHS):
train_loss, train_acc = train(
rnn_model, train_iterator, rnn_optimizer, rnn_criterion
)
valid_loss, valid_acc = evaluate(
rnn_model, validate_iterator, rnn_criterion
)
print(
f"| Epoch: {epoch+1:02} | Train Loss: {train_loss: .3f} |
Train Acc: {train_acc*100: .2f}% | Validation Loss:
{valid_loss: .3f} | Validation Acc: {valid_acc*100: .2f}% |"
)
else:
for epoch in range(N_EPOCHS):
train_loss, train_acc = train(
lstm_model, train_iterator, lstm_optimizer, lstm_criterion
)
valid_loss, valid_acc = evaluate(
lstm_model, validate_iterator, lstm_criterion
)
print(
f"| Epoch: {epoch+1:02} | Train Loss: {train_loss: .3f} |
Train Acc: {train_acc*100: .2f}% | Validation Loss:
{valid_loss: .3f} | Validation Acc: {valid_acc*100: .2f}% |"
)
# Training on our dataset
# | Epoch: 01 | Train Loss: 0.560 | Train Acc: 70.63% | Validation Loss:
# 0.574 | Validation Acc: 70.88% |
# | Epoch: 05 | Train Loss: 0.391 | Train Acc: 82.81% | Validation Loss:
# 0.368 | Validation Acc: 83.08% |
# | Epoch: 10 | Train Loss: 0.270 | Train Acc: 89.11% | Validation Loss:
# 0.315 | Validation Acc: 86.22% |
# | Epoch: 15 | Train Loss: 0.186 | Train Acc: 92.95% | Validation Loss:
# 0.381 | Validation Acc: 87.49% |
# | Epoch: 20 | Train Loss: 0.121 | Train Acc: 95.93% | Validation Loss:
# 0.444 | Validation Acc: 86.29% |
# | Epoch: 25 | Train Loss: 0.100 | Train Acc: 96.28% | Validation Loss:
# 0.451 | Validation Acc: 86.83% |
Inspect RNN and LSTM class implementations alongside tokenization, embeddings, and sentiment data preparation. section 2.2.7
