"""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') _ADAPTER_CACHE = {} # spec -> shared instance; keeps pool round-robin state # GLOBAL across benches (else each pool restarts at the # first backend and starves the rest) 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) if spec in _ADAPTER_CACHE: return _ADAPTER_CACHE[spec] cls = ADAPTER_REGISTRY.get(parsed['adapter']) key = parsed.get('api_base') and _key_for(parsed['api_base']) inst = cls(model=parsed['model'], api_base=parsed['api_base'], api_key=key) _ADAPTER_CACHE[spec] = inst return inst 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 --------------------------- def _payload_chars(payload: Dict[str, Any]) -> int: """Rough prompt size in characters (~3-4 chars/token). Used to decide the auto-stream path: gateways that buffer whole requests make a huge INPUT as hang-prone as a huge output budget.""" n = 0 for m in payload.get('messages') or []: c = m.get('content') n += len(c) if isinstance(c, str) else 256 return n def _parse_text_tool_calls(text: str) -> list: """Extract tool calls from a text reply. Handles both shapes: - JSON array: [{"name":..,"arguments":{..}}] - Qwen3 native XML: {"name":..,"arguments":{..}} (for vLLM builds whose qwen3_xml parser doesn't convert to tool_calls) """ import re as _re out = [] for m in _re.finditer(r'\s*(\{.*?\})\s*', text, _re.S): try: obj = json.loads(m.group(1)) if isinstance(obj, dict) and obj.get('name'): out.append({'id': '', 'type': 'function', 'function': {'name': obj['name'], 'arguments': json.dumps(obj.get('arguments', {}))}}) except (ValueError, TypeError): continue if out: return out candidates = _re.findall(r'\[[\s\S]*?\]', text) or [] for cand in candidates: try: arr = json.loads(cand) if isinstance(arr, list) and arr and all(isinstance(x, dict) and 'name' in x for x in arr): return [{'id': '', 'type': 'function', 'function': {'name': x['name'], 'arguments': json.dumps(x.get('arguments', {}))}} for x in arr] except (ValueError, TypeError): continue # python-call style: [func(a=1, b="x")] or nested [[{..}]] JSON strings -- # dp4/DeepSeek text-protocol output shape (es feeds the same text to its # official decoders). Parse func(name=args) via a safe regex + literal_eval. def _py_call(m_): name = m_.group(1) argstr = (m_.group(2) or '').strip() args = {} if argstr: import ast as _ast try: parsed = _ast.parse(f'dummy({argstr})', mode='eval').body for kw_ in parsed.keywords: try: args[kw_.arg] = _ast.literal_eval(kw_.value) except (ValueError, SyntaxError): args[kw_.arg] = _ast.unparse(kw_.value) except SyntaxError: return None # literal_eval happily returns Ellipsis (code like `f(key=...)`) and # other non-JSON constants; default=str keeps the serializer alive # instead of killing the whole benchmark at parse time return {'id': '', 'type': 'function', 'function': {'name': name, 'arguments': json.dumps(args, default=str)}} for m_ in _re.finditer(r'([A-Za-z_][A-Za-z0-9_]*)\((.*?)\)', text): c = _py_call(m_) if c and c['function']['name']: out.append(c) return out @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: import time as _time t0 = _time.time() payload = self._payload(messages, tools, kw) headers = {'Content-Type': 'application/json'} if self.api_key: headers['Authorization'] = f'Bearer {self.api_key}' # long generations cost minutes per attempt -- retry far less # (3 for normal requests, 1 for >8k-token budgets) _mt = int((kw.get('max_tokens') or payload.get('max_tokens') or 4096)) retries = self.extra.get('retries', 3 if _mt > 8192 else 6) last_exc: Exception = None stream = bool(self.extra.get('collect_perf') and not kw.get('no_stream')) if stream: payload['stream'] = True for attempt in range(retries + 1): try: if stream: out = await self._post_stream_perf( f'{self.api_base}/chat/completions', payload, headers, t0) elif (int(payload.get('max_tokens') or 0) > 100000 or _payload_chars(payload) > 300_000) \ and not os.environ.get('EVALHARNESS_NO_AUTOSTREAM'): # long generation OR LONG INPUT: stream and aggregate # (gateway-safe). Some gateways buffer the whole request # before answering on the non-stream path -- a 128k-token # longbench_v2 prompt sat there past every read timeout; # streaming starts emitting immediately, so a stuck # endpoint surfaces in ~60s instead of after 20 minutes. # Some gateways drop chat_template_kwargs on the STREAM # path only (non-stream honors it) -- append the /no_think # soft switch into the prompt itself as a belt-and-braces # fallback (it IS the prompt, cannot be stripped) payload['stream'] = True if payload.get('chat_template_kwargs', {}).get('enable_thinking') is False: msgs = payload.get('messages') or [] for m in reversed(msgs): if m.get('role') == 'user': if '/no_think' not in (m.get('content') or ''): m['content'] = (m.get('content') or '') + ' /no_think' break data = await self._post_stream_aggregate( f'{self.api_base}/chat/completions', payload, headers) out = self._parse(data, allow_text_calls=bool(tools)) else: data = await self._post(f'{self.api_base}/chat/completions', payload, headers) out = self._parse(data, allow_text_calls=bool(tools)) out.usage.latency_s = round(_time.time() - t0, 3) out.usage.retries = attempt return out except Exception as e: # 5xx/429/timeouts: worth retrying last_exc = e # context-overflow 400: OUR tokenizer counted fewer tokens than # the server's -> shrink the prompt by 15% and retry (cross- # tokenizer safety margin, converges in 1-2 attempts) msg = str(e) resp_body = '' try: resp_body = e.response.text or '' except AttributeError: pass if ('maximum context length' in msg or 'maximum context length' in resp_body) \ and ('reduce the length' in resp_body or 'reduce the length' in msg): msgs = payload.get('messages') or [] for m in reversed(msgs): if m.get('role') == 'user': c = m.get('content') or '' if len(c) > 4000: keep = int(len(c) * 0.85) // 2 m['content'] = (f'{c[:keep]}\n\n...[context trimmed]...\n\n' f'{c[-keep:]}') break continue retryable = 'Server error' in str(e) or '504' in str(e) or '502' in str(e) \ or '429' in str(e) or 'timeout' in str(e).lower() \ or 'TimeoutException' in type(e).__name__ if attempt >= retries or not retryable: raise import asyncio # surface the retry to the progress bar (user visibility) if self.extra.get('progress_reporter') is not None: self.extra['progress_reporter'].set_retries(attempt + 1) await asyncio.sleep(min(2 ** attempt * 3, 120)) raise last_exc # unreachable async def _post_stream_perf(self, url, payload, headers, t0) -> ModelOutput: """SSE streaming request collecting TTFT/ITL; reassembles a full response then reuses the standard parser.""" import time as _time try: import httpx except ImportError: data = await self._post(url, {k: v for k, v in payload.items() if k != 'stream'}, headers) out = self._parse(data) out.usage.http_status = 200 return out chunks: List[Dict[str, Any]] = [] ttft = None last_tok_t = None itl_vals: List[float] = [] status = None import json as _json import httpx as _hx async with _hx.AsyncClient(timeout=_hx.Timeout( connect=self.extra.get('connect_timeout', 15), read=self.extra.get('timeout', 300), write=30, pool=15)) as client: async with client.stream('POST', url, json=payload, headers=headers) as resp: status = resp.status_code resp.raise_for_status() async for line in resp.aiter_lines(): if not line.startswith('data:'): continue body = line[5:].strip() if body == '[DONE]': break try: ev = _json.loads(body) except ValueError: continue now = _time.time() delta = (ev.get('choices') or [{}])[0].get('delta') or {} piece = delta.get('content') or delta.get('reasoning_content') if piece: if ttft is None: ttft = now - t0 elif last_tok_t is not None: itl_vals.append(now - last_tok_t) last_tok_t = now chunks.append(ev) def _piece(ev): d = (ev.get('choices') or [{}])[0].get('delta') or {} return d.get('content') or d.get('reasoning_content') or '' text = ''.join(_piece(ev) for ev in chunks) finish = '' for ev in chunks: fr = (ev.get('choices') or [{}])[0].get('finish_reason') if fr: finish = fr break usage_ev = next((ev for ev in reversed(chunks) if ev.get('usage')), None) data = { 'choices': [{'message': {'role': 'assistant', 'content': text}, 'finish_reason': finish}], 'usage': (usage_ev or {}).get('usage') or {}, 'model': self.model, } out = self._parse(data, allow_text_calls=True) out.usage.ttft_s = round(ttft, 3) if ttft is not None else None out.usage.itl_mean_s = round(sum(itl_vals) / len(itl_vals), 4) if itl_vals else None out.usage.http_status = status return out 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: if self.extra.get('tools_mode', 'api') == 'text': # TEXT-PROTOCOL fallback for backends without --enable-auto-tool-choice: # declare tools in the prompt; the model outputs JSON calls as text decl = '\n\n'.join( f"- {t.get('name')}: {t.get('description', '')} args={t.get('parameters', {})}" for t in tools) payload['messages'][-1]['content'] += ( '\n\nYou can call these functions:\n' + decl + '\nTo call, output ONLY a JSON array like ' '[{"name": "...", "arguments": {...}}] and nothing else.') else: payload['tools'] = [ {'type': 'function', 'function': t} if 'function' not in t else t for t in tools ] payload.pop('chat_template_kwargs', None) for k in ('temperature', 'max_tokens', 'top_p', 'stop', 'seed', 'response_format', 'chat_template_kwargs'): if kw.get(k) is not None: payload[k] = kw[k] payload.setdefault('max_tokens', self.extra.get('max_tokens', 4096)) # CoT room if self.extra.get('no_think'): if payload.get('tools'): # tools payloads: kwargs popped above (template 400 issue) # -> keep the soft switch (appended marker) as before msgs = payload.get('messages') or [] for m in reversed(msgs): if m.get('role') == 'user': if '/no_think' not in (m.get('content') or ''): m['content'] = (m.get('content') or '') + ' /no_think' break else: # plain payloads: template-level switch (clean -- no prompt # pollution; verified on the 8123-8130 vLLM pool) payload['chat_template_kwargs'] = {'enable_thinking': False} return payload def _parse(self, data: Dict[str, Any], allow_text_calls: bool = True) -> ModelOutput: choice = (data.get('choices') or [{}])[0] msg = choice.get('message') or {} calls = [] raw_calls = list(msg.get('tool_calls') or []) if not raw_calls and allow_text_calls: # text-protocol fallback ONLY for requests that carried tools: # running it on plain prose/code (humaneval!) regex-matched # `f(key=...)` style code as "calls", literal_eval'd the `...` # into an Ellipsis and crashed json.dumps mid-generation raw_calls = _parse_text_tool_calls(msg.get('content') or '') for c in raw_calls: fn = c.get('function') or {} args = fn.get('arguments') or '{}' try: args_dict = json.loads(args) if isinstance(args_dict, str): # double-encoded JSON string args_dict = json.loads(args_dict) if not isinstance(args_dict, dict): args_dict = {'raw': args_dict} 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', '')) text = msg.get('content') or '' if not text.strip(): # Qwen3/DeepSeek thinking models may put EVERYTHING in reasoning_content rc = msg.get('reasoning_content') or msg.get('reasoning') if isinstance(rc, str) and rc.strip(): text = rc return ModelOutput(text=text, tool_calls=calls, usage=usage, raw=data, model=data.get('model', self.model)) async def _post_stream_aggregate(self, url: str, payload: Dict, headers: Dict) -> Dict[str, Any]: """STREAM a long generation and reassemble the non-stream response shape. Some gateways hang on large NON-streaming requests (the full 32k-token body must be buffered before any byte is sent); streaming starts emitting immediately, so a stuck endpoint surfaces in ~60s instead of after the whole (possibly 80-minute) read timeout.""" import json as _json import httpx as _hx # read timeout bounds the wait for the FIRST byte too: a 128k-token # prompt queued behind other prefills legitimately takes minutes to # start answering -- 60s killed exactly those (the gate's x0.7 and # sample-failure containment now handle real hangs) async with _hx.AsyncClient(timeout=_hx.Timeout( connect=self.extra.get('connect_timeout', 15), read=self.extra.get('stream_read_timeout', 240), write=30, pool=15)) as client: async with client.stream('POST', url, json=payload, headers=headers) as resp: if resp.status_code != 200: body = (await resp.aread()).decode('utf-8', 'replace')[:300] raise RuntimeError(f'HTTP {resp.status_code}: {body}') content = [] reasoning = [] tool_calls = {} usage = {} finish = None async for line in resp.aiter_lines(): if not line.startswith('data:'): continue chunk = line[5:].strip() if chunk in ('', '[DONE]'): continue try: ev = _json.loads(chunk) except ValueError: continue u = ev.get('usage') if u: usage = u for ch in ev.get('choices') or []: delta = ch.get('delta') or {} if delta.get('content'): content.append(delta['content']) if delta.get('reasoning_content'): reasoning.append(delta['reasoning_content']) for tc in delta.get('tool_calls') or []: i = tc.get('index', 0) slot = tool_calls.setdefault(i, {'id': '', 'type': 'function', 'function': {'name': '', 'arguments': ''}}) fn = tc.get('function') or {} slot['id'] = tc.get('id') or slot['id'] slot['function']['name'] += fn.get('name') or '' slot['function']['arguments'] += fn.get('arguments') or '' if ch.get('finish_reason'): finish = ch['finish_reason'] msg = {'role': 'assistant', 'content': ''.join(content)} if reasoning: msg['reasoning_content'] = ''.join(reasoning) if tool_calls: msg['tool_calls'] = [tool_calls[i] for i in sorted(tool_calls)] return {'choices': [{'index': 0, 'message': msg, 'finish_reason': finish or 'stop'}], 'usage': usage or {}, 'model': payload.get('model', '')} async def _post(self, url: str, payload: Dict, headers: Dict) -> Dict[str, Any]: try: import httpx # read timeout scales with the generation budget: a 32k-token # CoT legitimately takes 10+ minutes; a fixed 300s timeout would # kill and retry it forever (looks like a hang) # generous: GLM gateway takes 30+ seconds to start responding # on 100k+ token inputs, even before any generation begins _rt = max(self.extra.get('timeout', 600), int(payload.get('max_tokens') or 0) * 0.15) async with httpx.AsyncClient(timeout=httpx.Timeout( connect=self.extra.get('connect_timeout', 15), read=_rt, write=30, pool=15)) 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{target} (oracle channel: runner tags a MOCKTARGET message when the sample carries a target) oracle -- return the target verbatim (same channel; for coding benches whose target is the canonical solution) fc -- replay the target's ground-truth tool calls (oracle for function-calling benches; target JSON in runner dict form) 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 in ('fc', 'tool'): if mode == 'tool': return ModelOutput(text='', tool_calls=[ToolCall( name=self.extra.get('tool_name', 'dummy_tool'), arguments='{}', arguments_dict={})], model='mock') target = None already_played = any(m.role == 'tool' for m in messages) if already_played: # oracle replays ground truth ONCE, then wraps up like a # well-behaved agent (final turn, no more calls) return ModelOutput(text='Done.', model='mock', usage=Usage( input_tokens=1, output_tokens=1, total_tokens=2, finish_reason='stop')) for m in reversed(messages): if m.role == 'user' and m.content.startswith('MOCKTARGET::'): target = m.content[len('MOCKTARGET::'):] break calls = [] if target: try: gt = json.loads(target) raw_calls = gt.get('tool_calls', gt if isinstance(gt, list) else []) for c in raw_calls: fn = c.get('function', c) args = fn.get('arguments', {}) calls.append(ToolCall( name=fn.get('name', ''), arguments=json.dumps(args), arguments_dict=args if isinstance(args, dict) else {})) except (ValueError, TypeError): calls = [] if not calls: return ModelOutput(text='no tool needed', model='mock', usage=Usage( input_tokens=1, output_tokens=1, total_tokens=2, finish_reason='stop')) return ModelOutput(text='', tool_calls=calls, model='mock', usage=Usage( input_tokens=1, output_tokens=len(calls), total_tokens=1 + len(calls), finish_reason='tool_calls')) else: last = next((m.content for m in reversed(messages) if m.role == 'user'), '') text = last target = None for m in reversed(messages): if m.role == 'user' and m.content.startswith('MOCKTARGET::'): target = m.content[len('MOCKTARGET::'):] break if mode == 'oracle': text = target if target is not None else last elif mode == 'boxed': 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'))