Files
VianoBugAnalysis/scripts/measure_compact_investigation.py

52 lines
3.1 KiB
Python

"""Offline projection of archived v13/v14 requests into v15; no API or DB access."""
import argparse
import copy
import json
from pathlib import Path
from statistics import median
from viano_bug_analysis.package import PackageReader
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
from viano_bug_analysis.llm.gemini import GeminiProvider
def native_characters(request):
system,prefix,contents=GeminiProvider._messages(request,stable_case_context=True)
return sum(len(p['text']) for p in system['parts'])+sum(len(p['text']) for message in prefix+contents for p in message['parts'])
def measure(path):
rows=[]
with PackageReader(path) as reader:
for ref in reader.files:
if not ref.startswith('analysis/investigation/') or '/contexts/' not in ref or not ref.endswith('.json'):continue
c=json.loads(reader.read_bytes(ref))
if c.get('prompt_version') not in {'investigation_project_v13','investigation_project_v14'}:continue
old=build_prompt(c);current=copy.deepcopy(c);current['prompt_version']='investigation_project_v15'
current['other_working_briefs']={}
for hid,h in c['hypotheses'].items():
if hid!=c['selected_hypothesis'] and h.get('working_brief_ref'):
current['other_working_briefs'][hid]=json.loads(reader.read_bytes(h['working_brief_ref']))
new=build_prompt(current)
before=native_characters(old);after=native_characters(new)
view=json.loads(new['messages'][1]['content'])['input']
rows.append({'context':ref,'baseline_prompt_version':c['prompt_version'],'baseline_characters':before,'v15_characters':after,
'reduction_percent':round(100*(1-after/before),2),'target_characters':13125,
'above_target':after>13125,'evidence_count':len(c['evidence']),
'visible_raw_ids':list(view['evidence']),'schema_ids':list(view['known_schemas']),
'archive_ids_count':len(view['archive_ids'])})
return {'package':str(path),'samples':len(rows),'median_reduction_percent':median(r['reduction_percent'] for r in rows),
'median_v15_characters':median(r['v15_characters'] for r in rows),
'min_v15_characters':min(r['v15_characters'] for r in rows),'max_v15_characters':max(r['v15_characters'] for r in rows),
'above_target_samples':sum(r['above_target'] for r in rows),'rows':rows}
def main():
p=argparse.ArgumentParser(description=__doc__);p.add_argument('packages',nargs='+',type=Path);p.add_argument('--output',type=Path,required=True);a=p.parse_args()
result={'measurement':'investigation_v15_offline_projection_v1','notes':'Actual Gemini wire text characters, not measured tokens, cost, model output or proof of quality. Old stored model findings are reused without rewriting.',
'results':[measure(path) for path in a.packages]}
a.output.write_text(json.dumps(result,indent=2)+'\n',encoding='utf-8')
for row in result['results']:print(json.dumps({k:v for k,v in row.items() if k!='rows'}))
if __name__=='__main__':main()