Assignment 1: Neural Machine Translation¶

import os

os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' # Setting this env variable prevents TF warnings from showing up

​

import numpy as np

import tensorflow as tf

from collections import Counter

from utils import (sentences, train_data, val_data, english_vectorizer, portuguese_vectorizer, 

                   masked_loss, masked_acc, tokens_to_text)

import w1_unittest

portuguese_sentences, english_sentences = sentences

​

print(f"English (to translate) sentence:\n\n{english_sentences[-5]}\n")

print(f"Portuguese (translation) sentence:\n\n{portuguese_sentences[-5]}")

# sentences

# english_sentences[-2]

del portuguese_sentences

del english_sentences

del sentences

print(f"First 10 words of the english vocabulary:\n\n{english_vectorizer.get_vocabulary()[:10]}\n")

print(f"First 10 words of the portuguese vocabulary:\n\n{portuguese_vectorizer.get_vocabulary()[:10]}")

# Size of the vocabulary

vocab_size_por = portuguese_vectorizer.vocabulary_size()

vocab_size_eng = english_vectorizer.vocabulary_size()

​

print(f"Portuguese vocabulary is made up of {vocab_size_por} words")

print(f"English vocabulary is made up of {vocab_size_eng} words")

# This helps you convert from words to ids

word_to_id = tf.keras.layers.StringLookup(

    vocabulary=portuguese_vectorizer.get_vocabulary(), 

    mask_token="", 

    oov_token="[UNK]"

)

​

# This helps you convert from ids to words

id_to_word = tf.keras.layers.StringLookup(

    vocabulary=portuguese_vectorizer.get_vocabulary(),

    mask_token="",

    oov_token="[UNK]",

    invert=True,

)

unk_id = word_to_id("[UNK]")

sos_id = word_to_id("[SOS]")

eos_id = word_to_id("[EOS]")

baunilha_id = word_to_id("baunilha")

​

print(f"The id for the [UNK] token is {unk_id}")

print(f"The id for the [SOS] token is {sos_id}")

print(f"The id for the [EOS] token is {eos_id}")

print(f"The id for baunilha (vanilla) is {baunilha_id}")

for (to_translate, sr_translation), translation in train_data.take(1):

    print(f"Tokenized english sentence:\n{to_translate[0, :].numpy()}\n\n")

    print(f"Tokenized portuguese sentence (shifted to the right):\n{sr_translation[0, :].numpy()}\n\n")

    print(f"Tokenized portuguese sentence:\n{translation[0, :].numpy()}\n\n")

# train_data.take(1)

VOCAB_SIZE = 12000

UNITS = 256

# GRADED CLASS: Encoder

class Encoder(tf.keras.layers.Layer):

    def __init__(self, vocab_size, units):

        """Initializes an instance of this class

​

        Args:

            vocab_size (int): Size of the vocabulary

            units (int): Number of units in the LSTM layer

        """

        super(Encoder, self).__init__()

​

        ### START CODE HERE ###

​

        self.embedding = tf.keras.layers.Embedding(  

            input_dim=vocab_size,

            output_dim=units,

            mask_zero=True

        )  

​

        self.rnn = tf.keras.layers.Bidirectional(  

            layer=tf.keras.layers.LSTM(

                units=units,

                return_sequences=True

            ),  

            merge_mode="sum"

        )  

​

        ### END CODE HERE ###

​

    def call(self, context):

        """Forward pass of this layer

​

        Args:

            context (tf.Tensor): The sentence to translate

​

        Returns:

            tf.Tensor: Encoded sentence to translate

        """

​

        ### START CODE HERE ###

​

        # Pass the context through the embedding layer

        x = self.embedding(context)

​

        # Pass the output of the embedding through the RNN

        x = self.rnn(x)

​

        ### END CODE HERE ###

​

        return x

# Do a quick check of your implementation

​

# Create an instance of your class

encoder = Encoder(VOCAB_SIZE, UNITS)

​

# Pass a batch of sentences to translate from english to portuguese

encoder_output = encoder(to_translate)

​

print(f'Tensor of sentences in english has shape: {to_translate.shape}\n')

print(f'Encoder output has shape: {encoder_output.shape}')

# Test your code!

​

w1_unittest.test_encoder(Encoder)

# GRADED CLASS: CrossAttention

class CrossAttention(tf.keras.layers.Layer):

    def __init__(self, units):

        """Initializes an instance of this class

​

        Args:

            units (int): Number of units in the LSTM layer

        """

        super().__init__()

​

        ### START CODE HERE ###

​

        self.mha = tf.keras.layers.MultiHeadAttention( 

            key_dim=units,

            num_heads=1

        ) 

