File size: 18,153 Bytes
256a159 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 |
# flake8: noqa
# yapf: disable
import functools
import getpass
import math
import os.path as osp
from datetime import datetime
from typing import Any, Dict, List, Optional
import mmengine
import tabulate
from mmengine import ConfigDict
from opencompass.utils import (LarkReporter, dataset_abbr_from_cfg,
get_infer_output_path, get_logger,
model_abbr_from_cfg)
from opencompass.utils.prompt import get_prompt_hash
METRIC_WHITELIST = ['score', 'auc_score', 'accuracy', 'humaneval_pass@1', 'rouge1', 'avg_toxicity_score', 'bleurt_diff', 'matthews_correlation', 'truth', 'f1', 'exact_match']
METRIC_BLACKLIST = ['bp', 'sys_len', 'ref_len']
def model_abbr_from_cfg_used_in_summarizer(model):
if model.get('summarizer_abbr', None):
return model['summarizer_abbr']
else:
return model_abbr_from_cfg(model)
class DefaultSummarizer:
"""Default summarizer in OpenCompass.
Args:
config (ConfigDict): The configuration object of the evaluation task. It's expected to be filled out at runtime.
dataset_abbrs (list[str], optional): Dataset abbreviations to be listed in the summary.
summary_groups (list): The dataset groups whose results need to be averaged out. For example, mmlu. Each item it a dict with
'name' (str) and 'subsets' (list of dataset abbrs), and optionally
'weights' if weighted average is needed.
prompt_db: A deprecated field.
"""
def __init__(self, config: ConfigDict, dataset_abbrs: Optional[List[str]] = None, summary_groups: List = [], prompt_db = None) -> None:
self.tasks = []
self.cfg = config
self.logger = get_logger()
self.summary_groups = summary_groups
self.dataset_abbrs = dataset_abbrs
if prompt_db:
self.logger.warning('prompt_db is deprecated and no longer used. '
'Please remove it from your config.')
# Enable lark bot if lark_url is presented
self.lark_reporter = None
if self.cfg.get('lark_bot_url', None):
self.lark_reporter = LarkReporter(self.cfg['lark_bot_url'])
self.model_cfgs = self.cfg['models']
self.dataset_cfgs = self.cfg['datasets']
self.work_dir = self.cfg['work_dir']
model_abbrs = []
for model in self.model_cfgs:
model_abbr = model_abbr_from_cfg_used_in_summarizer(model)
if model_abbr in model_abbrs:
continue
model_abbrs.append(model_abbr)
self.model_abbrs = model_abbrs
def _pick_up_results(self):
"""The function reads the numerical results of evaluations from the
output folder based on the configuration file, and ultimately returns
four dictionaries, each containing processed information in different
formats. The contents of the four dictionaries are as follows:
- raw_results: contains the raw results of each model on each dataset (excluding details).
- parsed_results: contains the results of each model on each dataset for each metric, with metrics in METRIC_BLACKLIST being ignored.
- dataset_metrics: contains the list of metrics for each dataset, consistent with the metrics in parsed_results. The list is ordered according to the METRIC_WHITELIST,
with metrics appearing earlier considered more important.
- dataset_eval_mode: contains the evaluation mode for each dataset.
"""
# raw_results: {model_abbr: {dataset_abbr: result}}
raw_results : Dict[str, Dict[str, Any]] = {}
# parsed_results: {model_abbr: {dataset_abbr: {metric: score}}}
parsed_results : Dict[str, Dict[str, Dict[str, float]]] = {}
# dataset_metrics: {dataset_abbr: [metric]}
dataset_metrics : Dict[str, List[str]] = {}
for model in self.model_cfgs:
model_abbr = model_abbr_from_cfg_used_in_summarizer(model)
parsed_results.setdefault(model_abbr, {})
raw_results.setdefault(model_abbr, {})
for dataset in self.dataset_cfgs:
dataset_abbr = dataset_abbr_from_cfg(dataset)
filepath = get_infer_output_path(model, dataset, osp.join(self.work_dir, 'results'))
if not osp.exists(filepath):
continue
result = mmengine.load(filepath)
result.pop('details', None)
raw_results[model_abbr][dataset_abbr] = result
if 'error' in result:
self.logger.debug(f'error in {model_abbr} {dataset_abbr} {result["error"]}')
continue
_rst, _dm = {}, []
for metric, score in result.items():
if metric not in METRIC_BLACKLIST and isinstance(score, (int, float)):
_rst[metric] = score
_dm.append(metric)
else:
continue
if len(_rst) == 0:
self.logger.warning(f'unknown result format: {result}, continue')
continue
_dm = sorted(_dm, key=lambda i: METRIC_WHITELIST.index(i) if i in METRIC_WHITELIST else len(METRIC_WHITELIST))
if dataset_abbr in dataset_metrics:
assert tuple(dataset_metrics[dataset_abbr]) == tuple(_dm), \
f'{dataset_abbr} has different metrics: {dataset_metrics[dataset_abbr]} vs {_dm}'
else:
dataset_metrics[dataset_abbr] = _dm
parsed_results[model_abbr][dataset_abbr] = _rst
# dataset_eval_mode: {dataset_abbr: eval_mode}
dataset_eval_mode : Dict[str, str] = {}
for dataset in self.dataset_cfgs:
inferencer = dataset.get('infer_cfg', {}).get('inferencer', {}).get('type', '')
inferencer = inferencer if isinstance(inferencer, str) else inferencer.__name__
dataset_abbr = dataset_abbr_from_cfg(dataset)
if 'GenInferencer' in inferencer:
dataset_eval_mode[dataset_abbr] = 'gen'
elif 'PPLInferencer' in inferencer:
dataset_eval_mode[dataset_abbr] = 'ppl'
elif 'LLInferencer' in inferencer:
dataset_eval_mode[dataset_abbr] = 'll'
else:
dataset_eval_mode[dataset_abbr] = 'unknown'
self.logger.warning(f'unknown inferencer: {inferencer} - {dataset_abbr}')
return raw_results, parsed_results, dataset_metrics, dataset_eval_mode
def _calculate_group_metrics(self, raw_results, parsed_results, dataset_metrics, dataset_eval_mode):
"""The function calculates the numerical results for each group based
on the configuration in summary_groups, and updates the contents of
each dictionary accordingly."""
summary_groups = self.summary_groups
for sg in summary_groups:
for model_abbr in self.model_abbrs:
available_metrics, missing_metrics = [], []
for i in sg['subsets']:
if isinstance(i, (list, tuple)):
if i[0] in parsed_results[model_abbr] and i[1] in parsed_results[model_abbr][i[0]]:
available_metrics.append(i)
else:
missing_metrics.append(i)
else:
if i in parsed_results[model_abbr]:
available_metrics.append(i)
else:
missing_metrics.append(i)
if len(available_metrics) == 0:
continue
if len(missing_metrics) != 0:
raw_results[model_abbr][sg['name']] = {'error': 'missing metrics: {}'.format(missing_metrics)}
continue
if 'metric' in sg:
default_metric = sg['metric']
need_smart_metric = False
else:
need_smart_metric = True
if sg.get('std', False):
default_metric = 'standard_deviation'
elif sg.get('weights', []):
default_metric = 'weighted_average'
else:
default_metric = 'naive_average'
scores, eval_modes, group_metrics = {}, [], None
if any(isinstance(dataset_abbr, (list, tuple)) for dataset_abbr in sg['subsets']) and \
any(isinstance(dataset_abbr, str) for dataset_abbr in sg['subsets']):
raise NotImplementedError('mixed dataset_abbr type is not supported')
if all(isinstance(dataset_abbr, (list, tuple)) for dataset_abbr in sg['subsets']):
group_metrics = [default_metric]
for dataset_abbr, metric in sg['subsets']:
scores.setdefault(default_metric, {})[dataset_abbr + '@' + metric] = parsed_results[model_abbr][dataset_abbr][metric]
eval_modes.append(dataset_eval_mode.get(dataset_abbr, 'unknown'))
else:
group_metrics = list(functools.reduce(lambda a, b: a & b, [set(dataset_metrics[dataset_abbr]) for dataset_abbr in sg['subsets']]))
if need_smart_metric and len(group_metrics) > 1:
for metric in group_metrics:
for dataset_abbr in sg['subsets']:
scores.setdefault(metric, {})[dataset_abbr + '@' + metric] = parsed_results[model_abbr][dataset_abbr][metric]
eval_modes.append(dataset_eval_mode.get(sg['subsets'][0], 'unknown'))
else:
group_metrics = [default_metric]
for dataset_abbr in sg['subsets']:
metric = dataset_metrics[dataset_abbr][0]
scores.setdefault(default_metric, {})[dataset_abbr + '@' + metric] = parsed_results[model_abbr][dataset_abbr][metric]
eval_modes.append(dataset_eval_mode.get(dataset_abbr, 'unknown'))
result = {}
for metric in scores:
if default_metric == 'standard_deviation':
avg = sum(scores[metric].values()) / len(scores[metric])
variance = sum((scores[metric][k] - avg) ** 2 for k in scores[metric]) / len(scores[metric])
scores[metric] = result[metric] = math.sqrt(variance)
else:
if sg.get('weights', []):
# check sg['weights'][k] != 0 in case of scores[metric][k] is NaN
try:
numerator = sum(scores[metric][k] * sg['weights'][k] for k in sg['weights'] if sg['weights'][k] != 0)
except KeyError:
tmp_scores = {metric: {k.split('@')[0]: v for k, v in scores[metric].items()} for metric in scores}
numerator = sum(tmp_scores[metric][k] * sg['weights'][k] for k in sg['weights'] if sg['weights'][k] != 0)
denominator = sum(sg['weights'].values())
else:
numerator = sum(scores[metric].values())
denominator = len(scores[metric])
scores[metric] = result[metric] = numerator / denominator
eval_modes = list(set(eval_modes))
eval_mode = eval_modes[0] if len(eval_modes) == 1 else 'mixed'
# add to global results
raw_results[model_abbr].setdefault(sg['name'], {}).update(scores)
parsed_results[model_abbr].setdefault(sg['name'], {}).update(result)
dataset_metrics.setdefault(sg['name'], []).extend(group_metrics)
dataset_eval_mode[sg['name']] = eval_mode
return raw_results, parsed_results, dataset_metrics, dataset_eval_mode
def _format_table(self, parsed_results, dataset_metrics, dataset_eval_mode):
dataset_abbrs = [dataset_abbr_from_cfg(dataset) for dataset in self.dataset_cfgs]
prompt_version = {dataset_abbr_from_cfg(d): get_prompt_hash(d)[:6] for d in self.dataset_cfgs}
summarizer_dataset_abbrs = []
if self.dataset_abbrs is None:
# display all dataset metrics included in the config
for dataset_abbr in dataset_abbrs:
if dataset_abbr in dataset_metrics:
for metric in dataset_metrics[dataset_abbr]:
summarizer_dataset_abbrs.append((dataset_abbr, metric))
else:
summarizer_dataset_abbrs.append((dataset_abbr, None))
# along with all possible group metrics
for dataset_abbr in dataset_metrics:
for metric in dataset_metrics[dataset_abbr]:
if (dataset_abbr, metric) not in summarizer_dataset_abbrs:
summarizer_dataset_abbrs.append((dataset_abbr, metric))
else:
# follow the required order
for item in self.dataset_abbrs:
if isinstance(item, str):
summarizer_dataset_abbrs.append((item, None))
elif isinstance(item, (list, tuple)):
summarizer_dataset_abbrs.append((item[0], item[1]))
table = []
header = ['dataset', 'version', 'metric', 'mode'] + self.model_abbrs
table.append(header)
for dataset_abbr, metric in summarizer_dataset_abbrs:
if dataset_abbr not in dataset_metrics:
table.append([dataset_abbr, '-', '-', '-'] + ['-'] * len(self.model_abbrs))
continue
if metric is None:
metric = dataset_metrics[dataset_abbr][0]
elif metric in dataset_metrics[dataset_abbr]:
pass
else:
table.append([dataset_abbr, '-', '-', '-'] + ['-'] * len(self.model_abbrs))
continue
row = [dataset_abbr, prompt_version.get(dataset_abbr, '-'), metric, dataset_eval_mode.get(dataset_abbr, '-')]
for model_abbr in self.model_abbrs:
if dataset_abbr in parsed_results[model_abbr]:
row.append('{:.02f}'.format(parsed_results[model_abbr][dataset_abbr][metric]))
else:
row.append('-')
table.append(row)
return table
def _format_raw_txt(self, raw_results):
raw_dataset_abbrs = []
for model_abbr in self.model_abbrs:
for dataset_abbr in raw_results[model_abbr]:
if dataset_abbr not in raw_dataset_abbrs:
raw_dataset_abbrs.append(dataset_abbr)
raw_txts = []
for model_abbr in self.model_abbrs:
raw_txts.append('-------------------------------')
raw_txts.append(f'Model: {model_abbr}')
for dataset_abbr in raw_dataset_abbrs:
result = raw_results[model_abbr].get(dataset_abbr, '{}')
raw_txts.append(f'{dataset_abbr}: {result}')
raw_txts = '\n'.join(raw_txts)
return raw_txts
def _output_to_file(self, output_path, time_str, table, raw_txts):
# output to file
if output_path is None:
output_path = osp.join(self.work_dir, 'summary', f'summary_{time_str}.txt')
output_csv_path = osp.join(self.work_dir, 'summary', f'summary_{time_str}.csv')
else:
output_csv_path = output_path.replace('.txt', '.csv')
output_dir = osp.split(output_path)[0]
mmengine.mkdir_or_exist(output_dir)
with open(output_path, 'w', encoding='utf-8') as f:
text = f'{time_str}\n' + \
'tabulate format\n' + \
'^' * 128 + '\n' + \
tabulate.tabulate(table, headers='firstrow') + '\n' + \
'$' * 128 + '\n\n' + \
'-' * 128 + ' THIS IS A DIVIDER ' + '-' * 128 + '\n\n' + \
'csv format\n' + \
'^' * 128 + '\n' + \
'\n'.join([','.join(row) for row in table]) + '\n' + \
'$' * 128 + '\n\n' + \
'-' * 128 + ' THIS IS A DIVIDER ' + '-' * 128 + '\n\n' + \
'raw format\n' + \
'^' * 128 + '\n' + \
raw_txts + '\n' + \
'$' * 128 + '\n'
f.write(text)
self.logger.info(f'write summary to {osp.abspath(output_path)}')
with open(output_csv_path, 'w', encoding='utf-8') as f:
f.write('\n'.join([','.join(row) for row in table]) + '\n')
self.logger.info(f'write csv to {osp.abspath(output_csv_path)}')
def summarize(
self,
output_path: str = None,
time_str: str = datetime.now().strftime('%Y%m%d_%H%M%S')): # noqa
# pick up results
raw_results, parsed_results, dataset_metrics, dataset_eval_mode = self._pick_up_results()
# calculate group metrics
raw_results, parsed_results, dataset_metrics, dataset_eval_mode = \
self._calculate_group_metrics(raw_results, parsed_results, dataset_metrics, dataset_eval_mode)
# format table
table = self._format_table(parsed_results, dataset_metrics, dataset_eval_mode)
# format raw txt
raw_txts = self._format_raw_txt(raw_results)
# output to screen
print(tabulate.tabulate(table, headers='firstrow'))
# output to .text / .csv files
self._output_to_file(output_path, time_str, table, raw_txts)
if self.lark_reporter:
content = f'{getpass.getuser()} 的'
content += f'详细评测汇总已输出至 {osp.abspath(output_path)}'
self.lark_reporter.post(content)
|