"""Quantize DiffusionGemma to NVFP4 using LLM Compressor v0.11.0
Model: google/diffusiongemma-26B-A4B-it- Total parameters: ~25.8B- Expert parameters: 22.8B (88.4%)- Non-expert parameters: 3.0B (11.6%)
Note: This will require a local update to transformers to support the model definition."""
import torchfrom compressed_tensors.offload import dispatch_modelfrom datasets import load_datasetfrom transformers import AutoProcessorfrom transformers.models.diffusion_gemma import DiffusionGemmaForBlockDiffusionfrom compressed_tensors.offload import dispatch_model
from llmcompressor import oneshotfrom llmcompressor.modeling.diffusion_gemma4 import ( # noqa: F401 CalibrationDiffusionGemmaTextExperts,)from llmcompressor.modifiers.quantization import QuantizationModifier
# Load modelMODEL_ID = "google/diffusiongemma-26B-A4B-it"
model = DiffusionGemmaForBlockDiffusion.from_pretrained( MODEL_ID, dtype="auto", trust_remote_code=True)processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
# CalibrationDiffusionGemmaTextExperts replaces the original# DiffusionGemmaTextExperts class during calibration to:# 1. Linearize the 3D expert tensors into individual nn.Linear modules# 2. Ensure all experts are properly calibrated, even those not activated# for certain tokens during calibration
# Configure the quantization scheme# NVFP4 (4-bit weights, 4-bit activations) for Linear layersrecipe = QuantizationModifier( targets="Linear", scheme="NVFP4", ignore=[ "lm_head", "re:.*embed.*", "re:.*self_attn", "re:.*router", "re:.*vision_tower.*", "re:.*self_conditioning.*", ],)
DATASET_ID = "neuralmagic/calibration"NUM_CALIBRATION_SAMPLES = 256MAX_SEQUENCE_LENGTH = 4096
ds = load_dataset(DATASET_ID, name="LLM", split=f"train[:{NUM_CALIBRATION_SAMPLES}]")
def preprocess_function(example): messgages = [] for message in example["messages"]: messgages.append( { "role": message["role"], "content": [{"type": "text", "text": message["content"]}], } )
return processor.apply_chat_template( messgages, return_tensors="pt", padding=False, truncation=True, max_length=MAX_SEQUENCE_LENGTH, tokenize=True, add_special_tokens=False, return_dict=True, add_generation_prompt=False, )
ds = ds.map(preprocess_function, batched=False, remove_columns=ds.column_names)
def data_collator(batch): assert len(batch) == 1 return { key: ( torch.tensor(value) if key != "pixel_values" else torch.tensor(value, dtype=torch.bfloat16).squeeze(0) ) for key, value in batch[0].items() }
# Apply quantization with calibration dataoneshot( model=model, recipe=recipe, dataset=ds, max_seq_length=MAX_SEQUENCE_LENGTH, num_calibration_samples=NUM_CALIBRATION_SAMPLES, data_collator=data_collator, sequential_targets=[ "DiffusionGemmaDecoderTextLayer", "DiffusionGemmaEncoderTextLayer", ],)
# Test sample generationprint("========== SAMPLE GENERATION ==============")dispatch_model(model)
# "The reason the sky is blue is because" + chat templateinput_ids = torch.tensor( [[ 2, 105, 2364, 107, 818, 3282, 506, 7217, 563, 3730, 563, 1547, 106, 107, 105, 4368, 107 ]]).to(model.device)
output = model.generate( input_ids, max_new_tokens=100, max_denoising_steps=48,)print(processor.tokenizer.decode(output[0]))print("==========================================\n\n")
# Save to disk in compressed-tensors formatSAVE_DIR = MODEL_ID.rstrip("/").split("/")[-1] + "-NVFP4"model.save_pretrained(SAVE_DIR)processor.save_pretrained(SAVE_DIR)