Tale of two language models: Revisiting the evaluation

    import math
    
    def mmi_score(model_probs, source_probs, target_size):
    """
    Compute the match between model probabilities and ideal n-gram probabilities.
    :param model_probs: A list of word probability distributions predicted by the model (list of numpy arrays).
                       Each array contains an entry per word in the vocabulary, where each value is the probability of that word given its context in the sentence.
                       The number of entries should be equal to the size of the vocabulary.
    
    :param source_probs: A dictionary containing the frequency counts of all words in the training data, as computed using the empirical distribution.
                         This can be obtained using code similar to the following:
                            # Get frequencies from file or database etc.
                            freq = {}
                            with open('train_corpus', 'r') as f:
                                lines = f.readlines()
                                for line in lines:
                                    tokens = line.strip().split()
                                    for token in tokens:
                                        if token not in freq:
                                            freq[token] = 0
                                        freq[token] += 1
    
                            total_words = sum(freq.values())
    
                            source_probs = dict([(word, count / float(total_words)) for word, count in freq.items()])
    
    :param target_size: An integer representing the number of possible output symbols for the target language.
    
    :return: A floating point score indicating how well the model matches the expected probabilities under the assumption
                 that it's selecting based on n-grams. A higher score indicates better performance.
    """
    mm_scores = []
    
    for i in range(len(source_probs)):
        s_prob = source_probs[i]
    
        # Extract top k most probable target words
        sorted_probs = sorted(enumerate(model_probs[i]), key=lambda x: x[1], reverse=True)[:target_size]
        t_probs = [x[1] for x in sorted_probs]
    
        # Calculate MMI score
        mm_score = abs(math.log(sum([t * math.log(t_probs[i]) for i, t in enumerate(t_probs)])))
        mm_scores.append(mm_score)
    
    return sum(mm_scores) / len(source_probs)
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    from collections import defaultdict
    
    def mdsi_score(train_data, dev_data, test_data):
    """
    Compute the disparity between train and development sets.
    :param train_data: Dictionary containing the training sentences as lists of word indices, indexed by language pair.
                      e.g., {'en-de': [[1, 2, 3], [4, 5, 6]], 'de-fr': [[7, 8, 9], [10, 11, 12]]}
    :param dev_data: Dictionary containing the development set as lists of word indices, indexed by language pair.
                    Same format as `train_data`.
    :param test_data: List of input sentence pairs to evaluate against.
                     Format is [(src_sentence, trg_sentence)]. Example: [('The quick brown fox.', 'Der schnelle braune Fuchs.'), ('She sells seashells by the seashore.')].
    
    :return: A floating point score indicating the difference between the probability mass assigned to different subsets of training examples.
             A lower score indicates better performance on this subset. Higher scores are usually worse since they penalize suboptimal solutions due to non-uniformity.
    """
    # Count occurrences of words in training corpus
    vocab_count = defaultdict(int)
    num_tokens = 0
    lang_pairs = set(train_data.keys()).union(dev_data.keys())
    for lp in lang_pairs:
        src_sents = train_data[lp] + dev_data[lp]
        tgt_sents = train_data['{}-{}'.format(lp[-1], lp[:-1])] + dev_data['{}-{}'.format(lp[-1], lp[:-1])]
        assert len(src_sents) == len(tgt_sents)
        for sent_num in range(len(src_sents)):
            for word in src_sents[sent_num]:
                vocab_count[word] += 1
                num_tokens += 1
    
            for word in tgt_sents[sent_num]:
                vocab_count[word] += 1
                num_tokens += 1
    
    
    # Compute probability mass for each language pair
    pms = {}
    denominators = {}
    for lp in lang_pairs:
        train_count = 0
        dev_count = 0
        for w in vocab_count:
            train_count += train_data[lp].count(w)
            train_count += dev_data[lp].count(w)
    
            dev_count += train_data['{}-{}'.format(lp[-1], lp[:-1])].count(w)
            dev_count += dev_data['{}-{}'.format(lp[-1], lp[:-1])].count(w)
    
        denominators[lp] = max(train_count, dev_count)
        pms[lp] = min(train_count, dev_count) / float(denominators[lp])
    
    # Evaluate on test set
    tp = 0
    fp = 0
    fn = 0
    for src_sent, trg_sent in test_data:
        src_ids = [vocab_index[w] for w in src_sent.split()]
        trg_ids = [vocab_index[w] for w in trg_sent.split()]
    
        for lp in lang_pairs:
            if pms[lp] > 0:
                combined_probs = combine_probs(src_ids, trg_ids, src_sent, trg_sent, lp)
                pred_labels = np.array([combined_probs[_trg][:, _pred].argmax() for _pred, _trg in zip(*np.where((combined_probs[:, :, :] >= threshold) & ((combined_probs[:, :, :] < 1.) | (~mask))))])
                true_labels = torch.LongTensor([[vocab_index[w] for w in trg_sent.split()], ]).to(device)[0].tolist()
    
                tp += len([l for l in pred_labels if l in true_labels])
                fp += len([l for l in pred_labels if l not in true_labels])
                fn += len([l for l in true_labels if l not in pred_labels])
    
    precision = tp / (tp + fp)
    recall = tp / (tp + fn)
    f1score = 2 * precision * recall / (precision + recall)
    
    return -(f1score + (1 - pms['en-de']) + (1 - pms['de-fr'] + (1 - pms['fr-en'])))
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import numpy as np
    import torch
    
    
    class Net(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = torch.nn.Linear(hidden_size * hidden_size, K)
    
    
    def forward(self, x):
        x = self.fc1(x)
        return x
    
    
    def multi_tagging_analysis(test_loader, net, device):
    correct = 0
    total = 0
    y_true = None
    y_pred = None
    
    print("Evaluating...")
    with torch.no_grad():
        for step, batch in enumerate(test_loader):
            inputs, labels = tuple(t.to(device) for t in batch)
    
            outputs = net(inputs.view(-1, hidden_size * hidden_size))
            _, predicted = torch.max(outputs, dim=1)
    
            c = (predicted == labels).squeeze()
    
            correct += int(c.sum())
            total += int(labels.shape[0])
    
            if y_true is None:
                y_true = labels.cpu().numpy()
                y_pred = predicted.cpu().numpy()
            else:
                y_true = np.concatenate((y_true, labels.cpu().numpy()))
                y_pred = np.concatenate((y_pred, predicted.cpu().numpy()))
    
    
    accuracy = correct / total
    
    print("\nTest Accuracy:", round(accuracy, 3))
    cm = confusion_matrix(y_true, y_pred)
    plot_confusion_matrix(cm, classes=['positive', 'negative'], title='Confusion matrix')
    
    
    def plot_confusion_matrix(cm, classes, normalize=False, title='Confusion matrix', cmap=plt.cm.Blues):
    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title)
    plt.colorbar()
    tick_marks = np.arange(len(classes))
    plt.xticks(tick_marks, classes, rotation=45)
    plt.yticks(tick_marks, classes)
    
    if normalize:
        cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]
        print("Normalized confusion matrix")
    else:
        print('Confusion matrix, without normalization')
    
    thresh = cm.max() / 2.
    for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
        plt.text(j, i, cm[i, j], horizontalalignment="center", color="white" if cm[i, j] > thresh else "black")
    
    plt.tight_layout()
    plt.ylabel('True label')
    plt.xlabel('Predicted label')
    
    def combine_probs(src_ids, trg_ids, src_sent, trg_sent, language):
    """
    Combine source and target word probabilities into one joint probability distribution over all tags.
    """
    mask = create_mask(trg_ids, len(trg_ids)).unsqueeze(dim=0).expand(batch_size, len(trg_ids), len(src_ids)).contiguous()
    
    encoder_output = encode_sentences(src_sent, language)
    decoder_input = get_decoder_input(encoder_output, src_sent, vocab_index['<sos>']).unsqueeze(dim=0)
    
    tagger_output, attn_weights = decode_sentences(encoder_output, decoder_input, trg_ids, trg_sent, language)
    
    probs = softmax(tagger_output)
    
    combined_probs = []
    for i in range(probs.size()[0]):
        combined_probs.append(combine_dists(probs[i][:-1], src_sent, trg_sent))
    
    return combined_probs
    
    
    def create_mask(tensor, length):
    """
    Create a binary mask of dimensions (length,) for the specified tensor. All values except those corresponding to valid positions will be masked out.
    """
    mask = torch.ones_like(tensor)
    mask[length:] = 0
    return mask
    
    
    def encode_sentences(sent, language):
    """
    Encode a single sentence using the pre-trained embedding layer provided by spacy package. Return the final representation after applying dropout and reshaping.
    """
    doc = nlp_dict[language](sent)
    tensor = torch.tensor([nlp_dict[language](sent).vector]).to(device)
    tensor = dropout(embedding(tensor)).reshape(1, 1, -1)
    
    return tensor
    
    
    def decode_sentences(encoder_output, decoder_input, targets, target_sent, language):
    """
    Decode a sequence of word ids from start to end, updating the attention weights along the way. Return both the decoded sequences and attention weights.
    """
    trg_indexes = targets[:]
    decoder_hidden = encoder_output
    
    tagger_output = []
    attn_weights = []
    use_teacher_forcing = True if random.random() < teacher_forcing_ratio else False
    
    if use_teacher_forcing:
        for di in range(min(target_length, len(targets))):
            decoder_output, decoder_hidden, weight = decoder(decoder_input, decoder_hidden, encoder_output)
            attn_weights.append(weight)
    
            tagger_output.append(decoder_output)
            decoder_input = decoder_output
    
    else:
        for di in range(target_length):
            decoder_output, decoder_hidden, weight = decoder(decoder_input, decoder_hidden, encoder_output)
            attn_weights.append(weight)
    
            topv, topi = decoder_output.topk(1)
            ni = topi[0][0]
    
            if ni == vocab_index['<eos>']:
                break
    
            tagger_output.append(decoder_output)
            decoder_input = decoder_output.clone().detach().requires_grad_()
    
    tagger_output = torch.stack(tagger_output)
    attn_weights = torch.stack(attn_weights)
    
    return tagger_output, attn_weights
    
    
    
    def combine_dists(dist, src_sent, trg_sent):
    """
    Given a row vector dist over source words, return a (vocab_size, vocab_size)-dimensional matrix where element (i,j) represents 
    the conditional probability of target word j given source word i according to the learned model. If any conditioning information 
    such as the source or target sentence itself is available, we can incorporate it here to improve the estimate. For simplicity, we ignore these details.
    """
    return dist.repeat(vocab_size, 1).transpose(0, 1)
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读