"""Real GPT-2 / Qwen integration evaluation. Never substitutes mock generations. Run in an environment with Transformers and model downloads. On failure writes an explicit not_run report and exits nonzero. Contains no pretrained weights. """ from pathlib import Path import argparse,hashlib,json,os,platform,sys,time,traceback ROOT=Path(__file__).resolve().parents[1];sys.path.insert(0,str(ROOT)) from phillnet_recall import RecallMemory from phillnet_recall.context import prepare_prompt,prepare_messages,NOTICE def cases(): names=['Willow','Cobalt','Orchid','Juniper','Cedar','Lumen','Harbor','Aster','Mica','Birch','Fable','Ember'] return [dict(id=f'port_{i:02}',question=f'What service port does Project {n} use?',answer=str(7442+17*i), evidence=f'Project {n} uses service port {7442+17*i}.', distractors=[f'Project {n} has a planning meeting.',f'Project Other{n} uses service port 9001.']) for i,n in enumerate(names)] def main(): ap=argparse.ArgumentParser();ap.add_argument('--model',default='openai-community/gpt2',choices=['openai-community/gpt2','Qwen/Qwen3.5-0.8B']) ap.add_argument('--revision',default=None);ap.add_argument('--device',default='cpu',choices=['cpu','cuda']);ap.add_argument('--limit',type=int,default=12) ap.add_argument('--output');a=ap.parse_args() dest=Path(a.output) if a.output else ROOT/'reports'/('gpt2_integration.json' if 'gpt2' in a.model else 'qwen08_integration.json') dest.parent.mkdir(parents=True,exist_ok=True) report={'model_id':a.model,'requested_revision':a.revision,'status':'not_run','generated_answers':0,'rows':[], 'scope':'12 authored synthetic factual QA fixtures, not a general memory benchmark or a host-model certification', 'evaluation':'Strict exact match of first generated line, stripped of whitespace and one terminal period. All raw generations retained.', 'trained_or_finetuned':False,'private_data_used':False} started=time.perf_counter() try: import torch,transformers from transformers import AutoTokenizer,AutoModelForCausalLM,AutoConfig torch.manual_seed(0);torch.set_num_threads(min(4,os.cpu_count() or 1)) config=AutoConfig.from_pretrained(a.model,revision=a.revision,trust_remote_code=False) revision=getattr(config,'_commit_hash',None) or a.revision tok=AutoTokenizer.from_pretrained(a.model,revision=revision,trust_remote_code=False) kwargs=dict(revision=revision,trust_remote_code=False,torch_dtype=torch.float32 if a.device=='cpu' else torch.float16) if config.model_type=='qwen3_5': # The published full checkpoint includes a vision tower. This is a text-only test. from transformers import Qwen3_5ForConditionalGeneration model=Qwen3_5ForConditionalGeneration.from_pretrained(a.model,**kwargs) else:model=AutoModelForCausalLM.from_pretrained(a.model,**kwargs) model=model.to(a.device).eval() if tok.pad_token_id is None:tok.pad_token=tok.eos_token report['environment']={'torch':torch.__version__,'transformers':transformers.__version__,'python':sys.version,'platform':platform.platform(),'device':a.device} report['resolved_revision']=revision fixtures=cases()[:max(1,min(12,a.limit))] report['fixture_sha256']=hashlib.sha256(json.dumps(fixtures,sort_keys=True).encode()).hexdigest() conditions=['no_memory','recency','bm25','legacy','hybrid','semantic','oracle'] for case in fixtures: clock=[100.] with RecallMemory(clock=lambda:clock[0]) as mem: mem.remember(case['evidence'],source='synthetic:configuration') for d in case['distractors']: clock[0]+=100;mem.remember(d,source='synthetic:distractor') for condition in conditions: question=case['question']+' Return only the port number.' # Identical prompt/chat construction in all conditions. Only evidence differs. isolated=None selected=mem if condition in ('no_memory','oracle'): isolated=RecallMemory(clock=lambda:100.) if condition=='oracle':isolated.remember(case['evidence'],source='synthetic:configuration') selected=isolated backend=condition if condition not in ('no_memory','oracle') else 'hybrid' try: if getattr(tok,'chat_template',None): packet=prepare_messages(selected,question,tokenizer=tok,max_input_tokens=768,memory_tokens=450,top_k=1,backend=backend) prompt=tok.apply_chat_template(packet['messages'],tokenize=False,add_generation_prompt=True,enable_thinking=False) else: packet=prepare_prompt(selected,question,tokenizer=tok,max_input_units=768,memory_units=450,top_k=1,backend=backend) prompt=packet['prompt'] ids=packet['memory']['ids'] finally: if isolated is not None:isolated.close() x=tok(prompt,return_tensors='pt',add_special_tokens=False).to(a.device) if x['input_ids'].shape[1]>768:raise RuntimeError('full rendered prompt exceeds the declared input budget') t=time.perf_counter() with torch.inference_mode():y=model.generate(**x,max_new_tokens=24,do_sample=False,pad_token_id=tok.pad_token_id) if a.device=='cuda':torch.cuda.synchronize() raw=tok.decode(y[0,x['input_ids'].shape[1]:],skip_special_tokens=True) answer=raw.strip().split('\n')[0].strip().removesuffix('.') report['rows'].append({'id':case['id'],'condition':condition,'question':case['question'],'expected':case['answer'], 'output':raw,'strict_exact_match':answer==case['answer'],'input_tokens':x['input_ids'].shape[1], 'prompt':prompt,'selected_memory_ids':ids,'generation_seconds':time.perf_counter()-t}) report['generated_answers']+=1 report['status']='completed' report['results']={c:{'correct':sum(r['strict_exact_match'] for r in report['rows'] if r['condition']==c),'cases':sum(r['condition']==c for r in report['rows'])} for c in conditions} except Exception as e: report['status']='partial' if report['generated_answers'] else 'not_run' report['reason']=type(e).__name__+': '+str(e)[:800] report['note']='This is a failure receipt, not a zero-accuracy evaluation or a successful model test.' finally: report['elapsed_seconds']=time.perf_counter()-started dest.write_text(json.dumps(report,indent=2)+'\n') print(json.dumps({k:v for k,v in report.items() if k!='rows'},indent=2)) return 0 if report['status']=='completed' else 2 if __name__=='__main__':raise SystemExit(main())