Meta-Eval/EvalHarness
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Auto-stream long generations (max_tokens>8192) and aggregate to the non-stream response shape: gateways hang on large buffered NON-streaming requests (aime25's 32k budget stalled forever); streaming starts emitting immediately so a stuck endpoint surfaces in ~60s instead of the full adaptive read timeout; lazy httpx import fix

Browse Source 
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
sora 2026-09-11 03:41:24 +00:00
parent 5af8e2a7fd
commit 7473170784
1 changed files with 71 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
evalharness/model/adapter.py
View File

@ -202,6 +202,12 @@ class OpenAICompatible(ModelAdapter):
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) > 8192:
# long generation: stream and aggregate (gateway-safe)
payload['stream'] = True
data = await self._post_stream_aggregate(
f'{self.api_base}/chat/completions', payload, headers)
out = self._parse(data)
else:
data = await self._post(f'{self.api_base}/chat/completions', payload, headers)
out = self._parse(data)
@ -262,7 +268,9 @@ class OpenAICompatible(ModelAdapter):
status = None
import json as _json
async with httpx.AsyncClient(timeout=httpx.Timeout(
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:
@ -387,6 +395,68 @@ class OpenAICompatible(ModelAdapter):
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
async with _hx.AsyncClient(timeout=_hx.Timeout(
connect=self.extra.get('connect_timeout', 15),
read=60, 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