Support Qwix quantization on NNX

Author: hsuan-lun-chiang

src/maxtext/layers/nnx_decoders.py

@@ -671,22 +671,7 @@ def layer_fn(carry, scanned_vars):
       params = nnx_ensure_scan_leading_axis(params, length)
       state = nnx_ensure_scan_leading_axis(state, length)
 
-      # Linen FP8 ops keep amax_history in mutable Linen scope; jax.lax.scan
-      # leaks the tracer and hits UnexpectedTracerError. Use a Python for-loop
-      # for FP8 instead.
-      uses_linen_fp8_mutable_state = self.config.quantization in ("fp8_nanoo", "fp8_gpu")
-      if uses_linen_fp8_mutable_state:
-        carry = x_in
-        per_layer_states = []
-        for i in range(length):
-          current_params = jax.tree.map(lambda x, i=i: x[i], params)
-          current_state = jax.tree.map(lambda x, i=i: x[i], state)
-          carry, new_state_i = layer_fn(carry, (current_params, current_state))
-          per_layer_states.append(new_state_i)
-        final_carry = carry
-        scanned_state = jax.tree.map(lambda *xs: jnp.stack(list(xs)), *per_layer_states)
-      else:
-        final_carry, scanned_state = jax.lax.scan(layer_fn_wrapped, x_in, (params, state))
+      final_carry, scanned_state = jax.lax.scan(layer_fn_wrapped, x_in, (params, state))
       returned_kv_stacked = None
 
     if scan_axis != 0:

src/maxtext/layers/quantizations.py

@@ -707,6 +707,30 @@ def configure_kv_quant(config):
   return None if not config.quantize_kvcache else KVQuant(config)
 
 
+def _apply_linen_module_in_nnx(linen_module_cls, op_id, *args, **kwargs):
+  try:
+    from qwix._src import flax_util
+    parent = flax_util.get_current_module()
+    from flax import nnx
+    is_nnx = isinstance(parent, nnx.Module)
+  except Exception:
+    is_nnx = False
+
+  if is_nnx:
+    attr_name = f"_qwix_fp8_gpu_{op_id}"
+    if not hasattr(parent, attr_name):
+      from maxtext.layers import nnx_wrappers
+      rngs = getattr(parent, "qwix_rngs", None)
+      if rngs is None:
+        rngs = nnx.Rngs(0)
+      wrapper = nnx_wrappers.ToNNX(linen_module_cls(name=op_id), rngs=rngs)
+      wrapper.lazy_init(*args, **kwargs)
+      setattr(parent, attr_name, wrapper)
+    return getattr(parent, attr_name)(*args, mutable=["_overwrite_with_gradient"], **kwargs)
+  else:
+    return linen_module_cls(name=op_id)(*args, **kwargs)
+
+
 class NvidaFp8Provider(qwix.QtProvider):
   """Wraps nn.Fp8DirectDotGeneralOp with Qwix's provider interface."""
 
@@ -715,13 +739,13 @@ def dot_general(self, *args, **kwargs):
     rule, op_id = self._get_current_rule_and_op_id("dot_general")
     if rule is None:
       return jax.lax.dot_general(*args, **kwargs)
-    return nn.Fp8DirectDotGeneralOp(name=op_id)(*args, **kwargs)
+    return _apply_linen_module_in_nnx(nn.Fp8DirectDotGeneralOp, op_id, *args, **kwargs)
 
   def einsum(self, *args, **kwargs):
     rule, op_id = self._get_current_rule_and_op_id("einsum")
     if rule is None:
       return jnp.einsum(*args, **kwargs)
-    return nn.Fp8Einsum(name=op_id)(*args, **kwargs)
+    return _apply_linen_module_in_nnx(nn.Fp8Einsum, op_id, *args, **kwargs)
 
 
 class NANOOFp8Provider(qwix.QtProvider):
@@ -731,7 +755,7 @@ def dot_general(self, *args, **kwargs):
     rule, op_id = self._get_current_rule_and_op_id("dot_general")
     if rule is None:
       return jax.lax.dot_general(*args, **kwargs)
-    return nn.NANOOFp8DotGeneralOp(name=op_id)(*args, **kwargs)
+    return _apply_linen_module_in_nnx(nn.NANOOFp8DotGeneralOp, op_id, *args, **kwargs)
 
 
 def get_fp8_full_qwix_rule_w_sparsity(config: Config):
