irony detector

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
from typing import List, Optional
import spacy

class Tokenizer:
    """Tokenizes and pads a batch of input sentences."""
    def __init__(self, pad_symbol: Optional[str] = "<PAD>"):
        """Initializes the tokenizer.
        Args:
            pad_symbol (Optional[str], optional): The symbol for a pad. Defaults to "<PAD>".
        """
        self.pad_symbol = pad_symbol
        self.nlp = spacy.load("en_core_web_sm")
    
    def __call__(self, batch: List[str]) -> List[List[str]]:
        """Tokenizes each sentence in the batch, and pads them if necessary so
        that we have equal length sentences in the batch.
        Args:
            batch (List[str]): A List of sentence strings.
        Returns:
            List[List[str]]: A List of equal-length token Lists.
        """
        batch = self.tokenize(batch)
        batch = self.pad(batch)
        return batch

    def tokenize(self, sentences: List[str]) -> List[List[str]]:
        """Tokenizes the List of string sentences into a Lists of tokens using spacy tokenizer.
        Args:
            sentences (List[str]): The input sentence.
        Returns:
            List[str]: The tokenized version of the sentence.
        """
        tokens = []
        for sentence in sentences:
            t = ["<SOS>"]
            for token in self.nlp(sentence):
                t.append(token.text)
            t.append("<EOS>")
            tokens.append(t)
        return tokens

    def pad(self, batch: List[List[str]]) -> List[List[str]]:
        """Appends pad symbols to each tokenized sentence in the batch such that
        every List of tokens is the same length. This means that the max length sentence
        will not be padded.
        Args:
            batch (List[List[str]]): Batch of tokenized sentences.
        Returns:
            List[List[str]]: Batch of padded tokenized sentences. 
        """
        maxlen = max([len(sent) for sent in batch])
        for sent in batch:
            for i in range(maxlen - len(sent)):
                sent.append(self.pad_symbol)
        return batch
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 
from functions.util import load_datasets, split_data

# Set up paths and constants
embeddings_path = 'glove.twitter.27B.200d.txt'
vocab_path = "./vocab.txt"
SPECIAL_TOKENS = ['<UNK>', '<PAD>', '<SOS>', '<EOS>']

# Download and split data
train_sentences, train_labels, test_sentences, test_labels, label2i = load_datasets()
training_sentences, training_labels, dev_sentences, dev_labels = split_data(train_sentences, train_labels, split=0.85)

# Set up tokenizer and make vocab
tokenizer = Tokenizer()
all_data = train_sentences + test_sentences
tokenized_data = tokenizer.tokenize(all_data)
vocab = sorted(set([w for ws in tokenized_data + [SPECIAL_TOKENS] for w in ws]))
with open('vocab.txt', 'w') as vf:
    vf.write('\n'.join(vocab))
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
from typing import List, Dict, Tuple
import torch


def read_pretrained_embeddings(
    embeddings_path: str,
    vocab_path: str
) -> Tuple[Dict[str, int], torch.FloatTensor]:
    """Read the embeddings matrix and make a dict hashing each word.
    Note that we have provided the entire vocab for train and test, so that for practical purposes
    we can simply load those words in the vocab, rather than all 27B embeddings.
    Args:
        embeddings_path (str): Glove embeddings downloaded by wget, pretrained run on 200d.
        vocab_path (str): Vocab file created from tokenizer.
    Returns:
        Tuple[Dict[str, int], torch.FloatTensor]: One hot vector of vocab, embeddings tensor.
    """
    word2i = {}
    vectors = []
    with open(vocab_path, encoding='utf8') as vf:
        vocab = set([w.strip() for w in vf.readlines()]) 
    print(f"Reading embeddings from {embeddings_path}...")
    with open(embeddings_path, "r") as f:
        i = 0
        for line in f:
            word, *weights = line.rstrip().split(" ")
            if word in vocab:
                word2i[word] = i
                vectors.append(torch.Tensor([float(w) for w in weights]))
                i += 1
    
    return word2i, torch.stack(vectors)


