Performance profile: TTFT/ITL via SSE streaming (collect_perf flag), retries/http_status per request, pool success/retry/failure rates, perf_stats aggregator (P50/P95/P99 latency+ttft, tokens, retry_rate) in metric_groups['perf']; verified on fake SSE server
This commit is contained in:
parent
456d304a69
commit
f2a914d0ff
@ -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)."""
|
||||
|
||||
@ -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:
|
||||
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}
|
||||
|
||||
@ -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,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
@ -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'
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user