from __future__ import annotations import argparse import copy import importlib.util import json import os import string import sys import types from pathlib import Path from typing import Any, Mapping from scoring.verification import EXTERNAL_REQUEST_SCHEMA _EVALUATION_LIB_CACHE: dict[tuple[str, str, str], Any] = {} class VerificationError(ValueError): pass def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--ifbench-repo', type=Path, required=True) parser.add_argument('--ifeval-repo', type=Path, required=True) parser.add_argument('--nltk-data', type=Path, required=True) parser.add_argument('--jsonl-worker', action='store_true') return parser.parse_args() def _json_object(value: Any, *, name: str) -> dict[str, Any]: if isinstance(value, Mapping): return dict(value) if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError as exc: raise VerificationError(f'{name} is not valid JSON') from exc if isinstance(parsed, Mapping): return dict(parsed) raise VerificationError(f'{name} must be a JSON object') def _request_from_stdin() -> dict[str, Any]: try: value = json.loads(sys.stdin.read()) except json.JSONDecodeError as exc: raise VerificationError('stdin is not valid JSON') from exc if not isinstance(value, Mapping): raise VerificationError('request must be a JSON object') if value.get('schema') != EXTERNAL_REQUEST_SCHEMA: raise VerificationError('unsupported external verifier request schema') return dict(value) def _evaluation_lib_path(repo: Path) -> Path: path = repo.expanduser().resolve() if path.is_file(): if path.name != 'evaluation_lib.py': raise VerificationError(f'expected evaluation_lib.py, got {path}') return path candidates = (path / 'evaluation_lib.py', path / 'instruction_following_eval' / 'evaluation_lib.py', path / 'ifbench' / 'evaluation_lib.py', path / 'ifeval' / 'evaluation_lib.py') for candidate in candidates: if candidate.is_file(): return candidate raise VerificationError(f'evaluation_lib.py not found under {path}') def _load_evaluation_lib(mode: str, repo: Path) -> Any: module_path = _evaluation_lib_path(repo) cache_key = (mode, str(module_path.resolve())) cached = _EVALUATION_LIB_CACHE.get(cache_key) if cached is not None: return cached search_paths = [module_path.parent, module_path.parent.parent] for path in reversed(search_paths): value = str(path) if value not in sys.path: sys.path.insert(0, value) from langdetect import DetectorFactory DetectorFactory.seed = 0 package_name = f'_edgerock_{mode}_official' package = types.ModuleType(package_name) package.__path__ = [str(module_path.parent)] package.__package__ = package_name sys.modules[package_name] = package module_name = f'{package_name}.evaluation_lib' spec = importlib.util.spec_from_file_location(module_name, module_path) if spec is None or spec.loader is None: raise VerificationError(f'cannot import {module_path}') module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) for name in ('InputExample', 'test_instruction_following_strict'): if not hasattr(module, name): raise VerificationError(f'official evaluation_lib is missing {name}') _EVALUATION_LIB_CACHE[cache_key] = module return module def _normalize_kwarg(value: Any) -> Any: if isinstance(value, float) and value.is_integer(): return int(value) if isinstance(value, list): return [_normalize_kwarg(item) for item in value] if isinstance(value, Mapping): return {str(key): _normalize_kwarg(item) for key, item in value.items() if item is not None} return value def _literal_nonalpha_letter_frequency(candidate: str, kwargs: Mapping[str, Any]) -> bool | None: letter = kwargs.get('letter') frequency = kwargs.get('let_frequency') relation = kwargs.get('let_relation') if not isinstance(letter, str) or len(letter) != 1 or letter in string.ascii_letters or (not isinstance(frequency, int)) or isinstance(frequency, bool) or (frequency < 0): return None count = candidate.casefold().count(letter.casefold()) if relation == 'at least': return count >= frequency if relation == 'less than': return count < frequency return None def verify(request: Mapping[str, Any], args: argparse.Namespace) -> dict[str, Any]: mode = str(request.get('verifier_mode', '')).strip().lower() if mode not in {'ifbench', 'ifeval'}: raise VerificationError(f'unsupported verifier_mode: {mode}') problem = request.get('problem') candidate = request.get('candidate') if not isinstance(problem, str) or not problem.strip(): raise VerificationError('problem must contain the original prompt') if not isinstance(candidate, str): raise VerificationError('candidate must be the visible response text') payload = _json_object(request.get('verifier_payload'), name='verifier_payload') instruction_ids = payload.get('instruction_id_list') kwargs = payload.get('kwargs') if not isinstance(instruction_ids, list) or not instruction_ids: raise VerificationError('instruction_id_list must be a non-empty list') if not all((isinstance(value, str) and value for value in instruction_ids)): raise VerificationError('instruction_id_list contains an invalid id') if not isinstance(kwargs, list) or len(kwargs) != len(instruction_ids): raise VerificationError('kwargs must align one-to-one with instruction_id_list') normalized_kwargs = [] for item in kwargs: if not isinstance(item, Mapping): raise VerificationError('each kwargs item must be an object') normalized_kwargs.append(_normalize_kwarg(item)) repo = args.ifbench_repo if mode == 'ifbench' else args.ifeval_repo nltk_data = args.nltk_data if nltk_data is None: candidate_nltk = _evaluation_lib_path(repo).parent / '.nltk_data' if candidate_nltk.is_dir(): nltk_data = candidate_nltk if nltk_data is not None: os.environ['NLTK_DATA'] = str(nltk_data.expanduser().resolve()) evaluation_lib = _load_evaluation_lib(mode, repo) item = evaluation_lib.InputExample(key=payload.get('key', request.get('sample_id', 0)), instruction_id_list=list(instruction_ids), prompt=problem, kwargs=normalized_kwargs) outcome = evaluation_lib.test_instruction_following_strict(copy.deepcopy(item), {problem: candidate}) followed = [bool(value) for value in outcome.follow_instruction_list] if len(followed) != len(instruction_ids): raise VerificationError('official scorer returned the wrong instruction count') literal_overrides = [] for index, (instruction_id, instruction_kwargs) in enumerate(zip(instruction_ids, normalized_kwargs, strict=True)): if instruction_id != 'keywords:letter_frequency': continue literal_result = _literal_nonalpha_letter_frequency(candidate, instruction_kwargs) if literal_result is None: continue followed[index] = literal_result literal_overrides.append(index) passed = all(followed) return {'verdict': 'PASS' if passed else 'FAIL', 'details': {'method': f'{mode}_official_strict_with_literal_letter_frequency' if literal_overrides else f'{mode}_official_strict_prompt_level', 'instruction_id_list': list(instruction_ids), 'follow_instruction_list': followed, 'follow_all_instructions': passed, 'literal_letter_frequency_indices': literal_overrides, 'language_detection_seed': 0}} def main() -> int: result_stream = os.fdopen(os.dup(sys.stdout.fileno()), 'w', buffering=1) os.dup2(sys.stderr.fileno(), sys.stdout.fileno()) args = parse_args() def evaluate(request: Mapping[str, Any]) -> dict[str, Any]: try: return verify(request, args) except Exception as exc: return {'verdict': 'UNCERTAIN', 'details': {'method': 'instruction_verifier_fail_closed', 'error': f'{type(exc).__name__}: {str(exc)[:1000]}'}} if args.jsonl_worker: for line in sys.stdin: try: request = json.loads(line) if not isinstance(request, Mapping): raise VerificationError('request must be a JSON object') result = evaluate(request) except Exception as exc: result = evaluate({'verifier_mode': 'invalid', 'error': str(exc)}) print(json.dumps(result, ensure_ascii=False, sort_keys=True), file=result_stream, flush=True) return 0 result = evaluate(_request_from_stdin()) print(json.dumps(result, ensure_ascii=False, sort_keys=True), file=result_stream, flush=True) return 0 if __name__ == '__main__': raise SystemExit(main())