| """Post-implementation authored tests; not a public benchmark or an independent lab. |
| |
| Reports original known-development cases separately from fresh synthetic cases. |
| Only scoring reads expected labels. Runtime receives records + question/plan. |
| """ |
| from pathlib import Path |
| import json,sys,hashlib,statistics,time,platform,random,math |
| ROOT=Path(__file__).resolve().parents[1];sys.path.insert(0,str(ROOT)) |
| from phillnet_recall import RecallMemory,RecallPolicy |
| from phillnet_recall.evidence import semantic_retrieve,EvidencePolicy |
| from phillnet_recall.core import canonical_json |
| from scripts.evaluate import fixtures as prior_cases |
|
|
| PAIRS=[ |
| ('backup','Where can I recover files?','A recovery copy is stored in vault {v}.','The recover files discussion is pending.'), |
| ('port','What TCP port is configured?','The listener is listening on {v}.','The TCP port survey is still pending.'), |
| ('region','Which hosting region is selected?','The application is deployed in {v}.','The hosting region discussion is pending.'), |
| ('deadline','What is the delivery date?','The submission date is {v}.','The delivery date is undecided.'), |
| ('budget','What is the maximum spend?','Allocated funds are {v}.','Maximum spend survey is still pending.'), |
| ('editor','Which IDE is preferred?','The code editor is {v}.','Our IDE and preference survey is pending.'), |
| ('authentication','What should be used to log in?','Sign in requires {v}.','Log in is under discussion of authentication.'), |
| ('retention','How long do we keep records?','Retention lasts {v}.','The keep records survey is pending.'), |
| ('notifications','How do I get alerts?','Notifications are sent through {v}.','The alerts survey is pending.'), |
| ('quiet_hours','When should we silence alerts?','Do not disturb is enabled from {v}.','Silence alerts is still undecided.'), |
| ('license','Is commercial use permitted?','The licence is {v}.','Commercial use is still discussing options.'), |
| ('capacity','What is the storage limit?','Maximum entries are {v}.','The storage limit is unknown.'), |
| ] |
| OUTSIDE=[ |
| ('Which lodging arrangement is suitable?','An inn has been booked.'), |
| ('How should we deal with inclement weather?','Carry a raincoat.'), |
| ('Which optical aid is available?','A telescope is in the cabinet.'), |
| ('Which route provides nautical passage?','Take the ferry.'), |
| ('Which professional can mend footwear?','A cobbler is available.'), |
| ('What enables upright posture?','The spine supports the body.'), |
| ('What resource provides warmth?','Firewood is in the shed.'), |
| ('Which item preserves perishables?','A refrigerator is available.'), |
| ('How can I conserve hydration?','Take water in the flask.'), |
| ('What prevents ingress of precipitation?','Close the umbrella after entering.'), |
| ('Which implement cuts timber?','A saw is on the bench.'), |
| ('Where can I purchase baked goods?','The bakery is across the street.')] |
|
|
|
|
| def fresh_cases(): |
| result=[];rng=random.Random(131141) |
| for rep in range(5): |
| for idx,(topic,q,s,d) in enumerate(PAIRS): |
| ident=f'{topic}-{rep}' |
| |
| docs=[{'text':s.format(v=f'value_{idx}_{rep}'),'source':'confirmed','at':100}, |
| {'text':d,'source':'minutes','at':200}] |
| for j in range(28): |
| _,_,other,_=PAIRS[(idx+1+j%11)%12] |
| docs.append({'text':other.format(v=f'distractor_{j}_{rep}'),'source':'other','at':150}) |
| expected=docs[0];rng.shuffle(docs) |
| result.append({'id':ident,'family':'declared_topic_transfer','query':q,'docs':docs,'expected':[docs.index(expected)]}) |
| for i,(q,s) in enumerate(OUTSIDE): |
| result.append({'id':f'open-{i}','family':'outside_lexicon','query':q, |
| 'docs':[{'text':s,'source':'operator','at':100},{'text':q+' This request is pending.','source':'inbox','at':200}], 'expected':[0]}) |
| return result |
|
|
|
|
| def path_cases(): |
| result=[] |
| for i in range(72): |
| root=f'Project_{i}';mid=f'Team_{i}';ans=f'Person_{i}' |
| docs=[dict(text=f'{root} is assigned to {mid}.',source='ownership',at=100,entity=root,predicate='owner_team',value=mid), |
| dict(text=f'{ans} runs {mid}.',source='directory',at=100,entity=mid,predicate='lead',value=ans)] |
| expected=[0,1];status='supported';values=[ans];kw={} |
| family=['two_hop','revision','conflict','missing','alias','inverse'][i%6] |
| if family=='revision': |
| docs.append(dict(text=f'NewPerson_{i} now runs {mid}.',source='directory',at=200,entity=mid,predicate='lead',value=f'NewPerson_{i}',replaces_index=1)) |
| expected=[0,2];values=[f'NewPerson_{i}'] |
| if family=='conflict': |
| docs.append(dict(text=f'OtherPerson_{i} runs {mid}.',source='different_directory',at=200,entity=mid,predicate='lead',value=f'OtherPerson_{i}')) |
| expected=[];values=[];status='conflict' |
| if family=='missing':docs=docs[:1];expected=[];values=[];status='unknown' |
| if family=='alias': |
| docs.append(dict(text=f'Canary_{i} denotes {root}.',source='operator',at=100,entity=f'Canary_{i}',predicate='alias_of',value=root)) |
| expected=[0,1,2];root=f'Canary_{i}' |
| steps=['owner_team','lead'] |
| if family=='inverse': |
| docs[0].update(text=f'{mid} supports {root}.',entity=mid,predicate='supports',value=root) |
| steps=[{'predicate':'supports','direction':'in'},'lead'] |
| for j in range(40): |
| docs.append(dict(text=f'Distractor_{j} leads a team that owns unrelated project {i}.',source='noise',at=100,entity=f'Noise_{j}',predicate='lead',value=f'Unused_{j}')) |
| result.append(dict(id=f'path-{i}',family=family,query=f'Who leads the team that owns {root}?',entity=root,steps=steps,docs=docs,expected=expected,status=status,answer_values=values)) |
| return result |
|
|
|
|
| def setup(case,policy): |
| clock=[100.];m=RecallMemory(policy=policy,clock=lambda:clock[0]);ids=[] |
| for doc in case['docs']: |
| d=dict(doc);clock[0]=float(d.pop('at'));content=d.pop('text');r=d.pop('replaces_index',None) |
| if r is not None:d['replaces']=ids[r] |
| ids.append(m.remember(content,**d)) |
| clock[0]=100000.;return m,ids |
|
|
|
|
| def eval_retrieval(cases,pol): |
| rows=[];summary={} |
| for backend in ('bm25','hybrid','semantic_neural','semantic_linear','semantic_zero'): |
| measures=[];times=[] |
| for case in cases: |
| m,ids=setup(case,pol) |
| try: |
| t=time.perf_counter_ns() |
| if backend.startswith('semantic_'):r=semantic_retrieve(m,case['query'],top_k=5,backend=backend[9:],**case.get('retrieve_options',{})) |
| else:r=m.retrieve(case['query'],top_k=5,backend=backend,**case.get('retrieve_options',{})) |
| ms=(time.perf_counter_ns()-t)/1e6 |
| got=[ids.index(x['id']) for x in r['items']] |
| ok1=(r['status']==case['status'] and not got) if 'status' in case else bool(got and got[0] in case['expected']) |
| ok5=(r['status']==case['status'] and not got) if 'status' in case else set(case['expected']).issubset(got) |
| measures.append((case['family'],ok1,ok5));times.append(ms) |
| rows.append(dict(id=case['id'],backend=backend,status=r['status'],expected=case['expected'],got=got,hit1=ok1,all_support5=ok5,latency_ms=ms)) |
| finally:m.close() |
| family={} |
| for f,h,a in measures: |
| v=family.setdefault(f,dict(n=0,hit1=0,all_support5=0));v['n']+=1;v['hit1']+=h;v['all_support5']+=a |
| summary[backend]=dict(n=len(measures),hit1=sum(x[1] for x in measures),all_support5=sum(x[2] for x in measures),by_family=family,median_ms=statistics.median(times),p95_ms=sorted(times)[math.ceil(.95*len(times))-1]) |
| return dict(summary=summary,rows=rows) |
|
|
|
|
| def eval_paths(cases,pol): |
| rows=[];summary={} |
| for b in ('bm25','semantic_retrieval','path_neural','path_linear','path_zero'): |
| good=covered=0;families={};times=[] |
| for c in cases: |
| m,ids=setup(c,pol) |
| try: |
| t=time.perf_counter_ns() |
| if b.startswith('path_'):r=m.follow(c['entity'],c['steps'],backend=b[5:]) |
| elif b=='semantic_retrieval':r=semantic_retrieve(m,c['query'],top_k=5) |
| else:r=m.retrieve(c['query'],top_k=5,backend='bm25') |
| ms=(time.perf_counter_ns()-t)/1e6 |
| got=[ids.index(x['id']) for x in r['items']] |
| answer_ok=r.get('answer_values',[])==c['answer_values'] and r['status']==c['status'] |
| evidence_ok=(r['status']==c['status'] and not got) if not c['expected'] else set(c['expected']).issubset(got) |
| good+=answer_ok;covered+=evidence_ok;times.append(ms) |
| f=families.setdefault(c['family'],dict(n=0,answer_status=0,all_support_or_expected_status=0));f['n']+=1;f['answer_status']+=answer_ok;f['all_support_or_expected_status']+=evidence_ok |
| rows.append(dict(id=c['id'],backend=b,status=r['status'],answer_values=r.get('answer_values',[]),expected_status=c['status'],expected_values=c['answer_values'],expected_ids=c['expected'],got=got,all_support_or_expected_status=evidence_ok,answer_status=answer_ok,latency_ms=ms)) |
| finally:m.close() |
| summary[b]=dict(n=len(cases),answer_status=good,all_support_or_expected_status=covered,by_family=families,median_ms=statistics.median(times),p95_ms=sorted(times)[math.ceil(.95*len(times))-1]) |
| return dict(summary=summary,rows=rows,comparison_note='Baseline retrieval does not emit answers. Compare support coverage separately. Path joins use explicit same-scope claim metadata available in all records; BM25 only scores text.') |
|
|
|
|
| def main(): |
| pol=RecallPolicy();ep=EvidencePolicy();cases=fresh_cases();pc=path_cases() |
| fixture={'development':prior_cases(),'fresh_retrieval':cases,'fresh_paths':pc} |
| (ROOT/'reports/evidence_fixtures.json').write_text(json.dumps(fixture,indent=2)+'\n') |
| reports={'methodology':{'author':'same assistant as implementation; not independent validation', |
| 'development_note':'Original 240 cases were known before upgrade; diet topic intentionally repairs them.', |
| 'transfer_note':'Fresh authored combinations use declared topic vocabulary; outside-lexicon cases remain separate.', |
| 'public_benchmarks_run':[],'host_models_run':[],'gradient_training':False, |
| 'freeze':json.loads((ROOT/'reports/implementation_freeze.json').read_text()), |
| 'fixture_sha256':hashlib.sha256(canonical_json(fixture).encode()).hexdigest(), |
| 'core_policy_sha256':pol.bank.sha256,'extra_policy_sha256':ep.bank.sha256,'sensemap_sha256':ep.lexicon_sha256}, |
| 'environment':{'python':sys.version,'platform':platform.platform(),'native':bool(pol.bank._lib)}, |
| 'known_development':eval_retrieval(prior_cases(),pol),'fresh_retrieval':eval_retrieval(cases,pol),'fresh_paths':eval_paths(pc,pol)} |
| (ROOT/'reports/evidence_evaluation.json').write_text(json.dumps(reports,indent=2)+'\n') |
| print(json.dumps({k:v['summary'] for k,v in reports.items() if 'summary' in v},indent=2)) |
| if __name__=='__main__':main() |
|
|
