"""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)