| import ast |
| import json |
| import math |
| import os |
| import re |
|
|
| from datasets import Dataset |
|
|
| from opencompass.openicl.icl_evaluator.icl_base_evaluator import BaseEvaluator |
| from opencompass.registry import ICL_EVALUATORS, LOAD_DATASET |
| from opencompass.utils import get_data_path |
|
|
| from .base import BaseDataset |
|
|
|
|
| @LOAD_DATASET.register_module() |
| class OpenSWIDataset(BaseDataset): |
|
|
| @staticmethod |
| def load(path: str, name: str): |
| new_data = [] |
| path = os.path.join(get_data_path(path), name) |
| for file in os.listdir(path): |
| if file.endswith('.jsonl'): |
| final_path = os.path.join(path, file) |
| with open(final_path, 'r', encoding='utf-8') as f: |
| for line in f: |
| data = json.loads(line) |
| new_data.append({ |
| 'id': data['id_ddm'], |
| 'prompt': data['dialogs'][0]['content'], |
| 'ground_truth': data['ground_truth'], |
| 'subset': file.split('.')[0] |
| }) |
| dataset = Dataset.from_list(new_data) |
| return dataset |
|
|
|
|
| def extract_list(text): |
| |
| matches = re.findall(r'(\[.*?\])', text, re.DOTALL) |
| if matches: |
| raw_list_str = matches[-1] |
| |
| try: |
| question_list = ast.literal_eval(raw_list_str) |
| return question_list |
| except Exception: |
| return None |
| else: |
| return None |
|
|
|
|
| @ICL_EVALUATORS.register_module() |
| class OpenSWIMSEEvaluator(BaseEvaluator): |
| """Exact match evaluator for name conversion.""" |
|
|
| def __init__(self) -> None: |
| super().__init__() |
|
|
| def score(self, predictions, references): |
| if len(predictions) != len(references): |
| return { |
| 'error': 'predictions and references have different ' |
| 'length' |
| } |
|
|
| avg_score = 0 |
| avg_valid = [] |
| details = [] |
| for prediction, reference in zip(predictions, references): |
| pred = extract_list(prediction) |
| ans = reference |
|
|
| if not pred or all(isinstance(x, float) for x in pred) is False: |
| detail = {'pred': None, 'answer': ans, 'valid': False} |
| pred = [0] * len(ans) |
| else: |
| detail = {'pred': pred, 'answer': ans, 'valid': True} |
| if len(pred) < len(ans): |
| detail['valid'] = False |
| pred = pred + [0] * (len(ans) - len(pred)) |
| elif len(pred) > len(ans): |
| detail['valid'] = False |
| pred = pred[:len(ans)] |
| avg_valid.append(detail['valid']) |
| squared_errors = [(a - p)**2 for a, p in zip(ans, pred)] |
| rmse_score = math.sqrt(sum(squared_errors) / len(squared_errors)) |
| detail['score'] = rmse_score |
| avg_score += rmse_score |
| details.append(detail) |
|
|
| score = avg_score / len(predictions) |
| valid = sum(avg_valid) / len(avg_valid) |
|
|
| return {'score': score, 'valid': valid * 100, 'details': details} |
|
|
