Spaces:
Running
Running
File size: 1,836 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 | """Reproduce public frozen cases locally; retain synthetic native/HTTP outputs.
Historical qualification stored counters only. These are fresh reproductions,
not recovered historical confidence scores. Never executes a tool or opens ports.
"""
import asyncio
import json
from pathlib import Path
from fastapi.testclient import TestClient
import gateway
def main():
cases=json.loads(Path('benchmarks/qualification/scenarios.json').read_text())['cases']
chosen=[c for c in cases if c['id'] in {'ambiguous-0','ambiguous-2','ambiguous-5'}]
original=gateway.scheduler.schedule
native=[]
async def observe(tools,system,text,priority):
result=await original(tools,system,text,priority)
native.append({'candidate_names':[t['name'] for t in tools], 'candidate_required_fields':{t['name']:t['parameters'].get('required',[]) for t in tools},'native_result':result})
return result
gateway.scheduler.schedule=observe
gateway.API_KEY='';gateway.AUTH_MODE='optional'
rows=[]
try:
with TestClient(gateway.app) as client:
for repeat in range(3):
for case in chosen:
before=len(native)
r=client.post('/v1/chat/completions',json=case['request'])
rows.append({'id':case['id'],'repeat':repeat,'user_text':case['request']['messages'][-1]['content'],'policy_arguments':case['request']['decision_context'].get('arguments',{}),'status':r.status_code,'http_response':r.json(),'inference':native[before:]})
finally:gateway.scheduler.schedule=original
print(json.dumps({'methodology':'Fresh local actual native reproductions; synthetic public frozen inputs only. No historical raw native outputs existed. Host resources; not CPU benchmark.','rows':rows},indent=2))
if __name__=='__main__':main()
|