File size: 7,104 Bytes
3b3b2bd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """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())
|