File size: 2,969 Bytes
e29f88c | 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 | """Independent, seed-replayable graph panels for frozen pointer benchmarks."""
from dataclasses import replace
import json
from pathlib import Path
from .pointer import (PointerExample, example_seed, execute, generate_unconditioned_example,
orbit_structure, render_prompt, validate_example)
def count_variant(task: PointerExample, count: int) -> PointerExample:
"""Change requested horizon/labels only; never condition graph sampling on it."""
states = execute(dict(task.mapping), task.initial_state, count)
result = replace(task, example_id=f'{task.example_id}-steps-{count}', task_depth=count,
intermediate_states=states, final_state=states[-1],
prompt=render_prompt(task.mapping, task.initial_state, count))
validate_example(result)
return result
def independent_graphs(config: dict, excluded: set[str]) -> tuple[list[PointerExample], list[dict]]:
"""Balanced seed/mode strata; reject only graph identity overlap, never difficulty."""
if config['graphs_per_seed_mode'] < 1 or not 1 <= config['max_depth'] <= 256:
raise ValueError('Need positive graph count and a maximum depth in 1..256')
if len(set(config['seeds'])) != len(config['seeds']) or len(set(config['graph_modes'])) != len(config['graph_modes']):
raise ValueError('Repeated seeds or graph modes')
used = set(excluded)
tasks, metadata = [], []
for seed in config['seeds']:
for mode in config['graph_modes']:
for index in range(config['graphs_per_seed_mode']):
for attempt in range(1000):
task = generate_unconditioned_example(example_seed(seed, f'benchmark-{mode}', index, attempt),
config['max_depth'], 'benchmark', index, mode)
if task.mapping_sha256 not in used:
break
else:
raise ValueError('Unable to sample a new rule table after 1000 attempts')
used.add(task.mapping_sha256)
task = replace(task, example_id=f'benchmark-{seed}-{mode}-{index:06d}-{task.seed:016x}')
validate_example(task)
tasks.append(task)
metadata.append({'graph_index': len(tasks) - 1, 'example_id': task.example_id,
'mapping_sha256': task.mapping_sha256, 'dataset_seed': seed, 'graph_mode': mode,
**orbit_structure(dict(task.mapping), task.initial_state)})
return tasks, metadata
def exclusion_hashes(directory: Path) -> set[str]:
"""Read graph fingerprints, including reserved splits, solely to prevent overlap."""
hashes = set()
for path in sorted(directory.glob('*.jsonl')):
for line in path.open():
record = json.loads(line)
hashes.add(record['mapping_sha256'])
if not hashes:
raise ValueError('No old dataset graph fingerprints found')
return hashes
|