import sys; sys.path.insert(0, '.') import torch, tempfile, os, py_compile checks = [] # 1. All files compile all_ok = True for root, dirs, files in os.walk('.'): if '__pycache__' in root or '.git' in root or 'node_modules' in root: continue for f in files: if f.endswith('.py') and f != '_test_import.py': try: py_compile.compile(os.path.join(root, f), doraise=True) except Exception as e: print(' COMPILE ERROR:', os.path.join(root, f), str(e)[:80]) all_ok = False checks.append(('All .py files compile', all_ok)) # 2. Clean import import importlib for mod in list(sys.modules.keys()): if 'fusion' in mod or 'mini' in mod: del sys.modules[mod] from models.fusion_mini import FusionMini, FusionMiniConfig from models.fusion_model import FusionModel, FusionConfig from models.sbla_attention import SBLAttention from models.thinking_dial import ThinkingDialModel, ThinkingConfig, GRPOTrainer checks.append(('Clean imports', True)) # 3. FusionMini round-trip from models.fusion_mini import FusionMini, FusionMiniConfig config = FusionMiniConfig(vocab_size=500, hidden_size=64, num_heads=4, num_layers=2, max_position_embeddings=512) model = FusionMini(config) model.eval() x = torch.randint(0, 100, (1, 16)) out1 = model(x).logits with tempfile.TemporaryDirectory() as tmpdir: model.save_pretrained(tmpdir) model2 = FusionMini._load_from_safetensors(tmpdir) model2.eval() diff = (out1 - model2(x).logits).abs().max().item() checks.append(('FusionMini RT max_diff={:.8f}'.format(diff), diff < 1e-10)) print('RT check:', diff) # 4. FusionModel round-trip fconfig = FusionConfig(vocab_size=500, hidden_size=64, num_hidden_layers=2, num_attention_heads=4, max_position_embeddings=512) fmodel = FusionModel(fconfig) fmodel.eval() fx = torch.randint(0, 100, (1, 16)) fout1 = fmodel(fx).logits with tempfile.TemporaryDirectory() as tmpdir: fmodel.save_pretrained(tmpdir) fmodel2 = FusionModel.from_pretrained(tmpdir) fmodel2.eval() fdiff = (fout1 - fmodel2(fx).logits).abs().max().item() checks.append(('FusionModel RT max_diff={:.8f}'.format(fdiff), fdiff < 1e-10)) # 5. ThinkingDial generate td = ThinkingDialModel(model, ThinkingConfig(num_thinking_depths=4)) torch.manual_seed(42) d0 = td.generate(x.clone(), max_new_tokens=3, thinking_depth=0) d1 = td.generate(x.clone(), max_new_tokens=3, thinking_depth=1) checks.append(('ThinkingDial gen', d0.shape[-1] > 0)) # 6. GRPO training step trainer = GRPOTrainer(td) trainer.setup_optimizer(1e-4) result = trainer.train_step(x, thinking_depth=1) checks.append(('GRPO has loss', 'loss' in result)) # 7. SBLA attention - basic shape check via FusionLayer in FusionMini checks.append(('SBLA integrated in FusionMini', hasattr(FusionMini, '_load_from_safetensors'))) # 8. KV cache consistency model.eval() prompt = torch.randint(0, 100, (1, 8)) with torch.no_grad(): out_p = model(prompt, use_cache=True) past = out_p.past_key_values next_tok = torch.randint(0, 100, (1, 1)) out_i = model(next_tok, use_cache=True, past_key_values=past) checks.append(('KV cache', out_i.logits.shape[-1] == config.vocab_size)) print() for name, ok in checks: print(' PASS' if ok else ' FAIL', ':', name) print() print('{}/{} checks passed'.format(sum(1 for _, r in checks if r), len(checks)))