"""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 = 200, 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 # assemble() tokenizes for the max_input_tokens truncation -- on # long-context benches that is SECONDS of CPU per sample (2M-token # docs). Two failure modes fixed here: # - inline: froze the whole event loop behind one encode # - asyncio.to_thread (32-thread default pool): dozens of concurrent # tokenizers hogged the GIL and starved the progress renderer + # loop itself (bar froze, then jumped) # A DEDICATED BOUNDED pool: 8 encodes at a time, remaining workers # queue -- GIL pressure capped, everything stays responsive. global _ASSEMBLE_EXEC if _ASSEMBLE_EXEC is None: import concurrent.futures _ASSEMBLE_EXEC = concurrent.futures.ThreadPoolExecutor( max_workers=8, thread_name_prefix='assemble') text = await asyncio.get_running_loop().run_in_executor( _ASSEMBLE_EXEC, assemble, sample) \ if isinstance(sample.input, str) else None messages = ([ChatMessage(role='user', content=text)] 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)) # sandbox-image pull-ahead: register every pending sample's image the # moment the task list is known -- the background pool starts pulling # while generation is still in flight (swe: 1 image per sample) try: _imgs = {s.sandbox.image for s in work if getattr(s, 'sandbox', None) and s.sandbox.image} if _imgs: from ..sandbox.image_service import get_image_service get_image_service().register(sorted(_imgs)) except Exception: pass # let the adapter surface retry attempts to the bar members = getattr(adapter, 'adapters', [adapter]) for m_ in members: m_.extra['progress_reporter'] = progress_reporter # terminal (post-retry) sample failures are CONTAINED: one sample that # never makes it (server queue ate its first byte past every timeout) # must not kill the remaining hundreds -- it becomes an empty prediction # (scores as wrong, es-parity for timeouts), is NOT checkpointed (a # rerun retries it), and only a total wipeout fails the bench failed_samples: Dict[int, str] = {} async def run_one(i_s): i, s = i_s # NO outer retry: the adapter retries internally (and the pool # fails over per instance); a second loop here multiplied # worst-case time. One pass, one result or one contained error. try: pred = await one(s) except Exception as e: if progress_reporter is not None: progress_reporter.advance(success=False) failed_samples[i] = f'{type(e).__name__}: {str(e)[:120]}' return i, None # empty marker: no checkpoint write 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)) if failed_samples and len(failed_samples) >= len(pending): # every single fresh sample died: the endpoint is down, not flaky _f = next(iter(failed_samples.values())) raise RuntimeError(f'all {len(failed_samples)} generations failed ' f'(first: {_f})') if failed_samples: print(f'generation: {len(failed_samples)}/{len(pending)} samples ' 'failed after all retries (empty predictions, not ' 'checkpointed -- rerun to retry them); first: ' f'{next(iter(failed_samples.items()))[1][:100]}', flush=True) for i, pred in fresh: preds_by_key[keys[i]] = pred if pred is not None \ else {'raw': '', 'usage': {}, 'error': failed_samples.get(i, '')[:200]} 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') # ckpt info (store + per-position keys) so run_eval can read/write # SCORES bound to these predictions; None when checkpointing is off ckpt_info = (ckpt_store, keys) if ckpt_store is not None else None return preds, usages, total_usage, ckpt_info, len(fresh) - len(failed_samples) 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() _ASSEMBLE_EXEC = None # bounded truncation pool (lazy) 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 = 200, 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, on_scored=None, rescore: bool = False, ) -> 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, ckpt_info, n_fresh = 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() # SCORES ARE BOUND TO PREDICTIONS in the checkpoint: when every sample's # cached score matches the current scoring setup (recipe/extract/scorers/ # judge fingerprint), replay them without touching a single scorer -- # docker exec benches skip their containers entirely. --resume controls # the whole stack (no checkpoint -> nothing cached -> evaluate + backfill) _fp = None _records = None if ckpt_info is not None and not rescore: from ..eval.runner import score_fingerprint store, ck_keys = ckpt_info _fp = score_fingerprint(recipe, judge_spec or '') cached = store.scores() if ck_keys and all(cached.get(k, {}).get('fp') == _fp for k in ck_keys): _records = [cached[k] for k in ck_keys] if status_callback: status_callback('Scores cached in checkpoint -- replaying ' '(no scorers run; --rescore re-evaluates)') if judge is None and judge_spec and _records is None \ and _recipe_needs_judge(recipe): 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') # retarget the bar to scoring IMMEDIATELY (0/N): waiting for the first # on_scored left the stale GENERATION counters (100%, +N new) on screen # through minute-long preflights (image pulls) -- reading as 'done' if progress_reporter is not None and _records is None: _ss0 = getattr(progress_reporter, 'set_scoring', None) if _ss0 is not None: _ss0(0, len(samples)) _meta = {'gen_input_tokens': usage.input_tokens, 'gen_output_tokens': usage.output_tokens, 'gen_total_tokens': usage.total_tokens, # fresh=0 means the whole bench replayed from checkpoint: the # summary table then shows 'cached' instead of a ~0s time 'gen_fresh': n_fresh} if _records is not None: from ..eval.runner import evaluate_cached report = evaluate_cached(samples, preds, recipe, _records, model=model_spec, extra_metadata=_meta) else: # 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=_meta, on_scored=on_scored, ) # writeback: bind these scores to the predictions in the checkpoint if ckpt_info is not None: from ..eval.runner import score_fingerprint, score_record_of store, ck_keys = ckpt_info if _fp is None: _fp = score_fingerprint(recipe, judge_spec or '') store.put_scores({ck_keys[i]: score_record_of(report.samples[i], _fp) for i in range(min(len(ck_keys), len(report.samples)))}) 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 _recipe_needs_judge(recipe) -> bool: """Only construct the judge when a scorer actually consumes it -- rule-based benches (longbench_v2 etc.) never touch a judge, and building one there crashed on malformed specs for no benefit.""" try: for spec in (recipe.scorers or {}).values(): p = spec if isinstance(spec, dict) else {} if p.get('name') in ('llm_judge', 'judge'): return True except Exception: pass return False 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/'):] 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. # Segments WITHOUT a range are plain members: --auto-concurrency # wraps a single endpoint as a 1-member pool to get the gate. 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}') if not specs: raise ValueError('openai-pool needs at least one endpoint ' '(plain URL or {start..end} port range)') 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