File size: 8,890 Bytes
5374a2d |
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 |
"""
BIG-Bench Hard Benchmark Module
This module implements the BIGBenchHard benchmark evaluation framework.
BIGBenchHard is a challenging subset of 23 tasks from the BIG-bench evaluation suite,
designed to test reasoning capabilities of language models.
"""
import os
import random
import numpy as np
import torch
from typing import Any, List, Optional
from .benchmark import Benchmark
from .measures import exact_match_score
from ..core.logging import logger
from ..core.module_utils import load_json
from ..utils.utils import download_file
# Task categorization for different evaluation types
MULTIPLE_CHOICE_TASKS = [
'temporal_sequences', 'disambiguation_qa', 'date_understanding', 'tracking_shuffled_objects_three_objects', 'penguins_in_a_table',
'geometric_shapes', 'snarks', 'ruin_names', 'tracking_shuffled_objects_seven_objects', 'tracking_shuffled_objects_five_objects',
'logical_deduction_three_objects', 'hyperbaton', 'logical_deduction_five_objects', 'logical_deduction_seven_objects', 'movie_recommendation',
'salient_translation_error_detection', 'reasoning_about_colored_objects',
]
FREE_FORM_TASKS = [
'multistep_arithmetic_two', 'navigate', 'dyck_languages', 'word_sorting', 'sports_understanding',
'boolean_expressions', 'object_counting', 'formal_fallacies', 'causal_judgement', 'web_of_lies',
]
# Complete task mapping to data files
ALL_TASKS = {task: f"{task}.json" for task in MULTIPLE_CHOICE_TASKS + FREE_FORM_TASKS}
def download_raw_bigbenchhard_data(task_name: str, save_folder: str):
"""
Download raw BIGBenchHard data for a specific task.
Args:
task_name: The name of the task to download
save_folder: Directory to save the downloaded data file
Raises:
AssertionError: If task_name is not a valid BIGBenchHard task
"""
assert task_name in ALL_TASKS, f"'{task_name}' is an invalid bigbenchhard task name. Available tasks: {list(ALL_TASKS.keys())}"
file_name = ALL_TASKS[task_name]
url = f"https://raw.githubusercontent.com/suzgunmirac/BIG-Bench-Hard/main/bbh/{file_name}"
logger.info(f"Downloading BIGBenchHard '{task_name}' data from: {url}")
download_file(url=url, save_file=os.path.join(save_folder, file_name))
def set_seed(seed: int):
"""
Set random seeds for reproducibility across different libraries.
Args:
seed: The random seed value to use
"""
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
class BIGBenchHard(Benchmark):
"""
Benchmark class for BIGBenchHard dataset evaluation.
BIGBenchHard is a subset of 23 challenging tasks from the BIG-bench evaluation suite.
Each task example has the following structure:
{
"input": str, # The input question/problem
"target": str # The expected answer/output
}
The benchmark supports automatic data splitting for training/validation purposes
and evaluates predictions using exact match scoring.
"""
def __init__(self, task: str, path: str = None, mode: str = "all", dev_sample_num: int = 0, seed: int = 10, **kwargs):
"""
Initialize BIGBenchHard benchmark.
Args:
task: The specific BIGBenchHard task name
path: Path to store the dataset. Defaults to ~/.evoagentx/data/bigbenchhard/{task}
mode: Data loading mode. Defaults to "all"
dev_sample_num: Number of samples to use for dev set. If 0, all data goes to test set
seed: Random seed for reproducibility. Defaults to 10
**kwargs: Additional parameters for customization
Raises:
ValueError: If task is not a valid BIGBenchHard task name
"""
if task not in ALL_TASKS:
raise ValueError(f"Unknown task '{task}'. Available tasks: {list(ALL_TASKS.keys())}")
self.task = task
self.file_name = ALL_TASKS[task]
self.dev_sample_num = dev_sample_num
self.seed = seed
# Set default path if not provided
path = os.path.expanduser(path or f"~/.evoagentx/data/bigbenchhard/{task}")
super().__init__(name=f"BIGBenchHard-{self.task}", path=path, mode=mode, **kwargs)
def _load_data_from_file(self, file_name: str) -> Optional[List[dict]]:
"""
Load data from a specific file.
Args:
file_name: Name of the file to load
Returns:
List of loaded examples or None if file doesn't exist
"""
if file_name is None:
return None
file_path = os.path.join(self.path, file_name)
# Download data if not exists locally
if not os.path.exists(file_path):
download_raw_bigbenchhard_data(task_name=self.task, save_folder=self.path)
logger.info(f"Loading BIGBenchHard data from {file_path}...")
data = load_json(path=file_path, type="json")
return data.get("examples", [])
def _load_data(self):
"""
Load and split data according to mode and dev_sample_num settings.
Data splitting logic:
- If dev_sample_num > 0: randomly samples examples for dev set, rest go to test set
- If dev_sample_num = 0: all data goes to test set for evaluation
- No training data provided (BIGBenchHard is designed for few-shot evaluation)
"""
# Load the raw task data
task_data = self._load_data_from_file(file_name=self.file_name)
# Handle case where no data is loaded
if task_data is None:
logger.warning(f"No data loaded for task {self.task}")
self._train_data = []
self._dev_data = []
self._test_data = []
return
# BIGBenchHard doesn't provide training data - designed for few-shot evaluation
self._train_data = []
# Split data based on dev_sample_num parameter
if self.dev_sample_num > 0 and len(task_data) > self.dev_sample_num:
logger.info(f"Sampling {self.dev_sample_num} examples for dev set, rest for test set.")
if self.seed is not None:
set_seed(self.seed)
dev_subset = random.sample(task_data, self.dev_sample_num)
self._dev_data = dev_subset
self._test_data = [item for item in task_data if item not in dev_subset]
else:
# Handle edge cases
if self.dev_sample_num > 0:
logger.warning(f"dev_sample_num ({self.dev_sample_num}) >= total data size ({len(task_data)}). "
f"Using all data for dev set, none for test set.")
self._dev_data = task_data
self._test_data = []
else:
logger.info("dev_sample_num is 0, using all data for test set.")
self._dev_data = []
self._test_data = task_data
def get_input_keys(self) -> List[str]:
"""
Return the input keys expected by the benchmark.
Returns:
List containing "input" as the key for the problem text
"""
return ["input"]
def _get_label(self, example: Any) -> Any:
"""
Extract the ground truth label from an example.
Args:
example: The benchmark example
Returns:
The target answer/label
"""
return example["target"]
def _get_id(self, example: Any) -> Any:
"""
Extract the unique identifier from an example.
BIGBenchHard examples don't have explicit IDs, so we use input text as identifier.
Args:
example: The benchmark example
Returns:
The input text as a unique identifier
"""
return example.get("input", None)
def evaluate(self, prediction: Any, label: Any) -> dict:
"""
Score a prediction against the ground truth label.
Uses exact match scoring with task-specific handling for certain tasks.
Args:
prediction: The predicted answer
label: The ground truth answer
Returns:
Dictionary containing the exact match score
"""
if self.task == "dyck_languages":
# For Dyck languages, use special evaluation (ignore whitespace)
em = prediction.replace(' ', '') == label.replace(' ', '')
return {"em": em}
else:
# Standard exact match evaluation
em = exact_match_score(prediction=prediction, ground_truth=label)
return {"em": em} |