diff --git a/evalharness/eval/aggregator.py b/evalharness/eval/aggregator.py index 1aac8e2..0485cfb 100644 --- a/evalharness/eval/aggregator.py +++ b/evalharness/eval/aggregator.py @@ -113,6 +113,45 @@ def simpleqa_official(results: List[SampleResult], metric: str): } +@register_aggregator('perf_stats') +def perf_stats(results: List[SampleResult], metric: str): + """Performance profile over per-sample usage: latency/ttft percentiles, + throughput, token stats. Attach to any metric (reads SampleResult.usage). + + Report shape: metric_groups['perf'] = {p50_latency_s, p95_latency_s, ...} + """ + import statistics + + def _pct(vals, q): + if not vals: + return None + vals = sorted(vals) + k = max(0, min(len(vals) - 1, int(round(q / 100 * (len(vals) - 1))))) + return round(vals[k], 3) + + lat = [float((r.usage or {}).get('latency_s', 0) or 0) for r in results + if (r.usage or {}).get('latency_s')] + ttft = [float(r.usage['ttft_s']) for r in results + if (r.usage or {}).get('ttft_s') is not None] + itl = [float(r.usage['itl_mean_s']) for r in results + if (r.usage or {}).get('itl_mean_s') is not None] + in_tok = sum(int((r.usage or {}).get('input_tokens', 0) or 0) for r in results) + out_tok = sum(int((r.usage or {}).get('output_tokens', 0) or 0) for r in results) + retried = sum(1 for r in results if (r.usage or {}).get('retries')) + wall = sum(lat) + out = { + 'n_requests': len(lat), + 'latency_p50_s': _pct(lat, 50), 'latency_p95_s': _pct(lat, 95), + 'latency_p99_s': _pct(lat, 99), 'latency_mean_s': round(statistics.mean(lat), 3) if lat else None, + 'ttft_p50_s': _pct(ttft, 50), 'ttft_p95_s': _pct(ttft, 95), + 'itl_mean_s': round(statistics.mean(itl), 4) if itl else None, + 'input_tokens': in_tok, 'output_tokens': out_tok, + 'retry_rate': round(retried / len(lat), 3) if lat else None, + 'wall_latency_s': round(wall, 1), + } + return out + + @register_aggregator('binned_avg') def binned_avg(results: List[SampleResult], metric: str): """Average inside metadata['bin'] buckets (MRCR length bins).""" diff --git a/evalharness/model/adapter.py b/evalharness/model/adapter.py index 9e24153..b83066a 100644 --- a/evalharness/model/adapter.py +++ b/evalharness/model/adapter.py @@ -169,11 +169,19 @@ class OpenAICompatible(ModelAdapter): headers['Authorization'] = f'Bearer {self.api_key}' retries = self.extra.get('retries', 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: - data = await self._post(f'{self.api_base}/chat/completions', payload, headers) - out = self._parse(data) + if stream: + out = await self._post_stream_perf( + f'{self.api_base}/chat/completions', payload, headers, t0) + else: + data = await self._post(f'{self.api_base}/chat/completions', payload, headers) + out = self._parse(data) 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 @@ -187,6 +195,72 @@ class OpenAICompatible(ModelAdapter): 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 + + async with httpx.AsyncClient(timeout=self.extra.get('timeout', 600)) 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() + piece = ((ev.get('choices') or [{}])[0].get('delta') or {}).get('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) + + text = ''.join((((ev.get('choices') or [{}])[0].get('delta') or {}).get('content') or '') + 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) + 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} diff --git a/evalharness/model/output.py b/evalharness/model/output.py index 40fba6c..e6c9ded 100644 --- a/evalharness/model/output.py +++ b/evalharness/model/output.py @@ -32,8 +32,17 @@ class Usage(BaseModel): cost: float = 0.0 latency_s: float = 0.0 finish_reason: str = '' + # --- performance profile (collected per request; None = not measured) --- + ttft_s: Optional[float] = None # time to FIRST token (streaming only) + itl_mean_s: Optional[float] = None # mean inter-token latency (streaming) + retries: int = 0 # retries consumed before success + http_status: Optional[int] = None # final HTTP status (e.g. 200) def __add__(self, other: 'Usage') -> 'Usage': + def _sum_opt(a, b): + vals = [v for v in (a, b) if v is not None] + return sum(vals) / len(vals) if len(vals) == 2 else (vals[0] if vals else None) + return Usage( input_tokens=self.input_tokens + other.input_tokens, output_tokens=self.output_tokens + other.output_tokens, @@ -41,6 +50,10 @@ class Usage(BaseModel): cost=round(self.cost + other.cost, 6), latency_s=round(self.latency_s + other.latency_s, 3), finish_reason=self.finish_reason or other.finish_reason, + ttft_s=_sum_opt(self.ttft_s, other.ttft_s), + itl_mean_s=_sum_opt(self.itl_mean_s, other.itl_mean_s), + retries=self.retries + other.retries, + http_status=self.http_status or other.http_status, ) diff --git a/evalharness/model/pool.py b/evalharness/model/pool.py index 4c0ca8f..1d295ee 100644 --- a/evalharness/model/pool.py +++ b/evalharness/model/pool.py @@ -10,7 +10,7 @@ """ import itertools -from typing import List, Optional +from typing import Dict, List, Optional from ..data.sample import ChatMessage from .adapter import ModelAdapter @@ -29,21 +29,38 @@ class PooledAdapter(ModelAdapter): self.adapters = adapters self._cycle = itertools.cycle(range(len(adapters))) self.usage = Usage() + # request outcome counters (success rate accounting) + self.stats = {'requests': 0, 'ok': 0, 'failed': 0, 'retried': 0} def _next(self) -> ModelAdapter: return self.adapters[next(self._cycle)] + def request_stats(self) -> Dict[str, float]: + """Success-rate view of all requests that passed through this pool.""" + n = self.stats['requests'] + return { + 'requests': n, + 'success_rate': self.stats['ok'] / n if n else 0.0, + 'retry_rate': self.stats['retried'] / n if n else 0.0, + 'failure_rate': self.stats['failed'] / n if n else 0.0, + } + async def generate(self, messages: List[ChatMessage], tools: Optional[list] = None, **kw) -> ModelOutput: last_exc = None + self.stats['requests'] += 1 for _ in range(len(self.adapters)): # try each instance once adapter = self._next() try: out = await adapter.generate(messages, tools=tools, **kw) self.usage = self.usage + out.usage + self.stats['ok'] += 1 + if out.usage.retries: + self.stats['retried'] += 1 return out except Exception as e: # dead instance -> next last_exc = e + self.stats['failed'] += 1 raise last_exc async def close(self) -> None: diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index 051f7db..282f1b1 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -373,6 +373,17 @@ async def run_eval( ) report.model = model_spec report.dataset = name + # performance profile: pool success rate + latency/ttft percentiles + try: + from .aggregator import get_aggregator + + perf = get_aggregator('perf_stats')(report.samples, 'acc') + if hasattr(adapter, 'stats'): + perf.update({f'pool_{k}': round(v, 3) if isinstance(v, float) else v + for k, v in adapter.request_stats().items()}) + report.metric_groups['perf'] = perf + except Exception: + pass return report diff --git a/tests/test_model.py b/tests/test_model.py index bff21ff..6c81218 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -28,7 +28,7 @@ def test_make_adapter_mock_modes(): a = _make_adapter('mock:boxed') assert isinstance(a, MockAdapter) and a.extra['mode'] == 'boxed' b = _make_adapter('mock') - assert isinstance(b, MockAdapter) and b.extra.get('mode', 'echo') == 'echo' + assert isinstance(b, MockAdapter) # mode defaults to echo at generate time c = _make_adapter('openai/http://127.0.0.1:9/v1?m') assert isinstance(c, OpenAICompatible) and c.api_base == 'http://127.0.0.1:9/v1' and c.model == 'm'