From 6b3bb330c71e1e9fcfedd079771d30ffea579549 Mon Sep 17 00:00:00 2001 From: sora Date: Mon, 24 Aug 2026 06:28:00 +0000 Subject: [PATCH] Add model layer: async ModelAdapter (openai_compatible + mock) returning structured ModelOutput(text,tool_calls,usage), Deployer registry (vllm/sglang docker-pinned via models.yaml, external), async run_eval generate->score, CLI --model, agent-ready SampleResult.trajectory/env_state, tests --- README.md | 50 +++++++- evalharness/cli.py | 32 +++-- evalharness/eval/record.py | 7 ++ evalharness/eval/runner.py | 6 + evalharness/model/__init__.py | 45 +++++++ evalharness/model/adapter.py | 227 ++++++++++++++++++++++++++++++++++ evalharness/model/deployer.py | 187 ++++++++++++++++++++++++++++ evalharness/model/output.py | 59 +++++++++ evalharness/model/runner.py | 168 +++++++++++++++++++++++++ tests/test_model.py | 118 ++++++++++++++++++ 10 files changed, 885 insertions(+), 14 deletions(-) create mode 100644 evalharness/model/__init__.py create mode 100644 evalharness/model/adapter.py create mode 100644 evalharness/model/deployer.py create mode 100644 evalharness/model/output.py create mode 100644 evalharness/model/runner.py create mode 100644 tests/test_model.py diff --git a/README.md b/README.md index 82c63fd..760b747 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,10 @@ # EvalHarness -A plugin-based LLM/agent evaluation harness. **Currently: the data layer + -the evaluation layer** (datasets & eval recipes as plugins, lazy -materialization cache, official-aligned scorers, report artifacts & console -visualization). Sandbox/model/tool/skill layers land one at a time. +A plugin-based LLM/agent evaluation harness. **Currently: data + evaluation + +model layers** (datasets/eval-recipes/model-adapters/deployers as plugins, +lazy materialization cache, official-aligned scorers, async generation, +report artifacts & console visualization). Sandbox/agent/tool/skill layers +land one at a time. ## Features @@ -147,6 +148,36 @@ Per-sample results keep `raw_prediction` + extraction note + score details; and re-running `evaluate()` re-scores the same predictions — the model is never re-queried. +## Model layer (calling + deploying, separate plugins on purpose) + +```bash +# generate + score in one command (async, concurrent) +evalharness eval run gsm8k --model mock:boxed --limit 100 # offline pipeline check +evalharness eval run gsm8k --model openai/http://gpu03:8000/v1?qwen3-8b +evalharness eval run hle --model openai/...?qwen3-8b --judge openai/...?gpt-4o +# future: --model deploy:vllm/qwen3-8b (Deployer pulls a pinned docker env) +``` + +Model spec grammar (plain strings): + +| spec | meaning | +|---|---| +| `mock` / `mock:boxed` / `mock:tool` | offline adapter (echo / oracle-boxed / tool-call) | +| `openai/?` | any OpenAI-protocol endpoint: vllm, sglang, lmdeploy, ollama, cloud APIs | +| `deploy:/` | Deployer resolves the endpoint first (vllm/sglang: pinned docker image; external: models.yaml) | + +Design: + +- **ModelAdapter = how to call** (protocol). All adapters are `async` and + return structured `ModelOutput(text, tool_calls, usage)` — the hinge the + future agent loops hang on; single-turn recipes just read `.text`. +- **Deployer = how to run** (environment, separate lifecycle). Docker images + are pinned per model in `models.yaml`, so `vllm:v0.9.2` and `vllm:v0.6.6` + stacks coexist on one machine; `external` connects to existing endpoints. +- **Async boundary = waiting on the model**: `run_eval()` fans out calls with + a semaphore (default 32), collects raws + per-sample usage, then hands them + to the synchronous `evaluate()`. Data/scoring stay sync (fast, CPU/disk). + ## Built-in datasets (28, official sources) | Family | Datasets (source) | @@ -257,6 +288,11 @@ EvalHarness/ │ │ ├── loader.py # raw loading (local/HF/ModelScope native) │ │ ├── dataset.py # Dataset: lazy materialize + cache + derived views │ │ └── datasets/ # 28 built-in single-file dataset plugins +│ ├── model/ # ---- model layer ---- +│ │ ├── output.py # ModelOutput/ToolCall/Usage (agent hinge) +│ │ ├── adapter.py # @register_adapter: openai_compatible / mock +│ │ ├── deployer.py # @register_deployer: vllm / sglang / external +│ │ └── runner.py # async run_eval(): generate -> evaluate │ ├── eval/ # ---- evaluation layer ---- │ │ ├── record.py # SampleResult / EvalReport artifacts │ │ ├── extractor.py # answer-extraction primitives (+cascades) @@ -279,7 +315,11 @@ EvalHarness/ - [x] Data layer (28 dataset plugins, lazy cache, native HF/ModelScope loaders) - [x] Evaluation layer (extract/score/aggregate plugins, official scorers, recipes) - [x] Visualization (console/markdown renderers over report artifacts) -- [ ] Model layer (ModelAdapter: unified URL-based invocation; wires llm_judge) +- [x] Model layer (async ModelAdapter openai_compatible+mock, ModelOutput + with tool_calls, Deployer registry vllm/sglang/external + models.yaml + env pinning, run_eval generate->score) +- [ ] Agent layer (loops: single_turn fast path today, react/plan_execute; + environments: tau2 user-sim, swe docker; SampleResult.trajectory ready) - [ ] Sandbox layer (materialize `Sample.sandbox`: lazy per-instance image pull, refcounted image unload, container lifecycle; `requires` gating) - [ ] Tool layer (data filter/synthesis/dedup/export; Dataset in, Dataset out) diff --git a/evalharness/cli.py b/evalharness/cli.py index b789ce1..c2ae38c 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -108,18 +108,27 @@ def _cmd_eval_list(_args) -> int: def _cmd_eval_run(args) -> int: - from evalharness.eval import evaluate, get_eval + import asyncio + from evalharness.data import get_dataset + from evalharness.viz import render ds = get_dataset(args.dataset, **_overrides(args)) - preds = [json.loads(line) for line in open(args.predictions, encoding='utf-8') if line.strip()] - preds = [p.get('raw', p.get('prediction', '')) if isinstance(p, dict) else p for p in preds] - report = evaluate(ds, preds, model=args.model) + if args.model: # generate + score in one go + from evalharness.model import run_eval + + report = asyncio.run(run_eval( + ds, args.model, concurrency=args.concurrency, limit=args.limit, + judge_spec=args.judge)) + else: + from evalharness.eval import evaluate + + preds = [json.loads(line) for line in open(args.predictions, encoding='utf-8') if line.strip()] + preds = [p.get('raw', p.get('prediction', '')) if isinstance(p, dict) else p for p in preds] + report = evaluate(ds, preds, model=args.model or 'preds') if args.out: report.save(args.out) print(f'saved -> {args.out}') - from evalharness.viz import render - print(render(report, style=args.style)) return 0 @@ -171,10 +180,15 @@ def build_parser() -> argparse.ArgumentParser: p = esub.add_parser('list', help='list registered eval recipes') p.set_defaults(func=_cmd_eval_list) - p = esub.add_parser('run', help='score predictions against a dataset') + p = esub.add_parser('run', help='score predictions (file) or generate+score (--model)') p.add_argument('dataset', help='dataset name (recipe auto-resolved)') - p.add_argument('predictions', help='jsonl: one raw string or {"raw": ...} per sample') - p.add_argument('--model', default='', help='model tag recorded in the report') + p.add_argument('predictions', nargs='?', help='jsonl: one raw string or {"raw": ...} per sample') + p.add_argument('--model', default='', + help="generate with model spec: mock | mock:boxed | " + "openai/http://host:8000/v1?model | deploy:vllm/model") + p.add_argument('--judge', default='', help='judge model spec for llm_judge recipes') + p.add_argument('--concurrency', type=int, default=32, help='parallel model calls (default 32)') + p.add_argument('--limit', type=int, help='evaluate only the first N samples') p.add_argument('--out', help='save the EvalReport json here') p.add_argument('--style', default='text', help='result render style (text/md/radar/errors)') _add_override_flags(p) diff --git a/evalharness/eval/record.py b/evalharness/eval/record.py index 2ea7ed8..4adc651 100644 --- a/evalharness/eval/record.py +++ b/evalharness/eval/record.py @@ -36,6 +36,13 @@ class SampleResult(BaseModel): metadata: Dict[str, Any] = Field(default_factory=dict) error: str = '' # scorer/extractor exception (never silently dropped) + # agent hinge: when a future AgentLoop produces multi-turn trajectories, + # they land here (list of ChatMessage dumps) + final environment state; + # env_reward scorers consume these instead of extracted text + trajectory: Optional[List[Dict[str, Any]]] = None + env_state: Optional[Dict[str, Any]] = None + usage: Optional[Dict[str, Any]] = None # tokens/cost/latency per sample + class EvalReport(BaseModel): """Aggregated artifact: what a viewer/visualizer consumes.""" diff --git a/evalharness/eval/runner.py b/evalharness/eval/runner.py index 94ee648..f596b7d 100644 --- a/evalharness/eval/runner.py +++ b/evalharness/eval/runner.py @@ -69,6 +69,12 @@ def evaluate( ) if isinstance(pred, dict) and pred.get('metadata'): result.metadata.update(pred['metadata']) + if isinstance(pred, dict) and pred.get('trajectory'): + result.trajectory = pred['trajectory'] + if isinstance(pred, dict) and pred.get('env_state'): + result.env_state = pred['env_state'] + if isinstance(pred, dict) and pred.get('usage'): + result.usage = pred['usage'] try: value, ok, note = extractor(raw, sample) result.extracted_prediction = value diff --git a/evalharness/model/__init__.py b/evalharness/model/__init__.py new file mode 100644 index 0000000..e6ff289 --- /dev/null +++ b/evalharness/model/__init__.py @@ -0,0 +1,45 @@ +"""evalharness.model -- calling and deploying models. + +Two plugin families, deliberately separate lifecycles: + ModelAdapter HOW to call (protocol): openai_compatible, mock, ... + Deployer HOW to run (environment): vllm, sglang, external (docker-pinned) + +Model spec grammar (strings everywhere, no config ceremony): + mock / mock:boxed / mock:tool offline + openai/http://host:8000/v1?model_id any OpenAI-protocol endpoint (vllm, sglang, cloud) + deploy:vllm/qwen3-8b Deployer resolves endpoint (models.yaml pins env) + +All adapters are async and return structured ModelOutput(text, tool_calls, +usage) -- the hinge the future agent loops hang on. +""" + +from .adapter import ( + ADAPTER_REGISTRY, + ModelAdapter, + OpenAICompatible, + MockAdapter, + register_adapter, + resolve_adapter, + parse_model_spec, +) +from .deployer import ( + DEPLOYER_REGISTRY, + Deployer, + VLLMDeployer, + SGLangDeployer, + External, + deploy, + load_model_config, + register_deployer, + stop_all, +) +from .output import ModelOutput, ToolCall, Usage +from .runner import generate_predictions, run_eval + +__all__ = [ + 'ModelAdapter', 'OpenAICompatible', 'MockAdapter', 'register_adapter', + 'resolve_adapter', 'parse_model_spec', 'ADAPTER_REGISTRY', + 'Deployer', 'VLLMDeployer', 'SGLangDeployer', 'External', + 'register_deployer', 'deploy', 'stop_all', 'load_model_config', 'DEPLOYER_REGISTRY', + 'ModelOutput', 'ToolCall', 'Usage', 'generate_predictions', 'run_eval', +] diff --git a/evalharness/model/adapter.py b/evalharness/model/adapter.py new file mode 100644 index 0000000..2386211 --- /dev/null +++ b/evalharness/model/adapter.py @@ -0,0 +1,227 @@ +"""ModelAdapter: HOW to call a model (protocol level). Never deploys anything. + +Registry + two built-ins: + openai_compatible -- /v1/chat/completions; covers vllm, sglang, lmdeploy, + ollama, tgi, and every OpenAI-protocol cloud API + mock -- offline deterministic/testing adapter + +Model spec grammar (a single string, no config files needed for the common case): + openai/http://localhost:8000/v1?qwen3-8b adapter + api_base + model id + openai/https://api.openai.com/v1?gpt-4o (OPENAI_API_KEY read from env) + anthropic/claude-... (provider-native protocols later) + mock offline + deploy:vllm/qwen3-8b -> Deployer resolves to an endpoint, + then re-dispatches as openai/... +""" + +import json +import os +import re +from typing import Any, Dict, List, Optional + +from ..data.sample import ChatMessage +from ..eval.registry import EvalRegistry +from .output import ModelOutput, ToolCall, Usage + +ADAPTER_REGISTRY = EvalRegistry('model adapter') + + +def register_adapter(name: str): + def decorator(cls): + ADAPTER_REGISTRY.register(name, cls) + return cls + + return decorator + + +class ModelAdapter: + """Base class. Subclasses implement generate() (async, structured output).""" + + name = 'base' + + def __init__(self, model: str = '', api_base: str = '', api_key: str = '', **kwargs): + self.model = model + self.api_base = api_base.rstrip('/') + self.api_key = api_key + self.extra = kwargs + + async def generate( + self, + messages: List[ChatMessage], + tools: Optional[List[Dict[str, Any]]] = None, + **gen_kwargs, + ) -> ModelOutput: + raise NotImplementedError + + async def close(self) -> None: + pass + + def __repr__(self): + return f'{type(self).__name__}(model={self.model!r}, base={self.api_base!r})' + + +# --------------------------- spec parsing / dispatch --------------------------- + +_SPEC_RE = re.compile(r'^(?P[a-z_]+)(/(?P.+))?$') +_DEPLOY_RE = re.compile(r'^deploy:(?P[a-z0-9_-]+)/(?P.+)$') + + +def parse_model_spec(spec: str) -> Dict[str, str]: + """'openai/http://x:8000/v1?qwen' -> {adapter, api_base, model}.""" + m = _SPEC_RE.match(spec.strip()) + if not m: + raise ValueError(f'bad model spec: {spec!r} (expected adapter/rest?model)') + adapter = m.group('adapter') + rest = m.group('rest') or '' + api_base, _, model = rest.partition('?') + return {'adapter': adapter, 'api_base': api_base, 'model': model} + + +def resolve_adapter(spec: str, deploy_fn=None) -> ModelAdapter: + """Spec -> adapter instance. deploy:... specs go through a Deployer first.""" + m = _DEPLOY_RE.match(spec.strip()) + if m: + if deploy_fn is None: + from .deployer import deploy as default_deploy + deploy_fn = default_deploy + endpoint = deploy_fn(m.group('deployer'), m.group('model')) + spec = f"openai/{endpoint['api_base']}?{endpoint['model']}" + parsed = parse_model_spec(spec) + cls = ADAPTER_REGISTRY.get(parsed['adapter']) + key = parsed.get('api_base') and _key_for(parsed['api_base']) + return cls(model=parsed['model'], api_base=parsed['api_base'], api_key=key) + + +def _key_for(api_base: str) -> str: + """Best-effort API key by endpoint; explicit env always wins.""" + for host_hint, var in (('api.openai.com', 'OPENAI_API_KEY'), + ('anthropic.com', 'ANTHROPIC_API_KEY'), + ('dashscope', 'DASHSCOPE_API_KEY'), + ('bigmodel', 'ZAI_API_KEY')): + if host_hint in api_base: + return os.environ.get(var, '') + return os.environ.get('OPENAI_API_KEY', '') + + +# --------------------------- openai_compatible --------------------------- + + +@register_adapter('openai') +class OpenAICompatible(ModelAdapter): + """Async OpenAI chat-completions client with zero hard dependencies. + + Uses httpx if installed (proper async); falls back to urllib in a thread + so the core install stays dependency-free. + """ + + name = 'openai' + + async def generate(self, messages, tools=None, **kw) -> ModelOutput: + payload = self._payload(messages, tools, kw) + headers = {'Content-Type': 'application/json'} + if self.api_key: + headers['Authorization'] = f'Bearer {self.api_key}' + data = await self._post(f'{self.api_base}/chat/completions', payload, headers) + return self._parse(data) + + def _payload(self, messages, tools, kw) -> Dict[str, Any]: + msgs = [{'role': m.role, 'content': m.content} for m in messages] + payload: Dict[str, Any] = {'model': self.model, 'messages': msgs} + if tools: + payload['tools'] = [ + {'type': 'function', 'function': t} if 'function' not in t else t for t in tools + ] + for k in ('temperature', 'max_tokens', 'top_p', 'stop', 'seed', 'response_format'): + if kw.get(k) is not None: + payload[k] = kw[k] + return payload + + def _parse(self, data: Dict[str, Any]) -> ModelOutput: + choice = (data.get('choices') or [{}])[0] + msg = choice.get('message') or {} + calls = [] + for c in msg.get('tool_calls') or []: + fn = c.get('function') or {} + args = fn.get('arguments') or '{}' + try: + args_dict = json.loads(args) + except (ValueError, TypeError): + args_dict = {} + calls.append(ToolCall(id=c.get('id', ''), name=fn.get('name', ''), + arguments=args, arguments_dict=args_dict)) + u = data.get('usage') or {} + usage = Usage(input_tokens=u.get('prompt_tokens', 0), + output_tokens=u.get('completion_tokens', 0), + total_tokens=u.get('total_tokens', 0), + finish_reason=choice.get('finish_reason', '')) + return ModelOutput(text=msg.get('content') or '', tool_calls=calls, usage=usage, + raw=data, model=data.get('model', self.model)) + + async def _post(self, url: str, payload: Dict, headers: Dict) -> Dict[str, Any]: + try: + import httpx + + async with httpx.AsyncClient(timeout=self.extra.get('timeout', 600)) as client: + r = await client.post(url, json=payload, headers=headers) + r.raise_for_status() + return r.json() + except ImportError: + import asyncio + import urllib.request + + def _sync(): + req = urllib.request.Request(url, data=json.dumps(payload).encode(), + headers=headers, method='POST') + with urllib.request.urlopen(req, timeout=self.extra.get('timeout', 600)) as resp: + return json.loads(resp.read().decode()) + + return await asyncio.to_thread(_sync) + + +# --------------------------- mock --------------------------- + +_MOCK_PATTERNS = ( + (re.compile(r'\\boxed\{([^}]*)\}'), None), # echo any boxed target in the input +) + + +@register_adapter('mock') +class MockAdapter(ModelAdapter): + """Offline adapter for tests/CI/dev. + + Modes (extra['mode']): + echo -- return the input text (default) + boxed -- return \\boxed{} scraped from the last user message + (metadata['mock_target'] or first number found) + tool -- return one tool call named extra['tool_name'] + const -- return extra['text'] + """ + + name = 'mock' + + async def generate(self, messages, tools=None, **kw) -> ModelOutput: + mode = self.extra.get('mode', 'echo') + if mode == 'const': + text = self.extra.get('text', 'mock') + elif mode == 'tool': + name = self.extra.get('tool_name', 'dummy_tool') + return ModelOutput(text='', tool_calls=[ToolCall( + name=name, arguments='{}', arguments_dict={})], model='mock') + else: + last = next((m.content for m in reversed(messages) if m.role == 'user'), '') + text = last + if mode == 'boxed': + # cheat-mode for pipeline verification: an oracle message + # (metadata-free) can't know the target, so the runner tags + # the message with 'mock_target' when samples carry one + target = None + for m in reversed(messages): + if m.role == 'user' and m.content.startswith('MOCKTARGET::'): + target = m.content[len('MOCKTARGET::'):] + break + if target is None: + m = _MOCK_PATTERNS[0][0].search(last) + target = m.group(1) if m else (re.findall(r'-?\d+\.?\d*', last) or ['0'])[-1] + text = f'The answer is \\boxed{{{target}}}.' + return ModelOutput(text=text, model='mock', usage=Usage( + input_tokens=1, output_tokens=1, total_tokens=2, finish_reason='stop')) diff --git a/evalharness/model/deployer.py b/evalharness/model/deployer.py new file mode 100644 index 0000000..cf1b583 --- /dev/null +++ b/evalharness/model/deployer.py @@ -0,0 +1,187 @@ +"""Deployer: HOW to run a model (environment level). Never scores anything. + +Separate lifecycle from calling on purpose: a deployment is slow (minutes), +stateful (port/GPU), and shareable across eval jobs; calling is stateless HTTP. + +Built-ins: + vllm -- docker run vllm/vllm-openji: (image pin = environment pin; + override per-model via models.yaml so multiple versions coexist) + sglang -- docker run lmsysorg/sglang: + external -- nothing to do; endpoint already exists (default for cloud APIs) + +Environment binding is DECLARATIVE (models.yaml), never code: + + models: + qwen3-8b: + deployer: vllm + image: vllm/vllm-openai:v0.9.2 # pinned env + gpus: '0' + max_model_len: 32768 + qwen3-8b-old-stack: + deployer: vllm + image: vllm/vllm-openai:v0.6.6.post1 # same model, different env, coexists + port: 8001 + +resolve('vllm', 'qwen3-8b') -> {'api_base': ..., 'model': ...} +""" + +import os +import shlex +import subprocess +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +from ..eval.registry import EvalRegistry + +DEPLOYER_REGISTRY = EvalRegistry('deployer') + + +def register_deployer(name: str): + def decorator(cls): + DEPLOYER_REGISTRY.register(name, cls) + return cls + + return decorator + + +def _models_yaml_path() -> Path: + return Path(os.environ.get('EVALHARNESS_MODELS', 'models.yaml')).expanduser() + + +_CONFIG: Optional[Dict[str, Dict[str, Any]]] = None + + +def load_model_config() -> Dict[str, Dict[str, Any]]: + """models.yaml -> {model_name: {deployer, image, ...}} (cached; {} if absent).""" + global _CONFIG + if _CONFIG is None: + import yaml # optional; falls back to {} without it + + path = _models_yaml_path() + _CONFIG = {} + if path.exists(): + with open(path, encoding='utf-8') as f: + _CONFIG = (yaml.safe_load(f) or {}).get('models', {}) or {} + return _CONFIG + + +def _free_port() -> int: + import socket + + with socket.socket() as s: + s.bind(('', 0)) + return s.getsockname()[1] + + +class Deployer: + """Base class: deploy(name, cfg) -> {'api_base', 'model'} (idempotent).""" + + name = 'base' + + def deploy(self, model: str, cfg: Dict[str, Any]) -> Dict[str, str]: + raise NotImplementedError + + def stop(self, handle: Dict[str, Any]) -> None: + pass + + +@register_deployer('external') +class External(Deployer): + """Endpoint already exists; cfg: api_base, model, api_key.""" + + name = 'external' + + def deploy(self, model: str, cfg: Dict[str, Any]) -> Dict[str, str]: + api_base = cfg.get('api_base', '') + if not api_base: + raise ValueError(f"external deployer for {model!r} needs api_base in models.yaml") + return {'api_base': api_base, 'model': cfg.get('model', model)} + + +class DockerServeDeployer(Deployer): + """Shared docker-run logic for OpenAI-protocol serving engines.""" + + engine_args: List[str] = [] + + def deploy(self, model: str, cfg: Dict[str, Any]) -> Dict[str, str]: + image = cfg.get('image', self.default_image) + port = int(cfg.get('port', 0) or _free_port()) + hf = cfg.get('hf_home', os.environ.get('HF_HOME', '~/.cache/huggingface')) + gpus = cfg.get('gpus', 'all') + cmd = [ + 'docker', 'run', '-d', '--rm', + '--name', f'evalharness-{self.name}-{model}-{port}'.replace('/', '-'), + '--gpus', f'device={gpus}' if str(gpus).isdigit() else str(gpus), + '-p', f'{port}:8000', + '-v', f'{Path(hf).expanduser()}:/root/.cache/huggingface', + '-e', f'HF_ENDPOINT={os.environ.get("HF_ENDPOINT", "https://hf-mirror.com")}', + image, '--model', cfg.get('model_id', cfg.get('model', model)), + '--served-model-name', model, + *self.engine_args, + *shlex.split(cfg.get('extra_args', '')), + ] + if cfg.get('gpu_mem_util'): + cmd += ['--gpu-memory-utilization', str(cfg['gpu_mem_util'])] + if cfg.get('max_model_len'): + cmd += ['--max-model-len', str(cfg['max_model_len'])] + container = subprocess.run(cmd, capture_output=True, text=True, check=True).stdout.strip() + api_base = f'http://localhost:{port}/v1' + self._wait_healthy(api_base, cfg.get('timeout_s', 1800)) + return {'api_base': api_base, 'model': model, 'container': container} + + def _wait_healthy(self, api_base: str, timeout_s: int) -> None: + deadline = time.time() + timeout_s + while time.time() < deadline: + try: + import urllib.request + + req = urllib.request.Request(f'{api_base}/models') + with urllib.request.urlopen(req, timeout=5) as resp: + if resp.status == 200: + return + except Exception: + time.sleep(5) + raise TimeoutError(f'serving engine not healthy after {timeout_s}s at {api_base}') + + def stop(self, handle: Dict[str, Any]) -> None: + if handle.get('container'): + subprocess.run(['docker', 'rm', '-f', handle['container']], check=False) + + +@register_deployer('vllm') +class VLLMDeployer(DockerServeDeployer): + name = 'vllm' + default_image = 'vllm/vllm-openai:v0.9.2' + + +@register_deployer('sglang') +class SGLangDeployer(DockerServeDeployer): + name = 'sglang' + default_image = 'lmsysorg/sglang:latest' + + +_ACTIVE: Dict[str, Dict[str, str]] = {} + + +def deploy(deployer: str, model: str) -> Dict[str, str]: + """Resolve deploy:/ specs. Idempotent per (deployer, model).""" + key = f'{deployer}/{model}' + if key in _ACTIVE: + return _ACTIVE[key] + cfg = dict(load_model_config().get(model, {})) + cls = DEPLOYER_REGISTRY.get(deployer) + handle = cls().deploy(model, cfg) + _ACTIVE[key] = handle + return handle + + +def stop_all() -> None: + """Stop everything this process started (atexit-registered by runner).""" + for key, handle in _ACTIVE.items(): + cls = DEPLOYER_REGISTRY.get(key.split('/', 1)[0]) + try: + cls().stop(handle) + except Exception: + pass + _ACTIVE.clear() diff --git a/evalharness/model/output.py b/evalharness/model/output.py new file mode 100644 index 0000000..40fba6c --- /dev/null +++ b/evalharness/model/output.py @@ -0,0 +1,59 @@ +"""Structured model output -- the hinge between single-turn eval and agents. + +Every ModelAdapter returns ModelOutput, never a bare string: + - single-turn recipes read .text + - agent loops read .tool_calls and feed observations back + - accounting/monitoring reads .usage +""" + +import time +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + + +class ToolCall(BaseModel): + """One function call the model wants executed (OpenAI tool_calls shape).""" + + id: str = '' + name: str + arguments: str = '' # JSON-encoded args string + arguments_dict: Dict[str, Any] = Field(default_factory=dict) # parsed convenience + + def to_openai(self) -> Dict[str, Any]: + return {'id': self.id or f'call_{self.name}', 'type': 'function', + 'function': {'name': self.name, 'arguments': self.arguments or '{}'}} + + +class Usage(BaseModel): + input_tokens: int = 0 + output_tokens: int = 0 + total_tokens: int = 0 + cost: float = 0.0 + latency_s: float = 0.0 + finish_reason: str = '' + + def __add__(self, other: 'Usage') -> 'Usage': + return Usage( + input_tokens=self.input_tokens + other.input_tokens, + output_tokens=self.output_tokens + other.output_tokens, + total_tokens=self.total_tokens + other.total_tokens, + cost=round(self.cost + other.cost, 6), + latency_s=round(self.latency_s + other.latency_s, 3), + finish_reason=self.finish_reason or other.finish_reason, + ) + + +class ModelOutput(BaseModel): + """What every adapter returns. text may be '' when the model only calls tools.""" + + text: str = '' + tool_calls: List[ToolCall] = Field(default_factory=list) + usage: Usage = Field(default_factory=Usage) + raw: Optional[Dict[str, Any]] = None # provider response (audit/retry) + model: str = '' + created_at: str = Field(default_factory=lambda: time.strftime('%Y-%m-%d %H:%M:%S')) + + @property + def is_tool_call(self) -> bool: + return bool(self.tool_calls) diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py new file mode 100644 index 0000000..92158cd --- /dev/null +++ b/evalharness/model/runner.py @@ -0,0 +1,168 @@ +"""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, +) -> tuple: + """Fan out model calls; returns (raws, total_usage). + + Each sample becomes one user message (or its ChatMessage list is used + verbatim for multi-turn samples). Tool declarations from sample.tools + are passed through so fc/agent recipes degrade gracefully today and + agent loops can reuse this adapter untouched. + """ + gen_kwargs = gen_kwargs or {} + sem = asyncio.Semaphore(concurrency) + total_usage = Usage() + done_count = 0 + t0 = time.time() + raws: List[str] = [] + usages: List[Dict[str, Any]] = [] + + async def one(sample: Sample) -> tuple: + nonlocal done_count, total_usage + messages = ([ChatMessage(role='user', content=sample.input)] 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') == 'boxed' \ + and sample.target not in ('', None): + # oracle channel for mock:boxed so full pipelines verify 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) + usage = out.usage.model_dump() + done_count += 1 + if progress and (done_count % 20 == 0 or done_count == len(samples)): + rate = done_count / max(time.time() - t0, 1e-6) + print(f' [{done_count}/{len(samples)}] {rate:.1f} samples/s ' + f'tokens={total_usage.total_tokens}', flush=True) + return text, usage + + work = samples[:limit] if limit else samples + pairs = await asyncio.gather(*(one(s) for s in work)) + raws = [p[0] for p in pairs] + usages = [p[1] for p in pairs] + return raws, usages, total_usage + + +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, +) -> EvalReport: + """Generate + score in one call. Model spec examples: + 'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'. + """ + 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: + print(f'generating: {adapter} on {len(samples)} samples ' + f'(concurrency={concurrency})', flush=True) + try: + raws, usages, usage = await generate_predictions(adapter, samples, concurrency, + progress=progress, gen_kwargs=gen_kwargs) + finally: + await adapter.close() + if judge is None and judge_spec: + judge_adapter = _make_adapter(judge_spec) + judge = _judge_callable(judge_adapter) + + preds = [{'raw': r, 'usage': u} for r, u in zip(raws, usages)] + 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): + async def ask(messages) -> str: + out = await judge_adapter.generate([ChatMessage(role='user', content=str(m)) for m in messages] + if isinstance(messages, list) and messages and isinstance(messages[0], dict) + else messages) + return out.text + + import asyncio + + def sync_ask(messages): + return asyncio.run(ask(messages)) + + return sync_ask diff --git a/tests/test_model.py b/tests/test_model.py new file mode 100644 index 0000000..bff21ff --- /dev/null +++ b/tests/test_model.py @@ -0,0 +1,118 @@ +"""Model layer tests: spec parsing, adapters, async runner end-to-end. + +Run: .venv/bin/python tests/test_model.py +""" + +import asyncio +import json +import sys +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from evalharness.data.sample import Sample, ToolInfo # noqa: E402 +from evalharness.model import MockAdapter, OpenAICompatible, resolve_adapter, run_eval # noqa: E402 +from evalharness.model.adapter import parse_model_spec # noqa: E402 +from evalharness.model.runner import _make_adapter # noqa: E402 + + +def test_parse_spec(): + got = parse_model_spec('openai/http://gpu03:8000/v1?qwen3-8b') + assert got == {'adapter': 'openai', 'api_base': 'http://gpu03:8000/v1', 'model': 'qwen3-8b'} + assert parse_model_spec('mock') == {'adapter': 'mock', 'api_base': '', 'model': ''} + + +def test_make_adapter_mock_modes(): + a = _make_adapter('mock:boxed') + assert isinstance(a, MockAdapter) and a.extra['mode'] == 'boxed' + b = _make_adapter('mock') + assert isinstance(b, MockAdapter) and b.extra.get('mode', 'echo') == 'echo' + c = _make_adapter('openai/http://127.0.0.1:9/v1?m') + assert isinstance(c, OpenAICompatible) and c.api_base == 'http://127.0.0.1:9/v1' and c.model == 'm' + + +def test_mock_tool_mode(): + out = asyncio.run(MockAdapter().generate([Sample(input='x').input + and __import__('evalharness.data.sample', fromlist=['ChatMessage']).ChatMessage(role='user', content='x')])) + assert out.text == 'x' + + +def _fake_server(): + class Fake(BaseHTTPRequestHandler): + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + msg = body['messages'][-1]['content'] + tool_calls = [] + if body.get('tools'): + tool_calls = [{'id': 'c1', 'function': {'name': 't1', 'arguments': '{"x": 1}'}}] + resp = {'choices': [{'message': {'content': f'echo:{msg}', 'tool_calls': tool_calls}, + 'finish_reason': 'tool_calls' if tool_calls else 'stop'}], + 'usage': {'prompt_tokens': 3, 'completion_tokens': 5, 'total_tokens': 8}, + 'model': body['model']} + data = json.dumps(resp).encode() + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *a): + pass + + srv = HTTPServer(('127.0.0.1', 0), Fake) + threading.Thread(target=srv.serve_forever, daemon=True).start() + return srv + + +def test_openai_adapter_roundtrip(): + srv = _fake_server() + port = srv.server_address[1] + try: + rep = asyncio.run(run_eval([Sample(input='hello', target='echo:hello')], + f'openai/http://127.0.0.1:{port}/v1?fake-m', + progress=False)) + r = rep.samples[0] + assert 'echo:hello' in r.raw_prediction + assert r.usage['total_tokens'] == 8 + finally: + srv.shutdown() + + +def test_tools_pass_through(): + srv = _fake_server() + port = srv.server_address[1] + try: + s = Sample(input='call it', tools=[ToolInfo(name='f1', parameters={'type': 'object'})]) + rep = asyncio.run(run_eval([s], f'openai/http://127.0.0.1:{port}/v1?fake-m', + progress=False)) + assert '"t1"' in rep.samples[0].raw_prediction # tool call serialized into prediction + finally: + srv.shutdown() + + +def test_oracle_mock_pipeline(): + from evalharness import get_dataset + + ds = get_dataset('gsm8k') + rep = asyncio.run(run_eval(ds, 'mock:boxed', limit=50, concurrency=8, progress=False)) + assert rep.metrics['acc'] == 1.0 + assert rep.dataset == 'gsm8k' + assert rep.metric_groups['run_info']['gen_total_tokens'] == 100 + + +if __name__ == '__main__': + fails = 0 + for name, fn in sorted({k: v for k, v in globals().items() + if k.startswith('test_') and callable(v)}.items()): + try: + fn() + print(f'PASS {name}') + except AssertionError as e: + fails += 1 + print(f'FAIL {name}: {e}') + except Exception as e: + fails += 1 + print(f'ERROR {name}: {type(e).__name__}: {e}') + sys.exit(1 if fails else 0)