[Feature] Support DROP exact-match and F1 metrics

opencompass/datasets/drop_simple_eval.py

@@ -1,6 +1,7 @@
 import json
 import re
 import string
+from typing import List, Set, Tuple
 
 from datasets import Dataset, DatasetDict
 
@@ -15,6 +16,90 @@
 ANSWER_PATTERN = r'(?i)Answer\s*:\s*([^\n]+)'
 
 
+def _remove_articles(text: str) -> str:
+    regex = re.compile(r'\b(a|an|the)\b', re.UNICODE)
+    return re.sub(regex, ' ', text)
+
+
+def _white_space_fix(text: str) -> str:
+    return ' '.join(text.split())
+
+
+EXCLUDE = set(string.punctuation)
+
+
+def _is_number(text: str) -> bool:
+    try:
+        float(text)
+        return True
+    except ValueError:
+        return False
+
+
+def _remove_punc(text: str) -> str:
+    if _is_number(text):
+        return text
+    return ''.join(char for char in text if char not in EXCLUDE)
+
+
+def _normalize_number(text: str) -> str:
+    if _is_number(text):
+        return str(float(text))
+    return text
+
+
+def _normalize_answer(text: str) -> str:
+    parts = [
+        _white_space_fix(
+            _remove_articles(_normalize_number(_remove_punc(token.lower()))))
+        for token in re.split(r' |-', text)
+    ]
+    return ' '.join(part for part in parts if part.strip()).strip()
+
+
+def _answer_to_bag(answer: str) -> Tuple[str, Set[str]]:
+    normalized_answer = _normalize_answer(answer)
+    return normalized_answer, set(normalized_answer.split())
+
+
+def _compute_f1(predicted_bag: Set[str], gold_bag: Set[str]) -> float:
+    intersection = len(gold_bag.intersection(predicted_bag))
+    precision = intersection / len(predicted_bag) if predicted_bag else 1.0
+    recall = intersection / len(gold_bag) if gold_bag else 1.0
+    if precision == 0.0 and recall == 0.0:
+        return 0.0
+    return 100 * (2 * precision * recall) / (precision + recall)
+
+
+def _match_numbers_if_present(gold_bag: Set[str],
+                              predicted_bag: Set[str]) -> bool:
+    gold_numbers = {word for word in gold_bag if _is_number(word)}
+    predicted_numbers = {word for word in predicted_bag if _is_number(word)}
+    return not gold_numbers or bool(
+        gold_numbers.intersection(predicted_numbers))
+
+
+def get_drop_metrics(predicted: str, gold: str) -> Tuple[float, float]:
+    """Return official DROP exact-match and F1 for single-span answers."""
+    normalized_predicted, predicted_bag = _answer_to_bag(predicted)
+    normalized_gold, gold_bag = _answer_to_bag(gold)
+    exact_match = float(normalized_predicted == normalized_gold)
+    f1 = 0.0
+    if _match_numbers_if_present(gold_bag, predicted_bag):
+        f1 = _compute_f1(predicted_bag, gold_bag)
+    return exact_match, round(f1, 2)
+
+
+def drop_metric(sample: str, references: List[str]) -> Tuple[float, float]:
+    """Return the best DROP exact-match and F1 across valid references."""
+    scores = [
+        get_drop_metrics(sample, answer) for answer in references
+        if answer.strip()
+    ]
+    em_scores, f1_scores = zip(*scores)
+    return max(em_scores), max(f1_scores)
+
+
 def normalize(s: str) -> str:
     """Lower text and remove punctuation, articles and extra whitespace."""
     s = s.lower()
@@ -61,22 +146,38 @@ def score(self, predictions, references):
         if len(predictions) != len(references):
             return {'error': 'preds and refers have different length'}
         num_correct = 0
+        exact_match = 0.0
+        f1 = 0.0
         count = 0
         details = []
         for pred, refr in zip(predictions, references):
             match = re.search(ANSWER_PATTERN, pred)
             extracted_answer = match.group(1) if match else pred
             refrs = refr.split('|')
+            em_score, f1_score = drop_metric(extracted_answer, refrs)
             matches = [
                 fuzzy_match(extracted_answer, correct_answer)
                 for correct_answer in refrs
             ]
             correct = True in matches
             num_correct += correct
-
-            detail = {'pred': pred, 'answer': refr, 'correct': correct}
+            exact_match += em_score
+            f1 += f1_score
+
+            detail = {
+                'pred': pred,
+                'answer': refr,
+                'correct': correct,
+                'exact_match': em_score,
+                'f1': f1_score,
+            }
             count += 1
 
             details.append(detail)
-        result = {'accuracy': 100 * num_correct / count, 'details': details}
+        result = {
+            'accuracy': 100 * num_correct / count,
+            'exact_match': 100 * exact_match / count,
+            'f1': f1 / count,
+            'details': details,
+        }
         return result

tests/datasets/test_drop_simple_eval.py

@@ -0,0 +1,47 @@
+import unittest
+
+from opencompass.datasets.drop_simple_eval import (DropOpenAIEvaluator,
+                                                   drop_metric)
+
+
+class TestDropSimpleEval(unittest.TestCase):
+
+    def test_drop_metric_normalizes_numbers(self):
+        self.assertEqual(drop_metric('18.0', ['18']), (1.0, 100.0))
+
+    def test_drop_metric_normalizes_text(self):
+        cases = [
+            ('The U.S.', 'US'),
+            ('New\tYork', 'New York'),
+            ('New-York', 'New York'),
+        ]
+
+        for prediction, reference in cases:
+            with self.subTest(prediction=prediction, reference=reference):
+                self.assertEqual(drop_metric(prediction, [reference]),
+                                 (1.0, 100.0))
+
+    def test_drop_metric_rejects_overlap_with_different_numbers(self):
+        self.assertEqual(drop_metric('18 yards', ['19 yards']), (0.0, 0.0))
+
+    def test_drop_metric_handles_empty_prediction_and_reference(self):
+        self.assertEqual(drop_metric('', ['', 'answer']), (0.0, 0.0))
+
+    def test_evaluator_reports_em_and_f1_over_alternative_answers(self):
+        result = DropOpenAIEvaluator().score(
+            predictions=['Answer: Denver', 'Answer: Broncos'],
+            references=['Denver Broncos|Broncos', 'Denver Broncos|Broncos'],
+        )
+
+        self.assertEqual(result['accuracy'], 100.0)
+        self.assertEqual(result['exact_match'], 50.0)
+        self.assertAlmostEqual(result['f1'], 83.335)
+        self.assertTrue(result['details'][0]['correct'])
+        self.assertEqual(result['details'][0]['exact_match'], 0.0)
+        self.assertEqual(result['details'][0]['f1'], 66.67)
+        self.assertEqual(result['details'][1]['exact_match'], 1.0)
+        self.assertEqual(result['details'][1]['f1'], 100.0)
+
+
+if __name__ == '__main__':
+    unittest.main()