​

        ### END CODE HERE ###

​

        self.layernorm = tf.keras.layers.LayerNormalization()

        self.add = tf.keras.layers.Add()

​

    def call(self, context, target):

        """Forward pass of this layer

​

        Args:

            context (tf.Tensor): Encoded sentence to translate

            target (tf.Tensor): The embedded shifted-to-the-right translation

​

        Returns:

            tf.Tensor: Cross attention between context and target

        """

        ### START CODE HERE ###

​

        # Call the MH attention by passing in the query and value

        # For this case the query should be the translation and the value the encoded sentence to translate

        # Hint: Check the call arguments of MultiHeadAttention in the docs

        attn_output = self.mha(

            query=target,

            value=context

        )  

​

        ### END CODE HERE ###

​

        x = self.add([target, attn_output])

​

        x = self.layernorm(x)

​

        return x

# Do a quick check of your implementation

​

# Create an instance of your class

attention_layer = CrossAttention(UNITS)

​

# The attention layer expects the embedded sr-translation and the context

# The context (encoder_output) is already embedded so you need to do this for sr_translation:

sr_translation_embed = tf.keras.layers.Embedding(VOCAB_SIZE, output_dim=UNITS, mask_zero=True)(sr_translation)

​

# Compute the cross attention

attention_result = attention_layer(encoder_output, sr_translation_embed)

​

print(f'Tensor of contexts has shape: {encoder_output.shape}')

print(f'Tensor of translations has shape: {sr_translation_embed.shape}')

print(f'Tensor of attention scores has shape: {attention_result.shape}')

# Test your code!

​

w1_unittest.test_cross_attention(CrossAttention)

# GRADED CLASS: Decoder

class Decoder(tf.keras.layers.Layer):

    def __init__(self, vocab_size, units):

        """Initializes an instance of this class

​

        Args:

            vocab_size (int): Size of the vocabulary

            units (int): Number of units in the LSTM layer

        """

        super(Decoder, self).__init__()

​

        ### START CODE HERE ###

​

        # The embedding layer

        self.embedding = tf.keras.layers.Embedding(

            input_dim=vocab_size,

            output_dim=units,

            mask_zero=True

        )  

​

        # The RNN before attention

        self.pre_attention_rnn = tf.keras.layers.LSTM(

            units=units,

            return_sequences=True,

            return_state=True

        )  

​

        # The attention layer

        self.attention = CrossAttention(units)

​

        # The RNN after attention

        self.post_attention_rnn = tf.keras.layers.LSTM(

            units=units,

            return_sequences=True

        )  

​

        # The dense layer with logsoftmax activation

        self.output_layer = tf.keras.layers.Dense(

            units=vocab_size,

            activation=tf.nn.log_softmax

        )  

​

        ### END CODE HERE ###

​

    def call(self, context, target, state=None, return_state=False):

        """Forward pass of this layer

​

        Args:

            context (tf.Tensor): Encoded sentence to translate

            target (tf.Tensor): The shifted-to-the-right translation

            state (list[tf.Tensor, tf.Tensor], optional): Hidden state of the pre-attention LSTM. Defaults to None.

            return_state (bool, optional): If set to true return the hidden states of the LSTM. Defaults to False.

​

        Returns:

            tf.Tensor: The log_softmax probabilities of predicting a particular token

        """

        ### START CODE HERE ###

​

        # Get the embedding of the input

        x = self.embedding(target)

​

        # Pass the embedded input into the pre attention LSTM

        # Hints:

        # - The LSTM you defined earlier should return the output alongside the state (made up of two tensors)

        # - Pass in the state to the LSTM (needed for inference)

        x, hidden_state, cell_state = self.pre_attention_rnn(x, initial_state=state)

​

        # Perform cross attention between the context and the output of the LSTM (in that order)

        x = self.attention(context, x)

​

        # Do a pass through the post attention LSTM

        x = self.post_attention_rnn(x)

​

        # Compute the logits

        logits = self.output_layer(x)

​

        ### END CODE HERE ###

​

        if return_state:

            return logits, [hidden_state, cell_state]

​

        return logits

# Do a quick check of your implementation

​

# Create an instance of your class

decoder = Decoder(VOCAB_SIZE, UNITS)

​

# Notice that you don't need the embedded version of sr_translation since this is done inside the class

logits = decoder(encoder_output, sr_translation)

​

print(f'Tensor of contexts has shape: {encoder_output.shape}')

print(f'Tensor of right-shifted translations has shape: {sr_translation.shape}')

print(f'Tensor of logits has shape: {logits.shape}')

# Test your code!

​

