from __future__ import annotations import ast import hashlib import json import math import string from decimal import Decimal, InvalidOperation from typing import Any, Mapping, Sequence import re import unicodedata from fractions import Fraction from typing import Any, Sequence RESULT_SCHEMA = 'edgerock-verdict-v1' EXTERNAL_REQUEST_SCHEMA = 'edgerock-verifier-request-v1' TASK_TYPES = frozenset({'math', 'choice', 'code', 'stem', 'general', 'retrieval', 'ifeval', 'ifbench'}) VERDICTS = frozenset({'PASS', 'FAIL', 'UNCERTAIN'}) CPU_VERIFIER_REVISION = 'cpu_first_typed_tristate_final_answer_equivalence_v43' CODE_VERIFIER_MODES = frozenset({'python_harness', 'code'}) INSTRUCTION_VERIFIER_MODES = frozenset({'ifbench', 'ifeval'}) MATH_EXTERNAL_VERIFIER_MODES = frozenset({'math_equivalence', 'symbolic', 'expression', 'numeric_equivalence', 'set'}) DIRECT_REFERENCE_VERIFIER_MODES = frozenset({'numeric_equivalence', 'numeric', 'integer', 'fraction', 'decimal', 'exact_choice', 'choice', 'choice_set', 'exact_text', 'exact', 'text', 'set', 'symbolic', 'expression', 'drop_exact', 'drop_official', 'bbeh_official', 'python_literal_output'}) def _unwrap_complete_latex_reference(value: str) -> str: text = value.strip().strip('$').strip() for _ in range(4): previous = text for command in ('boxed', 'fbox', 'text', 'mbox', 'mathrm'): prefix = f'\\{command}{{' if not text.startswith(prefix) or not text.endswith('}'): continue depth = 0 closing = -1 for index, character in enumerate(text[len(prefix) - 1:], len(prefix) - 1): if character == '{': depth += 1 elif character == '}': depth -= 1 if depth == 0: closing = index break if closing == len(text) - 1: text = text[len(prefix):-1].strip() break if text == previous: break return text def _choice_reference_labels(reference: str) -> tuple[str, ...] | None: text = _unwrap_complete_latex_reference(reference) if len(text) > 2 and (text[0], text[-1]) in {('(', ')'), ('[', ']'), ('{', '}')}: text = text[1:-1].strip() if re.fullmatch('[A-Z]{2,26}', text): return tuple(text) text = text.replace('\\,', ' ') text = re.sub('\\band\\b|[;,、和及]', ',', text, flags=re.I) fields = [field for field in re.split('\\s*,\\s*', text) if field.strip()] labels = [choice_value(field) for field in fields] if not labels or any((label is None for label in labels)): return None return tuple((str(label) for label in labels)) def choice_set_value(value: Any) -> frozenset[str] | None: labels = _choice_reference_labels(str(value)) if not labels or len(labels) < 2 or len(set(labels)) != len(labels): return None return frozenset(labels) def _choice_problem_labels(problem: str) -> set[str]: labels: set[str] = set() patterns = ('(?[A-Z])\\)\\s*\\}|(?[A-Z])\\)(?![A-Za-z])|^[ \\t]*(?P[A-Z])[.)::]\\s+', re.MULTILINE) def choice_option_values(problem: str) -> dict[str, str] | None: question_starts = list(re.finditer('(?mi)^[ \\t]*(?:Q|Question|问题|题目)\\s*[::]', problem)) if question_starts: problem = problem[question_starts[-1].end():] headings = list(re.finditer('(?mi)^[ \\t]*(?:Options(?:\\s+are)?|Choices|选项)\\s*[::]', problem)) if headings: problem = problem[headings[-1].end():] tail = re.search("(?mi)^[ \\t]*(?:(?:give|enter|write|put|provide|return)\\b[^\\n]*(?:answer|letter|option|final|boxed)|reason through[^\\n]*finish|最后一行[^\\n]*答案|(?:请)?(?:在|将)[^\\n]*(?:答案|选项)|A:\\s*Let's think)", problem) if tail: problem = problem[:tail.start()] matches = list(_CHOICE_OPTION_MARKER.finditer(problem)) line_matches = [match for match in matches if not problem[problem.rfind('\n', 0, match.start()) + 1:match.start()].strip()] if len(line_matches) >= 2 and any(('A' in match.groupdict().values() for match in line_matches)): matches = line_matches markers = [] for match in matches: label = next((value for value in match.groupdict().values() if value)) markers.append((match.start(), match.end(), label.upper())) labels = [label for _, _, label in markers] if len(markers) < 2 or 'A' not in labels or len(labels) != len(set(labels)): return None options = {} for index, (_, end, label) in enumerate(markers): next_start = markers[index + 1][0] if index + 1 < len(markers) else len(problem) body = problem[end:next_start].strip().lstrip('::').strip() body = re.sub('^(?:\\$+|\\\\(?:quad|qquad|enspace)|\\\\[ ,;!])+', '', body) body = re.sub('(?:\\$+|\\\\(?:quad|qquad|enspace)|\\\\[ ,;!])+\\s*$', '', body) normalized = _choice_content_text(body) if not normalized: return None options[label] = normalized return options def compare_choice_option_content(candidate: Any, references: Sequence[Any], problem: str) -> dict[str, Any] | None: candidate_label = choice_value(candidate) options = choice_option_values(problem) if candidate_label is None or options is None or candidate_label not in options or any((choice_value(reference) is not None for reference in references)): return None reference_labels = {label for label, option in options.items() for reference in references if option == normalize_answer_text(reference)} if len(reference_labels) != 1: return None return {'verdict': 'PASS' if candidate_label in reference_labels else 'FAIL', 'method': 'choice_label_to_signed_option_content', 'candidate': candidate_label, 'references': sorted(reference_labels)} def _row_reference_values(row: Mapping[str, Any], extra: Mapping[str, Any]) -> list[Any]: values = extra.get('reference_answers') if values is not None: if not isinstance(values, list): return [] return [value for value in values if str(value).strip()] value = row.get('ground_truth') return [value] if value is not None and str(value).strip() else [] def executable_code_payload_contract(payload: Any, *, language: Any=None) -> dict[str, Any]: if isinstance(payload, list): cases = payload config: Mapping[str, Any] = {} elif isinstance(payload, Mapping): config = payload cases = payload.get('cases') else: return {'eligible': False, 'reason': 'code_verifier_payload_missing'} checker = str(config.get('checker', 'exact')).strip().lower() if checker not in {'exact', 'tokens'}: return {'eligible': False, 'reason': 'code_verifier_checker_invalid'} raw_timeout = config.get('time_limit_s', 2.0) try: timeout = math.nan if isinstance(raw_timeout, bool) else float(raw_timeout) except (TypeError, ValueError): timeout = math.nan if not math.isfinite(timeout) or timeout <= 0.0: return {'eligible': False, 'reason': 'code_verifier_timeout_invalid'} runtime = str(language or config.get('language') or 'python').strip().lower() python_runtime = runtime in {'', 'python', 'py', 'python3'} if not python_runtime and runtime not in {'cpp', 'c++', 'cxx', 'cc'}: return {'eligible': False, 'reason': 'code_verifier_language_invalid'} if config.get('lcb_payload_path') and config.get('lcb_runner_path'): if not python_runtime: return {'eligible': False, 'reason': 'livecodebench_python_required'} return {'eligible': True, 'reason': 'livecodebench_official_payload', 'mode': 'livecodebench'} harness = config.get('python_harness') if harness is not None: if not python_runtime or not isinstance(harness, str) or (not harness.strip()): return {'eligible': False, 'reason': 'code_verifier_python_harness_invalid'} return {'eligible': True, 'reason': 'code_verifier_python_harness', 'mode': 'python_harness'} if not isinstance(cases, list) or not cases: return {'eligible': False, 'reason': 'code_verifier_cases_missing'} for case in cases: if not isinstance(case, Mapping) or not isinstance(case.get('stdout', case.get('expected_stdout')), str) or ('stdin' in case and (not isinstance(case['stdin'], str))): return {'eligible': False, 'reason': 'code_verifier_case_invalid'} return {'eligible': True, 'reason': 'code_verifier_cases', 'mode': 'cases'} def bound_tool_verdict_contract(value: Any) -> dict[str, Any]: if value is None: return {'eligible': True, 'reason': 'absent'} if not isinstance(value, Mapping): return {'eligible': False, 'reason': 'code_tool_verdict_invalid'} hashes = (value.get('candidate_sha256'), value.get('tests_sha256')) if any((not isinstance(item, str) or len(item) != 64 or any((character not in '0123456789abcdef' for character in item)) for item in hashes)) or str(value.get('verdict') or '').upper() not in VERDICTS: return {'eligible': False, 'reason': 'code_tool_verdict_invalid'} return {'eligible': True, 'reason': 'bound_tool_verdict'} def deterministic_verifier_contract(row: Mapping[str, Any]) -> dict[str, Any]: extra = row['extra_info'] mode = 'math_equivalence' answer_type = '' references = _row_reference_values(row, extra) if not references or any((len(str(reference)) > 4096 for reference in references)): return {'eligible': False, 'reason': 'math_reference_empty_or_too_long', 'verifier_mode': mode, 'answer_type': answer_type, 'resolution': 'rejected'} problem = str(extra.get('problem') or '') if not problem: prompt = row.get('prompt') or [] if isinstance(prompt, list) and prompt and isinstance(prompt[-1], Mapping): problem = str(prompt[-1].get('content') or '') choice_labels = [_choice_reference_labels(str(reference)) for reference in references] problem_labels = _choice_problem_labels(problem) if all(choice_labels) and len(problem_labels) >= 2 and all((set(labels or ()).issubset(problem_labels) for labels in choice_labels)) and (len({len(labels or ()) for labels in choice_labels}) == 1): single = len(choice_labels[0] or ()) == 1 return {'eligible': True, 'reason': 'math_reference_routed_exact_choice' if single else 'math_reference_routed_choice_set', 'verifier_mode': 'exact_choice' if single else 'choice_set', 'answer_type': 'choice' if single else 'choice_set', 'resolution': 'choice_metadata_rebound'} unwrapped_references = [_unwrap_complete_latex_reference(str(reference)) for reference in references] fixed_text = all((2 < len(unwrapped) <= 128 and re.fullmatch("[A-Za-z\\u3400-\\u9fff][A-Za-z\\u3400-\\u9fff '\\u2019.,,。]*", unwrapped) and (not re.search('\\b(?:if|otherwise|when|where|such\\s+that)\\b|(?:如果|否则|当|其中|满足|时)', unwrapped, re.I)) for unwrapped in unwrapped_references)) if fixed_text: return {'eligible': True, 'reason': 'math_reference_routed_exact_text', 'verifier_mode': 'exact_text', 'answer_type': 'exact_text', 'resolution': 'fixed_text_metadata_rebound'} exact_text_fallback = all((re.search('\\\\begin\\{(?:cases|array)\\}|\\\\(?:text|mbox|mathrm)\\{|\\b(?:if|otherwise|when|where|such\\s+that)\\b|\\d\\s*\\\\?[!::]\\\\?!?\\s*\\d', str(reference), re.I) for reference in references)) if exact_text_fallback: return {'eligible': True, 'reason': 'math_reference_routed_exact_text_fallback', 'verifier_mode': 'exact_text', 'answer_type': 'exact_text', 'resolution': 'unparsed_reference_exact_text'} if all((symbolic_reference_is_supported(reference, problem=problem) for reference in references)): return {'eligible': True, 'reason': 'math_equivalence_symbolic', 'verifier_mode': 'math_equivalence', 'answer_type': answer_type or 'symbolic', 'resolution': 'symbolic_reference'} return {'eligible': False, 'reason': 'math_reference_not_supported_by_symbolic_verifier', 'verifier_mode': mode, 'answer_type': answer_type, 'resolution': 'rejected'} class ContractError(ValueError): pass def canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')) def sha256_json(value: Any) -> str: return hashlib.sha256(canonical_json(value).encode('utf-8')).hexdigest() def sha256_text(value: str) -> str: return hashlib.sha256(value.encode('utf-8')).hexdigest() def validate_request(record: Mapping[str, Any]) -> dict[str, Any]: if not isinstance(record, Mapping): raise ContractError('request must be a JSON object') sample_id = str(record.get('sample_id', '')).strip() task_type = str(record.get('task_type', '')).strip().lower() problem = record.get('problem') response = record.get('response') if not sample_id: raise ContractError('sample_id is required') if task_type not in TASK_TYPES: raise ContractError(f'unsupported task_type: {task_type}') if not isinstance(problem, str) or not problem.strip(): raise ContractError('problem must be non-empty text') if not isinstance(response, str): raise ContractError('response must be text') verifier_mode = str(record.get('verifier_mode') or ('code' if task_type == 'code' else 'exact' if task_type == 'choice' else 'semantic' if task_type == 'general' else 'auto')).lower() source_dataset = str(record.get('source_dataset') or '').strip().casefold() effective_drop = verifier_mode == 'drop_official' or (verifier_mode in {'exact', 'exact_text', 'text'} and (source_dataset == 'drop' or source_dataset.startswith('drop_'))) if (task_type == 'code') != (verifier_mode in CODE_VERIFIER_MODES): raise ContractError('code_task_verifier_mode_mismatch') if task_type != 'code' and (not (any((key in record for key in ('reference_answer', 'reference_answers', 'rubric'))) or (effective_drop and bool(drop_answer_sets(record))))): raise ContractError('non-code verification requires reference_answer, reference_answers, or rubric') if task_type == 'code' and (not any((key in record for key in ('tests', 'verifier_payload', 'tool_verdict')))): raise ContractError('code verification requires tests, verifier_payload, or a bound tool_verdict') if task_type == 'code': executable = executable_code_payload_contract(record.get('tests', record.get('verifier_payload')), language=record.get('language')) if not executable['eligible']: raise ContractError(str(executable['reason'])) bound = bound_tool_verdict_contract(record.get('tool_verdict')) if not bound['eligible']: raise ContractError(str(bound['reason'])) value = dict(record) value['sample_id'] = sample_id value['task_type'] = task_type value['problem'] = problem.strip() value['response'] = response value['verifier_mode'] = verifier_mode return value def _unwrap_latex_command(value: str, command: str) -> str: prefix = f'\\{command}{{' text = value.strip() if not text.startswith(prefix) or not text.endswith('}'): return text depth = 0 for index, character in enumerate(text[len(prefix) - 1:], len(prefix) - 1): if character == '{': depth += 1 elif character == '}': depth -= 1 if depth == 0 and index != len(text) - 1: return text return text[len(prefix):-1].strip() if depth == 0 else text def normalize_answer_text(value: Any) -> str: text = unicodedata.normalize('NFKC', str(value)).strip() text = re.sub('^\\s*(?:answer|答案|final\\s+answer)\\s*[::]\\s*', '', text, flags=re.I) for _ in range(8): previous = text text = text.strip().rstrip('。.;;').strip() if len(text) > 4 and text.startswith('$$') and text.endswith('$$'): text = text[2:-2].strip() elif len(text) > 2 and text.startswith('$') and text.endswith('$'): text = text[1:-1].strip() elif len(text) > 4 and text.startswith('\\(') and text.endswith('\\)'): text = text[2:-2].strip() elif len(text) > 4 and text.startswith('\\[') and text.endswith('\\]'): text = text[2:-2].strip() for command in ('boxed', 'fbox', 'text', 'mbox', 'mathrm'): text = _unwrap_latex_command(text, command) if text == previous: break text = text.replace('\\left', '').replace('\\right', '') text = text.replace('\\{', '{').replace('\\}', '}') text = text.replace('−', '-').replace('–', '-') text = re.sub('^([+-]?\\d+)\\s+(\\d+\\s*/\\s*\\d+)(\\\\?%)?$', '\\1又\\2\\3', text) text = re.sub('\\s+', '', text) text = _normalize_symbolic_number_format(text) text = text.rstrip('。.;;') return text.casefold() def _normalize_symbolic_number_format(text: str) -> str: text = re.sub('^\\\\\\\\?\\$(?=[+-]?(?:\\d|\\.\\d))', '', text) compact = re.sub('\\s+', '', text) if re.fullmatch('[+-]?\\d{1,3}(?:(?:,|\\{,\\}|,\\\\[!,; ])\\d{3})+(?:\\.\\d+)?(?:\\\\?%)?', compact): text = re.sub('\\{,\\}|,\\\\[!,; ]|,', '', compact) return text def symbolic_math_text(value: Any) -> str: value = re.sub('\\\\(?:displaystyle|textstyle|scriptstyle|scriptscriptstyle|[Bb]igg?[lrm]?)(?![A-Za-z])', '', str(value)) value = re.sub('\\\\sqrt\\s*([0-9A-Za-z])(?![A-Za-z])', '\\\\sqrt{\\1}', value) value = re.sub('\\\\log_\\s*(?:\\{([^{}]+)\\}|([0-9A-Za-z]))\\s*(?:\\{([^{}]+)\\}|\\(([^()]*)\\)|([0-9A-Za-z]+))', lambda m: '\\log(' + next((v for v in m.groups()[2:] if v is not None)) + ')/\\log(' + (m.group(1) or m.group(2)) + ')', value) text = normalize_answer_text(value) text = text.replace('\\dfrac', '\\frac').replace('\\tfrac', '\\frac') text = re.sub('\\\\frac\\s*\\{([^{}]+)\\}\\s*([0-9A-Za-z])', '\\\\frac{\\1}{\\2}', text) text = re.sub('\\\\frac\\s*([0-9A-Za-z])\\s*\\{([^{}]+)\\}', '\\\\frac{\\1}{\\2}', text) text = re.sub('\\\\frac\\s*([0-9A-Za-z])\\s*([0-9A-Za-z])', '\\\\frac{\\1}{\\2}', text) text = re.sub('\\\\sqrt\\s*([0-9A-Za-z])', '\\\\sqrt{\\1}', text) text = re.sub('(?<=[0-9A-Za-z})])(?:\\^\\{?\\\\circ\\}?|°)', '', text) text = re.sub('(?<=[0-9})])\\\\(?:text|mbox|mathrm)\\{[A-Za-z\\\\.\\s]{1,32}\\}\\s*$', '', text) text = re.sub('([+-]?(?:\\d+(?:\\.\\d*)?|\\.\\d+))\\\\?%', '(\\1/100)', text) text = re.sub('\\\\(?:quad|qquad|enspace|thinspace)', '', text) text = re.sub('\\\\[ ,;!]', '', text) text = text.replace('\\cdot', '*').replace('\\times', '*') text = text.replace('\\pi', 'pi').replace('\\infty', 'oo') text = text.replace('\\leq', '<=').replace('\\le', '<=') text = text.replace('\\geq', '>=').replace('\\ge', '>=') text = text.replace('\\cup', '|') text = text.replace('\\lfloor', 'floor(').replace('\\rfloor', ')') for name in ('sin', 'cos', 'tan', 'csc', 'arccos', 'arctan', 'log', 'ln', 'exp', 'alpha', 'beta', 'gamma', 'delta', 'theta', 'lambda', 'mu', 'phi', 'psi', 'omega', 'rho'): text = text.replace(f'\\{name}', name) text = re.sub('\\\\(?:operatorname|mathrm|text|mbox)\\{([^{}]*)\\}', '\\1', text) text = re.sub('_\\{([^{}]+)\\}', '_\\1', text) text = re.sub('\\^\\{([^{}]+)\\}', '^(\\1)', text) text = re.sub('log_(\\d+|[a-z])\\(([^()]*)\\)', '(log(\\2)/log(\\1))', text) text = re.sub('log_(\\d+|[a-z])\\{([^{}]*)\\}', '(log(\\2)/log(\\1))', text) fraction = re.compile('\\\\frac\\{([^{}]+)\\}\\{([^{}]+)\\}') square_root = re.compile('\\\\sqrt\\{([^{}]+)\\}') indexed_root = re.compile('\\\\sqrt\\[([^\\[\\]]+)\\]\\{([^{}]+)\\}') for _ in range(32): updated = indexed_root.sub('root(\\2,\\1)', text) updated = square_root.sub('sqrt(\\1)', updated) updated = fraction.sub('((\\1)/(\\2))', updated) if updated == text: break text = updated factorial = re.compile('(?P(?:\\d+|[a-zA-Z_]\\w*|\\([^()]+\\)))!') for _ in range(8): updated = factorial.sub('factorial(\\g)', text) if updated == text: break text = updated text = text.replace('{', '(').replace('}', ')') text = re.sub('(?<=[0-9a-zA-Z_)])(?=(?:sqrt|factorial|pi|oo)(?:\\b|\\())', '*', text) text = re.sub('(?<=\\))(?=[0-9a-zA-Z])', '*', text) text = re.sub('\\b(sin|cos|tan|csc|arccos|arctan|log|ln)(?=\\d)', '\\1*', text) return text def split_symbolic_top_level(text: str) -> list[str] | None: fields: list[str] = [] start = 0 stack: list[str] = [] pairs = {')': '(', ']': '[', '}': '{'} for index, character in enumerate(text): if character in '([{': stack.append(character) elif character in ')]}': if not stack or stack[-1] != pairs[character]: return None stack.pop() elif character in ',;' and (not stack): fields.append(text[start:index].strip()) start = index + 1 if stack: return None fields.append(text[start:].strip()) return fields if len(fields) > 1 and all(fields) else None def symbolic_solution_context(problem: str) -> bool: return bool(re.search('\\b(?:solve|solutions?|roots?|all\\s+(?:(?:real|complex|possible|integer)\\s+)*values)\\b|\\b(?:what|which|find)\\s+(?:(?:real|complex|possible|integer)\\s+)*values\\s+of\\b|解集|所有.{0,12}(?:解|根|值)|求.{0,12}(?:解|根)', problem, re.I)) and (not re.search('\\b(?:uncertainty|measurement|tolerance)\\b|margin of error|误差|测量', problem, re.I)) def symbolic_structured_values(value: Any, *, problem: str='', interval: bool=False) -> tuple[str, list[str]] | None: text = _normalize_symbolic_number_format(_unwrap_complete_latex_reference(str(value))) text = text.replace('\\left', '').replace('\\right', '') chain = re.fullmatch('\\s*(.+?)\\s*(<=|>=|\\\\leq?|\\\\geq?|[<>≤≥])\\s*(?:[A-Za-z]|\\\\[A-Za-z]+)\\s*(<=|>=|\\\\leq?|\\\\geq?|[<>≤≥])\\s*(.+?)\\s*', text) if chain and numeric_value(chain[1]) is not None and (numeric_value(chain[4]) is not None): lower, left, right, upper = chain.groups() left, right = [re.sub('\\\\leq?', '≤', re.sub('\\\\geq?', '≥', operator)) for operator in (left, right)] if left in {'<', '<=', '≤'} and right in {'<', '<=', '≤'}: return ('interval:' + ('(' if left == '<' else '[') + (')' if right == '<' else ']'), [lower, upper]) if left in {'>', '>=', '≥'} and right in {'>', '>=', '≥'}: return ('interval:' + ('(' if right == '>' else '[') + (')' if left == '>' else ']'), [upper, lower]) membership = re.fullmatch('\\s*(?:[A-Za-z](?:_\\{?[A-Za-z0-9]+\\}?)?|[A-Za-z]\\s*/\\s*[A-Za-z]|\\\\frac\\{[A-Za-z]\\}\\{[A-Za-z]\\})\\s*(?:\\\\in|∈)\\s*(.+)', text) if membership: text = membership.group(1) interval = text.startswith(('(', '[')) text = re.sub('\\\\{2,}\\s*', '', text) text = re.sub('\\\\(?:text|mathrm|mbox)\\{\\s*(?:or|或)\\s*\\}', ',', text, flags=re.I) text = re.sub('(?= 3 and text[0] in '([' and (text[-1] in ')]'): bounds = text[0] + text[-1] if interval or bounds in {'(]', '[)'} or re.search('\\binterval(?:s| notation)?\\b|区间', problem, re.I): fields = split_symbolic_top_level(text[1:-1]) if fields and len(fields) == 2: return ('interval:' + bounds, fields) for left, right, kind in (('(', ')', 'sequence'), ('[', ']', 'sequence'), ('\\{', '\\}', 'set'), ('{', '}', 'set')): if text.startswith(left) and text.endswith(right): body = text[len(left):-len(right)].strip() fields = split_symbolic_top_level(body) if fields is None and kind == 'set' and body and (',' not in body) and (';' not in body): fields = [body] if fields: if kind == 'set': expanded = [] for field in fields: token = '\\pm' if field.count('\\pm') == 1 else '±' if field.count(token) == 1 and '\\mp' not in field and ('∓' not in field): expanded.extend((field.replace(token, '+'), field.replace(token, '-'))) else: expanded.append(field) fields = expanded return (kind, fields) token = '\\pm' if text.count('\\pm') == 1 else '±' if solution_context and text.count(token) == 1 and ('\\mp' not in text) and ('∓' not in text): return ('set', [text.replace(token, '+'), text.replace(token, '-')]) fields = split_symbolic_top_level(text) return ('set', fields) if fields else None def symbolic_matrix_values(value: Any) -> list[list[str]] | None: text = _unwrap_complete_latex_reference(str(value)) text = text.replace('\\left', '').replace('\\right', '').strip() matched = re.fullmatch('\\\\begin\\{(?P[pbvBV]?matrix)\\}(?P[\\s\\S]*)\\\\end\\{(?P=environment)\\}', text) if matched is None: return None rows = re.split('\\\\\\\\(?:\\[[^\\]]*\\])?', matched.group('body')) matrix = [[cell.strip() for cell in row.split('&')] for row in rows] if not matrix or len(matrix) > 64 or (not matrix[0]) or (len(matrix[0]) > 64) or any((len(row) != len(matrix[0]) for row in matrix)) or any((not cell for row in matrix for cell in row)): return None return matrix def _symbolic_scalar_reference_is_supported(value: Any) -> bool: if numeric_value(value) is not None: return True text = symbolic_math_text(value) if not text or len(text) > 4096: return False if not re.fullmatch('[\\w\\s+\\-*/^().,=<>\\[\\]_:|]+', text, flags=re.UNICODE): return False if text.count('=') > 1: return False balanced_text = '' if '|' in text else text if balanced_text and len(text) >= 3 and (text[0] in '([') and (text[-1] in ')]') and (',' in text): balanced_text = text[1:-1] depth = 0 for character in balanced_text: if character == '(': depth += 1 elif character == ')': depth -= 1 if depth < 0: return False if depth: return False names = set(re.findall('[^\\W\\d]\\w*', text, flags=re.UNICODE)) functions = {'sqrt', 'sin', 'cos', 'tan', 'csc', 'arccos', 'arctan', 'log', 'ln', 'exp', 'abs', 'factorial', 'floor', 'root', 'pi', 'e', 'i', 'oo'} named_symbols = {'alpha', 'beta', 'gamma', 'delta', 'theta', 'lambda', 'mu', 'phi', 'psi', 'omega', 'rho'} return not any((name not in functions and len(name) > 2 and (name not in named_symbols) and (not ('_' in name and re.fullmatch('(?:[A-Za-z](?:_[A-Za-z0-9]{1,2})?)+', name))) for name in names)) def symbolic_reference_is_supported(value: Any, *, problem: str='') -> bool: matrix = symbolic_matrix_values(value) if matrix is not None: return all((_symbolic_scalar_reference_is_supported(cell) for row in matrix for cell in row)) structured = symbolic_structured_values(value, problem=problem) if structured is None: return _symbolic_scalar_reference_is_supported(value) return all((symbolic_structured_values(field) is None and _symbolic_scalar_reference_is_supported(field) for field in structured[1])) def _drop_number(value: str) -> str | None: text = unicodedata.normalize('NFKC', value).replace(',', '').strip() text = text.replace('−', '-').replace('–', '-') try: number = Decimal(text) except InvalidOperation: return None normalized = format(number.normalize(), 'f') if '.' in normalized: normalized = normalized.rstrip('0').rstrip('.') return '0' if normalized in {'-0', ''} else normalized def _drop_normalize_token(value: str) -> str: token = unicodedata.normalize('NFKC', value).lower().strip() token = token.replace('\\%', '%') if token.endswith('%') and _drop_number(token[:-1]) is not None: token = token[:-1] number = _drop_number(token) if number is not None: return number token = ''.join((character for character in token if character not in string.punctuation)) number = _drop_number(token) if number is not None: return number return '' if token in {'a', 'an', 'the'} else token def drop_normalize_answer(value: Any) -> str: text = unicodedata.normalize('NFKC', str(value)).lower().strip() for _ in range(4): normalized = re.sub('\\\\(?:text|mathrm|operatorname|mbox)\\{([^{}]*)\\}', '\\1', text) if normalized == text: break text = normalized raw_tokens = re.split('\\s+|(?<=[A-Za-z])-(?=[A-Za-z])', text) tokens = [_drop_normalize_token(token) for token in raw_tokens] return ' '.join((token for token in tokens if token)) def _drop_spans(value: Any) -> list[str]: if isinstance(value, Mapping): try: return drop_answer_json_to_strings(value)[0] except ContractError: return [] if isinstance(value, (list, tuple)): return [str(item).strip() for item in value if str(item).strip()] text = str(value).strip() if text.startswith(('[', '(')) and text.endswith((']', ')')): try: parsed = ast.literal_eval(text) except (SyntaxError, ValueError): parsed = None if isinstance(parsed, (list, tuple)): return [str(item).strip() for item in parsed if str(item).strip()] fields = [field.strip() for field in re.split('\\s*(?:;|\\|)\\s*', text) if field.strip()] return fields or ([text] if text else []) def drop_answer_json_to_strings(value: Mapping[str, Any]) -> tuple[list[str], str]: number = value.get('number') if number is not None and str(number).strip(): return ([str(number).strip()], 'number') spans = value.get('spans') if isinstance(spans, (list, tuple)) and any((str(item).strip() for item in spans)): normalized = [str(item).strip() for item in spans if str(item).strip()] return (normalized, 'span' if len(normalized) == 1 else 'spans') date = value.get('date') if isinstance(date, Mapping): fields = [str(date.get(key) or '').strip() for key in ('day', 'month', 'year')] joined = ' '.join((field for field in fields if field)) if joined: return ([joined], 'date') raise ContractError('DROP answer lacks number, spans, or date') def _parse_drop_object(value: Any) -> Any: parsed = value if isinstance(parsed, str) and parsed.strip().startswith('{'): try: parsed = json.loads(parsed) except json.JSONDecodeError: try: parsed = ast.literal_eval(parsed) except (SyntaxError, ValueError): parsed = value return parsed def _drop_answer_set(value: Any) -> list[str] | None: parsed = _parse_drop_object(value) if isinstance(parsed, Mapping): nested = parsed.get('answer_sets') if isinstance(nested, list) and len(nested) == 1: return _drop_answer_set(nested[0]) try: return drop_answer_json_to_strings(parsed)[0] except ContractError: return None return None def _drop_answer_sets_from_object(value: Any) -> list[list[str]]: parsed = _parse_drop_object(value) if not isinstance(parsed, Mapping): return [] nested = parsed.get('answer_sets') if isinstance(nested, list): answer_sets = [item for item in (_drop_answer_set(value) for value in nested) if item] if answer_sets: return answer_sets answers = [] primary = parsed.get('answer') if isinstance(parsed.get('answer'), Mapping) else parsed primary_set = _drop_answer_set(primary) if primary_set: answers.append(primary_set) validated = parsed.get('validated_answers') if isinstance(validated, list): answers.extend((item for item in (_drop_answer_set(value) for value in validated) if item)) return answers def drop_answer_sets(record: Mapping[str, Any]) -> list[list[str]]: explicit = record.get('drop_answer_sets') if isinstance(explicit, list) and explicit: answer_sets = [] for value in explicit: spans = _drop_answer_set(value) if spans is None and isinstance(value, (list, tuple)): spans = _drop_spans(value) if spans: answer_sets.append(spans) if answer_sets: return answer_sets for value in (record.get('verifier_payload'), record.get('reference_answer')): answer_sets = _drop_answer_sets_from_object(value) if answer_sets: return answer_sets references = record.get('reference_answers') if isinstance(references, list) and references: parsed_sets = [] for value in references: parsed_sets.extend(_drop_answer_sets_from_object(value)) parsed = _drop_answer_set(value) if parsed: parsed_sets.append(parsed) if parsed_sets: return parsed_sets return [spans for value in references if (spans := _drop_spans(value))] reference = record.get('reference_answer') parsed = _drop_answer_set(reference) if parsed: return [parsed] if reference is not None and str(reference).strip(): return [_drop_spans(reference)] return [] def _latex_fraction(value: str) -> str: match = re.fullmatch('[+-]?\\\\(?:[dt]?frac)\\{([^{}]+)\\}\\{([^{}]+)\\}', value) if not match: return value sign = '-' if value.startswith('-') else '' numerator = match.group(1) denominator = match.group(2) return f'{sign}{numerator}/{denominator}' def numeric_value(value: Any) -> Fraction | None: raw = _unwrap_complete_latex_reference(str(value)) text = normalize_answer_text(raw) percent = text.endswith(('%', '\\%')) text = re.sub('(?:\\\\%|%)$', '', text) mixed = re.fullmatch('([+-]?)(\\d+)(?:\\\\[dt]?frac\\{(\\d+)\\}\\{(\\d+)\\}|又(\\d+)/(\\d+))', text) if mixed: sign, whole, numerator, denominator, other_numerator, other_denominator = mixed.groups() denominator = int(denominator or other_denominator) if denominator == 0: return None number = Fraction(int(whole)) + Fraction(int(numerator or other_numerator), denominator) number = -number if sign == '-' else number return number / 100 if percent else number text = _latex_fraction(text) def bounded_fraction(raw: str) -> Fraction | None: if '_' in raw: return None decimal = Decimal(raw) if not decimal.is_finite(): return None digits = decimal.as_tuple().digits exponent = decimal.as_tuple().exponent if abs(exponent) > 10000 or len(digits) + max(exponent, 0) > 10000: return None return Fraction(decimal) try: if '/' in text and text.count('/') == 1: numerator, denominator = text.split('/', 1) numerator_value = bounded_fraction(numerator) denominator_value = bounded_fraction(denominator) if numerator_value is None or denominator_value in {None, 0}: return None number = numerator_value / denominator_value else: number = bounded_fraction(text) if number is None: return None return number / 100 if percent else number except (InvalidOperation, ValueError, ZeroDivisionError, OverflowError): return None def numeric_display_value(value: Any) -> Fraction | None: number = numeric_value(value) if number is None: return None normalized = normalize_answer_text(value) return number * 100 if normalized.endswith(('%', '\\%')) else number def choice_value(value: Any) -> str | None: text = normalize_answer_text(value).upper() match = re.fullmatch('(?:OPTION)?[\\(\\[]?([A-Z])[\\)\\].:]?', text) return match.group(1) if match else None def choice_answer_value(value: Any, problem: str='', *, allow_explanation: bool=False) -> str | None: text = re.sub('\\s*[✅✔✓]\\ufe0f?\\s*$', '', str(value).strip()) text = _choice_display_text(text) prefix = '^(?:(?:the\\s+)?(?:(?:correct|final)\\s+)?(?:answer|option)\\s*(?:is\\s*[::]?|[::])\\s*|(?:最终)?答案(?:字母)?\\s*(?:为|是|[::])\\s*|选项\\s*)' explicit = bool(re.match(prefix, text, re.I)) text = re.sub(prefix, '', text, flags=re.I) text = _choice_display_text(text) standalone = re.match('^\\(([A-Z])\\)\\s*[.。](?:\\s|$)', text, re.I) if standalone: return standalone.group(1).upper() label = choice_value(text) if label is not None: return label options = choice_option_values(problem) matches = [label for label, body in (options or {}).items() if _choice_content_text(text) == body] if matches: return matches[0] if len(matches) == 1 else None labels = choice_set_value(text) available = set(options) if options else _choice_problem_labels(problem) if labels is None and re.fullmatch('\\(?[A-Z]\\)?(?:\\s*(?:or|and|/|或|和)\\s*\\(?[A-Z]\\)?)+', text, re.I): labels = frozenset(re.findall('\\b[A-Z]\\b', text.upper())) - {'OR', 'AND'} if labels is not None and labels.issubset(available): return ','.join(sorted(labels)) labelled = re.fullmatch('\\(?([A-Z])(?:\\)|[.、::])\\s*(.+)', text, re.I) if labelled: label, body = labelled.groups() label = label.upper() if _choice_content_text(body) == (options or {}).get(label): return label if choice_value(body) == label: return label if (explicit or allow_explanation) and label in available: if not re.match('\\s*(?:or|and|或者|或|和)\\s*\\(?[A-Z]\\b', body, re.I): return label return None annotated = re.fullmatch('([A-Z])\\s*[((]([^A-Za-z]*?)[))]', text) if annotated: return annotated.group(1) return None def _choice_display_text(value: str) -> str: text = re.sub('^\\s*>+\\s*', '', value) text = _unwrap_complete_latex_reference(_strip_answer_formatting(text)) for command in ('text', 'mbox', 'mathrm'): for start, end, body in reversed(_latex_command_value_matches(text, command)): text = text[:start] + body + text[end:] return _strip_answer_formatting(text) def _choice_content_text(value: str) -> str: text = re.sub('\\\\(?:quad|qquad|enspace|thinspace)|\\\\[ ,;!]', '', _choice_display_text(value)) text = normalize_answer_text(text) text = text.replace('\\times', '*').replace('\\cdot', '*') text = re.sub('\\^(?:\\{\\\\circ\\}|\\\\circ)', '°', text) for name, letter in (('alpha', 'α'), ('beta', 'β'), ('gamma', 'γ'), ('delta', 'δ'), ('theta', 'θ'), ('lambda', 'λ'), ('mu', 'μ'), ('pi', 'π'), ('sigma', 'σ'), ('phi', 'φ'), ('omega', 'ω')): text = re.sub('\\\\' + name + '(?![a-z])', letter, text) if re.fullmatch('\\([a-z -]+\\)', text): text = text[1:-1] text = re.sub('\\^\\{([^{}]+)\\}', '^\\1', text) return re.sub('_(?:\\{(\\d+)\\}|(\\d+))', '\\1\\2', text) def set_value(value: Any, *, allow_sequence_brackets: bool=False) -> tuple[str, ...] | None: text = normalize_answer_text(value) if text.startswith('\\{') and text.endswith('\\}'): text = '{' + text[2:-2] + '}' accepted = {('{', '}')} if allow_sequence_brackets: accepted.add(('[', ']')) if len(text) < 2 or (text[0], text[-1]) not in accepted: return None fields = [field for field in re.split('[,,]', text[1:-1]) if field] if not fields: return None normalized = [] for field in fields: number = numeric_value(field) normalized.append(str(number) if number is not None else normalize_answer_text(field)) return tuple(sorted(normalized)) def reference_alternatives(record: Mapping[str, Any]) -> list[Any]: values = record.get('reference_answers') if values is not None: if not isinstance(values, list) or not values: raise ContractError('reference_answers must be a non-empty list') return values if 'reference_answer' in record: return [record['reference_answer']] return [] def code_blocks(response: str) -> list[dict[str, Any]]: blocks = [] pattern = re.compile('```([^\\n`]*)\\n([\\s\\S]*?)```', re.MULTILINE) for index, match in enumerate(pattern.finditer(response)): blocks.append({'index': index, 'language': match.group(1).strip().lower(), 'code': match.group(2).strip(), 'start': match.start(2), 'end': match.end(2)}) return blocks def _latex_brace_is_escaped(value: str, index: int) -> bool: backslashes = 0 cursor = index - 1 while cursor >= 0 and value[cursor] == '\\': backslashes += 1 cursor -= 1 return backslashes % 2 == 1 def _latex_command_value_matches(value: str, command: str) -> list[tuple[int, int, str]]: prefix = f'\\{command}{{' values: list[tuple[int, int, str]] = [] cursor = 0 while True: marker = '(?|<\\|im_end\\|>|\\n|$)' _STRONG_FINAL_ANSWER_PATTERN = re.compile('(?:final\\s+(?:answer|result)|最终答案|最终结果)\\s*(?:\\*\\*|__|`)?\\s*(?:is\\s*|equals?\\s*|=\\s*|[::]\\s*|为\\s*|是\\s*)(?:\\*\\*|__|`)?\\s*([^\\n]*?)' + _NEXT_ANSWER_DECLARATION, re.I) _TAGGED_ANSWER_PATTERN = re.compile('<(?:answer|final_answer)>\\s*([\\s\\S]*?)\\s*', re.I) _DECLARED_ANSWER_PATTERN = re.compile('(?:final\\s+(?:answer|result)|answer|最终答案|最终结果|答案字母|答案|结果)\\s*(?:\\*\\*|__|`)?\\s*(?:is\\s*|equals?\\s*|=\\s*|[::]\\s*|为\\s*|是\\s*)(?:\\*\\*|__|`)?\\s*([^\\n]*?)' + _NEXT_ANSWER_DECLARATION, re.I) def _non_overlapping_generic_answer_matches(response: str, strong_matches: list[re.Match[str]]) -> list[re.Match[str]]: generic = [] for match in _DECLARED_ANSWER_PATTERN.finditer(response): if any((match.start() >= strong.start() and match.end() <= strong.end() for strong in strong_matches)): continue generic.append(match) return generic def _last_balanced_latex_command_value(value: str, command: str) -> str | None: prefix = f'\\{command}{{' starts = [match.start() for match in re.finditer(re.escape(prefix), value)] for start in reversed(starts): body_start = start + len(prefix) depth = 1 for index in range(body_start, len(value)): character = value[index] if character == '{' and (not _latex_brace_is_escaped(value, index)): depth += 1 elif character == '}' and (not _latex_brace_is_escaped(value, index)): depth -= 1 if depth == 0: candidate = value[body_start:index].strip() if candidate: return candidate break return None def _answer_declarations(response: str, allow_hash_answer: bool=False) -> list[dict[str, Any]]: declarations = [] for command in ('boxed', 'fbox'): declarations.extend((dict(start=start, end=end, candidate_answer=body, method=f'last_latex_{command}', strong=False) for start, end, body in _latex_command_value_matches(response, command) if body)) strong_matches = sorted([*_STRONG_FINAL_ANSWER_PATTERN.finditer(response), *_TAGGED_ANSWER_PATTERN.finditer(response)], key=lambda match: match.start()) generic_matches = _non_overlapping_generic_answer_matches(response, strong_matches) headings = list(re.finditer('(?:final\\s+line|答案字母)\\s*[::]?\\s*\\n?\\s*([^\\n]+)', response, re.I)) for match in [*strong_matches, *generic_matches, *headings]: if not match.group(1).strip() or any((d['start'] <= match.start() < d['end'] for d in declarations)): continue declarations.append(dict(start=match.start(), end=match.end(), candidate_answer=match.group(1).strip(), method='last_final_answer_marker', strong=match in strong_matches or match in headings)) if allow_hash_answer: declarations.extend((dict(start=match.start(), end=match.end(), candidate_answer=match.group(1).strip(), method='last_hash_answer', strong=numeric_value(match.group(1)) is not None or symbolic_reference_is_supported(match.group(1))) for match in re.finditer('(?:^|\\n)\\s*####\\s*([^\\n]+)', response) if match.group(1).strip() and (not re.match('(?:[a-z][.)]\\s|(?:step|case|part|explanation|proof|analysis)\\b)', match.group(1).strip(), re.I)))) for declaration in declarations: nested = [d for d in declarations if declaration['start'] < d['start'] and d['end'] <= declaration['end']] if nested and len(nested) == 1: declaration['candidate_answer'] = nested[0]['candidate_answer'] return sorted(declarations, key=lambda d: d['start']) def extract_declared_answer(response: str, *, allow_hash_answer: bool=False) -> dict[str, Any] | None: declarations = _answer_declarations(response, allow_hash_answer) if declarations: return {**declarations[-1], 'confidence': 'high'} for command in ('boxed', 'fbox'): recovered = _last_balanced_latex_command_value(response, command) if recovered: return {'candidate_answer': recovered, 'confidence': 'high', 'method': f'last_latex_{command}_after_malformed_prefix'} return None def candidate_type(value: Any) -> str: text = str(value).strip() if choice_value(text) is not None: return 'choice' if text.endswith(('%', '\\%')) and numeric_value(text) is not None: return 'percent' if '/' in text or 'frac' in text: return 'fraction' if numeric_value(text) is not None else 'text' if numeric_value(text) is not None: return 'number' if len(_drop_spans(text)) > 1: return 'spans' return 'text' def _strip_answer_formatting(value: str) -> str: text = value.strip().lstrip('::').strip().rstrip('。.;;').rstrip() text = re.sub('^(?:\\*\\*|__|`)+\\s*', '', text) return re.sub('\\s*(?:\\*\\*|__|`)+$', '', text).strip().rstrip('。.;;').rstrip() def external_request(record: Mapping[str, Any], *, candidate: Any, language: str | None=None) -> dict[str, Any]: return {'schema': EXTERNAL_REQUEST_SCHEMA, 'sample_id': record['sample_id'], 'task_type': record['task_type'], 'verifier_mode': record['verifier_mode'], 'problem': record['problem'], 'candidate': candidate, 'language': language, 'reference_answer': record.get('reference_answer'), 'reference_answers': record.get('reference_answers'), 'tests': record['tests'] if record.get('tests') is not None else record.get('verifier_payload'), 'verifier_payload': record.get('verifier_payload')} def validate_external_result(value: Mapping[str, Any]) -> dict[str, Any]: if not isinstance(value, Mapping): raise ContractError('external verifier output must be a JSON object') verdict = str(value.get('verdict', '')).upper() if verdict not in VERDICTS: raise ContractError('external verifier returned an invalid verdict') return {'verdict': verdict, 'details': value.get('details'), 'raw': dict(value)} def compare_answers(candidate: Any, references: Sequence[Any], *, answer_type: str='auto', problem: str='', numeric_scale: str='', reference_unit: str='') -> dict[str, Any]: mode = answer_type.lower() if mode == 'exact_choice': mode = 'choice' if mode == 'numeric_equivalence': mode = 'numeric' if reference_unit and mode in {'numeric', 'integer', 'decimal', 'fraction'}: candidate = problem_numeric_answer(str(candidate), 'in ' + reference_unit) references = [value for value in references if str(value).strip()] if candidate is None or not str(candidate).strip() or (not references): return {'verdict': 'UNCERTAIN', 'method': 'missing_candidate_or_reference'} candidate_text = normalize_answer_text(candidate) normalized_references = [normalize_answer_text(value) for value in references] if mode in {'exact', 'exact_text', 'text'} and all((re.fullmatch('CODE-[0-9A-F]{12}', str(v).strip()) for v in references)): code = _unwrap_complete_latex_reference(_strip_answer_formatting(str(candidate))) return {'verdict': 'PASS' if code in references else 'FAIL', 'method': 'complete_exact_access_code', 'candidate': code} if candidate_text in normalized_references: return {'verdict': 'PASS', 'method': 'normalized_exact', 'candidate': candidate_text, 'references': normalized_references} option_content = compare_choice_option_content(candidate, references, problem) if option_content is not None: return option_content if mode == 'drop_exact': left = drop_normalize_answer(candidate) comparable = [drop_normalize_answer(value) for value in references] return {'verdict': 'PASS' if left in comparable else 'FAIL', 'method': 'drop_official_exact', 'candidate': left, 'references': comparable} if mode in {'choice', 'auto'}: selection = choice_answer_value(candidate, problem) left = choice_value(selection or candidate) right = [choice_value(value) for value in references] comparable = [value for value in right if value is not None] if left is not None and comparable: return {'verdict': 'PASS' if left in comparable else 'FAIL', 'method': 'choice', 'candidate': left, 'references': comparable} multiple = choice_set_value(selection) if selection else None if mode == 'choice' and multiple is not None and comparable: return {'verdict': 'FAIL', 'method': 'multiple_answers_in_single_choice', 'candidate': sorted(multiple), 'references': comparable} if mode == 'choice' and comparable: options = choice_option_values(problem) or {} number = numeric_value(candidate) numbers = {label: numeric_value(body) for label, body in options.items()} if number is not None and numbers and all((value is not None for value in numbers.values())): matches = [label for label, value in numbers.items() if value == number] return {'verdict': 'PASS' if len(matches) == 1 and matches[0] in comparable else 'FAIL' if len(matches) <= 1 else 'UNCERTAIN', 'method': 'numeric_content_against_all_options', 'candidate': str(number), 'matching_options': matches} if options and symbolic_reference_is_supported(candidate) and re.search('[0-9\\\\+*/^]', str(candidate)): from evaluation.math_verifier import symbolic_equivalence comparisons = {label: symbolic_equivalence(candidate, body) for label, body in options.items()} matches = [label for label, result in comparisons.items() if result['verdict'] == 'PASS'] uncertain = any((result['verdict'] == 'UNCERTAIN' for result in comparisons.values())) verdict = 'UNCERTAIN' if uncertain or len(matches) > 1 else 'PASS' if matches and matches[0] in comparable else 'FAIL' return dict(verdict=verdict, method='symbolic_content_against_all_options', matching_options=matches) if re.search('none of (?:the |these )?(?:options|choices)|no (?:listed )?(?:option|choice) matches|选项均不|没有正确选项', str(candidate), re.I): return {'verdict': 'FAIL', 'method': 'no_listed_option_submitted'} if re.fullmatch('[\\sA-Z()\\[\\],、,/或和及]+', str(candidate)) and len(set(re.findall('[A-Z]', str(candidate)))) > 1: return {'verdict': 'FAIL', 'method': 'multiple_answers_in_single_choice'} if mode == 'choice_set': left = choice_set_value(candidate) right = [choice_set_value(value) for value in references] comparable = [value for value in right if value is not None] if left is not None and comparable: return {'verdict': 'PASS' if left in comparable else 'FAIL', 'method': 'unordered_choice_set', 'candidate': sorted(left), 'references': [sorted(value) for value in comparable]} return {'verdict': 'UNCERTAIN', 'method': 'choice_set_unparseable', 'candidate': candidate_text, 'references': normalized_references} if mode in {'integer', 'numeric', 'decimal', 'fraction', 'auto', 'math_equivalence', 'symbolic', 'expression'}: left_interval = symbolic_structured_values(candidate, problem=problem) if left_interval and left_interval[0].startswith('interval:'): right_intervals = [symbolic_structured_values(value, problem=problem, interval=True) for value in references] if all((value and value[0].startswith('interval:') and all((numeric_value(x) is not None for x in value[1])) for value in [left_interval, *right_intervals])): signature = lambda value: (value[0], tuple((numeric_value(x) for x in value[1]))) return {'verdict': 'PASS' if any((signature(left_interval) == signature(value) for value in right_intervals)) else 'FAIL', 'method': 'numeric_interval_equivalence', 'candidate': str(candidate), 'references': list(references)} if re.search('volume|面积|体积', problem, re.I) and re.search('\\\\(?:leq?|geq?)\\b|[<>]', str(candidate)): return {'verdict': 'FAIL', 'method': 'region_submitted_for_requested_quantity'} value_parser = numeric_display_value if numeric_scale.strip().lower() == 'display_percent' else numeric_value left = value_parser(candidate) right = [value_parser(value) for value in references] comparable = [value for value in right if value is not None] if left is not None and comparable and (mode not in {'math_equivalence', 'symbolic', 'expression'} or left in comparable or (all((value is not None for value in right)) and (numeric_scale or not any(('%' in str(value) for value in [candidate, *references]))))): return {'verdict': 'PASS' if left in comparable else 'FAIL', 'method': 'numeric_display_percent' if value_parser is numeric_display_value else 'numeric_fraction', 'candidate': str(left), 'references': [str(value) for value in comparable]} if mode in {'set', 'auto'}: allow_sequence_brackets = mode == 'set' left = set_value(candidate, allow_sequence_brackets=allow_sequence_brackets) right = [set_value(value, allow_sequence_brackets=allow_sequence_brackets) for value in references] comparable = [value for value in right if value is not None] if left is not None and comparable: return {'verdict': 'PASS' if left in comparable else 'FAIL', 'method': 'finite_set', 'candidate': list(left), 'references': [list(value) for value in comparable]} if mode in {'exact', 'exact_text', 'text'}: return {'verdict': 'FAIL', 'method': 'normalized_exact', 'candidate': candidate_text, 'references': normalized_references} return {'verdict': 'UNCERTAIN', 'method': 'symbolic_or_semantic_equivalence_required', 'candidate': candidate_text, 'references': normalized_references} def typed_declared_answer(response: str, *, answer_type: str, allow_hash_answer: bool=False, problem: str='', reference_unit: str='') -> dict[str, Any] | None: mode = answer_type.strip().lower() terminal = terminal_choice_answer(response, problem) if mode in {'choice', 'exact_choice'} else None declarations = _answer_declarations(response, allow_hash_answer) if terminal and (not any((d['start'] <= terminal['start'] and terminal['end'] <= d['end'] for d in declarations))): declarations.append(terminal) declarations.sort(key=lambda d: d['start']) declared = dict(declarations[-1], confidence='high') if declarations else extract_declared_answer(response, allow_hash_answer=allow_hash_answer) if declared is None and mode in {'choice', 'exact_choice'}: lines = [(m.start(), m.end(), m.group().strip()) for m in re.finditer('[^\\n]+', response) if m.group().strip()] boundaries = lines[:1] + lines[-1:] if len(lines) > 1 else lines literals = [] for start, end, line in boundaries: label = choice_answer_value(line, problem) if label: literals.append(dict(start=start, end=end, candidate_answer=label, method='boundary_choice_literal', strong=True)) if literals: declarations = literals declared = {**literals[-1], 'confidence': 'high'} elif numeric_value(response) is not None: declared = dict(candidate_answer=response, confidence='high', method='complete_numeric_option_content', start=0, end=len(response)) if declared is None and mode in {'exact', 'exact_text', 'text'}: literal = re.fullmatch('\\s*(?:\\{(CODE-[0-9A-F]{12})\\}|(CODE-[0-9A-F]{12}))\\s*', response) if literal: declared = dict(candidate_answer=literal.group(1) or literal.group(2), confidence='high', method='complete_access_code_literal', start=0, end=len(response)) if declared is None and mode in {'auto', 'numeric', 'integer', 'fraction', 'decimal', 'symbolic', 'expression', 'math_equivalence'}: if numeric_value(response) is not None or symbolic_reference_is_supported(response): declared = dict(candidate_answer=response, confidence='high', method='complete_numeric_literal', start=0, end=len(response)) if declared is None: return None candidate = _strip_answer_formatting(str(declared['candidate_answer'])) candidate = re.sub('^(?:(?:the\\s+)?(?:final\\s+)?answer\\s*(?:is\\s*|[::]\\s*)|(?:最终)?答案(?:字母)?\\s*(?:是|为|[::])\\s*)', '', candidate, flags=re.I) candidate = _strip_answer_formatting(candidate) if not candidate or len(candidate) > 4096: return None raw_candidate = candidate if mode in {'choice', 'exact_choice'}: candidate = choice_answer_value(candidate, problem, allow_explanation=declared['method'] == 'last_final_answer_marker') or candidate if mode in {'auto', 'numeric', 'integer', 'fraction', 'decimal'}: candidate = problem_numeric_answer(candidate, 'in ' + reference_unit if reference_unit else problem, response) normalizer = drop_normalize_answer if mode in {'drop_exact', 'drop_official'} else normalize_answer_text normalized = normalizer(candidate) strong = [d for d in declarations if d['strong']] scope = strong[0]['start'] if strong else declared.get('start', 0) if mode in {'exact', 'exact_text', 'text'} and declared.get('end') is not None: if not response[declared['end']:].strip(' $*_`().。;;\n\r\t'): paragraph = response.rfind('\n\n', 0, declared['start']) + 2 scope = max(scope, paragraph) if mode not in {'choice', 'exact_choice'} and declared.get('method', '').startswith('last_latex_'): if not response[declared.get('end', len(response)):].strip(' $*_`\\[]().。;;\n\r\t'): scope = declared.get('start', scope) boxes = [d for d in declarations if 'latex_' in d['method'] and d['start'] < scope] for prior in reversed(boxes): gap = response[prior['end']:scope].strip(' $*_`\\[]().。;;\n\r\t') if gap.casefold() not in {'', ',', 'and', 'or', '或', '和', '、'}: break scope = prior['start'] if mode in {'choice', 'exact_choice'} and (not strong) and declarations and (not any(('latex_' in d['method'] for d in declarations))): scope = declarations[0]['start'] if mode in {'choice', 'exact_choice'} and (not strong): boxes = [d for d in declarations if 'latex_' in d['method']] if len(boxes) > 1: scope = boxes[0]['start'] corrections = list(re.finditer('\\b(?:correction|after correction|correcting)\\b|更正|修正为|改为', response[:declared.get('start', 0)], re.I)) if corrections: scope = max(scope, corrections[-1].start()) competing = set() for item in declarations: if item['start'] < scope: continue value = _strip_answer_formatting(item['candidate_answer']) value = re.sub('^(?:the\\s+)?(?:final\\s+)?answer\\s*(?:is\\s*|[::]\\s*)', '', value, flags=re.I) if not value.strip('$*_` ::') or all((unicodedata.category(c)[0] in {'P', 'Z'} for c in value)): continue if mode in {'choice', 'exact_choice'}: label = choice_answer_value(value, problem, allow_explanation=item['method'] == 'last_final_answer_marker') if label is None and item != declarations[-1]: if not re.match('^\\(?[A-Z](?:\\)|[.、::]|$)', value): continue value = label or value if mode in {'auto', 'numeric', 'integer', 'fraction', 'decimal'}: value = problem_numeric_answer(value, 'in ' + reference_unit if reference_unit else problem, response) competing.add(normalizer(value)) competing.discard('') return {**declared, 'raw': raw_candidate, 'visible_answer': candidate, 'candidate_answer': candidate, 'candidate_type': 'text' if mode in {'exact', 'exact_text', 'text'} else candidate_type(candidate), 'normalized': normalized, 'extraction_method': declared['method'], 'candidate_sha256': sha256_text(candidate), 'visible_suffix_sha256': sha256_text(response), 'ambiguous': len(competing) > 1, 'competing_candidate_count': len(competing) or 1} def problem_numeric_answer(candidate: str, problem: str, response: str='') -> str: original = candidate candidate = re.sub('\\\\(?:text|mbox|mathrm)\\{([^{}]*)\\}', ' \\1 ', candidate) candidate = re.sub('\\\\[ ,;!]', ' ', candidate).strip() requested = re.search('how many\\s+(seconds?|minutes?|hours?)\\b', problem, re.I) if requested is None: requested = re.search('(?:in|total(?: number of)?)\\s+(seconds?|minutes?|hours?)\\b', problem, re.I) if requested: units = {'second': 1, 'minute': 60, 'hour': 3600} pieces = list(re.finditer('([0-9]+(?:\\.[0-9]+)?)\\s*(seconds?|minutes?|hours?)\\b', candidate, re.I)) rest = re.sub('([0-9]+(?:\\.[0-9]+)?)\\s*(seconds?|minutes?|hours?)\\b', '', candidate, flags=re.I) if pieces and (not re.sub('\\band\\b|[ ,.;]', '', rest, flags=re.I)): total = sum((Fraction(m.group(1)) * units[m.group(2).lower().rstrip('s')] for m in pieces)) return str(total / units[requested.group(1).lower().rstrip('s')]) total = re.search('(?:total(?: of)?|altogether|in total|共|总计)\\s*[::=]?\\s*([0-9]+(?:\\.[0-9]+)?)\\b', candidate, re.I) if total and re.search('total|altogether|how many|总|共|多少', problem, re.I): return total.group(1) parts = re.fullmatch('(\\d+)\\s+([A-Za-z]+)\\s+and\\s+(\\d+)\\s+([A-Za-z]+)', candidate) asked = re.search('how many\\s+([A-Za-z]+)\\s+and\\s+([A-Za-z]+)', problem, re.I) if parts and asked and ({parts[2].lower(), parts[4].lower()} == {asked[1].lower(), asked[2].lower()}): a, b = (int(parts[1]), int(parts[3])) return str(a + b) labelled = re.fullmatch('([+-]?[0-9]+(?:\\.[0-9]+)?)\\s+([A-Za-z]+)[.]?', candidate) if labelled and re.search('how many\\s+' + re.escape(labelled.group(2).rstrip('s')) + 's?\\b', problem, re.I): return labelled.group(1) return original def evaluation_answer_text(response) -> str: text = str(response or '') if '' in text: text = text.rsplit('', 1)[-1] elif text.lstrip().startswith(''): return '' return re.sub('(?:<\\|im_end\\|>)?\\s*$', '', text).strip() def terminal_choice_answer(response: str, problem: str='') -> dict[str, Any] | None: end = len(response.rstrip()) if not end: return None start = response.rfind('\n', 0, end) + 1 value = response[start:end].strip() if re.fullmatch('(?:give (?:only )?the answer on the final line|on the final line|最后一行给出答案字母)[.。]?', value, re.I): end = len(response[:start].rstrip()) start = response.rfind('\n', 0, end) + 1 value = response[start:end].strip() tagged = re.search('<(answer|final_answer)>\\s*([\\s\\S]*?)\\s*\\s*$', response, re.I) if tagged: start, value = (tagged.start(), tagged.group(2)) value = _choice_display_text(value) prefix = '^(?:(?:therefore|thus|hence)[,,]?\\s*)?(?:(?:the\\s+)?(?:(?:correct|final|best)\\s+)?(?:answer(?:\\s+letter)?(?:\\s+on\\s+the\\s+final\\s+line)?|final\\s+line\\s+answer\\s+letter|choice|option|statement|letter|conclusion)\\s*(?:(?:is|equals?)\\s*[::]?|[::])\\s*|(?:故选|因此选|所以选|应选|选|(?:最终)?答案(?:字母)?)\\s*(?:是|为|[::])?\\s*)' for _ in range(2): value = _choice_display_text(re.sub(prefix, '', value, flags=re.I)) label = choice_answer_value(value, problem) if label is None: endorsements = list(re.finditer('(?m)^\\s*(?:[-*]\\s*)?(?:\\*\\*)?([A-Z])\\s*[::]?\\s*(?:\\*\\*)?(?:正确|is correct)\\b', response)) labels = {m[1] for m in endorsements} & _choice_problem_labels(problem) if len(labels) != 1: return None label = next(iter(labels)) start = next((m.start() for m in reversed(endorsements) if m[1] == label)) return dict(start=start, end=end, candidate_answer=label, method='explicit_option_endorsement', strong=True) return dict(start=start, end=end, candidate_answer=label, method='terminal_submitted_choice', strong=True)