第 67 天：微調配方行前預檢、零樣本基準基準與支出熔斷

import math

config = {
'r': 16,
'lora_alpha': 32,
'target_modules': ['q_proj', 'k_proj', 'v_proj', 'o_proj'],
'lr': 2e-4,
'warmup_ratio': 0.03,
'min_lr_ratio': 0.1, # cosine decays to 10% of the peak
'max_steps': 1000,
'seq_len': 1024,
'loss_masking': True # prompt tokens get label -100 and add nothing to the loss
}
assert config['r'] == 16 and config['lora_alpha'] == 32
assert config['lr'] == 2e-4 and config['warmup_ratio'] == 0.03
assert config['loss_masking'] is True
assert config['lora_alpha'] / config['r'] == 2.0 # the adapter output is scaled by 2

def lr_at(step, c=config):
warmup = round(c['warmup_ratio'] * c['max_steps']) # 30 steps
if step < warmup:
return c['lr'] * (step + 1) / warmup
progress = (step - warmup) / (c['max_steps'] - warmup)
floor = c['min_lr_ratio'] * c['lr']
return floor + 0.5 * (c['lr'] - floor) * (1 + math.cos(math.pi * progress))

assert abs(lr_at(29) - 2e-4) < 1e-12 # warmup ends at the peak
assert abs(lr_at(1000) - 2e-5) < 1e-12 # cosine ends at 10% of the peak
assert lr_at(0) < lr_at(10) < lr_at(29)

# Loss masking, worked on one example: 5 prompt tokens and 3 answer tokens plus EOS.
input_ids = [101, 7, 8, 9, 102, 55, 56, 57, 2]
labels = [-100] * 5 + input_ids[5:]
assert sum(l != -100 for l in labels) == 4 and labels[-1] == 2 # only the answer and EOS train
# Peak lr 2e-4 is about 10x a typical full fine-tuning lr (1e-5 to 2e-5) because only the adapter moves.

import json
import hashlib

DECODING = {'temperature': 0.1, 'max_new_tokens': 256, 'seed': 0} # Day 69 reuses these exactly

def score(output, expected_keys):
try:
obj = json.loads(output)
except json.JSONDecodeError:
return False, 'not JSON'
if not isinstance(obj, dict):
return False, 'JSON but not an object'
missing = [k for k in expected_keys if k not in obj]
return (not missing), (f'missing {missing}' if missing else 'ok')

def evaluate_prompts(generate, test_prompts):
# generate(prompt, DECODING) -> str; the base model with NO adapter loaded.
rows = []
for item in test_prompts:
out = generate(item['prompt'], DECODING)
ok, why = score(out, item['expected_keys'])
rows.append({'id': item['id'], 'ok': ok, 'why': why, 'output': out})
suite = json.dumps(test_prompts, sort_keys=True).encode()
return {'accuracy': sum(r['ok'] for r in rows) / len(rows), 'n': len(rows),
'decoding': DECODING, 'suite_sha256': hashlib.sha256(suite).hexdigest(),
'failures': [r for r in rows if not r['ok']]}

results = evaluate_prompts(base_generate, test_prompts)
print(f"Base model accuracy: {results['accuracy']:.1%}")
with open('baseline_results.json', 'w') as f:
json.dump(results, f, indent=2)

# Typical base-model failures to quote verbatim in your notes:
# 'Sure! Here is the JSON you asked for: {...' -> 'not JSON' (chatty prefix)
# '{"name": "Ada"' -> 'not JSON' (stopped mid-object)
# '{"name": "Ada"}' when 'date' was required -> "missing ['date']"
# With 10 prompts and 1 or 2 passes the accuracy is 10% or 20%; each prompt moves it by 10 points,
# so do not read a 10-point change on Day 69 as a real improvement without more prompts.

import math

CEILING_USD = 10.00
MAX_HOURS = 3.0 # hard validator bound; the plan is 2.0 h and the timeout guard is 2.5 h

def check_budget_limits(max_cost, max_hours):
for name, value in (('max_cost', max_cost), ('max_hours', max_hours)):
# Reject bools, NaN and infinity: NaN > 10.0 is False, so a plain comparison lets it through.
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
raise ValueError(f'{name} must be a positive finite number, got {value!r}')
if max_cost > CEILING_USD or max_hours > MAX_HOURS:
raise ValueError(f'Budget violation: ${max_cost} / {max_hours}hrs exceeds ceiling')
return True