@@ -812,7 +836,15 @@ def maybe_quantize_model(model, config):
   if config.use_qwix_quantization and not config.use_batch_split_schedule:
     quantization_provider = get_qt_provider(config)
     if quantization_provider:
-      model = qwix.quantize_model(model, quantization_provider)
+      if config.pure_nnx:
+        input_shape = (config.micro_batch_size_to_train_on, config.max_target_length)
+        import jax.numpy as jnp
+        dummy_tokens = jnp.ones(input_shape, dtype=jnp.int32)
+        dummy_positions = jnp.ones(input_shape, dtype=jnp.int32)
+        dummy_segment_ids = jnp.ones(input_shape, dtype=jnp.int32)
+        model = qwix.quantize_model(model, quantization_provider, dummy_tokens, dummy_positions, dummy_segment_ids, enable_dropout=False)
+      else:
+        model = qwix.quantize_model(model, quantization_provider)
   return model
 
 

src/maxtext/trainers/pre_train/train.py

@@ -349,7 +349,8 @@ def train_step(model, config, state_mesh_shardings, params_shardings, state, dat
           is_train=True,
       )
     else:
-      model_graphdef, curr_params, rest = nnx.split(state.model, nnx.Param, ...)
+      OverwriteWithGradient = nnx.variablelib.variable_type_from_name(maxtext_utils.OVERWRITE_WITH_GRADIENT, allow_register=True)
+      model_graphdef, curr_params, overwrite_vars, rest = nnx.split(state.model, nnx.Param, OverwriteWithGradient, ...)
       if config.parameter_memory_host_offload:
         # Params are kept on host (pinned_host) in in_shardings. Move only Param
         # variables to device before the forward/backward pass so that all dot_general
@@ -371,15 +372,16 @@ def train_step(model, config, state_mesh_shardings, params_shardings, state, dat
         )
         nnx.update(state.model, curr_params)
 
-      def diff_wrapper(param, rest, config, data):
-        local_model = nnx.merge(model_graphdef, param, rest, copy=True)
+      def diff_wrapper(param, overwrite_vars, rest, config, data):
+        local_model = nnx.merge(model_graphdef, param, overwrite_vars, rest, copy=True)
         loss, aux = loss_fn(local_model, config, data, None, None, is_train=True)
-        _, _, new_rest = nnx.split(local_model, nnx.Param, ...)
-        return loss, (aux, new_rest)
+        _, _, new_overwrite_vars, new_rest = nnx.split(local_model, nnx.Param, OverwriteWithGradient, ...)
+        return loss, (aux, new_overwrite_vars, new_rest)
 
-      grad_func = jax.value_and_grad(diff_wrapper, argnums=0, has_aux=True)
-      (loss, (aux, new_rest)), raw_grads = grad_func(curr_params, rest, config, data)
+      grad_func = jax.value_and_grad(diff_wrapper, argnums=(0, 1), has_aux=True)
+      (loss, (aux, new_overwrite_vars, new_rest)), (raw_grads, overwrite_grads) = grad_func(curr_params, overwrite_vars, rest, config, data)
       nnx.update(state.model, new_rest)
