Spaces:
Running
Running
| """Frozen synthetic qualification against loopback; no tool executes or deploys.""" | |
| import argparse | |
| import asyncio | |
| import hashlib | |
| import json | |
| import os | |
| import statistics | |
| import time | |
| from pathlib import Path | |
| from urllib.parse import urlparse | |
| import httpx | |
| def cases(): | |
| data=json.loads(Path('benchmarks/qualification/scenarios.json').read_text()) | |
| for c in data['cases']: | |
| gen=c.get('history_generator') | |
| if gen: | |
| n=gen['records'];pos={'beginning':0,'middle':n//2,'end':n}[gen['position']] | |
| if gen['kind']=='unique': | |
| lines=[f'item={i:06d} sha={hashlib.sha256(str(i).encode()).hexdigest()} status=observed\n' for i in range(n)] | |
| error='FAILED tests/test_x.py - ERR_X /repo/unique.py\n' | |
| else: | |
| lines=['PASSED tests/test_ok.py::test_ok\n']*n | |
| error='FAILED tests/test_x.py - ERR_X /repo/history.py\n' | |
| lines.insert(pos,error);c['request']['messages'][1]['content']=''.join(lines) | |
| yield c | |
| def aggregate(rows): | |
| latency=sorted(r['latency_ms'] for r in rows) | |
| q=lambda f:latency[min(len(latency)-1,__import__('math').ceil(len(latency)*f)-1)] | |
| routine=[r for r in rows if r['expected_tool']] | |
| safe=[r for r in rows if not r['expected_tool']] | |
| no_call=[r for r in rows if r['status']==200 and not r['calls'] and r['outcome'] in {'abstain','confirm'}] | |
| true_safe=sum(not r['expected_tool'] and r['status']==200 and not r['calls'] and r['outcome'] in {'abstain','confirm'} for r in rows) | |
| recall=[r['candidate_recall'] for r in routine if r['candidate_recall'] is not None] | |
| return {'samples':len(rows),'strict_accuracy':statistics.mean(r['correct'] for r in rows), | |
| 'routine_tool_accuracy':statistics.mean(r['tool_correct'] for r in routine) if routine else None, | |
| 'routine_exact_arguments_accuracy':statistics.mean(r['correct'] for r in routine) if routine else None, | |
| 'candidate_recall':statistics.mean(recall) if recall else None, | |
| 'incorrect_executable_calls':sum(bool(r['calls']) and not r['correct'] for r in rows), | |
| 'permission_violations':sum(r['group']=='safety' and bool(r['calls']) for r in rows), | |
| 'abstain_confirm_precision':true_safe/len(no_call) if no_call else None, | |
| 'abstain_confirm_recall':true_safe/len(safe) if safe else None, | |
| 'confirm_count':sum(r['outcome']=='confirm' for r in rows), | |
| 'body_413_count':sum(r['status']==413 for r in rows), | |
| 'latency_ms':{'p50':statistics.median(latency),'p95':q(.95),'p99':q(.99)}} | |
| async def main(): | |
| p=argparse.ArgumentParser();p.add_argument('--url',default='http://127.0.0.1:7860');p.add_argument('--concurrency',type=int,choices=[1,2,4],default=1);p.add_argument('--label',required=True);p.add_argument('--candidate-check',action='store_true');p.add_argument('--repeat',type=int,default=1) | |
| args=p.parse_args() | |
| if urlparse(args.url).hostname not in {'127.0.0.1','localhost','::1'}:raise SystemExit('Loopback test server only') | |
| sem=asyncio.Semaphore(args.concurrency) | |
| key=os.getenv('NEEDLE_API_KEY') or os.getenv('API_KEY') | |
| headers={'Authorization':'Bearer '+key} if key else {} | |
| async with httpx.AsyncClient(base_url=args.url,headers=headers,timeout=60) as client: | |
| assert (await client.get('/ready')).status_code==200 | |
| async def one(c,repeat): | |
| async with sem: | |
| recall=None | |
| if args.candidate_check and c['expected']['tool']: | |
| b=c['request'];query=b['messages'][-1]['content'] | |
| r=await client.post('/v1/gates/route',json={'query':query,'tools':b['tools'],'decision_context':b['decision_context']}) | |
| if r.status_code==200:recall=any(t['name']==c['expected']['tool'] for t in r.json().get('candidates',[])) | |
| start=time.perf_counter();calls=[];outcome='error';status=0;diagnostics={} | |
| try: | |
| r=await client.post('/v1/chat/completions',json=c['request']);status=r.status_code | |
| if status==200: | |
| data=r.json();calls=data['choices'][0]['message'].get('tool_calls',[]) or [] | |
| outcome=data.get('needle',{}).get('outcome','selected' if calls else 'abstain') | |
| diagnostics=data.get('needle',{}).get('diagnostics',{}) | |
| except httpx.TimeoutException:outcome='timeout' | |
| elapsed=(time.perf_counter()-start)*1000 | |
| expected=c['expected'];tool_ok=bool(len(calls)==1 and calls[0]['function']['name']==expected['tool']) | |
| args_ok=tool_ok and json.loads(calls[0]['function']['arguments'])==expected['arguments'] | |
| correct=(args_ok and outcome=='selected') if expected['tool'] else (status==200 and not calls and outcome in expected['safe_outcomes']) | |
| return {'id':c['id'],'group':c['group'],'repeat':repeat,'expected_tool':expected['tool'],'tool_correct':tool_ok,'correct':correct,'status':status,'outcome':outcome,'calls':len(calls),'candidate_recall':recall,'request_utf8_bytes':len(json.dumps(c['request']).encode()),'latency_ms':elapsed,'diagnostics':diagnostics} | |
| # At most four generated request bodies live concurrently, not all large histories. | |
| rows=[];batch=[] | |
| for repeat in range(args.repeat): | |
| for c in cases(): | |
| batch.append(one(c,repeat)) | |
| if len(batch)==args.concurrency: | |
| rows.extend(await asyncio.gather(*batch));batch=[] | |
| if batch:rows.extend(await asyncio.gather(*batch));batch=[] | |
| grouped={g:aggregate([r for r in rows if r['group']==g]) for g in sorted({r['group'] for r in rows})} | |
| stability={i:len({(r['status'],r['outcome'],r['tool_correct'],r['correct']) for r in rows if r['id']==i})==1 for i in {r['id'] for r in rows}} | |
| print(json.dumps({'label':args.label,'dataset_sha256':hashlib.sha256(Path('benchmarks/qualification/scenarios.json').read_bytes()).hexdigest(),'methodology':'Frozen authored synthetic scenarios; no supplied real traces; no routing tuning. Body rejections reported separately and not counted correct routing. Baseline outcome inferred from executable calls. Candidate checks are separate gate-route requests.','concurrency':args.concurrency,'summary':aggregate(rows),'groups':grouped,'repeat_stability':sum(stability.values())/len(stability) if args.repeat>1 else None,'rows':rows},indent=2)) | |
| if __name__=='__main__':asyncio.run(main()) | |