80194e845e
The adapter dropped reasoning_effort/thinking from the payload, so the
middle rung of the ladder (es reference: full 98.3 / effort_low 94 /
no-think 82.3 on humaneval) was unreachable. Both keys now pass
through; --reasoning-effort {minimal,low,medium,high,max} overrides
the YAML, and config/effort_low.yaml mirrors default.yaml with
reasoning_effort: low for one-command low-thinking runs.
Probe on the endpoint: same question, default = 319 chars reasoning /
262 tok, low = 47 chars / 117 tok -- the server honors it.
Co-Authored-By: Claude <noreply@anthropic.com>
621 lines
28 KiB
Python
621 lines
28 KiB
Python
"""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<adapter>[a-z_]+)(/(?P<rest>.+))?$')
|
|
_DEPLOY_RE = re.compile(r'^deploy:(?P<deployer>[a-z0-9_-]+)/(?P<model>.+)$')
|
|
|
|
|
|
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: <tool_call>{"name":..,"arguments":{..}}</tool_call>
|
|
(for vLLM builds whose qwen3_xml parser doesn't convert to tool_calls)
|
|
"""
|
|
import re as _re
|
|
|
|
out = []
|
|
for m in _re.finditer(r'<tool_call>\s*(\{.*?\})\s*</tool_call>', 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',
|
|
'thinking', 'reasoning_effort'):
|
|
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'))
|