def planned_cost(rate_per_hour, hours):
return round(rate_per_hour * hours, 2)

assert planned_cost(0.60, 2.0) == 1.20 # A10G for the planned 2 hours
assert planned_cost(1.50, 2.0) == 3.00 # A100 for the planned 2 hours
assert planned_cost(1.50, 2.5) == 3.75 # worst case if the timeout guard fires
assert check_budget_limits(planned_cost(1.50, 2.5), 2.5)

for bad in ((10.01, 2.0), (5.0, 3.5), (float('nan'), 2.0), (5.0, float('inf')), (True, 2.0), (-1.0, 2.0)):
try:
check_budget_limits(*bad)
except ValueError:
continue
raise AssertionError(f'accepted {bad}')
print('budget guard rejects every unsafe configuration')

import json
import math
from pathlib import Path

class FinetunePreflightValidator:
def __init__(self, dataset_path, baseline_path, budget_limit):
self.dataset_path = Path(dataset_path)
self.baseline_path = Path(baseline_path)
self.budget_limit = budget_limit

def verify_data(self):
# Each JSONL row has input_ids and labels of equal length; at least one prompt token is
# masked (-100) and at least one answer token is trained.
rows = [json.loads(line) for line in self.dataset_path.read_text().splitlines() if line.strip()]
bad = [i for i, r in enumerate(rows)
if len(r['input_ids']) != len(r['labels'])
or r['labels'][0] != -100
or all(l == -100 for l in r['labels'])]
return {'ok': bool(rows) and not bad, 'data_samples_count': len(rows), 'bad_rows': bad[:10]}

def verify_baseline(self):
if not self.baseline_path.exists():
return {'ok': False, 'reason': 'baseline_results.json missing'}
base = json.loads(self.baseline_path.read_text())
acc = base.get('accuracy')
ok = isinstance(acc, (int, float)) and 0 <= acc <= 1 and base.get('n', 0) > 0 and 'decoding' in base
return {'ok': ok, 'baseline_score': acc, 'n': base.get('n')}

def verify_budget(self):
b = self.budget_limit
ok = isinstance(b, (int, float)) and not isinstance(b, bool) and math.isfinite(b) and 0 < b <= 10.0
return {'ok': ok, 'budget_limit': b}

def validate(self):
checks = {'data': self.verify_data(), 'baseline': self.verify_baseline(), 'budget': self.verify_budget()}
verdict = 'GO' if all(c['ok'] for c in checks.values()) else 'NO-GO'
return {'verdict': verdict, 'checks': checks,
'baseline_score': checks['baseline'].get('baseline_score'),
'data_samples_count': checks['data']['data_samples_count']}

# Fixture: two masked rows and a baseline from Activity 2.
Path('train.jsonl').write_text('\n'.join(json.dumps(r) for r in [
{'input_ids': [101, 7, 8, 102, 55, 2], 'labels': [-100, -100, -100, -100, 55, 2]},
{'input_ids': [101, 9, 102, 56, 57, 2], 'labels': [-100, -100, -100, 56, 57, 2]}]))
Path('baseline_results.json').write_text(json.dumps({'accuracy': 0.2, 'n': 10, 'decoding': {'temperature': 0.1}}))

report = FinetunePreflightValidator('train.jsonl', 'baseline_results.json', 10.0).validate()
assert report['verdict'] == 'GO' and report['data_samples_count'] == 2 and report['baseline_score'] == 0.2
# Every check must be able to say NO-GO; prove it with one failing input per check.
assert FinetunePreflightValidator('train.jsonl', 'baseline_results.json', 12.0).validate()['verdict'] == 'NO-GO'
assert FinetunePreflightValidator('train.jsonl', 'missing.json', 10.0).validate()['verdict'] == 'NO-GO'
assert FinetunePreflightValidator('train.jsonl', 'baseline_results.json', float('nan')).validate()['verdict'] == 'NO-GO'
print(json.dumps(report, indent=2))