154 lines
4.4 KiB
Python
154 lines
4.4 KiB
Python
"""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:<tag> (image pin = environment pin;
|
|
override per-model via models.yaml so multiple versions coexist)
|
|
sglang -- docker run lmsysorg/sglang:<tag>
|
|
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):
|
|
"""Environment provisioning lives in the sandbox layer; this class only
|
|
declares the engine (image/args defaults + models.yaml overrides) and
|
|
acquires a shared, refcounted serve environment from it."""
|
|
|
|
engine_args: List[str] = []
|
|
default_image = ''
|
|
|
|
def deploy(self, model: str, cfg: Dict[str, Any]) -> Dict[str, str]:
|
|
from ..sandbox import serve_env
|
|
|
|
handle = serve_env(self.name, model, cfg, self.default_image, self.engine_args)
|
|
return {'api_base': handle.api_base, 'model': handle.model,
|
|
'container': handle.container, 'handle': handle}
|
|
|
|
|
|
@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:<deployer>/<model> 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()
|