| """Offline artifact-only validation: no source model weights are loaded.""" |
| import asyncio,json,time,resource |
| from pathlib import Path |
| import numpy as np |
| import psutil |
| from coreai.runtime import NDArray |
| from coreai_d3 import D3CoreAI,pack |
|
|
| async def main(): |
| start=time.perf_counter();engine=await D3CoreAI.load(Path(__file__).parent);load_seconds=time.perf_counter()-start |
| q=dict(type='choice',instructions='Which action best describes the user request?',criteria={'interrupt':'Stop the assistant speaking now','discard':'Ignore background noise','enqueue':'Handle a new question later'}) |
| state='The assistant is speaking. The user says: Stop talking now.' |
| result=await engine.system_one(state=state,questions={'action':q});assert result['answers']['action']['choice']=='interrupt' |
| |
| encoded=engine.tokenizer(['Hello '*600],add_special_tokens=False,return_tensors='np') |
| limit=engine.manifest['sequence_length'];ids=encoded['input_ids'][:,:limit] |
| arrays=pack({'input_ids':ids},engine.manifest) |
| output=await engine.fn({k:NDArray(v) for k,v in arrays.items()}) |
| assert np.isfinite(output[engine.manifest['output']].numpy()).all() |
| try:pack({'input_ids':np.pad(ids,((0,0),(0,1)))},engine.manifest) |
| except ValueError:pass |
| else:raise AssertionError('Capacity+1 must reject') |
| |
| engine._busy=True |
| try: |
| try:await engine.logits(state,q) |
| except RuntimeError:pass |
| else:raise AssertionError('Busy must reject') |
| finally:engine._busy=False |
| times=[] |
| for _ in range(20): |
| t=time.perf_counter();await engine.system_one(state=state,questions={'action':q});times.append(time.perf_counter()-t) |
| report=dict(offline=True,source_weights_loaded=False,result=result,warm_load_seconds=load_seconds,exact_capacity_finite=True,over_capacity_rejected=True,busy_rejected=True,warm_samples=times,median_seconds=float(np.median(times)),p95_seconds=float(np.percentile(times,95)),resident_rss_bytes=psutil.Process().memory_info().rss,peak_rss_bytes=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) |
| print(json.dumps(report,indent=2),flush=True) |
| if __name__=='__main__':asyncio.run(main()) |
|
|
