| |
| from __future__ import annotations |
| import argparse, json, math, os, statistics, time |
| from pathlib import Path |
| import requests |
| SYSTEM = 'Judge whether the Document meets the requirements based on the Query and the Instruction provided. Note that the answer can only be "yes" or "no".' |
| YES = {"yes", "yes.", '"yes"', "'yes'"}; NO = {"no", "no.", '"no"', "'no'"} |
| def score_case(base_url, model, case): |
| payload={'model':model,'messages':[{'role':'system','content':SYSTEM},{'role':'user','content':f"<Instruct>: {case['instruct']}\n\n<Query>: {case['query']}\n\n<Document>: {case['document']}"}],'max_tokens':1,'temperature':0.0,'logprobs':True,'top_logprobs':20,'stream':False} |
| t0=time.perf_counter(); r=requests.post(base_url.rstrip('/')+'/chat/completions',json=payload,timeout=180); r.raise_for_status(); latency=time.perf_counter()-t0 |
| data=r.json(); choice=data['choices'][0]; answer=str(choice['message']['content']).strip().lower(); top=choice['logprobs']['content'][0]['top_logprobs'] |
| zy=zn=-10.0; yp=np=False |
| for item in top: |
| token=str(item['token']).strip().lower(); value=float(item['logprob']) |
| if token in YES: zy=max(zy,value); yp=True |
| elif token in NO: zn=max(zn,value); np=True |
| score=math.exp(zy)/(math.exp(zy)+math.exp(zn)); pred=int(score>0.5); expected=case.get('expected_answer','yes' if case['label'] else 'no') |
| protocol=answer==expected and (not case.get('require_both_logprobs') or (yp and np)); out=dict(case) |
| out.update(score=score,prediction=pred,answer=answer,yes_logprob=zy,no_logprob=zn,yes_present_in_top_20=yp,no_present_in_top_20=np,protocol_valid=protocol,correct=(pred==int(case['label']) and protocol),latency_seconds=latency); return out |
| def main(): |
| root=Path(__file__).resolve().parents[1]; p=argparse.ArgumentParser(); p.add_argument('--cases',action='append',required=True); p.add_argument('--output',required=True); p.add_argument('--base-url',default=os.getenv('BASE_URL','http://127.0.0.1:18190/v1')); p.add_argument('--model',default=os.getenv('MODEL_ID',str(root))); a=p.parse_args() |
| cases=[] |
| for fn in a.cases: |
| cases.extend(json.loads(line) for line in Path(fn).read_text().splitlines() if line.strip()) |
| results=[] |
| for case in cases: |
| r=score_case(a.base_url,a.model,case); results.append(r); print(json.dumps(r,ensure_ascii=False)) |
| tp=sum(r['label']==1 and r['prediction']==1 for r in results); tn=sum(r['label']==0 and r['prediction']==0 for r in results); fp=sum(r['label']==0 and r['prediction']==1 for r in results); fn=sum(r['label']==1 and r['prediction']==0 for r in results); n=len(results) |
| precision=tp/(tp+fp) if tp+fp else 0.0; recall=tp/(tp+fn) if tp+fn else 0.0; f1=2*precision*recall/(precision+recall) if precision+recall else 0.0; lat=[r['latency_seconds'] for r in results]; req=[r for r in results if r.get('require_both_logprobs')] |
| metrics={'n':n,'accuracy':sum(r['correct'] for r in results)/n,'precision':precision,'recall':recall,'f1':f1,'mean_latency_seconds':statistics.mean(lat),'p95_latency_seconds':sorted(lat)[max(0,math.ceil(.95*n)-1)],'protocol_required_cases':len(req),'protocol_passed_cases':sum(r['protocol_valid'] for r in req),'missing_yes_or_no_in_top_20':sum(not(r['yes_present_in_top_20'] and r['no_present_in_top_20']) for r in req),'confusion':{'tp':tp,'tn':tn,'fp':fp,'fn':fn}} |
| out=Path(a.output); out.parent.mkdir(parents=True,exist_ok=True); out.write_text(json.dumps({'case_files':a.cases,'metrics':metrics,'results':results},indent=2,ensure_ascii=False)+'\n'); print(json.dumps(metrics,indent=2)) |
| if not all(r['correct'] for r in results): raise SystemExit(1) |
| if __name__=='__main__': main() |
|
|
