| """Full-depth real-weight component checks on small synthetic CPU fixtures.""" |
| import gc |
| import hashlib |
| import importlib.metadata as metadata |
| import importlib.util |
| import json |
| import math |
| import os |
| import sys |
| import time |
| from pathlib import Path |
|
|
| import torch |
| import torch.nn.functional as F |
| from safetensors import safe_open |
|
|
| ROOT = Path(__file__).resolve().parent |
|
|
|
|
| def main(name): |
| root = ROOT; started = time.monotonic(); checks = [] |
| torch.set_num_threads(5); torch.manual_seed(1027) |
| def record(label, passed, **details): |
| row = {'name': label, 'passed': bool(passed), **details}; checks.append(row) |
| print(json.dumps(row), flush=True) |
| assert passed, label |
| def close(label, actual, expected, atol=2e-4, rtol=2e-4): |
| record(label, torch.allclose(actual, expected, atol=atol, rtol=rtol), max_abs=float((actual-expected).abs().max()), atol=atol, rtol=rtol) |
| extraction = json.loads((root / 'extraction.json').read_text()) |
| with (root / 'model.safetensors').open('rb') as stream: |
| digest = hashlib.file_digest(stream, 'sha256').hexdigest() |
| record('export_weight_sha256', digest == extraction['sha256']) |
| shapes = {}; dtypes = {}; parameters = 0 |
| with safe_open(root / 'model.safetensors', framework='pt') as handle: |
| for key in handle.keys(): |
| value = handle.get_slice(key); shapes[key] = value.get_shape() |
| parameters += math.prod(shapes[key]); dtype = value.get_dtype() |
| dtypes[dtype] = dtypes.get(dtype, 0) + math.prod(shapes[key]) |
| spec = importlib.util.spec_from_file_location('priority_component_' + name, root / 'component.py') |
| module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) |
| print('LOAD_NATIVE', name, flush=True) |
| component = module.load_component(root, dtype=torch.float32, device='cpu') |
| native = component.native |
| record('strict_native_load_all_keys_and_shapes', {k: list(v.shape) for k, v in native.state_dict().items()} == shapes, tensors=len(shapes), parameters=parameters, original_dtypes=dtypes) |
| calls = [0]; original_sdpa = F.scaled_dot_product_attention |
| def explicit(q, k, v, attn_mask=None, dropout_p=0.0, is_causal=False, scale=None, enable_gqa=False): |
| calls[0] += 1 |
| if enable_gqa: |
| k = k.repeat_interleave(q.shape[-3] // k.shape[-3], -3) |
| v = v.repeat_interleave(q.shape[-3] // v.shape[-3], -3) |
| logits = (q @ k.transpose(-2, -1)) * (scale if scale is not None else q.shape[-1] ** -0.5) |
| if is_causal: |
| logits = logits.masked_fill(~torch.ones(q.shape[-2], k.shape[-2], dtype=torch.bool, device=q.device).tril(), -float('inf')) |
| if attn_mask is not None: |
| logits = logits.masked_fill(~attn_mask, -float('inf')) if attn_mask.dtype == torch.bool else logits + attn_mask |
| return logits.softmax(-1) @ v |
| with torch.inference_mode(): |
| if name == 'internvlu': |
| record('nonlearned_rope_cache_materialized', not native.decoder.pos_embed.all_inv_freqs.is_meta and torch.isfinite(native.decoder.pos_embed.all_inv_freqs).all()) |
| record('native_complete_20_layers', len(native.decoder.transformer_blocks) == 20) |
| x = torch.randn(1, 16, 4, 6) * .2; condition = torch.randn(1, 4, 4096) * .1 |
| y = component.predict(x, 500., condition) |
| record('rectangular_native_latent_forward', y.shape == x.shape and torch.isfinite(y).all(), rms=float(y.square().mean().sqrt())) |
| reference = native.decoder(hidden_states=x, encoder_hidden_states=condition, timestep=torch.tensor([500.]), encoder_attention_mask=torch.ones(1,4,dtype=torch.bool), encoder_image_token_mask=torch.zeros(1,4,dtype=torch.bool), image_fhw_cond=[torch.empty(0,3,dtype=torch.long)], return_dict=False)[0] |
| close('original_native_vs_exported_interface', y, reference, 0, 0) |
| F.scaled_dot_product_attention = explicit |
| try: |
| alternative = component.predict(x, 500., condition) |
| finally: |
| F.scaled_dot_product_attention = original_sdpa |
| close('all_layers_sdpa_vs_explicit_attention', y, alternative, 5e-4, 5e-4) |
| record('conditioning_changes_prediction', not torch.allclose(y, component.predict(x, 500., condition * -.7))) |
| record('time_changes_prediction', not torch.allclose(y, component.predict(x, 700., condition))) |
| condition2 = condition.clone(); condition2[:, -1] = torch.randn(1,4096) * 100 |
| mask = torch.tensor([[1,1,1,0]],dtype=torch.bool) |
| close('masked_condition_isolation', component.predict(x,500.,condition,condition_mask=mask), component.predict(x,500.,condition2,condition_mask=mask), 3e-4, 3e-4) |
| batch = component.predict(torch.cat([x,x*.7]), torch.tensor([500.,700.]), torch.cat([condition,condition*-.7])) |
| close('batch_isolation_first', batch[:1], y, 5e-4, 5e-4) |
| close('batch_isolation_second', batch[1:], component.predict(x*.7,700.,condition*-.7), 5e-4, 5e-4) |
| visual = component.predict(x,500.,condition,condition_image_mask=torch.ones(1,4,dtype=torch.bool),condition_grid=[torch.tensor([[1,2,2]])]) |
| record('semantic_visual_condition_grid', visual.shape == x.shape and torch.isfinite(visual).all()) |
| edit_x = torch.randn(1,16,8,8)*.2 |
| edited = component.predict(edit_x,500.,condition,reference_latents=[torch.randn(1,16,8,8)*.2]) |
| record('reference_vae_editing_input', edited.shape == edit_x.shape and torch.isfinite(edited).all()) |
| scheduler = component.make_scheduler(); scheduler.set_timesteps(3) |
| record('native_dpm_flow_scheduler', type(scheduler).__name__ == 'DPMSolverMultistepScheduler' and scheduler.config.prediction_type == 'flow_prediction' and scheduler.config.use_flow_sigmas and scheduler.config.flow_shift == 3.) |
| z = x.clone(); trajectory = [] |
| for timestep in scheduler.timesteps: |
| pred = component.predict(z,timestep,condition) |
| z = scheduler.step(pred,timestep,z,return_dict=False)[0] |
| trajectory.append(float(z.square().mean().sqrt())) |
| record('three_step_native_dpm_trajectory', torch.isfinite(z).all() and not torch.equal(z,x), latent_rms=trajectory) |
| else: |
| record('native_complete_eight_layers', len(native.blocks) == 8) |
| x = torch.randn(1,4,64)*.2; history = torch.randn(1,32,64)*.2; condition = torch.randn(1,1,896)*.1 |
| y = component.predict(x,.4,condition,history) |
| record('native_current_patch_prediction', y.shape == x.shape and torch.isfinite(y).all(), rms=float(y.square().mean().sqrt())) |
| reference = native(x,torch.tensor([.4]),condition,history,None)[:, -4:] |
| close('original_native_vs_exported_interface',y,reference,0,0) |
| F.scaled_dot_product_attention = explicit |
| try: |
| alternative = component.predict(x,.4,condition,history) |
| finally: |
| F.scaled_dot_product_attention = original_sdpa |
| close('all_layers_sdpa_vs_explicit_attention',y,alternative) |
| record('talker_condition_changes_prediction',not torch.allclose(y,component.predict(x,.4,condition*-.7,history))) |
| record('history_changes_prediction',not torch.allclose(y,component.predict(x,.4,condition,history*-.7))) |
| record('time_changes_prediction',not torch.allclose(y,component.predict(x,.7,condition,history))) |
| batch = component.predict(torch.cat([x,x*.7]),torch.tensor([.4,.7]),torch.cat([condition,condition*-.7]),torch.cat([history,history*-.7])) |
| close('batch_isolation_first',batch[:1],y) |
| close('batch_isolation_second',batch[1:],component.predict(x*.7,.7,condition*-.7,history*-.7)) |
| null = component.predict(x,.4,torch.zeros_like(condition),history) |
| packed = native.forward_with_cfg(x,torch.tensor(.4),condition,history,None) |
| close('native_cfg_condition_branch',packed[:1],y) |
| close('native_cfg_null_branch',packed[1:],null) |
| epss = component.times_function(10,x.device,x.dtype) |
| record('native_nonuniform_ten_step_times',torch.equal(epss,torch.tensor([0,2,4,6,8,12,16,20,24,28,32])/32)) |
| generated = component.sample(condition,history,x) |
| record('ten_step_native_cfm_trajectory',generated.shape==x.shape and torch.isfinite(generated).all() and not torch.equal(generated,x),rms=float(generated.square().mean().sqrt())) |
| |
| sway = epss - (torch.cos(torch.pi/2*epss)-1+epss) |
| z = x.clone() |
| for i in range(10): |
| cond = component.predict(z,sway[i],condition,history) |
| uncond = component.predict(z,sway[i],torch.zeros_like(condition),history) |
| z = z + (cond + 2*(cond-uncond)) * (sway[i+1]-sway[i]) |
| close('cfm_vs_independent_manual_cfg_integration',generated,z,4e-4,4e-4) |
| explicit_noise = [torch.randn_like(x) for _ in range(10)] |
| stochastic = component.sample(condition,history,x,sigma=.1,temperature=.2,sde_noise=explicit_noise) |
| record('native_sde_noise_path',torch.isfinite(stochastic).all() and not torch.allclose(stochastic,generated)) |
| record('independent_attention_reference_exercised',calls[0] >= (20 if name=='internvlu' else 8), calls=calls[0]) |
| versions = {k: metadata.version(k) for k in ('torch','safetensors','diffusers','transformers','einops','huggingface_hub')} |
| if name == 'ming_audio': |
| versions['x-transformers'] = metadata.version('x-transformers') |
| report = {'as_of':'2026-10-27','strict_load':True,'source_tensor_bytes_equal':True,'smoke':{'passed':True,'device':'cpu','runtime_dtype':'float32','synthetic_conditions':True,'real_text_or_audio_conditioning_tested':False,'codec_decoding_tested':False},'checks':checks,'tensors':len(shapes),'parameters':parameters,'original_dtype_elements':dtypes,'elapsed_seconds':round(time.monotonic()-started,3),'versions':versions,'limitations':['Full-depth real-weight CPU component fixtures using synthetic conditions and native latents.','No full VLM/Talker conditioning or image/audio codec decoding; no end-to-end generation-quality claim.','Checkpoint dtypes are preserved; float32 is a runtime validation cast only.']} |
| (root/'validation.json').write_text(json.dumps(report,indent=2)) |
| print('VALIDATED',name,len(checks),flush=True) |
|
|
|
|
| if __name__ == '__main__': |
| main(sys.argv[1]) |
|
|
