895 lines
42 KiB
Python
895 lines
42 KiB
Python
"""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',
|
|
) -> 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: <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
|
|
|
|
# one root for everything: --cache-dir > $EVALHARNESS_CACHE >
|
|
# ~/.cache/evalharness (data cache and checkpoints stay together)
|
|
ckpt = checkpoint_path(str(get_cache_root()),
|
|
f'{dataset_name}:{sub}' if sub else dataset_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))
|
|
|
|
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
|
|
for attempt in range(6):
|
|
try:
|
|
pred = await one(s)
|
|
break
|
|
except Exception:
|
|
if attempt == 5:
|
|
if progress_reporter is not None:
|
|
progress_reporter.advance(success=False)
|
|
raise
|
|
# minute-scale backoff: cluster routes flap in multi-minute
|
|
# bursts; short retries exhaust inside one dead window
|
|
await asyncio.sleep((10, 20, 40, 60, 90)[attempt])
|
|
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]
|
|
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 '<model>'
|
|
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 = '',
|
|
) -> 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)
|
|
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 .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:
|
|
pass
|
|
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/<base-url-template>?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
|