"""Paired API qualification. Run identically on baseline and MTP; synthetic data only.""" import argparse,base64,concurrent.futures,hashlib,json,random,time,urllib.request from pathlib import Path p=argparse.ArgumentParser();p.add_argument('--output',type=Path,required=True);p.add_argument('--soak-seconds',type=int,default=600);p.add_argument('--screen',action='store_true',help='Skip 16K output and soak for parameter screening');p.add_argument('--assets',type=Path,default=Path('/data/flash-next/cache/qualification'));a=p.parse_args() key=(Path(__file__).resolve().parents[1]/'secrets/api-key').read_text().strip();url='http://127.0.0.1:8000';model='qwen3.8-flash-next' def req(path,body=None): return urllib.request.urlopen(urllib.request.Request(url+path,data=None if body is None else json.dumps(body).encode(),headers={'Authorization':'Bearer '+key,'Content-Type':'application/json'}),timeout=1200) def run(name,messages,limit=4096,expected=None,forced=False,effort='low',extra=None): body=dict(model=model,messages=messages,temperature=0,seed=6000,max_tokens=limit,reasoning_effort=effort,stream=True,stream_options={'include_usage':True});body.update(extra or {}) if forced:body['ignore_eos']=True start=time.monotonic();first=None;content='';reasoning='';finish=None;usage={};events=0;max_gap=0;prev=None try: with req('/v1/chat/completions',body) 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) if x.get('error'):raise RuntimeError(str(x['error'])) 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 s or r: now=time.monotonic();first=first or now;events+=1 if prev is not None:max_gap=max(max_gap,now-prev) prev=now content+=s;reasoning+=r;finish=c.get('finish_reason') or finish end=time.monotonic();n=usage.get('completion_tokens',0) passed=bool(first and n and finish) if forced:passed=passed and n==limit and finish=='length' elif expected is not None: if isinstance(expected,list):passed=passed and all(str(v).lower() in content.lower() for v in expected) and finish=='stop' else:passed=passed and content.strip()==expected and finish=='stop' else:passed=passed and finish=='stop' and bool(content.strip()) return dict(name=name,passed=passed,finish=finish,usage=usage,ttft_s=None if first is None else first-start,elapsed_s=end-start,decode_tps=None if first is None else (n-1)/max(end-first,1e-6),max_stream_gap_s=max_gap,events=events,content=content,reasoning=reasoning,sha256=hashlib.sha256((content+reasoning).encode()).hexdigest()) except Exception as e: detail=e.read().decode(errors='replace') if hasattr(e,'read') else str(e) return dict(name=name,passed=False,error=detail,elapsed_s=time.monotonic()-start) def text(s):return [{'role':'user','content':s}] def media(kind,file): mime='image/png' if kind=='image' else 'video/mp4';f=a.assets/file return {'type':kind+'_url',kind+'_url':{'url':'data:'+mime+';base64,'+base64.b64encode(f.read_bytes()).decode()}} image=media('image','chart.png');video=media('video','sequence120.mp4') image_msgs=[{'role':'user','content':[{'type':'text','text':'Read the verification code and the three invoice amounts. Reply with the code and the sum of the amounts.'},image]}] video_msgs=[{'role':'user','content':[{'type':'text','text':'List the six background colors in chronological order. Reply only with the English color names.'},video]}] cases=[] for i in range(12): x=137+i*23;y=29+i*7 cases.append((f'arithmetic-{i}',text(f'Compute {x} * {y} + {i}. Reply only with the integer.'),2048,str(x*y+i))) cases += [('json',text('Return ONLY valid JSON with exactly these values: {"name":"设备A","count":7,"enabled":true}. No markdown.'),2048,None),('multiturn',[{'role':'user','content':'Remember the code AQUA-58321.'},{'role':'assistant','content':'I will remember it.'},{'role':'user','content':'What was the code? Reply only with the code.'}],2048,'AQUA-58321'),('image-ocr',image_msgs,4096,['SPARK-82741','1368']),('video-120s',video_msgs,4096,['red','green','blue','yellow','purple','white']),('multi-image',[{'role':'user','content':[{'type':'text','text':'These four copies show the same invoice. What is its verification code? Reply only with the code.'},image,image,image,image]}],4096,'SPARK-82741')] a.output.parent.mkdir(parents=True,exist_ok=True) with a.output.open('x') as out: def record(r): out.write(json.dumps(r,ensure_ascii=False)+'\n');out.flush();print(json.dumps({k:v for k,v in r.items() if k not in ('content','reasoning')},ensure_ascii=False),flush=True) def check(r): if r['name']=='json' and r.get('passed'): try:r['passed']=json.loads(r['content'])=={'name':'设备A','count':7,'enabled':True} except Exception:r['passed']=False if r['name'].startswith('video') and r.get('passed'): indices=[r['content'].lower().find(c) for c in ['red','green','blue','yellow','purple','white']];r['passed']=indices==sorted(indices) record(r) for name,msgs,limit,expected in cases:check(run(name,msgs,limit,expected)) # Different useful workloads, fixed-length timing includes reasoning, not a quality score. prompts=[('code','Write a production-quality Python CSV import utility with validation, duplicate handling, error reporting, unit tests, and a detailed explanation.'),('chinese','请写一份企业采购系统的完整需求说明，覆盖审批、预算、供应商、库存和审计，并逐项说明异常处理。'),('reasoning','Explain and prove why Dijkstra works with nonnegative weights. Include a counterexample with negative edges and compare Bellman-Ford.')] for tag,prompt in prompts: record(run('warmup-'+tag,text(prompt),256,forced=True,effort='xhigh')) for i in range(3):record(run(f'perf-{tag}-{i}',text(prompt),2048,forced=True,effort='xhigh')) if not a.screen:record(run('long-output-16384',text('Write an extensive technical textbook chapter explaining algorithms, with many worked examples and Python implementations. Continue with additional examples throughout.'),16384,forced=True,effort='xhigh')) # Exact token count, fresh random filler, two needles at separated positions. rng=random.Random(927);lines=[f'Item {i:06d}: batch {rng.randrange(10**8):08d}, archived in shelf {rng.randrange(1000):03d}.\n' for i in range(16000)] lines[20]='The northern access code is NORTH-49273.\n';lines[2500]='The southern access code is SOUTH-18362.\n' question='\nGive both northern and southern access codes only.' def longmsg(n):return text(''.join(lines[:n])+question) lo,hi=2501,len(lines) while lo