|
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 |
|
|
|
|
