File size: 10,636 Bytes
a130d59 | 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 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """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()))
# Independent manual integration validates sway, CFG semantics and SDE.
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])
|