| import torch |
| import sys |
| import os |
| import json |
|
|
| |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
|
|
| from safetensors import safe_open |
| from safetensors.torch import save_file |
| import comfy.model_management |
| import comfy.sd |
| import comfy.quant_ops |
|
|
| |
| INPUT_FILE = "/your model/weightst.safetensors" |
| OUTPUT_FILE = "/put something here/" |
| |
| TARGET_ARCH = "krea2" |
|
|
| |
| |
| |
| print("Reading metadata header...") |
| with safe_open(INPUT_FILE, framework="pt") as f: |
| metadata = f.metadata() or {} |
|
|
| |
| keys_to_remove = ['quantization', 'quant_method', 'unet_dtype', 'dtype'] |
| for key in list(metadata.keys()): |
| if key in keys_to_remove or 'quant' in key.lower(): |
| metadata.pop(key) |
|
|
| |
| metadata["architecture"] = TARGET_ARCH |
| metadata["unet_type"] = TARGET_ARCH |
| metadata["transformer_type"] = TARGET_ARCH |
| metadata["dtype"] = "bf16" |
| print("Cleaned Metadata:", json.dumps(metadata, indent=2)) |
|
|
| |
| |
| |
| print(f"\nLoading {INPUT_FILE}...") |
| try: |
| model_patcher = comfy.sd.load_diffusion_model(INPUT_FILE) |
| except Exception as e: |
| print(f"Failed to load: {e}") |
| sys.exit(1) |
|
|
| print("Patching model to instantiate ConvRot objects...") |
| try: |
| model_patcher.patch_model() |
| except Exception as e: |
| print(f"Failed to patch model: {e}") |
| sys.exit(1) |
|
|
| |
| |
| |
| new_sd = {} |
| dequant_count = 0 |
|
|
| |
| raw_sd = model_patcher.model_state_dict() |
| quantized_bases = set() |
| for k in raw_sd.keys(): |
| if k.endswith(".comfy_quant"): |
| quantized_bases.add(k[:-len(".comfy_quant")]) |
|
|
| print("Extracting and dequantizing weights from model tree...") |
|
|
| def process_module(module, prefix=""): |
| global dequant_count |
| for name, child in module.named_children(): |
| child_prefix = f"{prefix}.{name}" if prefix else name |
| |
| if hasattr(child, 'weight'): |
| weight_key = child_prefix + ".weight" |
| |
| |
| |
| if isinstance(child.weight, comfy.quant_ops.QuantizedTensor): |
| try: |
| dq = child.weight.dequantize() |
| |
| new_sd[weight_key] = dq.to(torch.bfloat16).contiguous() |
| dequant_count += 1 |
| except Exception as e: |
| print(f" ERROR dequantizing {weight_key}: {e}") |
| elif isinstance(child.weight, torch.Tensor): |
| new_sd[weight_key] = child.weight.to(torch.bfloat16).contiguous() |
| |
| if hasattr(child, 'bias') and child.bias is not None: |
| bias_key = child_prefix + ".bias" |
| new_sd[bias_key] = child.bias.to(torch.bfloat16).contiguous() |
| |
| |
| process_module(child, child_prefix) |
|
|
| |
| process_module(model_patcher.model) |
|
|
| |
| for k, v in raw_sd.items(): |
| if k.endswith(".comfy_quant"): continue |
| if k.endswith(".weight_scale") and k[:-len(".weight_scale")] in quantized_bases: continue |
| if k in new_sd: continue |
| |
| if isinstance(v, torch.Tensor): |
| if v.dtype == torch.int8: |
| print(f" WARNING: Raw int8 found at {k} without dequantize method. Skipping.") |
| continue |
| new_sd[k] = v.to(torch.bfloat16).contiguous() |
|
|
| |
| try: |
| model_patcher.unpatch_model() |
| except: |
| pass |
|
|
| |
| |
| |
| final_sd = {} |
| for k, v in new_sd.items(): |
| |
| new_key = "model." + k if not k.startswith("model.") else k |
| final_sd[new_key] = v |
|
|
| print(f"\nDequantized: {dequant_count} tensors") |
| print(f"Saving clean BF16 model to {OUTPUT_FILE}...") |
| save_file(final_sd, OUTPUT_FILE, metadata=metadata) |
| print("SUCCESS! File is a standard, unscrambled BF16 Krea2 model.") |
|
|