Files
VianoBugAnalysis/scripts/measure_investigation_context.py

62 lines
3.8 KiB
Python

"""Compare archived project investigation prompts offline; never invoke an LLM."""
import argparse
import copy
import json
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
from viano_bug_analysis.analysis.investigation_view import working_view, thin_view
from viano_bug_analysis.analysis.investigation_consolidation import consolidated_view
from viano_bug_analysis.analysis.investigation_blocks import block_view, restore_blocks
from viano_bug_analysis.case.contracts import read_json
from viano_bug_analysis.package import PackageReader
def restore_tables(view):
return restore_blocks(view)
def measure(path, target_version='investigation_project_v7'):
if target_version not in {'investigation_project_v5', 'investigation_project_v6', 'investigation_project_v7'}:
raise ValueError('Unsupported model view version')
records = []
with PackageReader(path) as reader:
for ref in sorted(reader.files):
if not ref.startswith('analysis/investigation/') or '/contexts/' not in ref or not ref.endswith('.json'):
continue
context = read_json(reader, ref, 'investigation-context', 4 * 1024**2)
if context.get('prompt_version') != 'investigation_project_v4':
continue
old = build_prompt(context)
projected = copy.deepcopy(context)
projected['prompt_version'] = target_version
new = build_prompt(projected)
old_view = working_view(projected)
new_view = block_view(projected) if target_version == 'investigation_project_v7' else consolidated_view(projected) if target_version == 'investigation_project_v6' else thin_view(projected)
if json.dumps(restore_tables(copy.deepcopy(new_view)), sort_keys=True) != json.dumps(old_view, sort_keys=True):
raise ValueError('Model view roundtrip failed')
records.append({'context_ref': ref,
'old_characters': len(old['messages'][1]['content']),
'new_characters': len(new['messages'][1]['content']),
'old_message_characters': sum(len(m['content']) for m in old['messages']),
'new_message_characters': sum(len(m['content']) for m in new['messages']),
'old_components': {k: len(json.dumps(v, ensure_ascii=False, separators=(',', ':'))) for k, v in old_view.items()},
'new_components': {k: len(json.dumps(v, ensure_ascii=False, separators=(',', ':'))) for k, v in new_view.items()}})
old = sum(r['old_characters'] for r in records)
new = sum(r['new_characters'] for r in records)
old_messages = sum(r['old_message_characters'] for r in records)
new_messages = sum(r['new_message_characters'] for r in records)
return {'target_version': target_version, 'stored_contexts': len(records), 'old_user_characters': old, 'new_user_characters': new,
'reduction_percent': round(100 * (1 - new / old), 2) if old else None,
'old_message_characters': old_messages, 'new_message_characters': new_messages,
'message_reduction_percent': round(100 * (1 - new_messages / old_messages), 2) if old_messages else None,
'lossless_roundtrip': True, 'unit': 'characters_not_provider_tokens',
'scope': 'stored_contexts_not_billed_calls', 'contexts': records}
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('package')
parser.add_argument('--target-version', choices=['investigation_project_v5', 'investigation_project_v6', 'investigation_project_v7'], default='investigation_project_v7')
args = parser.parse_args()
print(json.dumps(measure(args.package, args.target_version), indent=2))