| """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': |
| |
| 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.' |
| |
| 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()) |
|
|
