ERR.py

"""Token prediction metric."""

from typing import List, Tuple

import datasets
import numpy as np
from Levenshtein import distance as levenshtein_distance
from scipy.optimize import linear_sum_assignment

import evaluate


_DESCRIPTION = """
Unofficial implementation of the Error Reduction Rate (ERR) metric introduced for lexical normalization.
This implementation works on Seq2Seq models by aligning the predictions with the ground truth outputs.
"""


_KWARGS_DESCRIPTION = """
Args:
    predictions (`list` of `str`): Predicted labels.
    references (`list` of `Dict[str, str]`): Ground truth sentences, each with a field `input` and `output`.
Returns:
    `err` (`float` or `int`): Error Reduction Rate. See here: http://noisy-text.github.io/2021/multi-lexnorm.html
    `err_tp` (`int`): Number of true positives.
    `err_fn` (`int`): Number of false negatives.
    `err_tn` (`int`): Number of true negatives.
    `err_fp` (`int`): Number of false positives.
Examples:
    Example 1-A simple example
        >>> err = evaluate.load("err")
        >>> results = err.compute(predictions=[["The", "large", "dog"]], references=[{"input": ["The", "large", "dawg"], "output": ["The", "large", "dog"]}])
        >>> print(results)
        {'err': 1.0, 'err_tp': 2, 'err_fn': 0, 'err_tn': 1, 'err_fp': 0}
"""


_CITATION = """
@inproceedings{baldwin-etal-2015-shared,
    title = "Shared Tasks of the 2015 Workshop on Noisy User-generated Text: {T}witter Lexical Normalization and Named Entity Recognition",
    author = "Baldwin, Timothy  and
      de Marneffe, Marie Catherine  and
      Han, Bo  and
      Kim, Young-Bum  and
      Ritter, Alan  and
      Xu, Wei",
    booktitle = "Proceedings of the Workshop on Noisy User-generated Text",
    month = jul,
    year = "2015",
    address = "Beijing, China",
    publisher = "Association for Computational Linguistics",
    url = "https://aclanthology.org/W15-4319",
    doi = "10.18653/v1/W15-4319",
    pages = "126--135",
}
"""


@evaluate.utils.file_utils.add_start_docstrings(_DESCRIPTION, _KWARGS_DESCRIPTION)
class ErrorReductionRate(evaluate.Metric):
    def _info(self):
        return evaluate.MetricInfo(
            description=_DESCRIPTION,
            citation=_CITATION,
            inputs_description=_KWARGS_DESCRIPTION,
            features=datasets.Features(
                {
                    "predictions": datasets.Sequence(datasets.Value("string")),
                    "references": {
                        "input": datasets.Sequence(datasets.Value("string")),
                        "output": datasets.Sequence(datasets.Value("string")),
                    },
                }
            ),
        )

    def _compute(self, predictions, references):

        tp, fn, tn, fp = 0, 0, 0, 0
        for pred, ref in zip(predictions, references):
            inputs, outputs = ref["input"], ref["output"]

            labels = self._split_expressions_into_tokens(outputs)

            assert len(pred) == len(
                labels
            ), f"Number of predicted words ({len(pred)}) does not match number of target words ({len(labels)})"

            formatted_preds = self._align_predictions_with_labels(pred, labels)

            for i in range(len(inputs)):
                # Normalization was necessary
                if inputs[i].lower() != outputs[i]:
                    tp += formatted_preds[i] == outputs[i]
                    fn += formatted_preds[i] != outputs[i]
                else:
                    tn += formatted_preds[i] == outputs[i]
                    fp += formatted_preds[i] != outputs[i]

        err = (tp - fp) / (tp + fn)

        return {"err": err, "err_tp": tp, "err_fn": fn, "err_tn": tn, "err_fp": fp}

    def _align_predictions_with_labels(self, predictions: List[str], labels: List[Tuple[str, int]]) -> List[str]:
        levenshtein_matrix = np.zeros((len(labels), len(predictions)))

        for i, (label, _) in enumerate(labels):
            for j, pred in enumerate(predictions):
                levenshtein_matrix[i, j] = levenshtein_distance(label, pred)

        col_alignment, row_alignment = linear_sum_assignment(levenshtein_matrix)
        alignment = sorted(row_alignment, key=lambda i: col_alignment[i])

        num_outputs = max(map(lambda x: x[1], labels)) + 1
        formatted_preds = [[] for _ in range(num_outputs)]
        for i, aligned_idx in enumerate(alignment):
            formatted_preds[labels[i][1]].append(predictions[aligned_idx])

        formatted_preds = [" ".join(preds) for preds in formatted_preds]

        return formatted_preds

    def _split_expressions_into_tokens(self, outputs: List[str]) -> List[Tuple[str, int]]:
        labels = []
        for segment, normalized in enumerate(outputs):
            if normalized == "":
                labels.append((normalized, segment))
            else:
                for w in normalized.split():
                    labels.append((w, segment))

        return labels