def get_glove_embeddings(
    embeddings_path: str,
) -> Tuple[Dict[str, int], torch.FloatTensor]:
    """Read all glove embeddings and make a dict hashing each word instead.
    Args:
        embeddings_path (str): Glove embeddings downloaded by wget, pretrained run on 200d.
    Returns:
        Tuple[Dict[str, int], torch.FloatTensor]: One hot vector of vocab, embeddings tensor.
    """
    word2i = {}
    vectors = []
    print(f"Reading embeddings from {embeddings_path}...")
    with open(embeddings_path, "r") as f:
        i = 0
        for line in f:
            word, *weights = line.rstrip().split(" ")
            word2i[word] = i
            vectors.append(torch.Tensor([float(w) for w in weights]))
            i += 1

    return word2i, torch.stack(vectors)


def get_oovs(vocab_path: str, word2i: Dict[str, int]) -> List[str]:
    """Find the vocab items that do not exist in the glove embeddings (in word2i).
    Return the List of such (unique) words.
    Args:
        vocab_path: List of batches of sentences.
        word2i (Dict[str, int]): See above.
    Returns:
        List[str]: Words not in glove.
    """
    with open(vocab_path, encoding='utf8') as vf:
        vocab = set([w.strip() for w in vf.readlines()])
    glove_and_vocab = set(word2i.keys())
    vocab_and_not_glove = vocab - glove_and_vocab
    
    return list(vocab_and_not_glove)


def intialize_new_embedding_weights(num_embeddings: int, dim: int) -> torch.FloatTensor:
    """Xavier initialization for the embeddings of words in train, but not in glove."""
    weights = torch.empty(num_embeddings, dim)
    torch.nn.init.xavier_uniform_(weights)
    
    return weights


def update_embeddings(
    glove_word2i: Dict[str, int],
    glove_embeddings: torch.FloatTensor,
    oovs: List[str]
) -> Tuple[Dict[str, int], torch.FloatTensor]:
    i = len(glove_word2i)
    for oov in oovs:
        if oov not in glove_word2i:
            glove_word2i[oov] = i
            i += 1
    new_embeddings = torch.cat((glove_embeddings, intialize_new_embedding_weights(len(oovs), glove_embeddings.shape[1])))
    
    return glove_word2i, new_embeddings


# Load the pretrained embeddings, find the out-of-vocabularies, and add to word2i and embeddings
glove_word2i, glove_embeddings = read_pretrained_embeddings(embeddings_path, vocab_path)
oovs = get_oovs(vocab_path, glove_word2i)
word2i, embeddings = update_embeddings(glove_word2i, glove_embeddings, oovs)
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 
from typing import List, Dict
import torch

def encode_sentences(batch: List[List[str]], word2i: Dict[str, int]) -> torch.LongTensor:
    """Encode the tokens in each sentence in the batch with a dictionary.
    Args:
        batch (List[List[str]]): The padded and tokenized batch of sentences.
        word2i (Dict[str, int]): The encoding dictionary.
    Returns:
        torch.LongTensor: The tensor of encoded sentences.
    """
    UNK_IDX = word2i["<UNK>"]
    tensors = []
    for sent in batch:
        tensors.append(torch.LongTensor([word2i.get(w, UNK_IDX) for w in sent]))
        
    return torch.stack(tensors)
 

1
2
3
4
5
6
7
8
 
 def encode_labels(labels: List[int]) -> torch.FloatTensor:
    """Turns the batch of labels into a tensor.
    Args:
        labels (List[int]): List of all labels in the batch.
    Returns:
        torch.FloatTensor: Tensor of all labels in the batch.
    """
    return torch.LongTensor([int(l) for l in labels])
 

1
2
3
4
5
6
7
 
 def make_batches(sequences: List[str], batch_size: int) -> List[List[str]]:
    """Yield batch_size chunks from sequences."""
    batches = []
    for i in range(0, len(sequences), batch_size):
        batches.append(sequences[i:i + batch_size])
    
    return batches
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
import torch