w1_unittest.test_decoder(Decoder, CrossAttention)

# GRADED CLASS: Translator

class Translator(tf.keras.Model):

    def __init__(self, vocab_size, units):

        """Initializes an instance of this class

​

        Args:

            vocab_size (int): Size of the vocabulary

            units (int): Number of units in the LSTM layer

        """

        super().__init__()

​

        ### START CODE HERE ###

​

        # Define the encoder with the appropriate vocab_size and number of units

        self.encoder = Encoder(vocab_size, units)

​

        # Define the decoder with the appropriate vocab_size and number of units

        self.decoder = Decoder(vocab_size, units)

​

        ### END CODE HERE ###

​

    def call(self, inputs):

        """Forward pass of this layer

​

        Args:

            inputs (tuple(tf.Tensor, tf.Tensor)): Tuple containing the context (sentence to translate) and the target (shifted-to-the-right translation)

​

        Returns:

            tf.Tensor: The log_softmax probabilities of predicting a particular token

        """

​

        ### START CODE HERE ###

​

        # In this case inputs is a tuple consisting of the context and the target, unpack it into single variables

        context, target = inputs

​

        # Pass the context through the encoder

        encoded_context = self.encoder(context)

​

        # Compute the logits by passing the encoded context and the target to the decoder

        logits = self.decoder(encoded_context, target)

​

        ### END CODE HERE ###

​

        return logits

# Do a quick check of your implementation

​

# Create an instance of your class

translator = Translator(VOCAB_SIZE, UNITS)

​

# Compute the logits for every word in the vocabulary

logits = translator((to_translate, sr_translation))

​

print(f'Tensor of sentences to translate has shape: {to_translate.shape}')

print(f'Tensor of right-shifted translations has shape: {sr_translation.shape}')

print(f'Tensor of logits has shape: {logits.shape}')

w1_unittest.test_translator(Translator, Encoder, Decoder)

def compile_and_train(model, epochs=20, steps_per_epoch=500):

    model.compile(optimizer="adam", loss=masked_loss, metrics=[masked_acc, masked_loss])

​

    history = model.fit(

        train_data.repeat(),

        epochs=epochs,

        steps_per_epoch=steps_per_epoch,

        validation_data=val_data,

        validation_steps=50,

        callbacks=[tf.keras.callbacks.EarlyStopping(patience=3)],

    )

​

    return model, history

# Train the translator (this takes some minutes so feel free to take a break)

​

trained_translator, history = compile_and_train(translator)

def generate_next_token(decoder, context, next_token, done, state, temperature=0.0):

    """Generates the next token in the sequence

​

    Args:

        decoder (Decoder): The decoder

        context (tf.Tensor): Encoded sentence to translate

        next_token (tf.Tensor): The predicted next token

        done (bool): True if the translation is complete

        state (list[tf.Tensor, tf.Tensor]): Hidden states of the pre-attention LSTM layer

        temperature (float, optional): The temperature that controls the randomness of the predicted tokens. Defaults to 0.0.

​

    Returns:

        tuple(tf.Tensor, np.float, list[tf.Tensor, tf.Tensor], bool): The next token, log prob of said token, hidden state of LSTM and if translation is done

    """

    # Get the logits and state from the decoder

    logits, state = decoder(context, next_token, state=state, return_state=True)

    # Trim the intermediate dimension 

    logits = logits[:, -1, :]

    # If temp is 0 then next_token is the argmax of logits

    if temperature == 0.0:

        next_token = tf.argmax(logits, axis=-1)

    # If temp is not 0 then next_token is sampled out of logits

    else:

        logits = logits / temperature

        next_token = tf.random.categorical(logits, num_samples=1)

    # Trim dimensions of size 1

    logits = tf.squeeze(logits)

    next_token = tf.squeeze(next_token)

    # Get the logit of the selected next_token

    logit = logits[next_token].numpy()

    # Reshape to (1,1) since this is the expected shape for text encoded as TF tensors

    next_token = tf.reshape(next_token, shape=(1,1))

    # If next_token is End-of-Sentence token you are done

    if next_token == eos_id:

        done = True

    return next_token, logit, state, done

# PROCESS SENTENCE TO TRANSLATE AND ENCODE

​

# A sentence you wish to translate

eng_sentence = "I love languages"

​

# Convert it to a tensor

texts = tf.convert_to_tensor(eng_sentence)[tf.newaxis]

​

# Vectorize it and pass it through the encoder

context = english_vectorizer(texts).to_tensor()

context = encoder(context)

​

# SET STATE OF THE DECODER

​

# Next token is Start-of-Sentence since you are starting fresh

