Upload evaluation.py

8841a3f over 1 year ago

No virus

7.19 kB

	import numpy as np
	import torch

	from evaluate import load as load_metric

	from sklearn.metrics import mean_squared_error
	from tqdm.auto import tqdm

	MAX_TARGET_LENGTH = 128

	# load evaluation metrics
	sacrebleu = load_metric('sacrebleu')
	rouge = load_metric('rouge')
	meteor = load_metric('meteor')
	bertscore = load_metric('bertscore')

	# use gpu if it's available
	device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

	def flatten_list(l):
	"""
	Utility function to convert a list of lists into a flattened list
	Params:
	l (list of lists): list to be flattened
	Returns:
	A flattened list with the elements of the original list
	"""
	return [item for sublist in l for item in sublist]

	def parse_float(value):
	"""
	Utility function to parse a string into a float
	Params:
	value (string): value to be converted to float
	Returns:
	The float representation of the given string, or None if the string could
	not be converted to a float
	"""
	try:
	float_value = float(value)
	return float_value
	except ValueError:
	return None

	def extract_scores(predictions):
	"""
	Utility function to extract the scores from the predictions of the model
	Params:
	predictions (list): complete model predictions
	Returns:
	scores (list): extracted scores from the model's predictions
	"""
	scores = []
	# iterate through predictions and try to extract predicted score;
	# if score could not be extracted, set it to None
	for pred in predictions:
	try:
	score_string = pred.split(' ', 1)[0].strip()
	score = parse_float(score_string)
	except IndexError:
	score = None
	scores.append(score)

	return scores

	def extract_feedback(predictions):
	"""
	Utility function to extract the feedback from the predictions of the model
	Params:
	predictions (list): complete model predictions
	Returns:
	feedback (list): extracted feedback from the model's predictions
	"""
	feedback = []
	# iterate through predictions and try to extract predicted feedback
	for pred in predictions:
	try:
	fb = pred.split(':', 1)[1]
	except IndexError:
	try:
	fb = pred.split(' ', 1)[1]
	except IndexError:
	fb = pred
	feedback.append(fb.strip())

	return feedback

	def compute_rmse(predictions, labels):
	"""
	Utility function to compute the root mean squared error of the
	score predictions in relation to the golden label scores
	Params:
	predictions (list): model score predictions
	labels (list): golden label scores
	Returns:
	(float, int): rmse of valid samples and number of invalid samples
	"""
	# get indexes of valid score predictions
	# (i.e., where the score is not None)
	idx = np.where(np.array(predictions) != None)

	# get size of the golden labels list and of
	# the valid predictions array
	labels_size = np.array(labels).size
	valid_predictions_size = idx[0].size

	# only compute rmse if valid score predictions were generated,
	# otherwise set mse to 1
	if valid_predictions_size > 0:
	# calculate rmse from labels and predictions
	valid_predictions = np.array(predictions)[idx]
	score_labels = np.array(labels)[idx]
	rmse = mean_squared_error(score_labels, valid_predictions, squared=False)

	# cap mse at 1
	if rmse > 1:
	return 1, labels_size - valid_predictions_size

	# return computed rmse and number of invalid samples
	return rmse, labels_size - valid_predictions_size
	else:
	return 1, labels_size - valid_predictions_size

	def compute_metrics(predictions, labels):
	"""
	Compute evaluation metrics from the predictions of the model
	Params:
	predictions (list): complete model predictions
	labels (list): golden labels (previously tokenized)
	Returns:
	results (dict): dictionary with the computed evaluation metrics
	"""
	# extract feedback and labels from the model's predictions
	predicted_feedback = extract_feedback(predictions)
	predicted_scores = extract_scores(predictions)

	# extract feedback and labels from the golden labels
	reference_feedback = [x.split('Feedback:', 1)[1].strip() for x in labels]
	reference_scores = [float(x.split('Feedback:', 1)[0].strip()) for x in labels]

	# compute HF metrics
	sacrebleu_score = sacrebleu.compute(predictions=predicted_feedback, references=[[x] for x in reference_feedback])['score']
	rouge_score = rouge.compute(predictions=predicted_feedback, references=reference_feedback)['rouge2']
	meteor_score = meteor.compute(predictions=predicted_feedback, references=reference_feedback)['meteor']
	bert_score = bertscore.compute(
	predictions=predicted_feedback,
	references=reference_feedback,
	lang='de',
	model_type='bert-base-multilingual-cased',
	rescale_with_baseline=True)

	# compute rmse of score predictions
	rmse, _ = compute_rmse(predicted_scores, reference_scores)

	results = {
	'sacrebleu': sacrebleu_score,
	'rouge': rouge_score,
	'meteor': meteor_score,
	'bert_score': np.array(bert_score['f1']).mean().item(),
	'rmse': rmse
	}

	return results

	def evaluate(model, tokenizer, dataloader):
	"""
	Evaluate model on the given dataset
	Params:
	model (PreTrainedModel): seq2seq model
	tokenizer (PreTrainedTokenizer): tokenizer from HuggingFace
	dataloader (torch Dataloader): dataloader of the dataset to be used for evaluation
	Returns:
	results (dict): dictionary with the computed evaluation metrics
	predictions (list): list of the decoded predictions of the model
	"""
	decoded_preds, decoded_labels = [], []

	model.eval()
	# iterate through batchs in the dataloader
	for batch in tqdm(dataloader):
	with torch.no_grad():
	batch = {k: v.to(device) for k, v in batch.items()}
	# generate tokens from batch
	generated_tokens = model.generate(
	batch['input_ids'],
	attention_mask=batch['attention_mask'],
	max_length=MAX_TARGET_LENGTH
	)
	# get golden labels from batch
	labels_batch = batch['labels']

	# decode model predictions and golden labels
	decoded_preds_batch = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)
	decoded_labels_batch = tokenizer.batch_decode(labels_batch, skip_special_tokens=True)

	decoded_preds.append(decoded_preds_batch)
	decoded_labels.append(decoded_labels_batch)

	# convert predictions and golden labels into flattened lists
	predictions = flatten_list(decoded_preds)
	labels = flatten_list(decoded_labels)

	# compute metrics based on predictions and golden labels
	results = compute_metrics(predictions, labels)

	return results, predictions