diff --git a/evalharness/model/adapter.py b/evalharness/model/adapter.py index 4dea6a4..06a1417 100644 --- a/evalharness/model/adapter.py +++ b/evalharness/model/adapter.py @@ -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