| """Private cloud smoke test for the public adapter revision.""" |
| import hashlib |
| import json |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
|
|
| START = time.time() |
| OUT = Path('/kaggle/working/reload_check') |
| OUT.mkdir(parents=True, exist_ok=True) |
| REPO = 'sankalpsthakur/qwen3-06b-typed-decisions-cloud-pilot' |
| ADAPTER_REV = '4446265939ee157aa2a14e58bc8c6d23cb54701c' |
| BASE = 'Qwen/Qwen3-0.6B' |
| BASE_REV = 'c1899de289a04d12100db370d81485cdf75e47ca' |
| DATA = 'Praveenrajus/jev-bench' |
| DATA_REV = '18f88da81c28c2bec55edc31f63f2afdfba109ea' |
| EXPECTED_ADAPTER_SHA256 = '827fd0d08c308a89df55343ad7045d9d5b9ac22394e4fa876bdef8c2a001c845' |
|
|
| subprocess.check_call([sys.executable, '-m', 'pip', 'install', '-q', '--disable-pip-version-check', |
| 'transformers==4.57.6', 'peft==0.18.1', 'accelerate==1.12.0']) |
| import torch |
| from huggingface_hub import hf_hub_download |
| from peft import PeftModel |
| from transformers import AutoModelForCausalLM, AutoTokenizer |
|
|
| if not torch.cuda.is_available(): |
| raise RuntimeError('GPU unavailable') |
| weights = hf_hub_download(REPO, 'adapter_model.safetensors', revision=ADAPTER_REV) |
| actual_hash = hashlib.sha256(Path(weights).read_bytes()).hexdigest() |
| assert actual_hash == EXPECTED_ADAPTER_SHA256, actual_hash |
| pred_path = hf_hub_download(REPO, 'adapted-confirm-predictions.json', revision=ADAPTER_REV) |
| expected = json.loads(Path(pred_path).read_text())[:8] |
| source = hf_hub_download(DATA, 'data/boolq/test.jsonl', repo_type='dataset', revision=DATA_REV) |
| records = {r['id']: r for r in (json.loads(line) for line in Path(source).read_text().splitlines())} |
|
|
| tokenizer = AutoTokenizer.from_pretrained(BASE, revision=BASE_REV) |
| yes_id = tokenizer.encode('Yes', add_special_tokens=False) |
| no_id = tokenizer.encode('No', add_special_tokens=False) |
| assert len(yes_id) == len(no_id) == 1 |
| yes_id, no_id = yes_id[0], no_id[0] |
| base = AutoModelForCausalLM.from_pretrained(BASE, revision=BASE_REV, |
| torch_dtype=torch.float16, |
| attn_implementation='sdpa').to('cuda') |
| base.config.use_cache = False |
| model = PeftModel.from_pretrained(base, REPO, revision=ADAPTER_REV) |
| model.eval() |
|
|
| results = [] |
| with torch.inference_mode(): |
| for prior in expected: |
| state = json.loads(records[prior['id']]['state']) |
| content = ("Use only the evidence below to answer the question. " |
| "Reply with exactly Yes or No.\n\n" |
| f"Evidence: {state.get('passage', state.get('facts'))}\n\n" |
| f"Question: {state['question']}\n\nAnswer:") |
| def prefix(answer, token_id): |
| ids = tokenizer.apply_chat_template([ |
| {'role': 'user', 'content': content}, |
| {'role': 'assistant', 'content': answer} |
| ], tokenize=True, enable_thinking=False) |
| if hasattr(ids, 'keys'): |
| ids = ids['input_ids'] |
| pos = max(i for i, value in enumerate(ids) if value == token_id) |
| return ids[:pos] |
| yes_prefix = prefix('Yes', yes_id) |
| no_prefix = prefix('No', no_id) |
| assert yes_prefix == no_prefix |
| logits = model(torch.tensor([yes_prefix], device='cuda')).logits[0, -1] |
| p = float(torch.softmax(logits[[no_id, yes_id]].float(), dim=0)[1].cpu()) |
| delta = abs(p - prior['p_yes']) |
| results.append({'id': prior['id'], 'previous_p_yes': prior['p_yes'], |
| 'reloaded_p_yes': p, 'abs_delta': delta}) |
| print(json.dumps(results[-1]), flush=True) |
|
|
| max_delta = max(r['abs_delta'] for r in results) |
| assert max_delta < 0.002, max_delta |
| receipt = {'repo': REPO, 'revision': ADAPTER_REV, 'public_adapter_sha256': actual_hash, |
| 'gpu': torch.cuda.get_device_name(0), 'records_checked': len(results), |
| 'max_abs_probability_delta': max_delta, 'tolerance': 0.002, |
| 'elapsed_s': round(time.time() - START, 1), 'results': results} |
| (OUT / 'receipt.json').write_text(json.dumps(receipt, indent=2) + '\n') |
| print(json.dumps({'event': 'complete', **{k:v for k,v in receipt.items() if k != 'results'}}), flush=True) |
|
|
