| import os |
| import regex |
| import zipfile |
| import requests |
| from math import isclose |
| from typing import Any, List, Callable |
| from sympy import N, simplify |
| from sympy.parsing.latex import parse_latex |
| from sympy.parsing.sympy_parser import parse_expr |
|
|
| from ..core.logging import logger |
| from .benchmark import Benchmark |
| from ..utils.utils import make_parent_folder |
| from ..core.module_utils import load_json |
| from ..utils.aflow_utils.data_utils import AFLOW_DATASET_FILES_MAP, download_aflow_benchmark_data |
|
|
|
|
| def download_raw_math_data(save_folder: str): |
| """ |
| Download the MATH data from the modelscope website. |
| """ |
| url = "https://www.modelscope.cn/datasets/opencompass/competition_math/resolve/master/data/MATH.zip" |
| logger.info(f"Downloading MATH data from {url} ...") |
| save_file_path = os.path.join(save_folder, "MATH.zip") |
|
|
| make_parent_folder(save_file_path) |
| if not os.path.exists(save_file_path): |
| response = requests.get(url, stream=True) |
| response.raise_for_status() |
| with open(save_file_path, "wb") as file: |
| for chunk in response.iter_content(chunk_size=1024): |
| if chunk: |
| file.write(chunk) |
| |
| with zipfile.ZipFile(save_file_path, "r") as zip_ref: |
| zip_ref.extractall(save_folder) |
| if os.path.exists(save_file_path): |
| os.remove(save_file_path) |
|
|
|
|
| class MATH(Benchmark): |
|
|
| """Benchmark class for evaluating mathematical reasoning on the MATH dataset. |
| |
| MATH is a dataset of challenging competition mathematics problems, |
| spanning various difficulty levels and subject areas. This class handles |
| loading the dataset, extracting answers, evaluating solutions through |
| symbolic and numerical comparisons, and computing accuracy metrics. |
| |
| The dataset includes problems across 7 subject areas (Algebra, Geometry, etc.) |
| and 5 difficulty levels. Each problem contains LaTeX-formatted |
| questions and solutions. |
| |
| Each MATH example has the following structure: |
| { |
| "id": "test-1", |
| "problem": "the problem", |
| "solution": "the solution", |
| "level": "Level 1", # "Level 1", "Level 2", "Level 3", "Level 4", "Level 5", "Level ?" |
| "type": "Algebra", # 'Geometry', 'Algebra', 'Intermediate Algebra', 'Counting & Probability', 'Precalculus', 'Number Theory', 'Prealgebra' |
| } |
| |
| The benchmark evaluates answers using symbolic math equality checking |
| and numerical approximation to handle equivalent mathematical expressions. |
| """ |
|
|
| def __init__(self, path: str = None, mode: str = "all", **kwargs): |
| path = os.path.expanduser(path or "~/.evoagentx/data/math") |
| super().__init__(name=type(self).__name__, path=path, mode=mode, **kwargs) |
| |
| def _load_data_from_folders(self, data_folder: str) -> List[dict]: |
| if data_folder is None: |
| return None |
| data = [] |
| typ = "train" if "train" in data_folder else "test" |
| sub_data_folders = os.listdir(data_folder) |
| i = 0 |
| logger.info(f"loading MATH data from {data_folder} ...") |
| for sub_data_folder in sub_data_folders: |
| if os.path.isdir(os.path.join(data_folder, sub_data_folder)): |
| files = os.listdir(os.path.join(data_folder, sub_data_folder)) |
| for file in files: |
| if file.endswith(".json"): |
| example = {"id": f"{typ}-{i+1}"} |
| example.update(load_json(os.path.join(data_folder, sub_data_folder, file), type="json")) |
| data.append(example) |
| i += 1 |
| return data |
| |
| def _load_data(self): |
| if not os.path.exists(os.path.join(self.path, "MATH")): |
| download_raw_math_data(save_folder=self.path) |
| data_folder = os.path.join(self.path, "MATH") |
|
|
| |
| if self.mode == "train" or self.mode == "all": |
| self._train_data = self._load_data_from_folders(data_folder=os.path.join(data_folder, "train")) |
| if self.mode == "dev" or self.mode == "all": |
| self._dev_data = None |
| if self.mode == "test" or self.mode == "all": |
| self._test_data = self._load_data_from_folders(data_folder=os.path.join(data_folder, "test")) |
| |
| def _get_label(self, example: Any) -> Any: |
| return example["solution"] |
| |
| def _get_id(self, example: Any) -> Any: |
| return example["id"] |
| |
| def extract_answer(self, text: str) -> str: |
|
|
| pattern = r"\\boxed{((?:[^{}]|{[^{}]*})*)}" |
| boxed_matches = regex.findall(pattern, text, regex.DOTALL) |
| if boxed_matches: |
| return boxed_matches[-1].strip() |
| |
| sentence_end_pattern = r"(?<!\d)[.!?]\s+" |
| sentences = regex.split(sentence_end_pattern, text) |
| sentences = [s.strip() for s in sentences if s.strip()] |
| return sentences[-1] if sentences else "" |
| |
| |
| def math_equal(self, prediction: Any, reference: Any) -> bool: |
| if str(prediction) == str(reference): |
| return True |
| |
| try: |
| if self.is_digit(prediction) and self.is_digit(reference): |
| prediction = self.parse_digits(prediction) |
| reference = self.parse_digits(reference) |
| return isclose(prediction, reference, abs_tol=1e-3) |
| except Exception: |
| pass |
|
|
| try: |
| return self.symbolic_equal(prediction, reference) |
| except Exception: |
| pass |
|
|
| return False |
| |
| def is_digit(self, num: Any) -> bool: |
| return self.parse_digits(num) is not None |
| |
| def parse_digits(self, num: Any) -> float: |
| num = regex.sub(",", "", str(num)) |
| try: |
| return float(num) |
| except Exception: |
| if num.endswith("%"): |
| num = num[:-1] |
| if num.endswith("\\"): |
| num = num[:-1] |
| try: |
| return float(num) / 100 |
| except Exception: |
| pass |
| return None |
|
|
| def symbolic_equal(self, a: Any, b: Any) -> bool: |
| def _parse(s: Any) -> Any: |
| for f in [parse_latex, parse_expr]: |
| try: |
| return f(s) |
| except Exception: |
| pass |
| return s |
|
|
| a = _parse(a) |
| b = _parse(b) |
|
|
| try: |
| if simplify(a - b) == 0: |
| return True |
| except Exception: |
| pass |
|
|
| try: |
| if isclose(N(a), N(b), abs_tol=1e-3): |
| return True |
| except Exception: |
| pass |
| return False |
|
|
| def evaluate(self, prediction: Any, label: Any) -> dict: |
| ground_truth_answer = self.extract_answer(label) |
| predicted_answer = self.extract_answer(prediction) |
| solve_rate = 1.0 if self.math_equal(predicted_answer, ground_truth_answer) else 0.0 |
| return {"solve_rate": solve_rate} |
| |
|
|
| class AFlowMATH(MATH): |
|
|
| def __init__(self, path: str = None, mode: str = "all", **kwargs): |
| path = os.path.expanduser(path or "~/.evoagentx/data/aflow/math") |
| super().__init__(path=path, mode=mode, **kwargs) |
|
|
| def _load_data_from_file(self, file_name: str): |
| if file_name is None: |
| return None |
| file_path = os.path.join(self.path, file_name) |
| if not os.path.exists(file_path): |
| download_aflow_benchmark_data(dataset="math", save_folder=self.path) |
| return load_json(path=file_path, type="jsonl") |
|
|
| def _load_data(self): |
|
|
| if self.mode == "train" or self.mode == "all": |
| logger.info(f"Loading train data from {AFLOW_DATASET_FILES_MAP['math']['train']}") |
| self._train_data = self._load_data_from_file(file_name=AFLOW_DATASET_FILES_MAP["math"]["train"]) |
| if self.mode == "dev" or self.mode == "all": |
| logger.info(f"Loading dev data from {AFLOW_DATASET_FILES_MAP['math']['dev']}") |
| self._dev_data = self._load_data_from_file(file_name=AFLOW_DATASET_FILES_MAP["math"]["dev"]) |
| if self.mode == "test" or self.mode == "all": |
| logger.info(f"Loading test data from {AFLOW_DATASET_FILES_MAP['math']['test']}") |
| self._test_data = self._load_data_from_file(file_name=AFLOW_DATASET_FILES_MAP["math"]["test"]) |
| |
| async def async_evaluate(self, graph: Callable, example: Any) -> float: |
|
|
| problem = example["problem"] |
| label = self._get_label(example) |
| output = await graph(problem) |
| metrics = await super().async_evaluate(prediction=output, label=label) |
| return metrics["solve_rate"] |
|
|
