"""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 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, env_factory=None, system: str = '', max_turns: int = 8, max_input_chars: int = 0, attach_context_keys: tuple = ('passage', 'context'), ) -> tuple: """Fan out model calls; returns (pred-dicts, total_usage). Without env_factory: single-turn generation (text or tool-call JSON). With env_factory(sample)->Environment: the agent message pump runs per sample and predictions carry trajectory/env_state/usage. max_input_chars: hard cap on the assembled input (anti-OOM for 128k contexts); 0 = no cap. Truncation keeps the head AND the question tail. attach_context_keys: metadata fields (passage/context) prepended to the question at generation time -- the data layer keeps them separate, the runner assembles the actual prompt. """ gen_kwargs = gen_kwargs or {} def assemble(sample: Sample) -> str: parts = [] for key in attach_context_keys: ctx = (sample.metadata or {}).get(key) if ctx: parts.append(str(ctx)) parts.append(sample.input_text) text = '\n\n'.join(parts) 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 sem = asyncio.Semaphore(concurrency) total_usage = Usage() done_count = 0 t0 = time.time() usages: List[Dict[str, Any]] = [] 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 async with sem: traj = await drive(adapter, sample, env=env_factory(), max_turns=max_turns, system=system) total_usage = total_usage + traj.usage pred = trajectory_to_prediction(traj) pred['group_key'] = str(sample.metadata.get('test_category') or sample.metadata.get('category') or sample.metadata.get('id') or sample.id or '') done_count += 1 _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)) 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: out = await adapter.generate(messages, tools=tools, **gen_kwargs) 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 _progress(progress, done_count, len(samples), t0, total_usage) return {'raw': text, 'usage': out.usage.model_dump()} work = samples[:limit] if limit else samples preds = list(await asyncio.gather(*(one(s) for s in work))) usages = [p.get('usage', {}) for p in preds] return preds, usages, total_usage def _progress(progress: bool, done: int, total: int, t0: float, usage: Usage) -> None: if progress and (done % 20 == 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) 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, judge_spec: Optional[str] = None, judge: Optional[Any] = None, progress: bool = True, env: str = '', system: str = '', max_turns: int = 8, max_input_chars: int = 0, ) -> 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. """ adapter = _make_adapter(model_spec) spec = getattr(dataset, 'spec', None) 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'}}) samples = list(dataset)[:limit] if limit else list(dataset) if progress: 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 BFCLEnvironment envs = {'bfcl_mock': BFCLEnvironment} if env not in envs: raise KeyError(f"unknown env {env!r}; available: {', '.join(envs)}") env_factory = envs[env] try: preds, _usages, usage = await generate_predictions( adapter, samples, concurrency, progress=progress, gen_kwargs=gen_kwargs, env_factory=env_factory, system=system, max_turns=max_turns, max_input_chars=max_input_chars) finally: await adapter.close() if judge is None and judge_spec: judge_adapter = _make_adapter(judge_spec) judge = _judge_callable(judge_adapter) report = 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 return report def _make_adapter(spec: str) -> ModelAdapter: """'mock:boxed' -> MockAdapter(mode='boxed'); else resolve_adapter(). The colon-mode syntax exists ONLY for 'mock': adapter names contain no scheme/colon, so 'mock:xxx' is safe while URLs ('openai/http://...') must never be split on ':'. """ base, sep, mode = spec.partition(':') if sep and '/' not in base and base == 'mock': adapter = resolve_adapter('mock') adapter.extra['mode'] = mode or 'echo' return adapter return resolve_adapter(spec) 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