next_token = tf.fill((1,1), sos_id)

​

# Hidden and Cell states of the LSTM can be mocked using uniform samples

state = [tf.random.uniform((1, UNITS)), tf.random.uniform((1, UNITS))]

​

# You are not done until next token is EOS token

done = False

​

# Generate next token

next_token, logit, state, done = generate_next_token(decoder, context, next_token, done, state, temperature=0.5)

print(f"Next token: {next_token}\nLogit: {logit:.4f}\nDone? {done}")

# GRADED FUNCTION: translate

def translate(model, text, max_length=50, temperature=0.0):

    """Translate a given sentence from English to Portuguese

​

    Args:

        model (tf.keras.Model): The trained translator

        text (string): The sentence to translate

        max_length (int, optional): The maximum length of the translation. Defaults to 50.

        temperature (float, optional): The temperature that controls the randomness of the predicted tokens. Defaults to 0.0.

​

    Returns:

        tuple(str, np.float, tf.Tensor): The translation, logit that predicted <EOS> token and the tokenized translation

    """

    # Lists to save tokens and logits

    tokens, logits = [], []

​

    ### START CODE HERE ###

    # PROCESS THE SENTENCE TO TRANSLATE

    # Convert the original string into a tensor

    texts = tf.convert_to_tensor(text)[tf.newaxis]

    # Vectorize the text using the correct vectorizer

    context = english_vectorizer(texts).to_tensor()

    # Get the encoded context (pass the context through the encoder)

    # Hint: Remember you can get the encoder by using model.encoder

    context = model.encoder(context)

    # INITIAL STATE OF THE DECODER

    # First token should be SOS token with shape (1,1)

    next_token = tf.fill((1,1), sos_id)

    # Initial hidden and cell states should be tensors of zeros with shape (1, UNITS)

    state = [tf.zeros((1, UNITS)), tf.zeros((1, UNITS))]

    # You are done when you draw a EOS token as next token (initial state is False)

    done = False

​

    # Iterate for max_length iterations

    for _ in range(max_length):

        # Generate the next token

        try:                     

            next_token, logit, state, done = generate_next_token(

                model.decoder, context, next_token, done, state, temperature)

        except:

             raise Exception("Problem generating the next token")

        # If done then break out of the loop

        if done:

            break

        # Add next_token to the list of tokens

        tokens.append(next_token)

        # Add logit to the list of logits

        logits.append(logit)

    ### END CODE HERE ###

    # Concatenate all tokens into a tensor

    tokens = tf.concat(tokens, axis=-1)

    # Convert the translated tokens into text

    translation = tf.squeeze(tokens_to_text(tokens, id_to_word))

    translation = translation.numpy().decode()

    return translation, logits[-1], tokens

# Running this cell multiple times should return the same output since temp is 0

​

temp = 0.0 

original_sentence = "I love languages"

​

translation, logit, tokens = translate(trained_translator, original_sentence, temperature=temp)

​

print(f"Temperature: {temp}\n\nOriginal sentence: {original_sentence}\nTranslation: {translation}\nTranslation tokens:{tokens}\nLogit: {logit:.3f}")

# Running this cell multiple times should return different outputs since temp is not 0

# You can try different temperatures

​

temp = 0.7

original_sentence = "I love languages"

​

translation, logit, tokens = translate(trained_translator, original_sentence, temperature=temp)

​

print(f"Temperature: {temp}\n\nOriginal sentence: {original_sentence}\nTranslation: {translation}\nTranslation tokens:{tokens}\nLogit: {logit:.3f}")

w1_unittest.test_translate(translate, trained_translator)

def generate_samples(model, text, n_samples=4, temperature=0.6):

    samples, log_probs = [], []

​

    # Iterate for n_samples iterations

    for _ in range(n_samples):

        # Save the logit and the translated tensor

        _, logp, sample = translate(model, text, temperature=temperature)

        # Save the translated tensors

        samples.append(np.squeeze(sample.numpy()).tolist())

        # Save the logits

        log_probs.append(logp)

    return samples, log_probs

samples, log_probs = generate_samples(trained_translator, 'I love languages')

​

for s, l in zip(samples, log_probs):

    print(f"Translated tensor: {s} has logit: {l:.3f}")

def jaccard_similarity(candidate, reference):

    # Convert the lists to sets to get the unique tokens

    candidate_set = set(candidate)

    reference_set = set(reference)

    # Get the set of tokens common to both candidate and reference

    common_tokens = candidate_set.intersection(reference_set)

    # Get the set of all tokens found in either candidate or reference

    all_tokens = candidate_set.union(reference_set)

    # Compute the percentage of overlap (divide the number of common tokens by the number of all tokens)

    overlap = len(common_tokens) / len(all_tokens)

    return overlap

