| | from collections import defaultdict |
| | from typing import Dict, List |
| |
|
| | import mmengine |
| | from mmengine import ConfigDict, track_parallel_progress |
| |
|
| | from opencompass.registry import EVALUATORS, MODELS |
| | from opencompass.utils import dataset_abbr_from_cfg, model_abbr_from_cfg |
| |
|
| |
|
| | @EVALUATORS.register_module() |
| | class ModelEvaluator: |
| | """TODO: Finish the implementation""" |
| |
|
| | def __init__( |
| | self, |
| | config: ConfigDict, |
| | ) -> None: |
| | self.tasks = [] |
| | self.cfg = config |
| | self.parse_cfg(self.cfg.pop('evaluator', ConfigDict({}))) |
| | self.dataset_abbrs = [ |
| | dataset_abbr_from_cfg(d) for d in self.cfg['datasets'] |
| | ] |
| | self.model_abbrs = [model_abbr_from_cfg(m) for m in self.cfg['models']] |
| | assert len(self.model_abbrs) > 1 |
| |
|
| | def parse_cfg(self, cfg: ConfigDict): |
| | |
| | self.judger = MODELS.build(cfg['judger']) |
| | |
| | self.max_num_workers = cfg.get('max_num_workers', 4) |
| |
|
| | def evaluate(self): |
| | model_scores = defaultdict(int) |
| | all_partial_scores = track_parallel_progress( |
| | self._evaluate_dataset, |
| | self.dataset_abbrs, |
| | nproc=self.max_num_workers, |
| | keep_order=True) |
| | for partial_scores in all_partial_scores: |
| | for model_idx, score in partial_scores.items(): |
| | model_scores[self.model_abbrs[model_idx]] += score |
| | print(model_scores) |
| |
|
| | def _load_dataset(self, dataset_abbr: str): |
| | |
| | original_datasets = [] |
| | self.responses: List[List[str]] = [] |
| | self.questions: List[str] = [] |
| | for model_abbr in self.model_abbrs: |
| | filename = f'output_model/{model_abbr}/{dataset_abbr}.json' |
| | original_datasets.append(mmengine.load(filename)) |
| | for key in original_datasets[-1].keys(): |
| | self.questions.append(original_datasets[-1][key]['origin_prompt']) |
| | responses = [] |
| | for i in range(len(self.model_abbrs)): |
| | responses.append(original_datasets[i][key]['prediction']) |
| | self.responses.append(responses) |
| |
|
| | def _evaluate_dataset(self, dataset_abbr: str): |
| | self._load_dataset(dataset_abbr=dataset_abbr) |
| | model_scores = defaultdict(int) |
| | for question, responses in zip(self.questions, self.responses): |
| | prompt = self._make_prompt(question, responses) |
| | print(prompt) |
| | output = self.judger.generate(prompt, |
| | max_out_len=2 * |
| | len(self.model_abbrs)) |
| | model_scores = self._rank_models(output, model_scores) |
| | return model_scores |
| |
|
| | def _make_prompt(self, question: str, responses: List[str]) -> str: |
| | prompt = ('Below are a question and a set of answers, each numbered by' |
| | ' a digit. Please sort the answers from least to most ' |
| | 'appropriate to the question. Only return the digit ' |
| | 'seperated by a blank space. For example, when there are ' |
| | 'three answers presented, you should say "1 0 2" when the ' |
| | 'second answer is the best and the third is the worst.\n' |
| | f'Q: {question}\n') |
| | for i, response in enumerate(responses): |
| | prompt += f'A{i + 1}: {response}\n' |
| | return prompt |
| |
|
| | def _rank_models(self, output: str, |
| | model_scores: defaultdict) -> Dict[str, int]: |
| | """Returns model ranking.""" |
| | output = output.strip().split(' ') |
| | for score, model_idx in enumerate(output): |
| | model_scores[model_idx] += int(score) |
| | return model_scores |
| |
|
