File size: 4,468 Bytes
2359bda |
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 102 103 104 105 106 107 108 109 110 111 112 |
from . import SentenceEvaluator
import logging
from ..util import pytorch_cos_sim
import os
import csv
import numpy as np
import scipy.spatial
from typing import List
import torch
logger = logging.getLogger(__name__)
class TranslationEvaluator(SentenceEvaluator):
"""
Given two sets of sentences in different languages, e.g. (en_1, en_2, en_3...) and (fr_1, fr_2, fr_3, ...),
and assuming that fr_i is the translation of en_i.
Checks if vec(en_i) has the highest similarity to vec(fr_i). Computes the accurarcy in both directions
"""
def __init__(self, source_sentences: List[str], target_sentences: List[str], show_progress_bar: bool = False, batch_size: int = 16, name: str = '', print_wrong_matches: bool = False, write_csv: bool = True):
"""
Constructs an evaluator based for the dataset
The labels need to indicate the similarity between the sentences.
:param source_sentences:
List of sentences in source language
:param target_sentences:
List of sentences in target language
:param print_wrong_matches:
Prints incorrect matches
:param write_csv:
Write results to CSV file
"""
self.source_sentences = source_sentences
self.target_sentences = target_sentences
self.name = name
self.batch_size = batch_size
self.show_progress_bar = show_progress_bar
self.print_wrong_matches = print_wrong_matches
assert len(self.source_sentences) == len(self.target_sentences)
if name:
name = "_"+name
self.csv_file = "translation_evaluation"+name+"_results.csv"
self.csv_headers = ["epoch", "steps", "src2trg", "trg2src"]
self.write_csv = write_csv
def __call__(self, model, output_path: str = None, epoch: int = -1, steps: int = -1) -> float:
if epoch != -1:
if steps == -1:
out_txt = " after epoch {}:".format(epoch)
else:
out_txt = " in epoch {} after {} steps:".format(epoch, steps)
else:
out_txt = ":"
logger.info("Evaluating translation matching Accuracy on "+self.name+" dataset"+out_txt)
embeddings1 = torch.stack(model.encode(self.source_sentences, show_progress_bar=self.show_progress_bar, batch_size=self.batch_size, convert_to_numpy=False))
embeddings2 = torch.stack(model.encode(self.target_sentences, show_progress_bar=self.show_progress_bar, batch_size=self.batch_size, convert_to_numpy=False))
cos_sims = pytorch_cos_sim(embeddings1, embeddings2).detach().cpu().numpy()
correct_src2trg = 0
correct_trg2src = 0
for i in range(len(cos_sims)):
max_idx = np.argmax(cos_sims[i])
if i == max_idx:
correct_src2trg += 1
elif self.print_wrong_matches:
print("i:", i, "j:", max_idx, "INCORRECT" if i != max_idx else "CORRECT")
print("Src:", self.source_sentences[i])
print("Trg:", self.target_sentences[max_idx])
print("Argmax score:", cos_sims[i][max_idx], "vs. correct score:", cos_sims[i][i])
results = zip(range(len(cos_sims[i])), cos_sims[i])
results = sorted(results, key=lambda x: x[1], reverse=True)
for idx, score in results[0:5]:
print("\t", idx, "(Score: %.4f)" % (score), self.target_sentences[idx])
cos_sims = cos_sims.T
for i in range(len(cos_sims)):
max_idx = np.argmax(cos_sims[i])
if i == max_idx:
correct_trg2src += 1
acc_src2trg = correct_src2trg / len(cos_sims)
acc_trg2src = correct_trg2src / len(cos_sims)
logger.info("Accuracy src2trg: {:.2f}".format(acc_src2trg*100))
logger.info("Accuracy trg2src: {:.2f}".format(acc_trg2src*100))
if output_path is not None and self.write_csv:
csv_path = os.path.join(output_path, self.csv_file)
output_file_exists = os.path.isfile(csv_path)
with open(csv_path, newline='', mode="a" if output_file_exists else 'w', encoding="utf-8") as f:
writer = csv.writer(f)
if not output_file_exists:
writer.writerow(self.csv_headers)
writer.writerow([epoch, steps, acc_src2trg, acc_trg2src])
return (acc_src2trg+acc_trg2src)/2
|