File size: 5,471 Bytes
5adf069 | 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 101 102 103 104 105 106 107 108 | """Leakage, label and measurement checks independent of CUDA."""
import copy
import json
import tempfile
import unittest
from pathlib import Path
from fev4b.data import split_records, require_disjoint, target_index
from fev4b.metrics import rows_for, summarize
from fev4b.calibrate import fit_temperature, rescale
from fev4b.benchmark import evaluate
from fev4b.adaptation import base_identity, verify_adaptation
from fev4b.integrity import file_sha256
from tools.compare_runs import compare
def record(i, group=None):
return {'id': str(i), 'group': group or str(i), 'state': str(i), 'questions': {
'q': {'type': 'choice', 'criteria': {'yes': None, 'no': None}, 'label': 'yes'}}}
class TrainingToolsTests(unittest.TestCase):
def test_overlay_rejects_other_base_or_modified_payload(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root/'adapter').mkdir()
config = {'r': 8, 'lora_alpha': 16}
(root/'adapter/adapter_config.json').write_text(json.dumps(config))
manifest = {'files': {'checkpoint/model.safetensors': {'bytes': 4, 'sha256': 'base'},
'adapter/adapter_model.safetensors': {'bytes': 4, 'sha256': 'shared'}}}
(root/'manifest.json').write_text(json.dumps(manifest))
overlay = root/'overlay'
overlay.mkdir()
(overlay/'adapter_config.json').write_text(json.dumps(config))
(overlay/'adapter_model.safetensors').write_bytes(b'test')
meta = {'format': 'fev-adaptation-v1', 'baseIdentity': base_identity(root), 'temperature': 1.5,
'files': {name: file_sha256(overlay/name) for name in
('adapter_config.json', 'adapter_model.safetensors')}}
(overlay/'adaptation.json').write_text(json.dumps(meta))
self.assertEqual(verify_adaptation(root, overlay)['temperature'], 1.5)
(overlay/'adapter_model.safetensors').write_bytes(b'changed')
with self.assertRaises(ValueError):
verify_adaptation(root, overlay)
(overlay/'adapter_model.safetensors').write_bytes(b'test')
manifest['files']['checkpoint/model.safetensors']['sha256'] = 'other'
(root/'manifest.json').write_text(json.dumps(manifest))
with self.assertRaises(ValueError):
verify_adaptation(root, overlay)
def test_comparison_rejects_different_labels_and_preserves_paired_counts(self):
a = rows_for(record(0), {'answers': {'q': {'type': 'choice', 'probabilities': {'yes': .9, 'no': .1}}}})
b = rows_for(record(0), {'answers': {'q': {'type': 'choice', 'probabilities': {'yes': .1, 'no': .9}}}})
result = compare({'a': a, 'b': b})
self.assertEqual(result['pairedOutcomes']['a vs b']['aOnlyCorrect'], 1)
b[0]['target'] = 1
with self.assertRaises(ValueError):
compare({'a': a, 'b': b})
def test_group_bridge_does_not_leak(self):
records = [record(i) for i in range(30)]
duplicate = record(31, group='0')
duplicate['state'] = '1'
records.append(duplicate)
split = split_records(records)
require_disjoint(*split.values())
partition = next(rows for rows in split.values() if any(r['id'] == '0' for r in rows))
self.assertTrue({'0', '1', '31'} <= {r['id'] for r in partition})
with self.assertRaises(ValueError):
require_disjoint([record(0)], [record(1, group='0')])
def test_label_types_are_strict(self):
with self.assertRaises(ValueError):
target_index({'type': 'noul', 'label': 1})
with self.assertRaises(ValueError):
target_index({'type': 'score', 'criteria': ['low', 'high'], 'label': True})
def test_errors_remain_in_denominator_and_confidence_is_top_probability(self):
response = {'answers': {'q': {'type': 'choice', 'probabilities': {'yes': .9, 'no': .1}}}}
rows = rows_for(record(0), response) + rows_for(record(1), error='Timeout')
report = summarize(rows)
self.assertEqual(report['accuracyAllQuestions'], .5)
self.assertEqual(report['coverageByFixedThreshold'][3]['coverage'], .5)
response['answers']['q']['probabilities'] = {'yes': 2., 'no': -1.}
with self.assertRaises(ValueError):
rows_for(record(0), response)
def test_temperature_does_not_change_argmax(self):
values = [.05, .15, .8]
for factor in (.2, 1., 5.):
self.assertEqual(max(range(3), key=rescale(values, factor).__getitem__), 2)
rows = [{'error': None, 'probabilities': [.99, .01], 'target': 1} for _ in range(3)]
_, report = fit_temperature(rows)
self.assertLessEqual(report['afterNll'], report['beforeNll'])
def test_benchmark_keeps_failed_records(self):
def predict(request):
if request['state'] == '1':
raise RuntimeError('failed')
return {'model': 'fake', 'answers': {'q': {'type': 'choice', 'probabilities': {'yes': .8, 'no': .2}}}}
with tempfile.TemporaryDirectory() as directory:
report = evaluate([record(0), record(1)], predict, Path(directory)/'run', provenance={})
self.assertEqual(report['overall']['questions'], 2)
self.assertEqual(report['overall']['errors'], 1)
if __name__ == '__main__':
unittest.main()
|