"""Async generation runner: model + dataset -> predictions -> scored report. The async boundary is exactly "waiting on the model". Data loading and scoring stay synchronous (fast, CPU/disk bound); this coroutine fans out model calls with a semaphore, streams progress, then hands the collected raw strings to the sync evaluate(). from evalharness.model import run_eval report = asyncio.run(run_eval(ds, 'mock', limit=50)) # offline smoke report = asyncio.run(run_eval(ds, 'openai/http://gpu03:8000/v1?qwen3-8b')) """ import asyncio import os import re import time from typing import Any, Dict, List, Optional, Union from ..data.dataset import Dataset from ..data.sample import ChatMessage, Sample from ..eval.recipe import EvalRecipe from ..eval.record import EvalReport from ..eval.runner import evaluate from .adapter import ModelAdapter, resolve_adapter from .output import Usage async def generate_predictions( adapter: ModelAdapter, samples: List[Sample], concurrency: int = 32, limit: Optional[int] = None, gen_kwargs: Optional[Dict[str, Any]] = None, progress: bool = True, progress_reporter=None, status_callback=None, env_factory=None, env_user_spec: str = '', no_shuffle: bool = False, system: str = '', max_turns: int = 8, max_input_chars: int = 0, max_input_tokens: int = 0, tokenizer_path: str = '', attach_context_keys: tuple = ('passage', 'context'), dataset_spec=None, limit_per_task: Optional[int] = None, checkpoint: Union[bool, str] = False, dataset_name: str = 'adhoc', few_shot_num: int = 0, few_shot_samples: Optional[List[Sample]] = None, few_shot_text: Optional[str] = None, prompt_style: str = 'strict_letter', repeat: int = 1, ) -> tuple: """Fan out model calls; returns (pred-dicts, total_usage). MCQ samples are generated with the strict-letter contract ('ANSWER: X', evalscope parity). few_shot: official exemplar text (few_shot_text) or dev/train-split samples (few_shot_samples) are prepended. """ gen_kwargs = gen_kwargs or {} def _default_max_tokens() -> int: return 4096 sem = asyncio.Semaphore(concurrency) total_usage = Usage() done_count = 0 t0 = time.time() hle_system = [''] # mutable cell: answer_type-specific system prompt (hle) extra_system = [''] # mutable cell: renderer-provided system message (lcb etc.) def assemble(sample: Sample) -> str: parts = [] math_glue = False # es math few-shot: single \n before the test Problem: hle_system[0] = '' # reset per sample (es: answer_type-specific system role) extra_system[0] = '' # reset per sample (renderer system, e.g. lcb) if few_shot_text: parts.append(few_shot_text.strip()) # official exemplars, verbatim elif few_shot_num and few_shot_samples: letters_fs = 'ABCDEFGHIJ' es_style = getattr(dataset_spec, 'prompt_style', '') in ('cot_letter', 'cot_letter_zh', 'cot_letter_plain') plain_style = getattr(dataset_spec, 'prompt_style', '') == 'cot_letter_plain' drop_style = getattr(dataset_spec, 'prompt_style', '') == 'drop_es' if es_style and len(few_shot_samples) > few_shot_num: # domain-matched selection (es parity): exemplars sharing the # current sample's category first, global first-N as fallback. # key: 'category' (cmmlu/mmlu_pro) OR 'subject' (mmlu) -- # es reformat_subset regroups fewshot by subset_key def _cat_of(md): return ((md or {}).get('category') or (md or {}).get('subject') or (md or {}).get('level')) # math: per-Level exemplars cat = _cat_of(sample.metadata) pool = [s for s in few_shot_samples if _cat_of(s.metadata) == cat] if len(pool) < few_shot_num: pool = pool + [s for s in few_shot_samples if _cat_of(s.metadata) != cat] sel = pool[:few_shot_num] else: sel = few_shot_samples[:few_shot_num] blocks = [] for fs in sel: if drop_style: # es drop exemplar: full Passage + Question + bare-span Answer # (multi-span gold joins with ', ' -- teaches the model the # exact answer FORM the Hungarian EM compares against) line = f"Passage: {(fs.metadata or {}).get('passage', '')}\nQuestion: {fs.input_text}" ans = fs.target if not isinstance(fs.target, list) else ', '.join(str(t) for t in fs.target) line += f'\nAnswer: {ans}' elif plain_style: # es mmlu-pro exemplar (adapter sample_to_fewshot, verbatim): # Question:/Options:/A x + cot_content transformed # 'The answer is (X).' -> 'ANSWER: X.' -- exactly ONE answer # marker, no appended ANSWER line line = f'Question:\n{fs.input_text}' if fs.choices: line += '\nOptions:\n' + '\n'.join(f'{letters_fs[j]} {c}' for j, c in enumerate(fs.choices)) ans = fs.target if not isinstance(fs.target, list) else fs.target[0] cot = (fs.metadata or {}).get('cot_content') if cot: ans_str = str(cot).strip().replace('The answer is', 'ANSWER:') ans_opt = ans_str.split('ANSWER:')[-1].split('.')[0].strip().strip('(').strip(')') ans_str = ans_str.replace(f'ANSWER: ({ans_opt})', f'ANSWER: {ans_opt}') line += f'\n{ans_str}' else: line += f'\nANSWER: {ans}' elif es_style: # es exemplar rendering: bare question + 'A) opt' + 'ANSWER: X' # (mimicry target for the CoT-last-line contract) line = fs.input_text if fs.choices: line += '\n' + '\n'.join(f'{letters_fs[j]}) {c}' for j, c in enumerate(fs.choices)) ans = fs.target if not isinstance(fs.target, list) else fs.target[0] cot = (fs.metadata or {}).get('cot_content') if cot: line += f'\n{str(cot).strip()}' line += f'\nANSWER: {ans}' elif (fs.metadata or {}).get('reasoning') and not fs.choices: # es qa few-shot (gsm8k): question + Reasoning + ANSWER: boxed line = (f"{fs.input_text}\n\nReasoning:\n{str((fs.metadata or {}).get('reasoning', '')).strip()}\n\n" f'ANSWER: \\boxed{{{fs.target}}}') elif (fs.metadata or {}).get('es_math_fewshot'): # es math: Problem:/Solution: bare-answer exemplars line = f'Problem:\n{fs.input_text}\nSolution:\n{fs.target}' else: line = f'Question: {fs.input_text}' if fs.choices: line += '\n' + '\n'.join(f'{letters_fs[j]}. {c}' for j, c in enumerate(fs.choices)) ans = fs.target if not isinstance(fs.target, list) else fs.target[0] line += f'\nAnswer: {ans}' blocks.append(line) if drop_style and few_shot_text: # hook 版范例已含完整 es 结构, 直接用 parts.append(few_shot_text.strip() + '\n\n# Your Task\n---\n') elif drop_style: parts.append('You will be asked to read a passage and answer a question. ' 'Some examples of passages and Q&A are provided below.\n\n' '# Examples\n---\n' + '\n---\n'.join(blocks) + '\n\n# Your Task\n---\n') elif plain_style: # es mmlu-pro: subject header FIRST, then exemplars, then the # user template (SYSTEM_W_EXAMPLES_PROMPT_TEMPLATE + '\n' + USER) subj = (sample.metadata or {}).get('category') or 'knowledge' parts.append( f'The following are multiple choice questions (with answers) about ' f'{subj}. Think step by step and then finish your answer with ' f"'ANSWER: [LETTER]' (without quotes) where [LETTER] is the correct " f'letter choice.\n\n' + '\n\n'.join(blocks)) elif es_style: parts.append('Here are some examples of how to answer similar questions:\n\n' + '\n\n'.join(blocks)) elif blocks and ('\nReasoning:\n' in blocks[0] or blocks[0].startswith('Problem:\n')): # es gsm8k/math FEWSHOT_TEMPLATE header parts.append('Here are some examples of how to solve similar problems:\n\n' + '\n\n'.join(blocks)) if blocks[0].startswith('Problem:\n') and '\nReasoning:\n' not in blocks[0]: math_glue = True # es math: ONE newline before the test Problem: else: parts.extend(blocks) for key in attach_context_keys: ctx = (sample.metadata or {}).get(key) if ctx: parts.append(str(ctx)) question = sample.input_text spec_style = getattr(dataset_spec, 'prompt_style', '') if dataset_spec is not None else '' # prompt-style PLUGINS: each registered renderer rewrites the question # (and may set a system message); unknown styles -> generic handling from .prompt_renderers import render as _render out = _render(spec_style, question, sample, dataset_spec, prompt_style) if out: question = out.get('question', question) if out.get('system'): extra_system[0] = out['system'] elif sample.choices: if prompt_style in ('strict_letter', 'auto'): # evalscope/OpenAI-style contract: reply ONLY 'ANSWER: X' # rendering is VERBATIM es: 'A) option' + 'one of A,B,C,D' -- # 'A.' vs 'A)' alone swings hswag by 22 points on no-think Qwen3 letters = 'ABCDEFGHIJ' opts = '\n'.join(f'{letters[i]}) {c}' for i, c in enumerate(sample.choices) if i < len(letters)) question = (f'Answer the following multiple choice question. The entire ' f'content of your response should be of the following format: ' f"'ANSWER: [LETTER]' (without quotes) where [LETTER] is one of " f'{",".join(letters[:len(sample.choices)])}.\n\n{question}\n\n{opts}') else: letters = 'ABCDEFGHIJ' opts = '\n'.join(f'{letters[i]}. {c}' for i, c in enumerate(sample.choices) if i < len(letters)) question = (f'{question}\n\n{opts}\n\n' 'Answer with the letter of the correct option.') elif sample.task_type in ('qa',): # hle OFFICIAL protocol: answer_type-specific SYSTEM contract (es # puts it in the system role; injected as a system message in # one(), the question itself stays bare) at = (sample.metadata or {}).get('answer_type') if at == 'exactMatch': hle_system[0] = ( 'Your response should be in the following format:\n' 'Explanation: {your explanation for your final answer}\n' 'Exact Answer: {your succinct, final answer}\n' 'Confidence: {your confidence score between 0% and 100% for your answer}') elif at == 'multipleChoice': hle_system[0] = ( 'Your response should be in the following format:\n' 'Explanation: {your explanation for your answer choice}\n' 'Answer: {your chosen answer}\n' 'Confidence: {your confidence score between 0% and 100% for your answer}') elif not getattr(dataset_spec, 'prompt_suffix', ''): question = (f'{question}\n\n' 'End your reply with the final answer on its own last line ' 'in the form "Answer: ".') ds_spec = dataset_spec if ds_spec is not None and getattr(ds_spec, 'prompt_suffix', ''): question = question + ds_spec.prompt_suffix if math_glue and parts: # es competition_math: exactly ONE newline between the last # exemplar and the test 'Problem:' (FEWSHOT_TEMPLATE tail) parts[-1] = parts[-1] + '\n' + question else: parts.append(question) text = '\n\n'.join(parts) if max_input_tokens: # reserve room for the OUTPUT budget + safety margin, else the # server rejects input+max_tokens > context_limit by 1 token budget = max(1024, max_input_tokens - int(gen_kwargs.get('max_tokens') or 4096) - 2048) try: from .truncation import truncate_middle_tokens, default_tokenizer_path text = truncate_middle_tokens(text, budget, tokenizer_path or default_tokenizer_path()) except Exception as e: # no tokenizer/transformers: degrade to a CHARS budget that # approximates the token cap (never send the raw 2M-token input) approx_chars = budget * 3 if len(text) > approx_chars: keep = approx_chars // 2 text = f'{text[:keep]}\n\n...[truncated {len(text) - 2 * keep} chars]...\n\n{text[-keep:]}' print(f'truncation degraded to chars ({type(e).__name__})', flush=True) if max_input_chars and len(text) > max_input_chars: keep = max_input_chars // 2 head = text[:keep] tail = text[-keep:] text = f'{head}\n\n...[truncated {len(text) - 2 * keep} chars]...\n\n{tail}' return text async def one(sample: Sample) -> Dict[str, Any]: nonlocal done_count, total_usage if env_factory is not None: from ..agent import drive, trajectory_to_prediction from ..agent.loop import Environment, Usage as _U # noqa: F401 async with sem: if progress_reporter is not None: # begin AFTER acquiring the slot: "in flight" must mean # actually generating, not queued on the semaphore progress_reporter.begin_sample(f'sample {sample.id}') try: env = env_factory() if type(env).run_task is not Environment.run_task: # self-running env (official engine bundles: tau2/swe) pred = await env.run_task(adapter, sample, max_turns=max_turns, system=system, user_adapter=_env_user_adapter(env_user_spec) if env_user_spec else None, gen_kwargs=gen_kwargs) if pred is None: traj = await drive(adapter, sample, env=env, max_turns=max_turns, system=system) pred = trajectory_to_prediction(traj) else: traj = await drive(adapter, sample, env=env, max_turns=max_turns, system=system) pred = trajectory_to_prediction(traj) except Exception: if progress_reporter is not None: progress_reporter.rollback() raise if not pred.get('usage'): pred['usage'] = traj.usage.model_dump() if 'traj' in dir() else {} pred.setdefault('group_key', str(sample.metadata.get('test_category') or sample.metadata.get('category') or sample.metadata.get('domain') or sample.metadata.get('id') or sample.id or '')) u = pred.get('usage') or {} total_usage = total_usage + Usage( input_tokens=int(u.get('input_tokens', 0) or 0), output_tokens=int(u.get('output_tokens', 0) or 0), total_tokens=int(u.get('total_tokens', 0) or 0), latency_s=float(u.get('latency_s', 0) or 0)) done_count += 1 if progress_reporter is not None: progress_reporter.advance(success=True) else: _progress(progress, done_count, len(samples), t0, total_usage) return pred messages = ([ChatMessage(role='user', content=assemble(sample))] if isinstance(sample.input, str) else list(sample.input)) if not system and extra_system[0] and isinstance(sample.input, str): # renderer-provided SYSTEM contract (es lcb expert-programmer) messages.insert(0, ChatMessage(role='system', content=extra_system[0])) if not system and hle_system[0]: # es hle: answer_type-specific format contract in the SYSTEM role messages.insert(0, ChatMessage(role='system', content=hle_system[0])) tools = None if sample.tools: tools = [{'name': t.name, 'description': t.description or '', 'parameters': t.parameters} for t in sample.tools] if getattr(adapter, 'name', '') == 'mock' \ and adapter.extra.get('mode') in ('boxed', 'oracle', 'fc') \ and sample.target not in ('', None): # oracle channel for mock verification so full pipelines run offline messages = messages + [ChatMessage(role='user', content=f'MOCKTARGET::{sample.target}')] async with sem: if progress_reporter is not None: progress_reporter.begin_sample(f'sample {sample.id}') try: out = await adapter.generate(messages, tools=tools, **gen_kwargs) except Exception: # retry path re-enters one() and begins again: pair this # begin here or the in-flight count leaks upward if progress_reporter is not None: progress_reporter.rollback() raise total_usage = total_usage + out.usage text = out.text if out.tool_calls: # fc tasks: serialize calls as the prediction import json text = (text + '\n' if text else '') + json.dumps( [c.to_openai()['function'] for c in out.tool_calls], ensure_ascii=False) done_count += 1 if progress_reporter is not None: progress_reporter.advance(success=True) else: _progress(progress, done_count, len(samples), t0, total_usage) return {'raw': text, 'usage': out.usage.model_dump()} work = _apply_limits(samples, limit, limit_per_task, shuffle=not no_shuffle) # checkpointing: restore completed samples, generate only the rest ckpt_store = None if checkpoint: from ..eval.checkpoint import CheckpointStore, checkpoint_path if isinstance(checkpoint, str): ckpt = checkpoint else: # include subset in the checkpoint key: same dataset under # different subsets (bbh tasks, lb2 lengths) must not share state sub = getattr(dataset_spec, 'subset', '') or '' from ..data.dataset import get_cache_root # repeats are INDEPENDENT samples of a temp>0 run: repeat 2 must # never reuse repeat 1's predictions from the shared checkpoint # (that made repeats 2..N finish instantly with identical scores) ckpt_name = f'{dataset_name}:{sub}' if sub else dataset_name if repeat > 1: ckpt_name = f'{ckpt_name}:rep{repeat}' # one root for everything: --cache-dir > $EVALHARNESS_CACHE > # ~/.cache/evalharness (data cache and checkpoints stay together) ckpt = checkpoint_path(str(get_cache_root()), ckpt_name, adapter.model or str(adapter)) ckpt_store = CheckpointStore(ckpt, model=adapter.model or str(adapter)) restored = ckpt_store.load() else: restored = {} keys = [] pending = [] preds_by_key: Dict[str, Dict[str, Any]] = {} for i, s in enumerate(work): # NB: must be `is not None` -- an EMPTY store is falsy via __len__, # which silently switched the key scheme between fresh runs (str(i)) # and resumed runs (key_for) and broke every restore k = CheckpointStore.key_for(s, i) if ckpt_store is not None else str(i) keys.append(k) if k in restored: preds_by_key[k] = restored[k] else: pending.append((i, s)) if status_callback: if restored: status_callback(f'Checkpoint: {len(restored)}/{len(work)} predictions already generated, ' f'{len(pending)} samples left to run') else: status_callback(f'{len(work)} samples to evaluate') elif restored: print(f'checkpoint: restored {len(restored)} predictions ' f'({len(pending)} to generate) -> {ckpt_store.path}', flush=True) if progress_reporter is not None: progress_reporter.reset_samples(len(work), dataset_name, completed=len(restored)) # let the adapter surface retry attempts to the bar members = getattr(adapter, 'adapters', [adapter]) for m_ in members: m_.extra['progress_reporter'] = progress_reporter async def run_one(i_s): i, s = i_s # transient network flaps (cluster routes re-converge): retry with # backoff so one ConnectError burst cannot kill the whole batch -- # the adapter already retries 5xx/429 and the pool fails over per # instance; this is the last line of defense around asyncio.gather # NO outer retry: the adapter retries internally; a second loop # here multiplied worst-case time (42+ attempts before this fix). # One pass, one result or one error. try: pred = await one(s) except Exception: if progress_reporter is not None: progress_reporter.advance(success=False) raise if ckpt_store is not None: ckpt_store.append(keys[i], pred) return i, pred try: if status_callback: if pending: status_callback(f'Generating {len(pending)} model responses') else: status_callback('Generation skipped: the checkpoint already covers every sample') fresh = await asyncio.gather(*(run_one((i, s)) for i, s in pending)) for i, pred in fresh: preds_by_key[keys[i]] = pred preds = [preds_by_key[k] for k in keys] usages = [p.get('usage', {}) for p in preds] # include RESTORED predictions' usage (they carry it in the ckpt); # previously only fresh generations counted -> restored benches showed 0 fresh_keys = {keys[i] for i, _ in pending} for k, p in preds_by_key.items(): if k in fresh_keys: continue # already counted via one()'s total_usage updates u = p.get('usage') or {} if not any(u.get(kk) for kk in ('input_tokens', 'output_tokens')): continue total_usage = total_usage + Usage( input_tokens=int(u.get('input_tokens', 0) or 0), output_tokens=int(u.get('output_tokens', 0) or 0), total_tokens=int(u.get('total_tokens', 0) or 0), latency_s=float(u.get('latency_s', 0) or 0)) if status_callback and pending: status_callback(f'Generation complete: {len(preds)} responses collected') return preds, usages, total_usage finally: # reporter lifecycle belongs to the CALLER (CLI reuses one reporter # across benchmarks and closes it after the whole run); only close # here when nobody external passed it in if progress_reporter is not None and not getattr(progress_reporter, 'owned_externally', False): progress_reporter.close() def _apply_limits(samples: List[Sample], total: Optional[int], per_task: Optional[int], dataset=None, shuffle: bool = True, seed: int = 42) -> List[Sample]: """total: cap the WHOLE run (ours semantics). per_task: cap each subset/ category (evalscope's --limit semantics) -- first N per group_key. shuffle+seed mirror evalscope run.py: dataset_args.shuffle=True with --seed 42 -> random.Random(seed).shuffle BEFORE limiting, so both frameworks evaluate the IDENTICAL sample subset.""" if shuffle and not per_task: import random random.Random(seed).shuffle(samples) if per_task: # evalscope semantics: each subset's records are shuffled with a # fresh Random(seed) INDEPENDENTLY, then capped at N (builder.py: # build_dataset_from_records per subset). Emulate exactly: group, # per-group shuffle, first-N. For single-pool datasets this is # identical to the global shuffle above. import random from collections import OrderedDict def _key(s: Sample) -> str: return str((s.metadata or {}).get('subset') or (s.metadata or {}).get('category') or (s.metadata or {}).get('subject') or (s.metadata or {}).get('test_category') or (s.metadata or {}).get('length') or (s.metadata or {}).get('level') or getattr(getattr(dataset, 'spec', None), 'subset', 'default')) groups: Dict[str, List[Sample]] = OrderedDict() for s in samples: groups.setdefault(_key(s), []).append(s) out: List[Sample] = [] for lst in groups.values(): if shuffle: # no_shuffle => raw first-N per group (same-questions) random.Random(seed).shuffle(lst) out.extend(lst[:per_task]) samples = out if total: samples = samples[:total] return samples def _progress(progress: bool, done: int, total: int, t0: float, usage: Usage) -> None: interval = max(1, min(20, total)) if progress and (done % interval == 0 or done == total): rate = done / max(time.time() - t0, 1e-6) print(f' [{done}/{total}] {rate:.1f} samples/s tokens={usage.total_tokens}', flush=True) _PROBED_SPECS = set() async def _probe_model(adapter, model_spec: str) -> None: """Delegate to the registered PROBER plugin ('ping' by default). @register_prober('custom') async def probe(adapter): ... replaces the whole reachability strategy without touching the runner.""" from .probers import get_prober if getattr(adapter, 'name', '') != 'mock': await get_prober(os.environ.get('EVALHARNESS_PROBER', 'ping'))(adapter) return async def _default_ping_probe(adapter, model_spec=None): """Fail fast on an unreachable model endpoint. One 1-token request before any dataset work: a wrong api-url/model name surfaces in seconds (with a clear fix hint) instead of the multi-minute retry ladder. Cached per spec so multi-benchmark runs probe only once. Mock adapters are exempt. """ if getattr(adapter, 'name', '') == 'mock': return _probe_model._t0 = time.monotonic() members = getattr(adapter, 'adapters', [adapter]) if model_spec in _PROBED_SPECS: return bad = [] for a in members: try: out = await asyncio.wait_for( a.generate([ChatMessage(role='user', content='ping')], max_tokens=1, temperature=0.0), timeout=30) if out is None or (not out.text and not out.tool_calls): raise RuntimeError('empty response') except Exception as e: bad.append(f'{a.api_base}: {type(e).__name__} {str(e)[:80]}') if bad and len(bad) == len(members): import json as _json model_name = getattr(members[0], 'model', '') or '' body = _json.dumps({'model': model_name, 'messages': [{'role': 'user', 'content': 'ping'}], 'max_tokens': 1}) curl = f'curl -m 5 {members[0].api_base}/chat/completions -H "Content-Type: application/json" -d {body!r}' raise RuntimeError( 'Model endpoint unreachable -- aborted before running any samples.\n' f' endpoint: {members[0].api_base}\n' f' reason: {bad[0]}\n' 'Fix: check that --api-url points to a running OpenAI-compatible server\n' ' and --model matches the served model name. Verify manually:\n' f' {curl}') if members and not bad: import sys as _sys dur = time.monotonic() - _probe_model._t0 if hasattr(_probe_model, '_t0') else 0.0 name = getattr(members[0], 'model', '') or '?' url = members[0].api_base txt = (f'· Model endpoint verified: "{name}" responded at {url} ' f'in {dur:.1f}s ({len(members)} instance(s) in pool) -- ' f'generation will use this endpoint') if _sys.stdout.isatty(): # color the machine-relevant facts on terminals txt = (f'· Model endpoint verified: "\x1b[1m{name}\x1b[0m" responded at ' f'\x1b[36m{url}\x1b[0m in {dur:.1f}s ' f'({len(members)} instance(s) in pool) -- ' f'generation will use this endpoint') print(txt, flush=True) # probe runs before status_callback exists _PROBED_SPECS.add(model_spec) async def run_eval( dataset: Union[Dataset, List[Sample]], model_spec: str, recipe: Optional[EvalRecipe] = None, *, concurrency: int = 32, limit: Optional[int] = None, gen_kwargs: Optional[Dict[str, Any]] = None, api_key: str = '', judge_api_key: str = '', judge_spec: Optional[str] = None, judge: Optional[Any] = None, progress: bool = True, progress_reporter=None, status_callback=None, env: str = '', env_user_spec: str = '', no_shuffle: bool = False, # fixed-order selection: raw first-N (same-questions parity) system: str = '', max_turns: int = 8, max_input_chars: int = 0, max_input_tokens: int = 0, limit_per_task: Optional[int] = None, checkpoint: Union[bool, str] = False, dataset_name: str = 'adhoc', few_shot_num: int = -1, prompt_style: str = 'strict_letter', gen_profile: str = '', repeat: int = 1, ) -> EvalReport: """Generate + score in one call. Model spec examples: 'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'. env: environment plugin name ('bfcl_mock') -> agent message pump per sample; omit for single-turn generation. 'auto' = logprob when the adapter supports it. few_shot_num: -1 = the dataset's declared paper default (mmlu 5, bbh 3, gsm8k 4, ...); 0 = zero-shot; N = explicit override. prompt_style: 'strict_letter' (default, evalscope-style 'ANSWER: X') | 'cot' (reasoning-friendly). """ spec = getattr(dataset, 'spec', None) if few_shot_num < 0: few_shot_num = (spec.few_shot_num if spec is not None else 0) adapter = _make_adapter(model_spec, api_key=api_key) await _probe_model(adapter, model_spec if isinstance(model_spec, str) else repr(adapter)) # reports carry a string model label: pre-built adapter objects need one model_spec = model_spec if isinstance(model_spec, str) \ else (getattr(model_spec, 'model', '') or repr(model_spec)) name = spec.name if spec is not None else 'adhoc' if recipe is None: from ..eval.recipe import EvalRecipe, get_eval try: recipe = get_eval(name) except KeyError: if name != 'adhoc': raise recipe = EvalRecipe(name='adhoc', extract='identity', scorers={'acc': {'name': 'exact', 'mode': 'raw'}}) # materialize in a worker thread: hub downloads here are synchronous # (requests/ssl) and would otherwise stall the whole event loop raw_samples = await asyncio.to_thread(lambda: list(dataset)) if limit: raw_samples = raw_samples[:limit] # generate_predictions applies the SAME deterministic limiting internally; # recompute on an equal copy so evaluate() zips against the exact work # list (positional pairing) instead of relying on in-place aliasing. samples = _apply_limits(list(raw_samples), limit, limit_per_task, shuffle=not no_shuffle) # MUST mirror generate_predictions if progress and not status_callback: mode = f'agent env={env}' if env else 'single-turn' print(f'generating: {adapter} on {len(samples)} samples ' f'({mode}, concurrency={concurrency})', flush=True) env_factory = None if env: from ..agent import ENV_REGISTRY, get_env if env not in ENV_REGISTRY: raise KeyError(f'unknown env {env!r}; available: {", ".join(ENV_REGISTRY.names())}') probe = get_env(env) if getattr(probe, 'needs_adapter', False): env_factory = lambda: get_env(env, adapter=adapter) # noqa: E731 else: env_factory = lambda: get_env(env) # noqa: E731 # paper-faithful few-shot exemplars: official hand-written hooks first # (bbh CoT), else the dataset's own dev/train split few_shot_samples = None few_shot_text = None if few_shot_num: if status_callback: status_callback(f'Loading {few_shot_num} few-shot exemplars') from ..data.registry import get_dataset_provider prov = get_dataset_provider(name) hook = getattr(prov, 'few_shot_hook', None) if hook is not None: few_shot_text = hook('', spec.subset if spec else 'default', few_shot_num) if few_shot_text is None: fs_split = (spec.few_shot_split if spec is not None else None) or 'dev' try: import dataclasses fs_spec = dataclasses.replace(spec, split=fs_split) if spec is not None else None if fs_spec is not None: # go through the CACHED materialization path (not raw hub # loads): a cached few-shot split never touches the # network; first use downloads and caches it for offline # runs afterwards from ..data.dataset import Dataset fn = prov.resolve_record_fn() fs_ds = Dataset(fs_spec, fn) fs_ds.materialize() fs_samples_all = list(fs_ds) # keep the WHOLE dev split when samples carry a category: # es selects domain-MATCHED exemplars per subject (mmlu # biology questions get biology exemplars), we do the same # at assemble time; global first-N otherwise def _lv_of(md): return (md or {}).get('category') or (md or {}).get('level') cats = {_lv_of(s.metadata) for s in fs_samples_all[:200]} style_is = getattr(spec, 'prompt_style', '') if spec is not None else '' if len(cats) > 1 and spec is not None and \ (style_is.startswith('cot_letter') or style_is == 'imo_es'): # mmlu-style per-subject OR math per-Level exemplars: # load the WHOLE few-shot split; assemble-time picks # domain-matched first-N (es reformat_subset semantics) few_shot_samples = fs_samples_all else: few_shot_samples = fs_samples_all[:few_shot_num] except Exception as e: print(f'few-shot: could not load {fs_split} split ({type(e).__name__}: ' f'{str(e)[:80]}); continuing 0-shot', flush=True) try: from .gen_profiles import merge_gen_kwargs preds, _usages, usage = await generate_predictions( adapter, list(raw_samples), concurrency, progress=progress, progress_reporter=progress_reporter, status_callback=status_callback, gen_kwargs=merge_gen_kwargs(name, spec, gen_kwargs, gen_profile), env_factory=env_factory, env_user_spec=env_user_spec, no_shuffle=no_shuffle, system=system, max_turns=max_turns, max_input_chars=max_input_chars, max_input_tokens=max_input_tokens, dataset_spec=spec, limit_per_task=limit_per_task, checkpoint=checkpoint, dataset_name=name, few_shot_num=few_shot_num, few_shot_samples=few_shot_samples, few_shot_text=few_shot_text, prompt_style=prompt_style, repeat=repeat) finally: await adapter.close() if judge is None and judge_spec: if status_callback: status_callback('loading judge model') judge_adapter = _make_adapter(judge_spec, api_key=judge_api_key or api_key) judge = _judge_callable(judge_adapter) if status_callback: status_callback('Scoring predictions against the benchmark recipe') # scoring off the event loop: math_equal/sympy equivalence can chew a # single hard problem for minutes (es's checker famously hangs on one) -- # running it inline froze the progress bar's clock for the whole bench report = await asyncio.to_thread( evaluate, samples, preds, recipe, model=model_spec, judge=judge, extra_metadata={'gen_input_tokens': usage.input_tokens, 'gen_output_tokens': usage.output_tokens, 'gen_total_tokens': usage.total_tokens}, ) report.model = model_spec report.dataset = name if status_callback: _m = next(((k, v) for k, v in report.metrics.items() if k != 'extraction_failure_rate'), None) status_callback(f'Scoring complete: {_m[0]} {_m[1] * 100:.1f}% ' f'over {report.num_samples} samples' if _m else 'Scoring complete') # performance profile: pool success rate + latency/ttft percentiles try: from ..eval.aggregator import get_aggregator perf = get_aggregator('perf_stats')(report.samples, 'acc') if hasattr(adapter, 'stats'): perf.update({f'pool_{k}': round(v, 3) if isinstance(v, float) else v for k, v in adapter.request_stats().items()}) report.metric_groups['perf'] = perf except Exception as e: # pragma: no cover import sys print(f'perf stats skipped: {type(e).__name__}: {str(e)[:120]}', file=sys.stderr) return report def _env_user_adapter(spec: str): """Build (once per spec) the separate USER-simulator adapter for env benches (tau2 strong-user parity mode).""" global _ENV_USER_CACHE if spec not in _ENV_USER_CACHE: _ENV_USER_CACHE[spec] = _make_adapter(spec) return _ENV_USER_CACHE[spec] _ENV_USER_CACHE = {} def _make_adapter(spec: str, api_key: str = '') -> ModelAdapter: """Model spec forms: - 'mock[:mode]' offline adapter - 'openai-pool/?model' with {port} placeholder: e.g. 'openai-pool/http://127.0.0.1:{8123..8130}/v1?Qwen3-8B' -> N ports - else resolve_adapter(spec) single endpoint Pooled specs are CACHED per spec: all benches share one pool so the round-robin counter stays global (independent pools would each restart at the first backend and starve the rest). """ from .adapter import _ADAPTER_CACHE as _CACHE, ModelAdapter if isinstance(spec, ModelAdapter): # pre-built adapter (tests/custom) return spec cache_key = spec if cache_key in _CACHE: return _CACHE[cache_key] opts = {} while True: for f in ('!nothink', '!textools', '!perf'): if spec.endswith(f): spec = spec[:-len(f)] opts[f] = True break else: break if spec.startswith('openai-pool/'): from .pool import pooled rest = spec[len('openai-pool/'):] ms = __import__('re').findall(r'\{(\d+)\.\.(\d+)\}', rest) if not ms: raise ValueError("openai-pool needs a {start..end} port range") base_url, _, model = rest.partition('?') # expand EACH comma-separated segment's OWN range independently -- # a global sub(count=1) would keep replacing only the FIRST range # and emit URLs with literal '{8200..8203}' in later segments specs = [] for seg in base_url.split(','): seg = seg.strip() m = __import__('re').search(r'\{(\d+)\.\.(\d+)\}', seg) if m: lo, hi = int(m.group(1)), int(m.group(2)) for port in range(lo, hi + 1): u = seg[:m.start()] + str(port) + seg[m.end():] specs.append(f'openai/{u}?{model}') elif seg: specs.append(f'openai/{seg}?{model}') adapter = pooled(specs, api_key=api_key) if api_key else pooled(specs) elif re.fullmatch(r'mock[-:](boxed|oracle|fc|tool|echo|const)?', spec): # mock-boxed (preferred) == legacy mock:boxed; bare 'mock' == echo. # NEVER reuse the cached singleton: resolve_adapter memoizes and a # shared instance would leak this run's mode into the next one mode = re.fullmatch(r'mock[-:]?(.*)', spec).group(1) or 'echo' from .adapter import ADAPTER_REGISTRY adapter = ADAPTER_REGISTRY.get('mock')(model='mock', api_base='') adapter.extra['mode'] = mode return adapter else: adapter = resolve_adapter(spec) members = adapter.adapters if hasattr(adapter, 'adapters') else [adapter] for a in members: if opts.get('!nothink'): a.extra['no_think'] = True if opts.get('!textools'): a.extra['tools_mode'] = 'text' if opts.get('!perf'): a.extra['collect_perf'] = True _CACHE[cache_key] = adapter return adapter def _judge_callable(judge_adapter: ModelAdapter): """Sync judge bridge. Works inside a running event loop (evaluate() may be called from async run_eval): the coroutine runs on a private loop in a worker thread.""" def ask(messages) -> str: import asyncio if isinstance(messages, list) and messages and isinstance(messages[0], dict): messages = [ChatMessage(role=m.get('role', 'user'), content=m.get('content', '')) for m in messages] async def go(): out = await judge_adapter.generate(messages) return out.text try: asyncio.get_running_loop() except RuntimeError: return asyncio.run(go()) import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: return pool.submit(asyncio.run, go()).result() return ask