+      nnx.update(state.model, overwrite_grads)
 
   raw_grads = jax.tree_util.tree_map(
       lambda x: x.astype(config.grad_dtype) if x.dtype == jnp.float32 else x,

tests/unit/quantizations_test.py

@@ -387,17 +387,117 @@ def compare_fn(path, x, y):
 
     jax.tree_util.tree_map_with_path(compare_fn, a, b)
 
-  def quantization_config(self, quant, logits_tolerance=2e-1, grad_tolerance=5e-1):
+  def quantization_config(self, quant, logits_tolerance=2e-1, grad_tolerance=5e-1, **kwargs):
     """Run forward pass and backward pass for quantized model and compare with base model."""
     # pylint: disable=protected-access
-    cfg = self.init_pyconfig(quantization=quant)
-    qt_model = model_creation_utils.create_model(cfg, self.mesh)
-
+    cfg = self.init_pyconfig(quantization=quant, **kwargs)
     ids, decoder_segment_ids, decoder_positions = self.get_data()
 
-    if not hasattr(self.__class__, "_cached_base_results"):
-      model = model_creation_utils.create_model(self.cfg, self.mesh)
-      var = model.init(
+    if cfg.pure_nnx:
+      qt_model = model_creation_utils.create_model(cfg, self.mesh, rngs=nnx.Rngs(0))
+      if getattr(self.__class__, "_cached_base_results_nnx", None) is None:
+        base_cfg = self.init_pyconfig(quantization="", **kwargs)
+        base_model = model_creation_utils.create_model(base_cfg, self.mesh, rngs=nnx.Rngs(0))
+
+        def loss_base(model):
+          logits = model(
+              decoder_input_tokens=ids,
+              decoder_positions=decoder_positions,
+              decoder_segment_ids=decoder_segment_ids,
+              enable_dropout=False,
+          )
+          return jnp.mean((logits) ** 2)
+
+        grads_base = nnx.grad(loss_base)(base_model)
+        logits_base = base_model(
+            decoder_input_tokens=ids,
+            decoder_positions=decoder_positions,
+            decoder_segment_ids=decoder_segment_ids,
+            enable_dropout=False,
+        )
+        self.__class__._cached_base_results_nnx = (grads_base, logits_base)
+
+      grads_base, logits = self.__class__._cached_base_results_nnx
+
+      def loss_quant(model):
+        logits_q = model(
+            decoder_input_tokens=ids,
+            decoder_positions=decoder_positions,
+            decoder_segment_ids=decoder_segment_ids,
+            enable_dropout=False,
+        )
+        return jnp.mean((logits_q) ** 2)
+
+      grads_quant = nnx.grad(loss_quant)(qt_model)
+      quant_logits = qt_model(
+          decoder_input_tokens=ids,
+          decoder_positions=decoder_positions,
+          decoder_segment_ids=decoder_segment_ids,
+          enable_dropout=False,
+      )
+
+      print("relative error in logits:" f" {jnp.abs(quant_logits - logits).mean() / jnp.abs(logits).mean()}")
+      assert jnp.abs(quant_logits - logits).mean() / jnp.abs(logits).mean() < logits_tolerance
+
+      # nnx.grad returns a State object which is a mapping of paths to gradients.
+      # Flatten them to check for tolerance.
+      from flax.nnx import traversals
+      grads_base_flat = traversals.flatten_mapping(grads_base)
+      grads_quant_flat = traversals.flatten_mapping(grads_quant)
+      
+      # Filter for param collections to compare only parameters and not stats/buffers if any
+      # Note: NNX grads structure might contain variables like 'kernel', 'bias'.
+      # For simplicity we compare all matching keys.
+      def flatten_and_filter(grads_flat):
+        return {k: v for k, v in grads_flat.items() if hasattr(v, "shape") and "quant_stats" not in str(k)}
+
+      gb_f = flatten_and_filter(grads_base_flat)
+      gq_f = flatten_and_filter(grads_quant_flat)
+
+      for k in gb_f:
+        if k in gq_f:
+          diff = jnp.abs(gb_f[k] - gq_f[k]).mean() / (jnp.abs(gb_f[k]).mean() + 1e-8)
+          if diff > grad_tolerance:
+            print(f"Gradient mismatch for {k}: rel_error = {diff}")
+            assert diff <= grad_tolerance
+    else:
+      qt_model = model_creation_utils.create_model(cfg, self.mesh)
+      if not hasattr(self.__class__, "_cached_base_results"):
+        model = model_creation_utils.create_model(self.cfg, self.mesh)
+        var = model.init(
+            {"params": self.rng, "aqt": self.rng, "dropout": self.rng},
+            ids,
+            decoder_positions,
+            decoder_segment_ids,
+            enable_dropout=False,
+            mutable=True,
+        )
+
+        def loss_base_linen(all_vars, inputs):
+          logits_b, _ = model.apply(
+              all_vars,
+              *inputs,
+              enable_dropout=False,
+              rngs={"params": self.rng},
+              mutable=True,
+          )
+          return jnp.mean((logits_b) ** 2)
+
+        grads_base_linen = jax.grad(loss_base_linen)(var, (ids, decoder_positions, decoder_segment_ids))
+        logits_b, _ = model.apply(
+            var,
+            ids,
+            decoder_positions,
+            decoder_segment_ids,
+            enable_dropout=False,
+            rngs={"params": self.rng},
+            mutable=True,
+        )
+        self.__class__._cached_base_results = (grads_base_linen, logits_b)
+
+      grads_base_linen, logits = self.__class__._cached_base_results
+
+      quantized_vars = qt_model.init(
           {"params": self.rng, "aqt": self.rng, "dropout": self.rng},
           ids,
           decoder_positions,
@@ -406,71 +506,37 @@ def quantization_config(self, quant, logits_tolerance=2e-1, grad_tolerance=5e-1)
           mutable=True,
       )
 
-      def loss_base(all_vars, inputs):
-        logits, _ = model.apply(
+      def loss_quant_linen(all_vars, inputs):
+        logits_q, _ = qt_model.apply(
             all_vars,
             *inputs,
             enable_dropout=False,
             rngs={"params": self.rng},
             mutable=True,
         )
-        return jnp.mean((logits) ** 2)
+        return jnp.mean((logits_q) ** 2)
+
+      grads_quant_linen = jax.grad(loss_quant_linen)(quantized_vars, (ids, decoder_positions, decoder_segment_ids))
 
-      grads_base = jax.grad(loss_base)(var, (ids, decoder_positions, decoder_segment_ids))
-      logits, _ = model.apply(
-          var,
+      quant_logits, _ = qt_model.apply(
+          quantized_vars,
           ids,
           decoder_positions,
           decoder_segment_ids,
           enable_dropout=False,
           rngs={"params": self.rng},
           mutable=True,
       )
-      self.__class__._cached_base_results = (grads_base, logits)
-
-    grads_base, logits = self.__class__._cached_base_results
-
-    quantized_vars = qt_model.init(
-        {"params": self.rng, "aqt": self.rng, "dropout": self.rng},
-        ids,
-        decoder_positions,
-        decoder_segment_ids,
-        enable_dropout=False,
-        mutable=True,
-    )
-
-    def loss_quant(all_vars, inputs):
-      logits, _ = qt_model.apply(
-          all_vars,
-          *inputs,
-          enable_dropout=False,
-          rngs={"params": self.rng},
-          mutable=True,
+      print("relative error in logits:" f" {jnp.abs(quant_logits - logits).mean() / jnp.abs(logits).mean()}")
+      assert jnp.abs(quant_logits - logits).mean() / jnp.abs(logits).mean() < logits_tolerance
+      self.print_grad_diff(grads_base_linen["params"], grads_quant_linen["params"])
+      self.assertTrue(
+          self.pytree_allclose(
+              grads_base_linen["params"],
+              grads_quant_linen["params"],
+              tolerance=grad_tolerance,
+          )
       )
-      return jnp.mean((logits) ** 2)
-
-    # Compute gradients w.r.t. both models
-    grads_quant = jax.grad(loss_quant)(quantized_vars, (ids, decoder_positions, decoder_segment_ids))
-
-    quant_logits, _ = qt_model.apply(
-        quantized_vars,
-        ids,
-        decoder_positions,
-        decoder_segment_ids,
-        enable_dropout=False,
-        rngs={"params": self.rng},
-        mutable=True,
-    )
-    print("relative error in logits:" f" {jnp.abs(quant_logits - logits).mean() / jnp.abs(logits).mean()}")
-    assert jnp.abs(quant_logits - logits).mean() / jnp.abs(logits).mean() < logits_tolerance
-    self.print_grad_diff(grads_base["params"], grads_quant["params"])
-    self.assertTrue(
-        self.pytree_allclose(
-            grads_base["params"],
-            grads_quant["params"],
-            tolerance=grad_tolerance,
-        )
-    )
 
   @pytest.mark.tpu_only
   def test_int8_quantization(self):
@@ -489,6 +555,11 @@ def test_fp8_full_quantization(self):
   def test_fp8_gpu_quantization(self):
     self.quantization_config("fp8_gpu", grad_tolerance=1.5)
 
+  # @pytest.mark.gpu_only
+  @pytest.mark.external_serving
+  def test_fp8_gpu_quantization(self):
+    self.quantization_config("fp8_gpu", grad_tolerance=1.5, enable_nnx=True, pure_nnx_decoder=True, pure_nnx=True)
+
   @pytest.mark.gpu_only
   @pytest.mark.external_serving
   def test_fp8_nanoo_quantization(self):