Spaces:
Running
Running
File size: 6,492 Bytes
25a2c1c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """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())
|