| import json |
| import os.path as osp |
| import re |
|
|
| from datasets import Dataset |
|
|
| from opencompass.openicl.icl_evaluator import BaseEvaluator |
| from opencompass.registry import (ICL_EVALUATORS, LOAD_DATASET, |
| TEXT_POSTPROCESSORS) |
|
|
| from .base import BaseDataset |
|
|
|
|
| @LOAD_DATASET.register_module() |
| class BBHDataset(BaseDataset): |
|
|
| @staticmethod |
| def load(path: str, name: str): |
| with open(osp.join(path, f'{name}.json'), 'r') as f: |
| data = json.load(f)['examples'] |
| dataset = Dataset.from_list(data) |
| return dataset |
|
|
|
|
| @TEXT_POSTPROCESSORS.register_module('bbh-mcq') |
| def bbh_mcq_postprocess(text: str) -> str: |
| ans = text |
| ans_line = ans.split('answer is ') |
| if len(ans_line) != 1: |
| ans = ans_line[1].strip() |
| match = re.search(r'\(([A-Z])\)*', ans) |
| if match: |
| return match.group(1) |
| match = re.search(r'([A-Z])', ans) |
| if match: |
| return match.group(1) |
| return ans |
|
|
|
|
| @TEXT_POSTPROCESSORS.register_module('bbh-freeform') |
| def bbh_freeform_postprocess(text: str) -> str: |
| ans = text |
| ans_line = ans.split('answer is ') |
| if len(ans_line) != 1: |
| ans = ans_line[1].strip() |
| ans = ans.split('\n')[0] |
| if ans.endswith('.'): |
| ans = ans[:-1] |
| return ans |
|
|
|
|
| @ICL_EVALUATORS.register_module() |
| class BBHEvaluator(BaseEvaluator): |
|
|
| def score(self, predictions, references): |
| if len(predictions) != len(references): |
| return { |
| 'error': 'predictions and references have different ' |
| 'length' |
| } |
|
|
| predictions = [bbh_freeform_postprocess(pred) for pred in predictions] |
|
|
| details = [] |
| cnt = 0 |
| for pred, ref in zip(predictions, references): |
| detail = {'pred': pred, 'answer': ref, 'correct': False} |
| if pred == ref: |
| cnt += 1 |
| detail['correct'] = True |
| details.append(detail) |
|
|
| score = cnt / len(predictions) * 100 |
|
|
| return {'score': score, 'details': details} |
|
|
|
|
| @ICL_EVALUATORS.register_module() |
| class BBHEvaluator_mcq(BaseEvaluator): |
|
|
| def score(self, predictions, references): |
| if len(predictions) != len(references): |
| return { |
| 'error': 'predictions and references have different ' |
| 'length' |
| } |
| details = [] |
| cnt = 0 |
| for pred, ref in zip(predictions, references): |
| detail = {'pred': pred, 'answer': ref, 'correct': False} |
| if pred == ref: |
| cnt += 1 |
| detail['correct'] = True |
| details.append(detail) |
|
|
| score = cnt / len(predictions) * 100 |
|
|
| return {'score': score, 'details': details} |
|
|