l1 = [1, 2, 3]

l2 = [1, 2, 3, 4]

​

js = jaccard_similarity(l1, l2)

​

print(f"jaccard similarity between lists: {l1} and {l2} is {js:.3f}")

# GRADED FUNCTION: rouge1_similarity

def rouge1_similarity(candidate, reference):

    """Computes the ROUGE 1 score between two token lists

​

    Args:

        candidate (list[int]): Tokenized candidate translation

        reference (list[int]): Tokenized reference translation

​

    Returns:

        float: Overlap between the two token lists

    """

    ### START CODE HERE ###

    # Make a frequency table of the candidate and reference tokens

    # Hint: use the Counter class (already imported)

    candidate_word_counts = Counter(candidate)

    reference_word_counts = Counter(reference)

    # Initialize overlap at 0

    overlap = 0

    # Iterate over the tokens in the candidate frequency table

    # Hint: Counter is a subclass of dict and you can get the keys 

    #       out of a dict using the keys method like this: dict.keys()

    for token in candidate_word_counts.keys():

        # Get the count of the current token in the candidate frequency table

        # Hint: You can access the counts of a token as you would access values of a dictionary

        token_count_candidate = candidate_word_counts[token]

        # Get the count of the current token in the reference frequency table

        # Hint: You can access the counts of a token as you would access values of a dictionary

        token_count_reference = reference_word_counts[token]

        # Update the overlap by getting the minimum between the two token counts above

        overlap += min(token_count_candidate, token_count_reference)

    # Compute the precision

    # Hint: precision = overlap / (number of tokens in candidate list) 

    precision = overlap / len(candidate)

    # Compute the recall

    # Hint: recall = overlap / (number of tokens in reference list) 

    recall = overlap / len(reference)

    if precision + recall != 0:

        # Compute the Rouge1 Score

        # Hint: This is equivalent to the F1 score

        f1_score = 2 * (precision * recall) / (precision + recall)

        return f1_score

    ### END CODE HERE ###

    return 0 # If precision + recall = 0 then return 0

l1 = [1, 2, 3]

l2 = [1, 2, 3, 4]

​

r1s = rouge1_similarity(l1, l2)

​

print(f"rouge 1 similarity between lists: {l1} and {l2} is {r1s:.3f}")

w1_unittest.test_rouge1_similarity(rouge1_similarity)

# GRADED FUNCTION: average_overlap

def average_overlap(samples, similarity_fn):

    """Computes the arithmetic mean of each candidate sentence in the samples

​

    Args:

        samples (list[list[int]]): Tokenized version of translated sentences

        similarity_fn (Function): Similarity function used to compute the overlap

​

    Returns:

        dict[int, float]: A dictionary mapping the index of each translation to its score

    """

    # Initialize dictionary

    scores = {}

    # Iterate through all samples (enumerate helps keep track of indexes)

    for index_candidate, candidate in enumerate(samples):    

        ### START CODE HERE ###

        # Initially overlap is zero

        overlap = 0

        # Iterate through all samples (enumerate helps keep track of indexes)

        for index_sample, sample in enumerate(samples):

​

            # Skip if the candidate index is the same as the sample index

            if index_candidate == index_sample:

                continue

            # Get the overlap between candidate and sample using the similarity function

            sample_overlap = similarity_fn(candidate, sample)

            # Add the sample overlap to the total overlap

            overlap += sample_overlap

​

        ### END CODE HERE ###

        # Get the score for the candidate by computing the average

        score = overlap / (len(samples) - 1)

​

        # Only use 3 decimal points

        score = round(score, 3)

        # Save the score in the dictionary. use index as the key.

        scores[index_candidate] = score

    return scores

# Test with Jaccard similarity

​

l1 = [1, 2, 3]

l2 = [1, 2, 4]

l3 = [1, 2, 4, 5]

​

avg_ovlp = average_overlap([l1, l2, l3], jaccard_similarity)

​

print(f"average overlap between lists: {l1}, {l2} and {l3} using Jaccard similarity is:\n\n{avg_ovlp}")

# Test with Rouge1 similarity

​

l1 = [1, 2, 3]

l2 = [1, 4]

l3 = [1, 2, 4, 5]

l4 = [5,6]

​

avg_ovlp = average_overlap([l1, l2, l3, l4], rouge1_similarity)

​

print(f"average overlap between lists: {l1}, {l2}, {l3} and {l4} using Rouge1 similarity is:\n\n{avg_ovlp}")

w1_unittest.test_average_overlap(average_overlap)