| import asyncio |
| import random |
| import numpy as np |
| from abc import ABC, abstractmethod |
| from typing import Tuple, Optional, List, Dict, Any |
|
|
| from ..core.logging import logger |
| from ..core.callbacks import timeout, TimeoutException |
| from ..utils.sanitize import sanitize |
| from .lcb_utils.evaluation import estimate_pass_at_k |
|
|
|
|
| class Benchmark(ABC): |
|
|
| """ |
| Abstract base class for defining benchmarks. This class provides methods to load, |
| retrieve, and evaluate benchmark data, with train, dev, and test splits. |
| """ |
|
|
| def __init__(self, name: str, path: str, mode: str = "all", **kwargs): |
| """ |
| Initializes the benchmark with a name and data path. |
| |
| Args: |
| name (str): The name of the benchmark. |
| path (str): The path to the dataset. |
| mode (str): which type of data to load, choices: ["all", "train", "dev", "test"] |
| **kwargs: Additional parameters for customization. |
| """ |
| valid_mode = ["all", "train", "dev", "test"] |
| assert mode in valid_mode, f"Invalid value for model: {mode}. Available choices: {valid_mode}" |
|
|
| self.name = name |
| self.path = path |
| self.mode = mode |
| self.kwargs = kwargs |
|
|
| |
| self._train_data: Optional[List[dict]] = None |
| self._dev_data: Optional[List[dict]] = None |
| self._test_data: Optional[List[dict]] = None |
|
|
| |
| self._load_data() |
| |
| @abstractmethod |
| def _load_data(self): |
| """ |
| Abstract method to load data from `self.path` and assign it to `_train_data`, `_dev_data`, and `_test_data` if applicable. |
| """ |
| pass |
|
|
| @abstractmethod |
| def _get_id(self, example: Any) -> Any: |
| """ |
| Abstract method to return the id for a given example. |
| """ |
| pass |
|
|
| @abstractmethod |
| def _get_label(self, example: Any) -> Any: |
| """ |
| Abstract method to return the ground-truth label for a given example. |
| |
| Args: |
| example (Any): The input example for which the label is needed. |
| |
| Returns: |
| Any: The ground-truth label associated with the example. |
| """ |
| pass |
|
|
| @abstractmethod |
| def evaluate(self, prediction: Any, label: Any) -> dict: |
| """ |
| Abstract method to evaluate a single prediction against the ground-truth label. |
| |
| Args: |
| prediction (Any): The predicted output. |
| label (Any): The actual ground-truth label. |
| |
| Returns: |
| dict: A dictionary containing evaluation metrics. |
| """ |
| pass |
|
|
| async def async_evaluate(self, prediction: Any, label: Any) -> dict: |
| """ |
| Asynchronous version of evaluate method that internally calls the synchronous evaluate. |
| |
| Args: |
| prediction (Any): The predicted output. |
| label (Any): The actual ground-truth label. |
| |
| Returns: |
| dict: A dictionary containing evaluation metrics. |
| """ |
| return await asyncio.to_thread(self.evaluate, prediction, label) |
| |
| def get_label(self, example: List[Any]) -> Any: |
| return self._get_label(example=example) |
| |
| def get_labels(self, examples: List[Any]) -> List[Any]: |
| return [self._get_label(example=example) for example in examples] |
| |
| def get_id(self, example: List[Any]) -> Any: |
| return self._get_id(example=example) |
| |
| def get_ids(self, examples: List[Any]) -> List[Any]: |
| return [self._get_id(example=example) for example in examples] |
| |
| def get_data_by_mode(self, mode: str = "test") -> List[Any]: |
| """ |
| Get the data from the benchmark by mode. |
| """ |
| assert mode in ["train", "dev", "test"], f"Invalid value for mode: {mode}. Available choices: ['train', 'dev', 'test']" |
| if mode == "train": |
| if self._train_data is None: |
| logger.warning(f"Train data for benchmark {type(self).__name__} is not loaded or None. Return an empty list.") |
| return [] |
| data = self._train_data |
| elif mode == "dev": |
| if self._dev_data is None: |
| logger.warning(f"Dev data for benchmark {type(self).__name__} is not loaded or None. Return an empty list.") |
| return [] |
| data = self._dev_data |
| else: |
| if self._test_data is None: |
| logger.warning(f"Test data for benchmark {type(self).__name__} is not loaded or None. Return an empty list.") |
| return [] |
| data = self._test_data |
| return data |
| |
| def get_example_by_id(self, example_id: Any, mode: str = None) -> Optional[Any]: |
| """ |
| Get an example from the benchmark by its id. |
| |
| Args: |
| example_id (Any): The id of the example to retrieve. |
| mode (str): The mode to retrieve the example from, choices: ["train", "dev", "test", "all"] |
| |
| Returns: |
| Optional[Any]: The example if found, otherwise None. |
| """ |
| |
| if mode is not None and mode not in ["train", "dev", "test", "all"]: |
| raise ValueError(f"Invalid value for mode: {mode}. Available choices: ['train', 'dev', 'test', 'all']") |
| |
| if mode is None or mode == "all": |
| data = [] |
| if self._train_data is not None: |
| data.extend(self._train_data) |
| if self._dev_data is not None: |
| data.extend(self._dev_data) |
| if self._test_data is not None: |
| data.extend(self._test_data) |
| else: |
| data = self.get_data_by_mode(mode=mode) |
| for example in data: |
| if self._get_id(example=example) == example_id: |
| return example |
| return None |
| |
| def get_example_by_index(self, index: int, mode: str = "test") -> Optional[Any]: |
| """ |
| Get an example from the benchmark by its index. |
| |
| Args: |
| index (int): The index of the example to retrieve. |
| mode (str): The mode to retrieve the example from, choices: ["train", "dev", "test"] |
| |
| Returns: |
| Optional[Any]: The example if found, otherwise None. |
| """ |
| data = self.get_data_by_mode(mode=mode) |
| return data[index] if index < len(data) else None |
| |
| def _get_data(self, data: List[dict], indices: Optional[List[int]]=None, sample_k: Optional[int]=None, seed: Optional[int]=None) -> List[dict]: |
| """ |
| Retrieves a subset of data based on provided indices or a random sample. |
| |
| Args: |
| data (List[dict]): The list of data examples. |
| indices (List[int], optional): Specific indices of data to retrieve. Defaults to None. |
| sample_k (int, optional): The number of random samples to retrieve. Defaults to None. |
| seed (int, optional): The seed for random sampling. Defaults to None. If provided, the random sampling will be deterministic. |
| Returns: |
| List[dict]: The selected subset of data. If both `indices` and `sample_k` are None, it will return the original `data`. |
| """ |
| if indices is None: |
| indices = list(range(len(data))) |
| if sample_k is not None: |
| if seed is not None: |
| random.seed(seed) |
| indices = random.sample(indices, k=min(sample_k, len(indices))) |
| return_data = [data[idx] for idx in indices] |
| return return_data |
|
|
| def get_train_data(self, indices: Optional[List[int]] = None, sample_k: Optional[int] = None, seed: Optional[int] = None) -> List[dict]: |
| |
| if self._train_data is None: |
| logger.warning(f"Train data for benchmark {type(self).__name__} is not loaded or None. Return an empty list.") |
| return [] |
| |
| train_data = self._get_data(self._train_data, indices=indices, sample_k=sample_k, seed=seed) |
| return train_data |
| |
| def get_dev_data(self, indices: Optional[List[int]] = None, sample_k: Optional[int] = None, seed: Optional[int] = None) -> List[dict]: |
| |
| if self._dev_data is None: |
| logger.warning(f"Dev data for benchmark {type(self).__name__} is not loaded or None. Return an empty list.") |
| return [] |
| |
| dev_data = self._get_data(self._dev_data, indices=indices, sample_k=sample_k, seed=seed) |
| return dev_data |
|
|
| def get_test_data(self, indices: Optional[List[int]] = None, sample_k: Optional[int] = None, seed: Optional[int] = None) -> List[dict]: |
| |
| if self._test_data is None: |
| logger.warning(f"Test data for benchmark {type(self).__name__} is not loaded or None. Return an empty list.") |
| return [] |
| |
| test_data = self._get_data(self._test_data, indices=indices, sample_k=sample_k, seed=seed) |
| return test_data |
|
|
|
|
| class CodingBenchmark(Benchmark): |
|
|
| """ |
| Abstract base class for defining coding benchmarks. This class provides methods to check the solution code. |
| """ |
| |
| def __init__(self, name: str, path: str, mode: str = "all", timeout: int = 60, **kwargs): |
| |
| self.SUCCESS = 0 |
| self.FAILED = 1 |
| self.TIMEOUT = 2 |
| self.timeout = timeout |
| super().__init__(name=name, path=path, mode=mode, **kwargs) |
|
|
| def handle_special_cases(self, task_id: str, solution: str, test: str) -> bool: |
| return solution, test |
| |
| def _check_evaluation_inputs(self, prediction: Any, label: Any) -> bool: |
| """ |
| Check if the inputs are valid for evaluation. |
| """ |
| assert isinstance(prediction, str) or isinstance(prediction, list), "prediction must be a string or a list of strings, but got {}".format(type(prediction)) |
| assert isinstance(label, dict) or isinstance(label, list), "label must be a string or a list of strings, but got {}".format(type(label)) |
| prediction = [prediction] if isinstance(prediction, str) else prediction |
| label = [label] if isinstance(label, dict) else label |
| return prediction, label |
|
|
| def check_solution(self, task_id: str, solution: str, test: str, entry_point: Optional[str] = None, use_entrypoint_as_input: bool = True) -> Tuple[int, str]: |
| """ |
| Execute the solution code and check if it passes the unit test. |
| |
| Args: |
| task_id (str): The task id. |
| solution (str): The solution code. |
| test (str): The unit test code in HumanEval format. |
| entry_point (str): The entry point of the solution code. |
| Returns: |
| Tuple[int, str]: A tuple containing an integer indicating whether the solution passes the unit test (0: success, 1: failed, 2: timeout) and a string containing the success/error message. |
| """ |
| solution = sanitize(solution, entrypoint=entry_point) |
|
|
| try: |
| global_dict = { |
| "math": __import__("math"), |
| "hashlib": __import__("hashlib"), |
| "re": __import__("re"), |
| "List": List, |
| "Dict": Dict, |
| "Tuple": Tuple, |
| "Optional": Optional, |
| "Any": Any, |
| } |
| solution, test = self.handle_special_cases(task_id=task_id, solution=solution, test=test) |
| exec(solution, global_dict) |
| if entry_point not in global_dict: |
| raise ValueError(f"Function {entry_point} not found in the solution code.") |
| exec(test, global_dict) |
| |
| unit_test_func = global_dict["check"] |
| |
| with timeout(seconds=self.timeout): |
| if use_entrypoint_as_input: |
| unit_test_func(global_dict[entry_point]) |
| else: |
| unit_test_func() |
| result = (self.SUCCESS, "The solution passed the unit test.") |
| |
| except TimeoutException: |
| result = (self.TIMEOUT, "Execution timed out.") |
| |
| except Exception as e: |
| error_msg = f"An error occurred: {e}\nSolution:\n{solution}\nTest:\n{test}" |
| result = (self.FAILED, error_msg) |
| |
| return result |
| |
| def check_solution_scicode(self, task_id: str, solution: str, test: str, entry_point: Optional[str] = None, use_entrypoint_as_input: bool = True) -> Tuple[int, str]: |
| """ |
| Execute the solution code and check if it passes the unit test. |
| |
| Args: |
| task_id (str): The task id. |
| solution (str): The solution code. |
| test (str): The unit test code in HumanEval format. |
| entry_point (str): The entry point of the solution code. |
| Returns: |
| Tuple[int, str]: A tuple containing an integer indicating whether the solution passes the unit test (0: success, 1: failed, 2: timeout) and a string containing the success/error message. |
| """ |
| solution = sanitize(solution, entrypoint=entry_point) |
|
|
| try: |
| global_dict = { |
| "math": __import__("math"), |
| "hashlib": __import__("hashlib"), |
| "re": __import__("re"), |
| "List": List, |
| "Dict": Dict, |
| "Tuple": Tuple, |
| "Optional": Optional, |
| "Any": Any, |
| } |
| solution, test = self.handle_special_cases(task_id=task_id, solution=solution, test=test) |
| exec(solution, global_dict) |
| if entry_point not in global_dict: |
| raise ValueError(f"Function {entry_point} not found in the solution code.") |
| with timeout(seconds=self.timeout): |
| if use_entrypoint_as_input: |
| exec(test, global_dict) |
| else: |
| result = (self.FAILED, error_msg) |
| |
| result = (self.SUCCESS, "The solution passed the unit test.") |
| |
| except TimeoutException: |
| result = (self.TIMEOUT, "Execution timed out.") |
| |
| except Exception as e: |
| error_msg = f"An error occurred: {e}\nSolution:\n{solution}\nTest:\n{test}" |
| result = (self.FAILED, error_msg) |
| |
| return result |
| |
| def check_solution_plus(self, task_id: str, solution: str, test: str, entry_point: Optional[str] = None, use_entrypoint_as_input: bool = True) -> Tuple[int, str]: |
| """ |
| Execute the solution code and check if it passes the unit test. |
| |
| Args: |
| task_id (str): The task id. |
| solution (str): The solution code. |
| test (str): The unit test code in HumanEval format. |
| entry_point (str): The entry point of the solution code. |
| Returns: |
| Tuple[int, str]: A tuple containing an integer indicating whether the solution passes the unit test (0: success, 1: failed, 2: timeout) and a string containing the success/error message. |
| """ |
| solution = sanitize(solution, entrypoint=entry_point) |
| result = (self.FAILED, "The code running is wrong") |
| try: |
| global_dict = { |
| "math": __import__("math"), |
| "hashlib": __import__("hashlib"), |
| "re": __import__("re"), |
| "List": List, |
| "Dict": Dict, |
| "Tuple": Tuple, |
| "Optional": Optional, |
| "Any": Any, |
| } |
| solution, test = self.handle_special_cases(task_id=task_id, solution=solution, test=test) |
| exec(solution, global_dict) |
| if entry_point not in global_dict: |
| raise ValueError(f"Function {entry_point} not found in the solution code.") |
| with timeout(seconds=self.timeout): |
| exec(test, global_dict) |
| result = (self.SUCCESS, "The solution passed the unit test.") |
| |
| except TimeoutException: |
| result = (self.TIMEOUT, "Execution timed out.") |
| |
| except Exception as e: |
| error_msg = f"An error occurred: {e}\nSolution:\n{solution}\nTest:\n{test}" |
| result = (self.FAILED, error_msg) |
| |
| return result |
| |
| def compute_pass_at_k(self, results: List[bool], k_list: List[int]) -> Dict[str, float]: |
| """ |
| Compute the pass@k for the given results. |
| """ |
| pass_at_k = {} |
| n = len(results) |
| c = sum(results) |
| for k in k_list: |
| if n >= k: |
| pass_at_k[f"pass@{k}"] = float(estimate_pass_at_k(np.array([n]), np.array([c]), k)[0]) |
| |
| return pass_at_k |