class IronyDetector(torch.nn.Module):
    def __init__(
        self,
        input_dim: int,
        hidden_dim: int,
        embeddings_tensor: torch.FloatTensor,
        pad_idx: int,
        output_size: int,
        dropout_val: float = 0.5,
    ):
        super().__init__()
        self.input_dim = input_dim
        self.hidden_dim = hidden_dim
        self.pad_idx = pad_idx
        self.dropout_val = dropout_val
        self.output_size = output_size
        # Initialize the embeddings from the weights matrix.
        self.embeddings = torch.nn.Embedding.from_pretrained(embeddings_tensor, True, pad_idx)
        # Dropout regularization
        self.dropout_layer = torch.nn.Dropout(p=self.dropout_val, inplace=False)
        # 2-layer Bi-LSTM
        self.lstm = torch.nn.LSTM(
            self.input_dim,
            self.hidden_dim,
            num_layers=3,
            dropout=dropout_val,
            batch_first=True,
            bidirectional=True,
        )
        # For classification over the final LSTM state.
        self.classifier = torch.nn.Linear(hidden_dim*2, self.output_size)
        self.log_softmax = torch.nn.LogSoftmax(dim=2)
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
from tqdm.notebook import tqdm as tqdm
from functions.util import f1_score, avg_f1_score
import random
import torch

def training_loop(
    num_epochs,
    train_features,
    train_labels,
    dev_features,
    dev_labels,
    optimizer,
    model,
    label2i,
):
    print("Training...")
    loss_func = torch.nn.NLLLoss()
    batches = list(zip(train_features, train_labels))
    random.shuffle(batches)
    for i in range(num_epochs):
        losses = []
        for features, labels in tqdm(batches):
            # Empty the dynamic computation graph
            optimizer.zero_grad()
            logits = model(features).squeeze(1)
            loss = loss_func(logits, labels)
            # Backpropogate the loss through our model
            loss.backward()
            optimizer.step()
            losses.append(loss.item())
        print(f"epoch {i+1}, loss: {sum(losses)/len(losses)}")
        # Estimate the f1 score for the development set
        print("Evaluating dev...")
        preds = predict(model, dev_features)
        dev_f1 = f1_score(preds, dev_labels, label2i['1'])
        dev_avg_f1 = avg_f1_score(preds, dev_labels, set(label2i.values()))
        print(f"Dev F1 {dev_f1}")
        print(f"Avg Dev F1 {dev_avg_f1}")
    # Return the trained model
    return model
 

1
2
3
4
5
6
7
 
 model = IronyDetector(
    input_dim=embeddings.shape[1],
    hidden_dim=128,
    embeddings_tensor=embeddings,
    pad_idx=word2i["<PAD>"],
    output_size=len(label2i),
)
 

1
2
3
4
5
6
 
# Hyperparameters
batch_size = 8
epochs = 15
learning_rate = 0.00005
weight_decay = 0
optimizer = torch.optim.AdamW(model.parameters(), learning_rate, weight_decay=weight_decay)
 

1
2
3
4
5
6
7
8
9
10
11
12
13
 
# Create batches
batch_tokenized = []
for batch in make_batches(training_sentences, batch_size):
    batch_tokenized.append(tokenizer(batch))
batch_labels = make_batches(training_labels, batch_size)
dev_sentences = tokenizer(dev_sentences)
test_sentences = tokenizer(test_sentences)

# Encode data
train_features = [encode_sentences(batch, word2i) for batch in batch_tokenized]
train_labels = [encode_labels(batch) for batch in batch_labels]
dev_features = encode_sentences(dev_sentences, word2i)
dev_labels = [int(l) for l in dev_labels]
 

1
2
3
4
5
6
7
8
9
10
11
 
# Train model
trained_model = training_loop(
    epochs,
    train_features,
    train_labels,
    dev_features,
    dev_labels,
    optimizer,
    model,
    label2i,
)
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
 
from typing import List
from functions.util import f1_score, avg_f1_score
import torch

def predict(model: torch.nn.Module, sequences: List[torch.Tensor]):
    """Run prediction with the model
    Args: model (torch.nn.Module): Model to use.
        dev_sequences (List[torch.Tensor]): List of encoded sequences to analyze.
    Returns:
        List[int]: List of predicted labels.
    """
    preds = []
    with torch.no_grad():
        model.eval()
        logits = model(sequences)
        preds = list(torch.argmax(logits, axis=2).squeeze().numpy())
    return preds

# Test model
test_features = encode_sentences(test_sentences, word2i)
test_labels = [int(l) for l in test_labels]
preds = predict(trained_model, test_features)
dev_f1 = f1_score(preds, test_labels, label2i['1'])
dev_avg_f1 = avg_f1_score(preds, test_labels, set(label2i.values()))
print(f"Test F1 {dev_f1}")
print(f"Avg Test F1 {dev_avg_f1}")