| from . import SentenceEvaluator |
| import logging |
| import os |
| import csv |
| from sklearn.metrics.pairwise import paired_cosine_distances, paired_euclidean_distances, paired_manhattan_distances |
| from sklearn.metrics import average_precision_score |
| import numpy as np |
| from typing import List |
| from ..readers import InputExample |
|
|
|
|
| logger = logging.getLogger(__name__) |
|
|
| class BinaryClassificationEvaluator(SentenceEvaluator): |
| """ |
| Evaluate a model based on the similarity of the embeddings by calculating the accuracy of identifying similar and |
| dissimilar sentences. |
| The metrics are the cosine similarity as well as euclidean and Manhattan distance |
| The returned score is the accuracy with a specified metric. |
| |
| The results are written in a CSV. If a CSV already exists, then values are appended. |
| |
| The labels need to be 0 for dissimilar pairs and 1 for similar pairs. |
| |
| :param sentences1: The first column of sentences |
| :param sentences2: The second column of sentences |
| :param labels: labels[i] is the label for the pair (sentences1[i], sentences2[i]). Must be 0 or 1 |
| :param name: Name for the output |
| :param batch_size: Batch size used to compute embeddings |
| :param show_progress_bar: If true, prints a progress bar |
| :param write_csv: Write results to a CSV file |
| """ |
|
|
| def __init__(self, sentences1: List[str], sentences2: List[str], labels: List[int], name: str = '', batch_size: int = 32, show_progress_bar: bool = False, write_csv: bool = True): |
| self.sentences1 = sentences1 |
| self.sentences2 = sentences2 |
| self.labels = labels |
|
|
| assert len(self.sentences1) == len(self.sentences2) |
| assert len(self.sentences1) == len(self.labels) |
| for label in labels: |
| assert (label == 0 or label == 1) |
|
|
| self.write_csv = write_csv |
| self.name = name |
| self.batch_size = batch_size |
| if show_progress_bar is None: |
| show_progress_bar = (logger.getEffectiveLevel() == logging.INFO or logger.getEffectiveLevel() == logging.DEBUG) |
| self.show_progress_bar = show_progress_bar |
|
|
| self.csv_file = "binary_classification_evaluation" + ("_"+name if name else '') + "_results.csv" |
| self.csv_headers = ["epoch", "steps", |
| "cossim_accuracy", "cossim_accuracy_threshold", "cossim_f1", "cossim_precision", "cossim_recall", "cossim_f1_threshold", "cossim_ap", |
| "manhattan_accuracy", "manhattan_accuracy_threshold", "manhattan_f1", "manhattan_precision", "manhattan_recall", "manhattan_f1_threshold", "manhattan_ap", |
| "euclidean_accuracy", "euclidean_accuracy_threshold", "euclidean_f1", "euclidean_precision", "euclidean_recall", "euclidean_f1_threshold", "euclidean_ap", |
| "dot_accuracy", "dot_accuracy_threshold", "dot_f1", "dot_precision", "dot_recall", "dot_f1_threshold", "dot_ap"] |
|
|
|
|
| @classmethod |
| def from_input_examples(cls, examples: List[InputExample], **kwargs): |
| sentences1 = [] |
| sentences2 = [] |
| scores = [] |
|
|
| for example in examples: |
| sentences1.append(example.texts[0]) |
| sentences2.append(example.texts[1]) |
| scores.append(example.label) |
| return cls(sentences1, sentences2, scores, **kwargs) |
|
|
| def __call__(self, model, output_path: str = None, epoch: int = -1, steps: int = -1) -> float: |
|
|
| if epoch != -1: |
| if steps == -1: |
| out_txt = f" after epoch {epoch}:" |
| else: |
| out_txt = f" in epoch {epoch} after {steps} steps:" |
| else: |
| out_txt = ":" |
|
|
| logger.info("Binary Accuracy Evaluation of the model on " + self.name + " dataset" + out_txt) |
|
|
| scores = self.compute_metrices(model) |
|
|
|
|
| |
| main_score = max(scores[short_name]['ap'] for short_name in scores) |
|
|
| file_output_data = [epoch, steps] |
|
|
| for header_name in self.csv_headers: |
| if '_' in header_name: |
| sim_fct, metric = header_name.split("_", maxsplit=1) |
| file_output_data.append(scores[sim_fct][metric]) |
|
|
| if output_path is not None and self.write_csv: |
| csv_path = os.path.join(output_path, self.csv_file) |
| if not os.path.isfile(csv_path): |
| with open(csv_path, newline='', mode="w", encoding="utf-8") as f: |
| writer = csv.writer(f) |
| writer.writerow(self.csv_headers) |
| writer.writerow(file_output_data) |
| else: |
| with open(csv_path, newline='', mode="a", encoding="utf-8") as f: |
| writer = csv.writer(f) |
| writer.writerow(file_output_data) |
|
|
| return main_score |
|
|
|
|
| def compute_metrices(self, model): |
| sentences = list(set(self.sentences1 + self.sentences2)) |
| embeddings = model.encode(sentences, batch_size=self.batch_size, show_progress_bar=self.show_progress_bar, convert_to_numpy=True) |
| emb_dict = {sent: emb for sent, emb in zip(sentences, embeddings)} |
| embeddings1 = [emb_dict[sent] for sent in self.sentences1] |
| embeddings2 = [emb_dict[sent] for sent in self.sentences2] |
|
|
| cosine_scores = 1 - paired_cosine_distances(embeddings1, embeddings2) |
| manhattan_distances = paired_manhattan_distances(embeddings1, embeddings2) |
| euclidean_distances = paired_euclidean_distances(embeddings1, embeddings2) |
|
|
| embeddings1_np = np.asarray(embeddings1) |
| embeddings2_np = np.asarray(embeddings2) |
| dot_scores = [np.dot(embeddings1_np[i], embeddings2_np[i]) for i in range(len(embeddings1_np))] |
|
|
|
|
| labels = np.asarray(self.labels) |
| output_scores = {} |
| for short_name, name, scores, reverse in [['cossim', 'Cosine-Similarity', cosine_scores, True], ['manhattan', 'Manhattan-Distance', manhattan_distances, False], ['euclidean', 'Euclidean-Distance', euclidean_distances, False], ['dot', 'Dot-Product', dot_scores, True]]: |
| acc, acc_threshold = self.find_best_acc_and_threshold(scores, labels, reverse) |
| f1, precision, recall, f1_threshold = self.find_best_f1_and_threshold(scores, labels, reverse) |
| ap = average_precision_score(labels, scores * (1 if reverse else -1)) |
|
|
| logger.info("Accuracy with {}: {:.2f}\t(Threshold: {:.4f})".format(name, acc * 100, acc_threshold)) |
| logger.info("F1 with {}: {:.2f}\t(Threshold: {:.4f})".format(name, f1 * 100, f1_threshold)) |
| logger.info("Precision with {}: {:.2f}".format(name, precision * 100)) |
| logger.info("Recall with {}: {:.2f}".format(name, recall * 100)) |
| logger.info("Average Precision with {}: {:.2f}\n".format(name, ap * 100)) |
|
|
| output_scores[short_name] = { |
| 'accuracy' : acc, |
| 'accuracy_threshold': acc_threshold, |
| 'f1': f1, |
| 'f1_threshold': f1_threshold, |
| 'precision': precision, |
| 'recall': recall, |
| 'ap': ap |
| } |
|
|
|
|
| return output_scores |
|
|
|
|
|
|
| @staticmethod |
| def find_best_acc_and_threshold(scores, labels, high_score_more_similar: bool): |
| assert len(scores) == len(labels) |
| rows = list(zip(scores, labels)) |
|
|
| rows = sorted(rows, key=lambda x: x[0], reverse=high_score_more_similar) |
|
|
| max_acc = 0 |
| best_threshold = -1 |
|
|
| positive_so_far = 0 |
| remaining_negatives = sum(labels == 0) |
|
|
| for i in range(len(rows)-1): |
| score, label = rows[i] |
| if label == 1: |
| positive_so_far += 1 |
| else: |
| remaining_negatives -= 1 |
|
|
| acc = (positive_so_far + remaining_negatives) / len(labels) |
| if acc > max_acc: |
| max_acc = acc |
| best_threshold = (rows[i][0] + rows[i+1][0]) / 2 |
|
|
| return max_acc, best_threshold |
|
|
| @staticmethod |
| def find_best_f1_and_threshold(scores, labels, high_score_more_similar: bool): |
| assert len(scores) == len(labels) |
|
|
| scores = np.asarray(scores) |
| labels = np.asarray(labels) |
|
|
| rows = list(zip(scores, labels)) |
|
|
| rows = sorted(rows, key=lambda x: x[0], reverse=high_score_more_similar) |
|
|
| best_f1 = best_precision = best_recall = 0 |
| threshold = 0 |
| nextract = 0 |
| ncorrect = 0 |
| total_num_duplicates = sum(labels) |
|
|
| for i in range(len(rows)-1): |
| score, label = rows[i] |
| nextract += 1 |
|
|
| if label == 1: |
| ncorrect += 1 |
|
|
| if ncorrect > 0: |
| precision = ncorrect / nextract |
| recall = ncorrect / total_num_duplicates |
| f1 = 2 * precision * recall / (precision + recall) |
| if f1 > best_f1: |
| best_f1 = f1 |
| best_precision = precision |
| best_recall = recall |
| threshold = (rows[i][0] + rows[i + 1][0]) / 2 |
|
|
| return best_f1, best_precision, best_recall, threshold |
|
|
|
|
