34 lines
2.0 KiB
Python
34 lines
2.0 KiB
Python
"""Fixed 256-token single-stream timing; not a task-quality score."""
|
|
import argparse
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
p=argparse.ArgumentParser(); p.add_argument('--output',type=Path,required=True); a=p.parse_args()
|
|
key=(Path(__file__).resolve().parents[1]/'secrets/api-key').read_text().strip()
|
|
body=dict(model='qwen3.8-flash-next',messages=[dict(role='user',content='Write a Python function that merges overlapping intervals. Explain the algorithm, edge cases and complexity.')],
|
|
temperature=0,seed=6000,reasoning_effort='low',max_tokens=256,ignore_eos=True,
|
|
stream=True,stream_options={'include_usage':True})
|
|
with a.output.open('x') as out:
|
|
for i in range(4):
|
|
req=urllib.request.Request('http://127.0.0.1:8000/v1/chat/completions',data=json.dumps(body).encode(),
|
|
headers={'Content-Type':'application/json','Authorization':'Bearer '+key})
|
|
start=time.monotonic(); first=None; usage={}; content=''; reasoning=''
|
|
with urllib.request.urlopen(req,timeout=600) as resp:
|
|
for raw in resp:
|
|
if not raw.startswith(b'data: '): continue
|
|
data=raw[6:].strip()
|
|
if data==b'[DONE]': break
|
|
x=json.loads(data); usage=x.get('usage') or usage
|
|
for c in x.get('choices',[]):
|
|
d=c.get('delta',{}); s=d.get('content') or ''; r=d.get('reasoning') or d.get('reasoning_content') or ''
|
|
if first is None and (s or r): first=time.monotonic()
|
|
content+=s; reasoning+=r
|
|
end=time.monotonic(); tokens=usage['completion_tokens']
|
|
if tokens != 256 or first is None: raise RuntimeError('Incomplete timing sample')
|
|
row=dict(run=i,warmup=i==0,ttft_s=first-start,elapsed_s=end-start,
|
|
decode_tps=(tokens-1)/(end-first),usage=usage,content=content,reasoning=reasoning)
|
|
line=json.dumps(row); out.write(line+'\n'); out.flush()
|
|
print(json.dumps({k:v for k,v in row.items() if k not in ['content','reasoning']}),flush=True)
|