| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import re |
| import signal |
| import sys |
| from collections import Counter |
| from contextlib import contextmanager |
| from dataclasses import dataclass |
| from functools import lru_cache |
| from pathlib import Path |
| from typing import Any, Iterable, Optional |
|
|
| try: |
| from tqdm import tqdm |
| except Exception: |
| tqdm = None |
|
|
| try: |
| from datasets import load_dataset |
| except Exception as exc: |
| load_dataset = None |
| DATASETS_IMPORT_ERROR = exc |
| else: |
| DATASETS_IMPORT_ERROR = None |
|
|
| try: |
| from math_verify import parse, verify |
| from math_verify.parser import ExprExtractionConfig, LatexExtractionConfig |
| except Exception as exc: |
| parse = None |
| verify = None |
| ExprExtractionConfig = None |
| LatexExtractionConfig = None |
| MATH_VERIFY_IMPORT_ERROR = exc |
| else: |
| MATH_VERIFY_IMPORT_ERROR = None |
|
|
| try: |
| import sympy |
| from sympy.parsing.latex import parse_latex |
| except Exception as exc: |
| sympy = None |
| parse_latex = None |
| SYMPY_IMPORT_ERROR = exc |
| else: |
| SYMPY_IMPORT_ERROR = None |
|
|
|
|
| SUBSTITUTIONS = [ |
| ("an ", ""), |
| ("a ", ""), |
| (".$", "$"), |
| ("\\$", ""), |
| (r"\ ", ""), |
| (" ", ""), |
| ("mbox", "text"), |
| (",\\text{and}", ","), |
| ("\\text{and}", ","), |
| ("\\text{m}", "\\text{}"), |
| ] |
|
|
| REMOVED_EXPRESSIONS = [ |
| "square", |
| "ways", |
| "integers", |
| "dollars", |
| "mph", |
| "inches", |
| "ft", |
| "hours", |
| "km", |
| "units", |
| "\\ldots", |
| "points", |
| "feet", |
| "minutes", |
| "digits", |
| "cents", |
| "degrees", |
| "cm", |
| "gm", |
| "pounds", |
| "meters", |
| "meals", |
| "edges", |
| "students", |
| "multiples", |
| "\\text{s}", |
| "\\text{.}", |
| "\\text{\ns}", |
| "\\text{}^2", |
| "\\text{}^3", |
| "\\text{\n}", |
| "\\text{}", |
| r"\mathrm{th}", |
| r"^\circ", |
| r"^{\circ}", |
| r"\;", |
| r",\!", |
| "{,}", |
| '"', |
| "\\dots", |
| ] |
|
|
|
|
| @dataclass |
| class MatchResult: |
| correct: bool |
| method: str |
| gold_candidate: Optional[str] = None |
| pred_candidate: Optional[str] = None |
|
|
|
|
| class EvaluationTimeoutError(TimeoutError): |
| pass |
|
|
|
|
| @contextmanager |
| def time_limit(seconds: Optional[float]): |
| if seconds is None or seconds <= 0: |
| yield |
| return |
|
|
| def _handle_timeout(signum, frame): |
| raise EvaluationTimeoutError(f"example evaluation exceeded {seconds} seconds") |
|
|
| previous_handler = signal.getsignal(signal.SIGALRM) |
| signal.signal(signal.SIGALRM, _handle_timeout) |
| signal.setitimer(signal.ITIMER_REAL, seconds) |
| try: |
| yield |
| finally: |
| signal.setitimer(signal.ITIMER_REAL, 0) |
| signal.signal(signal.SIGALRM, previous_handler) |
|
|
|
|
| def read_jsonl_to_list(path: str, encoding: str = "utf-8") -> list[dict[str, Any]]: |
| rows: list[dict[str, Any]] = [] |
| with open(path, "r", encoding=encoding) as fh: |
| for lineno, line in enumerate(fh, start=1): |
| stripped = line.strip() |
| if not stripped: |
| continue |
| try: |
| obj = json.loads(stripped) |
| except json.JSONDecodeError as exc: |
| raise ValueError(f"Invalid JSON at {path}:{lineno}: {exc}") from exc |
| if not isinstance(obj, dict): |
| raise ValueError(f"Expected JSON object at {path}:{lineno}, got {type(obj).__name__}") |
| rows.append(obj) |
| return rows |
|
|
|
|
| def normalize_final_answer(text: str) -> str: |
| answer = text.split("=")[-1] |
|
|
| for before, after in SUBSTITUTIONS: |
| answer = answer.replace(before, after) |
| for expr in REMOVED_EXPRESSIONS: |
| answer = answer.replace(expr, "") |
|
|
| answer = re.sub(r"(.*?)(\$)(.*?)(\$)(.*)", "$\\3$", answer) |
| answer = re.sub(r"(\\text\{)(.*?)(\})", "\\2", answer) |
| answer = re.sub(r"(\\textbf\{)(.*?)(\})", "\\2", answer) |
| answer = re.sub(r"(\\overline\{)(.*?)(\})", "\\2", answer) |
| answer = re.sub(r"(\\boxed\{)(.*)(\})", "\\2", answer) |
| answer = re.sub(r"(frac)([^{])(.)", "frac{\\2}{\\3}", answer) |
| answer = re.sub(r"(sqrt)([^{])", "sqrt{\\2}", answer) |
| answer = answer.replace("$", "") |
|
|
| if answer.replace(",", "").isdigit(): |
| answer = answer.replace(",", "") |
|
|
| return answer |
|
|
|
|
| def dedupe_keep_order(items: Iterable[str]) -> list[str]: |
| seen: set[str] = set() |
| output: list[str] = [] |
| for item in items: |
| if item not in seen: |
| seen.add(item) |
| output.append(item) |
| return output |
|
|
|
|
| def strip_markdown(text: str) -> str: |
| stripped = text.strip() |
| stripped = stripped.replace("**", "").replace("__", "") |
| stripped = stripped.strip("`") |
| return stripped |
|
|
|
|
| def strip_answer_prefix(text: str) -> str: |
| cleaned = strip_markdown(text) |
| cleaned = re.sub(r"^\s*####\s*", "", cleaned) |
| cleaned = re.sub(r"(?is)^\s*<answer>\s*", "", cleaned) |
| cleaned = re.sub(r"(?is)\s*</answer>\s*$", "", cleaned) |
| cleaned = re.sub(r"(?is)^\s*(?:final answer|answer)\s*[:：]\s*", "", cleaned) |
| cleaned = re.sub(r"(?is)^\s*the final answer is\s*", "", cleaned) |
| cleaned = re.sub(r"(?is)\s*I hope it is correct\.?\s*$", "", cleaned) |
| cleaned = cleaned.strip() |
| return cleaned |
|
|
|
|
| def is_balanced_bracket_wrap(text: str, left: str, right: str) -> bool: |
| if not (text.startswith(left) and text.endswith(right)): |
| return False |
| depth = 0 |
| body = text[len(left) : len(text) - len(right)] |
| for char in body: |
| if char == left: |
| depth += 1 |
| elif char == right: |
| depth -= 1 |
| if depth < 0: |
| return False |
| return depth == 0 |
|
|
|
|
| def extract_braced_content(text: str, open_index: int) -> tuple[Optional[str], Optional[int]]: |
| if open_index >= len(text) or text[open_index] != "{": |
| return None, None |
|
|
| depth = 0 |
| content: list[str] = [] |
| for idx in range(open_index, len(text)): |
| char = text[idx] |
| if char == "{": |
| depth += 1 |
| if depth > 1: |
| content.append(char) |
| elif char == "}": |
| depth -= 1 |
| if depth == 0: |
| return "".join(content), idx |
| content.append(char) |
| else: |
| content.append(char) |
| return None, None |
|
|
|
|
| def unwrap_known_wrappers(text: str) -> str: |
| current = text.strip() |
| while True: |
| previous = current |
| current = current.strip() |
|
|
| if current.startswith("$$") and current.endswith("$$") and len(current) >= 4: |
| current = current[2:-2].strip() |
| elif current.startswith("\\[") and current.endswith("\\]"): |
| current = current[2:-2].strip() |
| elif current.startswith("\\(") and current.endswith("\\)"): |
| current = current[2:-2].strip() |
| elif current.startswith("$") and current.endswith("$") and len(current) >= 2: |
| current = current[1:-1].strip() |
| elif current.startswith("\\boxed"): |
| brace_start = current.find("{") |
| content, brace_end = extract_braced_content(current, brace_start) |
| if content is not None and brace_end == len(current) - 1: |
| current = content.strip() |
| elif current.startswith("\\fbox"): |
| brace_start = current.find("{") |
| content, brace_end = extract_braced_content(current, brace_start) |
| if content is not None and brace_end == len(current) - 1: |
| current = content.strip() |
| elif current.startswith("\\text{") or current.startswith("\\mathrm{") or current.startswith("\\textbf{"): |
| brace_start = current.find("{") |
| content, brace_end = extract_braced_content(current, brace_start) |
| if content is not None and brace_end == len(current) - 1: |
| current = content.strip() |
|
|
| current = current.strip(" \n\t\r,.;:!。；，") |
| if current == previous: |
| break |
|
|
| return current |
|
|
|
|
| def cleanup_candidate(text: str) -> str: |
| cleaned = strip_answer_prefix(text) |
| cleaned = cleaned.replace("\u2212", "-").replace("−", "-") |
| cleaned = cleaned.replace("\u00d7", "*").replace("\u00f7", "/") |
| cleaned = cleaned.replace("\\dfrac", "\\frac").replace("\\tfrac", "\\frac") |
| cleaned = cleaned.replace("\\left", "").replace("\\right", "") |
| cleaned = cleaned.replace("\\!", "").replace("\\,", "").replace("\\;", "") |
| cleaned = cleaned.replace("\\%", "%") |
| cleaned = unwrap_known_wrappers(cleaned) |
| cleaned = re.sub(r"\s+", " ", cleaned).strip() |
| return cleaned |
|
|
|
|
| def canonicalize_for_compare(text: str) -> str: |
| cleaned = cleanup_candidate(text) |
| cleaned = normalize_final_answer(cleaned) |
| cleaned = cleaned.replace("\\{", "{").replace("\\}", "}") |
| cleaned = cleaned.replace(" ", "") |
| cleaned = cleaned.strip(" \n\t\r,.;:!。；，") |
| return cleaned |
|
|
|
|
| def extract_boxed_contents(text: str) -> list[str]: |
| matches: list[str] = [] |
| for command in ("\\boxed", "\\fbox"): |
| start = 0 |
| while True: |
| idx = text.find(command, start) |
| if idx == -1: |
| break |
| cursor = idx + len(command) |
| while cursor < len(text) and text[cursor].isspace(): |
| cursor += 1 |
| if cursor < len(text) and text[cursor] == "{": |
| content, end_idx = extract_braced_content(text, cursor) |
| if content is not None and end_idx is not None: |
| matches.append(content.strip()) |
| start = end_idx + 1 |
| continue |
| start = cursor + 1 |
| return matches |
|
|
|
|
| def extract_last_boxed_with_position(text: str) -> tuple[Optional[str], Optional[int]]: |
| last_content: Optional[str] = None |
| last_end: Optional[int] = None |
|
|
| for command in ("\\boxed", "\\fbox"): |
| start = 0 |
| while True: |
| idx = text.find(command, start) |
| if idx == -1: |
| break |
| cursor = idx + len(command) |
| while cursor < len(text) and text[cursor].isspace(): |
| cursor += 1 |
| if cursor < len(text) and text[cursor] == "{": |
| content, end_idx = extract_braced_content(text, cursor) |
| if content is not None and end_idx is not None: |
| if last_end is None or end_idx > last_end: |
| last_content = content.strip() |
| last_end = end_idx |
| start = end_idx + 1 |
| continue |
| start = cursor + 1 |
|
|
| return last_content, last_end |
|
|
|
|
| def extract_answer_by_patterns(text: str) -> list[str]: |
| patterns = [ |
| r"(?is)<answer>\s*(.*?)\s*</answer>", |
| r"(?is)Final Answer\s*[:：]\s*(.*?)(?=\n\s*\n|$)", |
| r"(?is)The final answer is\s*(.*?)(?:\.?\s*I hope it is correct\.?|$)", |
| r"(?im)^\s*Answer\s*[:：]\s*(.+?)\s*$", |
| r"(?im)^\s*####\s*(.+?)\s*$", |
| ] |
| matches: list[str] = [] |
| for pattern in patterns: |
| for match in re.findall(pattern, text): |
| if isinstance(match, tuple): |
| for piece in match: |
| if piece and piece.strip(): |
| matches.append(piece.strip()) |
| elif match and match.strip(): |
| matches.append(match.strip()) |
| return matches |
|
|
|
|
| def has_substantial_suffix(text: str, start_index: int) -> bool: |
| suffix = text[start_index:].strip() |
| if not suffix: |
| return False |
|
|
| suffix = re.sub(r"[\s`*_>#\-\.,;:!?\)\]\}]+", "", suffix) |
| signal_chars = re.findall(r"[A-Za-z0-9\u4e00-\u9fff\\\\]", suffix) |
| return len(signal_chars) >= 20 |
|
|
|
|
| def extract_last_pattern_answer_near_end(text: str) -> Optional[str]: |
| patterns = [ |
| r"(?is)<answer>\s*(.*?)\s*</answer>", |
| r"(?is)Final Answer\s*[:：]\s*(.*?)(?=\n\s*\n|$)", |
| r"(?is)The final answer is\s*(.*?)(?:\.?\s*I hope it is correct\.?|$)", |
| r"(?im)^\s*Answer\s*[:：]\s*(.+?)\s*$", |
| r"(?im)^\s*####\s*(.+?)\s*$", |
| ] |
|
|
| last_match_text: Optional[str] = None |
| last_match_end: Optional[int] = None |
|
|
| for pattern in patterns: |
| for match in re.finditer(pattern, text): |
| candidate = match.group(1).strip() |
| if not candidate: |
| continue |
| if last_match_end is None or match.end() > last_match_end: |
| last_match_text = candidate |
| last_match_end = match.end() |
|
|
| if last_match_text is None or last_match_end is None: |
| return None |
| if has_substantial_suffix(text, last_match_end): |
| return None |
| return cleanup_candidate(last_match_text) |
|
|
|
|
| def extract_final_marked_candidates(text: str) -> list[str]: |
| candidates: list[str] = [] |
|
|
| boxed = extract_boxed_contents(text) |
| if boxed: |
| candidates.append(boxed[-1]) |
|
|
| pattern_matches = extract_answer_by_patterns(text) |
| for match in reversed(pattern_matches[-3:]): |
| candidates.append(match) |
| first_line = match.splitlines()[0].strip() |
| if first_line and first_line != match: |
| candidates.append(first_line) |
|
|
| cleaned = [cleanup_candidate(item) for item in candidates] |
| cleaned = [item for item in cleaned if item and item.lower() not in {"answer", "final answer"}] |
| return dedupe_keep_order(cleaned) |
|
|
|
|
| def extract_latex_blocks(text: str) -> list[str]: |
| blocks: list[str] = [] |
| patterns = [ |
| r"(?s)\$\$(.*?)\$\$", |
| r"(?s)\\\[(.*?)\\\]", |
| r"(?s)\\\((.*?)\\\)", |
| r"(?s)(?<!\$)\$(?!\$)(.+?)(?<!\$)\$(?!\$)", |
| ] |
| for pattern in patterns: |
| for match in re.findall(pattern, text): |
| if match and str(match).strip(): |
| blocks.append(str(match).strip()) |
| return blocks[-6:] |
|
|
|
|
| def extract_line_candidates(text: str) -> list[str]: |
| lines = [line.strip() for line in text.splitlines() if line.strip()] |
| candidates: list[str] = [] |
| for line in lines[-8:]: |
| stripped = strip_markdown(line).lstrip("-*") |
| if not stripped: |
| continue |
| candidates.append(stripped) |
| if "=" in stripped: |
| rhs = stripped.split("=")[-1].strip() |
| if rhs: |
| candidates.append(rhs) |
| return candidates[-8:] |
|
|
|
|
| def extract_answer_candidates(text: str) -> list[str]: |
| if not text: |
| return [] |
|
|
| raw_text = str(text) |
| candidates: list[str] = [] |
|
|
| boxed = extract_boxed_contents(raw_text) |
| candidates.extend(reversed(boxed[-6:])) |
|
|
| pattern_matches = extract_answer_by_patterns(raw_text) |
| for match in reversed(pattern_matches[-6:]): |
| candidates.append(match) |
| first_line = match.splitlines()[0].strip() |
| if first_line and first_line != match: |
| candidates.append(first_line) |
|
|
| for block in reversed(extract_latex_blocks(raw_text)): |
| candidates.append(block) |
|
|
| candidates.extend(reversed(extract_line_candidates(raw_text))) |
|
|
| raw_text_stripped = raw_text.strip() |
| if raw_text_stripped and len(raw_text_stripped) <= 200: |
| candidates.append(raw_text_stripped) |
|
|
| cleaned = [cleanup_candidate(item) for item in candidates] |
| cleaned = [item for item in cleaned if item and item.lower() not in {"answer", "final answer"}] |
| return dedupe_keep_order(cleaned)[:20] |
|
|
|
|
| def load_ground_truth_examples(dataset_name: str, split: str, gt_jsonl_path: Optional[str]) -> list[dict[str, Any]]: |
| if gt_jsonl_path: |
| return read_jsonl_to_list(gt_jsonl_path) |
|
|
| if load_dataset is None: |
| raise RuntimeError( |
| "datasets is not installed, so ground truth cannot be loaded from Hugging Face. " |
| "Please install datasets or pass --gt-jsonl-path." |
| ) from DATASETS_IMPORT_ERROR |
|
|
| return list(load_dataset(dataset_name, split=split)) |
|
|
|
|
| def extract_ground_truth_answer_candidates(example: dict[str, Any]) -> list[str]: |
| answer_text = str(example.get("answer", "") or "").strip() |
| if not answer_text: |
| return [] |
|
|
| candidates = [cleanup_candidate(answer_text)] |
| candidates.extend(extract_final_marked_candidates(answer_text)) |
| if len(answer_text) <= 200: |
| candidates.extend(extract_answer_candidates(answer_text)) |
|
|
| return [item for item in dedupe_keep_order(candidates) if item] |
|
|
|
|
| def extract_llm_final_answer_candidates(text: str) -> list[str]: |
| if not text: |
| return [] |
|
|
| raw_text = str(text) |
| raw_text_stripped = raw_text.strip() |
| tail_text = raw_text_stripped[-800:] |
| tail_lines = [line.strip() for line in tail_text.splitlines() if line.strip()] |
|
|
| candidates: list[str] = [] |
|
|
| near_end_pattern = extract_last_pattern_answer_near_end(raw_text) |
| if near_end_pattern: |
| candidates.append(near_end_pattern) |
|
|
| last_boxed, last_boxed_end = extract_last_boxed_with_position(raw_text) |
| if last_boxed and last_boxed_end is not None and not has_substantial_suffix(raw_text, last_boxed_end + 1): |
| candidates.append(last_boxed) |
|
|
| tail_latex_blocks = extract_latex_blocks(tail_text) |
| if tail_latex_blocks: |
| candidates.append(tail_latex_blocks[-1]) |
|
|
| for line in reversed(tail_lines[-3:]): |
| stripped = strip_markdown(line).lstrip("-*").strip() |
| if not stripped: |
| continue |
| candidates.append(stripped) |
| if "=" in stripped: |
| rhs = stripped.split("=")[-1].strip() |
| if rhs: |
| candidates.append(rhs) |
|
|
| if raw_text_stripped and len(raw_text_stripped) <= 200: |
| candidates.append(raw_text_stripped) |
|
|
| cleaned = [cleanup_candidate(item) for item in candidates] |
| cleaned = [item for item in cleaned if item and item.lower() not in {"answer", "final answer"}] |
| return dedupe_keep_order(cleaned)[:10] |
|
|
|
|
| def maybe_parse_numeric(text: str) -> Optional[Any]: |
| if sympy is None: |
| return None |
|
|
| candidate = cleanup_candidate(text) |
| if not candidate: |
| return None |
|
|
| if candidate.endswith("%"): |
| candidate = f"({candidate[:-1]})/100" |
|
|
| if any(token in candidate for token in ("\\frac", "\\sqrt", "\\pi", "\\cdot", "\\pm", "{", "}")): |
| try: |
| return parse_latex(candidate) |
| except Exception: |
| pass |
|
|
| ascii_candidate = candidate |
| ascii_candidate = ascii_candidate.replace("^", "**") |
| ascii_candidate = ascii_candidate.replace("\\pi", "pi").replace("\\cdot", "*") |
| ascii_candidate = ascii_candidate.replace("{", "(").replace("}", ")") |
|
|
| try: |
| return sympy.sympify(ascii_candidate) |
| except Exception: |
| return None |
|
|
|
|
| def are_sympy_equivalent(left: str, right: str) -> bool: |
| left_expr = maybe_parse_numeric(left) |
| right_expr = maybe_parse_numeric(right) |
| if left_expr is None or right_expr is None: |
| return False |
|
|
| try: |
| diff = sympy.simplify(left_expr - right_expr) |
| except Exception: |
| return False |
| return diff == 0 |
|
|
|
|
| @lru_cache(maxsize=16384) |
| def cached_parse(text: str, mode: str) -> Any: |
| if parse is None: |
| return None |
|
|
| if mode == "default": |
| return parse(text) |
|
|
| if mode == "snippet": |
| return parse(text, extraction_config=[LatexExtractionConfig(), ExprExtractionConfig()]) |
|
|
| raise ValueError(f"Unknown parse mode: {mode}") |
|
|
|
|
| def try_math_verify(gold_text: str, pred_text: str, snippet_mode: bool) -> bool: |
| if parse is None or verify is None: |
| return False |
|
|
| gold_variants: list[tuple[str, str]] = [] |
| pred_variants: list[tuple[str, str]] = [] |
|
|
| if snippet_mode: |
| cleaned_gold = cleanup_candidate(gold_text) |
| cleaned_pred = cleanup_candidate(pred_text) |
| for candidate in dedupe_keep_order([gold_text, cleaned_gold, f"${cleaned_gold}$"]): |
| if candidate: |
| gold_variants.append(("snippet", candidate)) |
| for candidate in dedupe_keep_order([pred_text, cleaned_pred, f"${cleaned_pred}$"]): |
| if candidate: |
| pred_variants.append(("snippet", candidate)) |
| else: |
| gold_variants.append(("default", gold_text)) |
| pred_variants.append(("default", pred_text)) |
|
|
| for gold_mode, gold_variant in gold_variants: |
| gold_parsed = cached_parse(gold_variant, gold_mode) |
| if gold_parsed is None: |
| continue |
| for pred_mode, pred_variant in pred_variants: |
| pred_parsed = cached_parse(pred_variant, pred_mode) |
| if pred_parsed is None: |
| continue |
| try: |
| if verify(gold_parsed, pred_parsed): |
| return True |
| except Exception: |
| continue |
|
|
| return False |
|
|
|
|
| def compare_candidates( |
| gold_candidates: list[str], |
| pred_candidates: list[str], |
| ) -> MatchResult: |
| for gold in gold_candidates: |
| gold_norm = canonicalize_for_compare(gold) |
| if not gold_norm: |
| continue |
| for pred in pred_candidates: |
| pred_norm = canonicalize_for_compare(pred) |
| if gold_norm and gold_norm == pred_norm: |
| return MatchResult(True, "normalized_exact", gold, pred) |
|
|
| for gold in gold_candidates: |
| for pred in pred_candidates: |
| if are_sympy_equivalent(gold, pred): |
| return MatchResult(True, "sympy_equiv", gold, pred) |
|
|
| for gold in gold_candidates: |
| for pred in pred_candidates: |
| if try_math_verify(gold, pred, snippet_mode=True): |
| return MatchResult(True, "math_verify_snippet", gold, pred) |
|
|
| return MatchResult(False, "no_match") |
|
|
|
|
| def evaluate_example(example: dict[str, Any], prediction_row: dict[str, Any]) -> tuple[MatchResult, list[str], list[str]]: |
| pred_raw = str(prediction_row.get("answer", "")) |
| pred_candidates = extract_llm_final_answer_candidates(pred_raw) |
| gold_candidates = extract_ground_truth_answer_candidates(example) |
|
|
| if not pred_candidates and pred_raw.strip(): |
| pred_candidates = [cleanup_candidate(pred_raw)] |
|
|
| result = compare_candidates( |
| gold_candidates=gold_candidates, |
| pred_candidates=pred_candidates, |
| ) |
| return result, gold_candidates, pred_candidates |
|
|
|
|
| def build_argparser() -> argparse.ArgumentParser: |
| parser = argparse.ArgumentParser(description="High-precision MATH-500 answer evaluator") |
| parser.add_argument( |
| "--pred-path", |
| default="outputs/minerva_math500/rank_0.jsonl", |
| help="Path to the model prediction jsonl file.", |
| ) |
| parser.add_argument( |
| "--dataset-name", |
| default="HuggingFaceH4/MATH-500", |
| help="Hugging Face dataset name for ground truth.", |
| ) |
| parser.add_argument("--split", default="test", help="Dataset split.") |
| parser.add_argument( |
| "--gt-jsonl-path", |
| default=None, |
| help="Optional local jsonl ground truth path. If set, datasets will not be used.", |
| ) |
| parser.add_argument("--limit", type=int, default=None, help="Evaluate only the first N examples.") |
| parser.add_argument( |
| "--print-wrong", |
| type=int, |
| default=0, |
| help="How many mismatched examples to print for debugging.", |
| ) |
| parser.add_argument( |
| "--details-path", |
| default=None, |
| help="Optional jsonl path for saving per-example evaluation details. " |
| "If not set, a file will be created next to the prediction file automatically.", |
| ) |
| parser.add_argument( |
| "--per-example-timeout", |
| type=float, |
| default=8.0, |
| help="Maximum seconds allowed for one example before it is skipped as a timeout.", |
| ) |
| return parser |
|
|
|
|
| def main(argv: Optional[list[str]] = None) -> int: |
| args = build_argparser().parse_args(argv) |
|
|
| pred_path = Path(args.pred_path) |
| if not pred_path.exists(): |
| print(f"Prediction file not found: {pred_path}", file=sys.stderr) |
| return 1 |
|
|
| gt_examples = load_ground_truth_examples(args.dataset_name, args.split, args.gt_jsonl_path) |
| pred_examples = read_jsonl_to_list(str(pred_path)) |
|
|
| if args.limit is not None: |
| gt_examples = gt_examples[: args.limit] |
| pred_examples = pred_examples[: args.limit] |
|
|
| total = min(len(gt_examples), len(pred_examples)) |
| if total == 0: |
| print("No examples to evaluate.", file=sys.stderr) |
| return 1 |
|
|
| if len(gt_examples) != len(pred_examples): |
| print( |
| f"[warning] ground truth count = {len(gt_examples)}, prediction count = {len(pred_examples)}; " |
| f"evaluating the first {total} pairs only.", |
| file=sys.stderr, |
| ) |
|
|
| method_counter: Counter[str] = Counter() |
| wrong_printed = 0 |
| correct = 0 |
| details_path = Path(args.details_path) if args.details_path else pred_path.with_name(f"{pred_path.stem}_eval_details.jsonl") |
| details_fh = open(details_path, "w", encoding="utf-8") |
| indices: Iterable[int] = range(total) |
|
|
| if tqdm is not None: |
| indices = tqdm(indices, total=total, desc="Evaluating MATH-500", unit="sample") |
|
|
| try: |
| for idx in indices: |
| example = gt_examples[idx] |
| prediction_row = pred_examples[idx] |
| try: |
| with time_limit(args.per_example_timeout): |
| result, gold_candidates, pred_candidates = evaluate_example(example, prediction_row) |
| except EvaluationTimeoutError: |
| result = MatchResult(False, "example_timeout") |
| gold_candidates = extract_ground_truth_answer_candidates(example) |
| pred_candidates = [] |
| except Exception as exc: |
| result = MatchResult(False, f"example_error:{type(exc).__name__}") |
| gold_candidates = extract_ground_truth_answer_candidates(example) |
| pred_candidates = [] |
|
|
| if result.correct: |
| correct += 1 |
| method_counter[result.method] += 1 |
|
|
| detail_row = { |
| "index": idx, |
| "correct": result.correct, |
| "method": result.method, |
| "problem": example.get("problem"), |
| "ground_truth_answer": example.get("answer"), |
| "llm_response": prediction_row.get("answer", ""), |
| "gold_candidate": result.gold_candidate, |
| "pred_candidate": result.pred_candidate, |
| "ground_truth_answer_candidates": gold_candidates, |
| "llm_final_answer_candidates": pred_candidates, |
| } |
|
|
| details_fh.write(json.dumps(detail_row, ensure_ascii=False) + "\n") |
|
|
| if not result.correct and wrong_printed < args.print_wrong: |
| wrong_printed += 1 |
| print("=" * 80) |
| print(f"Index: {idx}") |
| print(f"Problem: {example.get('problem', '')}") |
| print(f"Gold candidates: {gold_candidates[:5]}") |
| print(f"Pred candidates: {pred_candidates[:5]}") |
| print("Raw prediction tail:") |
| print(str(prediction_row.get("answer", ""))[-800:]) |
|
|
| finally: |
| if details_fh is not None: |
| details_fh.close() |
|
|
| accuracy = correct / total |
| print("=" * 80) |
| print(f"Total: {total}") |
| print(f"Correct: {correct}") |
| print(f"Accuracy: {accuracy:.4%}") |
| print(f"Saved details: {details_path}") |
| print("Match breakdown:") |
| for method, count in method_counter.most_common(): |
| print(f" {method}: {count}") |
|
|
| if MATH_VERIFY_IMPORT_ERROR is not None: |
| print( |
| f"[warning] math_verify unavailable, matching quality is reduced: {MATH_VERIFY_IMPORT_ERROR}", |
| file=sys.stderr, |
| ) |
| if SYMPY_IMPORT_ERROR is not None: |
| print( |
| f"[warning] sympy unavailable, symbolic fallback is reduced: {SYMPY_IMPORT_ERROR}", |
| file=sys.stderr, |
| ) |
|
|
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
