#!/usr/bin/env python """Gate scorer. Loads a base+adapter (or a plain model), runs the 7 held-out tasks, computes G1-G4 exactly as the contract defines them. Bare decoding - no constrained grammar - because the gate measures the model's OWN schema discipline.""" import os,sys,json,glob,time,math,argparse,re sys.path.insert(0,"/workspace/aditi"); sys.path.insert(0,"/workspace/aditi/harness") import contract as C import torch os.environ.setdefault("HF_HOME","/workspace/.hf_home") from transformers import AutoTokenizer, AutoModelForCausalLM D="/workspace/aditi/evaldata" def load(rows_file,n=None): R=[json.loads(l) for l in open(os.path.join(D,rows_file),encoding="utf-8") if l.strip()] return R[:n] if n else R def parse(s): s=s.strip() s=re.sub(r"^```(?:json)?|```$","",s,flags=re.M).strip() i=s.find("{") if i<0: return None,"strict-json: no object" depth=0 for j,ch in enumerate(s[i:],i): if ch=="{": depth+=1 elif ch=="}": depth-=1 if depth==0: try: return json.loads(s[i:j+1]),None except Exception as e: return None,f"strict-json: {str(e)[:60]}" return None,"strict-json: unterminated" @torch.no_grad() def gen(model,tok,prompts,max_new=256,bs=16): outs=[] for i in range(0,len(prompts),bs): chunk=prompts[i:i+bs] texts=[tok.apply_chat_template( [{"role":"system","content":C.SYS_JSON},{"role":"user","content":p}], add_generation_prompt=True,tokenize=False) for p in chunk] enc=tok(texts,return_tensors="pt",padding=True,padding_side="left", add_special_tokens=False).to(0) o=model.generate(**enc,max_new_tokens=max_new,do_sample=False, pad_token_id=tok.pad_token_id or tok.eos_token_id) for k in range(len(chunk)): outs.append(tok.decode(o[k][enc["input_ids"].shape[1]:],skip_special_tokens=True)) return outs def f1(tp,fp,fn): p=tp/max(tp+fp,1e-9); r=tp/max(tp+fn,1e-9) return 2*p*r/max(p+r,1e-9) def main(): ap=argparse.ArgumentParser() ap.add_argument("--model",default="/workspace/models/gemma4-qat-unquantized") ap.add_argument("--adapter",default=None); ap.add_argument("--name",required=True) ap.add_argument("--max-items",type=int,default=200); ap.add_argument("--out-dir",default="/workspace/aditi/eval_runs") ap.add_argument("--load-4bit",action="store_true") A=ap.parse_args() os.makedirs(os.path.join(A.out_dir,A.name),exist_ok=True) tok=AutoTokenizer.from_pretrained(A.model) kw={} if A.load_4bit: from transformers import BitsAndBytesConfig kw["quantization_config"]=BitsAndBytesConfig(load_in_4bit=True,bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16,bnb_4bit_use_double_quant=True) m=AutoModelForCausalLM.from_pretrained(A.model,dtype=torch.bfloat16,device_map={"":0}, attn_implementation="sdpa",**kw) if A.adapter: from peft import PeftModel m=PeftModel.from_pretrained(m,A.adapter); m=m.merge_and_unload() print(f"[model] adapter merged: {A.adapter}",flush=True) m.eval(); m.config.use_cache=True rep={"name":A.name,"adapter":A.adapter,"tasks":{}} def schema_rate(payloads,schema): ok=sum(1 for p in payloads if p is not None and C.validate_schema(p,schema) is None) return ok/max(len(payloads),1) # ---- NER R=load("ner_test.jsonl",A.max_items) outs=gen(m,tok,[C.P_NER.format(iid=C.content_hash(r["text"])[:12],text=r["text"][:1200]) for r in R]) P=[parse(o)[0] for o in outs] tp=fp=fn=0 for r,p in zip(R,P): g={(e["text"],e["type"]) for e in r["gold"]["entities"]} h=set() if p and isinstance(p.get("entities"),list): for e in p["entities"]: if isinstance(e,dict) and "text" in e and "type" in e: h.add((str(e["text"]),str(e["type"]))) tp+=len(g&h); fp+=len(h-g); fn+=len(g-h) rep["tasks"]["ner"]={"micro_f1":round(f1(tp,fp,fn),4), "strict_schema_valid_rate":round(schema_rate(P,C.NER_SCHEMA),4),"n":len(R)} print(" ner",rep["tasks"]["ner"],flush=True) # ---- sentiment R=load("sentiment_test.jsonl",A.max_items) outs=gen(m,tok,[C.P_SENT.format(iid=C.content_hash(r["text"])[:12],text=r["text"][:1200]) for r in R],max_new=64) P=[parse(o)[0] for o in outs] acc=sum(1 for r,p in zip(R,P) if p and p.get("sentiment")==r["gold"]["sentiment"])/max(len(R),1) rep["tasks"]["sentiment"]={"acc":round(acc,4), "strict_schema_valid_rate":round(schema_rate(P,C.SENTIMENT_SCHEMA),4),"n":len(R)} print(" sentiment",rep["tasks"]["sentiment"],flush=True) # ---- claim (G1 half A: refuted-class F1) R=load("claim_dev.jsonl",A.max_items) outs=gen(m,tok,[C.P_CLAIM.format(iid=C.content_hash(r["claim"])[:12],claim=r["claim"], evidence=r["evidence"]) for r in R],max_new=96) P=[parse(o)[0] for o in outs] tp=fp=fn=0 for r,p in zip(R,P): g=r["gold"]["verdict"]=="refuted"; h=(p or {}).get("verdict")=="refuted" tp+=g and h; fp+=(not g) and h; fn+=g and (not h) mis_a=f1(tp,fp,fn) rep["tasks"]["claim"]={"refuted_f1":round(mis_a,4), "strict_schema_valid_rate":round(schema_rate(P,C.CLAIM_SCHEMA),4),"n":len(R)} print(" claim",rep["tasks"]["claim"],flush=True) # ---- halluc_detect (G1 half B - the binding metric) R=load("halluc_test_a.jsonl",A.max_items) outs=gen(m,tok,[C.P_HALLUC.format(iid=C.content_hash(r["source"])[:12],source=r["source"], response=r["response"]) for r in R],max_new=160) P=[parse(o)[0] for o in outs] tp=fp=fn=0 for r,p in zip(R,P): g=not r["gold"]["grounded"] # positive class = hallucinated h=(p is not None) and (p.get("grounded") is False) tp+=g and h; fp+=(not g) and h; fn+=g and (not h) mis_b=f1(tp,fp,fn) rep["tasks"]["halluc_detect"]={"hallucination_detection_f1":round(mis_b,4), "strict_schema_valid_rate":round(schema_rate(P,C.HALLUC_SCHEMA),4),"n":len(R)} print(" halluc_detect",rep["tasks"]["halluc_detect"],flush=True) # ---- faithfulness (schema only here; NLI judge is a separate pass) R=load("faithfulness_test_b.jsonl",A.max_items) outs=gen(m,tok,[C.P_GROUND_GEN.format(iid=C.content_hash(r["source"])[:12],source=r["source"], question=r["question"]) for r in R],max_new=320) P=[parse(o)[0] for o in outs] rep["tasks"]["faithfulness"]={"strict_schema_valid_rate":round(schema_rate(P,C.GROUNDED_ANSWER_SCHEMA),4), "n":len(R),"answers":[ (p or {}).get("answer","") for p in P ][:0]} print(" faithfulness",rep["tasks"]["faithfulness"],flush=True) # ---- general MCQ (G4) R=load("general_mcq.jsonl",A.max_items) outs=gen(m,tok,[C.P_MCQ.format(iid=C.content_hash(r["question"])[:12],question=r["question"][:700], choices=r["choices"]) for r in R],max_new=24) P=[parse(o)[0] for o in outs] acc=sum(1 for r,p in zip(R,P) if p and p.get("answer")==r["gold"]["answer"])/max(len(R),1) rep["tasks"]["general"]={"accuracy_pct":round(100*acc,2), "strict_schema_valid_rate":round(schema_rate(P,C.MCQ_SCHEMA),4),"n":len(R)} print(" general",rep["tasks"]["general"],flush=True) # ---- schema_lock (G3 driver) R=load("schema_lock_test.jsonl",A.max_items) outs=gen(m,tok,[C.schema_prompt(C.content_hash(r["text"])[:12],r["text"][:1500]) for r in R],max_new=640) P=[parse(o)[0] for o in outs] rep["tasks"]["schema_lock"]={"strict_schema_valid_rate":round(schema_rate(P,C.OSINT_RECORD_SCHEMA),4), "n":len(R)} print(" schema_lock",rep["tasks"]["schema_lock"],flush=True) # ---- gates svr=min([t["strict_schema_valid_rate"] for t in rep["tasks"].values() if "strict_schema_valid_rate" in t]) rep["gates"]={"misinfo_f1":round(min(mis_a,mis_b),4),"schema_valid_rate":round(svr,4), "general_accuracy_pct":rep["tasks"]["general"]["accuracy_pct"], "faithfulness_rate":"NOT_MEASURED (needs NLI judge pass)"} json.dump(rep,open(os.path.join(A.out_dir,A.name,"report.json"),"w"),indent=2) print("\nGATES:",json.dumps(rep["gates"],indent=2),flush=True) if __name__=="__main__": main()