Skip to content

Commit 02122fd

Browse files
committed
NNX: flip pure_nnx/enable_nnx/pure_nnx_decoder defaults to True
PR6-PR10 promoted every routed-to-Linen feature to NNX-native. This PR flips the three defaults in base.yml so NNX is the production path, pins Linen-coupled tests so the flip doesn't silently swap their backend, and bundles the NNX-only fixes that surface once pure_nnx=True (DiLoCo merge/checkpoint, Zero-1 input shardings on flat nnx.State, MTP sown-Variable handling, generate_param_only_checkpoint NNX flow, maxengine Linen-parity removal). NNX pipeline parallelism deferred to PR11.5; train_compile fails fast under pure_nnx=True with pipeline configured.
1 parent e92ca84 commit 02122fd

31 files changed

Lines changed: 665 additions & 495 deletions

src/maxtext/checkpoint_conversion/to_maxtext.py

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -319,22 +319,20 @@ def get_maxtext_model_info(config):
319319
# Get abstract model structure (name, shape) without materializing the weights to save memory
320320
abstract_params_tree = maxtext_utils.get_abstract_param(maxtext_model_flax, config)["params"]
321321

322-
abstract_params_flat, _ = jax.tree_util.tree_flatten_with_path(abstract_params_tree)
323-
# Standardize abstract tree for later unflattening
324-
abstract_params_tree = jax.tree.map(
325-
lambda _: 0,
326-
abstract_params_tree,
327-
is_leaf=lambda x: isinstance(x, nn.LogicallyPartitioned),
322+
abstract_params_flat, abstract_params_treedef = jax.tree_util.tree_flatten_with_path(
323+
abstract_params_tree, is_leaf=lambda x: isinstance(x, nn.LogicallyPartitioned)
328324
)
329-
abstract_params_treedef = jax.tree_util.tree_structure(abstract_params_tree)
330325

331326
max_logging.log("MaxText abstract model and state initialized.")
332327

333328
# preprocess state
334329
maxtext_abstract_dict = {}
335330
for mt_target_idx, (path_tuple, abstract_leaf_value) in enumerate(abstract_params_flat):
336331
mt_param_key = "params-" + "-".join(param_key_parts_from_path(path_tuple))
337-
mt_target_shape = abstract_leaf_value.shape
332+
if isinstance(abstract_leaf_value, nn.LogicallyPartitioned):
333+
mt_target_shape = abstract_leaf_value.value.shape
334+
else:
335+
mt_target_shape = abstract_leaf_value.shape
338336
maxtext_abstract_dict[mt_param_key] = (mt_target_idx, mt_target_shape)
339337

340338
return maxtext_abstract_dict, abstract_params_treedef

src/maxtext/checkpoint_conversion/utils/utils.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -950,7 +950,7 @@ def extract_nnx_weights(weights_dict: dict) -> dict[str, np.ndarray]:
950950
for path_tuple, leaf_value in leaves_with_paths:
951951
path_keys = param_key_parts_from_path(path_tuple)
952952
# Skip NNX RNG state variables (not model weights)
953-
if "to_nnx__rngs" in path_keys or any(k.endswith("_rngs") for k in path_keys):
953+
if "to_nnx__rngs" in path_keys or any(k == "rngs" or k.endswith("_rngs") for k in path_keys):
954954
continue
955955
maxtext_param_key = "params-" + "-".join(path_keys)
956956
if not isinstance(leaf_value, (jax.Array, np.ndarray)):

src/maxtext/common/checkpointing.py

Lines changed: 22 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
# See the License for the specific language governing permissions and
1313
# limitations under the License.
1414

15+
1516
"""Create an Orbax CheckpointManager with specified (Async or not) Checkpointer."""
1617

1718
import time
@@ -355,11 +356,16 @@ def combine_sharding(sds, shardings):
355356
use_ocdbt=use_ocdbt,
356357
use_zarr3=use_zarr3,
357358
)
359+
# NNX checkpoints are saved as pure dicts (see maybe_save_checkpoint); the
360+
# restore target must match — a boxed nnx.State wouldn't.
361+
restore_target = abstract_unboxed_pre_state
362+
if isinstance(abstract_unboxed_pre_state, nnx.State):
363+
restore_target = abstract_unboxed_pre_state.to_pure_dict()
358364
# Provide sharding info to ensure restoration returns JAX arrays (not NumPy arrays).
359365
restore_args = jax.tree_util.tree_map(
360-
lambda x: ocp.type_handlers.ArrayRestoreArgs(sharding=x.sharding), abstract_unboxed_pre_state
366+
lambda x: ocp.type_handlers.ArrayRestoreArgs(sharding=x.sharding), restore_target
361367
)
362-
return ocp.Checkpointer(handler).restore(p, abstract_unboxed_pre_state, restore_args=restore_args)
368+
return ocp.Checkpointer(handler).restore(p, restore_target, restore_args=restore_args)
363369

364370

365371
def create_orbax_checkpoint_manager(
@@ -838,9 +844,7 @@ def map_to_pspec(data):
838844
(EmergencyCheckpointManager, EmergencyReplicatorCheckpointManager),
839845
):
840846
checkpoint_path = str(checkpoint_manager.directory / str(step) / "items")
841-
with handle_checkpoint_mismatch(
842-
"restore NNX checkpoint", checkpoint_path
843-
):
847+
with handle_checkpoint_mismatch("restore NNX checkpoint", checkpoint_path):
844848
restored_nnx = _load_linen_checkpoint_into_nnx(
845849
checkpoint_path,
846850
abstract_unboxed_pre_state,
@@ -876,9 +880,7 @@ def map_to_pspec(data):
876880
EmergencyReplicatorCheckpointManager,
877881
),
878882
):
879-
restored = checkpoint_manager.restore(
880-
step, args=Composite(state=checkpoint_args)
881-
).state
883+
restored = checkpoint_manager.restore(step, args=Composite(state=checkpoint_args)).state
882884
_assert_no_shaped_dtype_struct(restored)
883885
return (
884886
restored,
@@ -906,9 +908,7 @@ def map_to_pspec(data):
906908
# Case 3: Default/Fallback case.
907909
# This case acts as a wildcard ('_') and matches if none of the preceding cases were met.
908910
case _:
909-
restored = checkpoint_manager.restore(
910-
step, args=Composite(items=checkpoint_args)
911-
)
911+
restored = checkpoint_manager.restore(step, args=Composite(items=checkpoint_args))
912912
_assert_no_shaped_dtype_struct(restored)
913913
return (restored, None)
914914

@@ -918,9 +918,7 @@ def map_to_pspec(data):
918918
else:
919919
params = abstract_unboxed_pre_state.params
920920

921-
with handle_checkpoint_mismatch(
922-
"load parameters", load_parameters_from_path
923-
):
921+
with handle_checkpoint_mismatch("load parameters", load_parameters_from_path):
924922
restored_params = load_params_from_path(
925923
load_parameters_from_path,
926924
params,
@@ -932,9 +930,7 @@ def map_to_pspec(data):
932930
return None, restored_params
933931
elif load_full_state_from_path != "":
934932
max_logging.log(f"Loading full state from path: {load_full_state_from_path}")
935-
with handle_checkpoint_mismatch(
936-
"load full state", load_full_state_from_path
937-
):
933+
with handle_checkpoint_mismatch("load full state", load_full_state_from_path):
938934
restored_state = _load_full_state_from_path(
939935
path=load_full_state_from_path,
940936
abstract_unboxed_pre_state=abstract_unboxed_pre_state,
@@ -1034,7 +1030,8 @@ def maybe_save_checkpoint(checkpoint_manager, state, config, data_iterator, step
10341030
actual_step = int(step)
10351031
else:
10361032
if config.pure_nnx:
1037-
actual_step = int(state.optimizer.step) - 1
1033+
# Under DiLoCo the step lives on the DiLoCoTrainState; otherwise on the optimizer.
1034+
actual_step = int(state.step if config.enable_diloco else state.optimizer.step) - 1
10381035
else:
10391036
# Linen TrainState has .step attribute
10401037
actual_step = int(state.step) - 1
@@ -1045,7 +1042,13 @@ def maybe_save_checkpoint(checkpoint_manager, state, config, data_iterator, step
10451042

10461043
if config.pure_nnx:
10471044
# Save in the Linen on-disk layout so pure_nnx and Linen checkpoints are interchangeable.
1048-
state = train_state_nnx.to_linen_checkpoint_dict(state.to_pure_dict())
1045+
if config.enable_diloco:
1046+
# DiLoCoTrainState: persist the synchronized global model (outer params).
1047+
# The per-replica inner optimizer / outer-momentum state is not checkpointed.
1048+
step_value = state.step.get_value() if hasattr(state.step, "get_value") else state.step
1049+
state = train_state_nnx.to_linen_checkpoint_dict({"model": state.params, "optimizer": {"step": step_value}})
1050+
else:
1051+
state = train_state_nnx.to_linen_checkpoint_dict(state.to_pure_dict())
10491052

10501053
# Determine if a checkpoint save should be forced, overriding the usual `config.checkpoint_period` logic.
10511054
# This occurs if this function was called:

src/maxtext/configs/base.yml

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1184,9 +1184,9 @@ position_id_per_seconds: 25
11841184
subslice_shape: ""
11851185

11861186
# NNX
1187-
enable_nnx: false
1188-
pure_nnx_decoder: false
1189-
pure_nnx: false
1187+
enable_nnx: true
1188+
pure_nnx_decoder: true
1189+
pure_nnx: true
11901190

11911191
################################## Qwen3-Next Specific Configs ##################################
11921192
# Kernel size for the 1D convolution in the Gated Delta Net

src/maxtext/configs/types.py

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2789,6 +2789,15 @@ def calculate_global_batch_sizes(per_device_batch_size, expansion_factor, num_de
27892789

27902790
self.using_pipeline_parallelism = self.ici_pipeline_parallelism > 1 or self.dcn_pipeline_parallelism > 1
27912791
if self.using_pipeline_parallelism:
2792+
if self.pure_nnx:
2793+
# The NNX decoder has no pipeline path yet, so the scanned-layers axis ends up
2794+
# sharded by 'stage' and fails with a cryptic IndivisibleError at state init.
2795+
# Fail fast with a clear message instead. NNX pipeline support is tracked as PR11.5.
2796+
raise NotImplementedError(
2797+
"Pipeline parallelism is not yet supported on the NNX path. Set "
2798+
"ici_pipeline_parallelism=1 and dcn_pipeline_parallelism=1, or use the Linen path "
2799+
"(pure_nnx=False enable_nnx=False)."
2800+
)
27922801
num_stages = int(self.ici_pipeline_parallelism * self.dcn_pipeline_parallelism)
27932802
if self.num_pipeline_repeats == -1:
27942803
num_pipeline_repeats, remainder = divmod(

src/maxtext/layers/nnx_wrappers.py

Lines changed: 10 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@
2626
from flax.core import FrozenDict
2727
from flax.core import meta
2828
from flax.nnx import graph
29-
from flax.nnx import tracers as nnx_tracers
3029
from flax.nnx import variablelib
3130
from flax.nnx.bridge import module as bdg_module
3231
from flax.nnx.module import Module
@@ -180,19 +179,6 @@ def is_linen_initializing() -> bool:
180179
return False
181180

182181

183-
def _refresh_variable_trace_state(module: Module) -> None:
184-
"""Resets stale ``_trace_state`` on Variables to unblock downstream ``nnx.split``.
185-
186-
``nnx.update`` called with JAX tracer values uses ``_unsafe_bypass_check=True``,
187-
which leaves Variables with a stale ``_trace_state`` from the outer Python
188-
context and breaks ``nnx.split`` with "Cannot extract graph node from different
189-
trace level". Resets ``_trace_state`` on any Variable whose ``_can_update`` is False.
190-
"""
191-
for _, v in nnx.graph.iter_graph(module):
192-
if isinstance(v, variablelib.Variable) and not v._can_update: # pylint: disable=protected-access
193-
object.__setattr__(v, "_trace_state", nnx_tracers.TraceState())
194-
195-
196182
class ToNNX(Module):
197183
"""A wrapper to turn any Linen module into an NNX module.
198184
@@ -511,9 +497,17 @@ def maybe_unbox(x):
511497

512498
warnings.warn(f"Found unknown module paths in incoming state:{paths_str}")
513499

500+
# Filter out unknown paths so we don't try to assign them to static attributes
501+
filtered_state_flat = {k: v for k, v in new_state_flat.items() if k not in unknown_state_flat}
502+
new_state = nnx.State(nnx.traversals.unflatten_mapping(filtered_state_flat))
503+
504+
# Rebind the module to the current trace via split / update / merge.
505+
# nnx.update directly on the live module can leave stale tracers.
506+
graphdef, full_state = nnx.split(module)
507+
nnx.update(full_state, new_state)
508+
module = nnx.merge(graphdef, full_state)
509+
514510
_fix_for_qwix_quantization(module)
515-
nnx.update(module, new_state)
516-
_refresh_variable_trace_state(module)
517511
method_fn = _get_module_method(module, nnx_method)
518512
out = method_fn(module, *args, **kwargs)
519513
self._update_variables(module)

src/maxtext/trainers/diloco/diloco.py

Lines changed: 43 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -28,13 +28,13 @@
2828
import drjax
2929
from flax import nnx
3030
from flax import struct
31-
from flax.training import train_state
3231
import jax
3332
import jax.numpy as jnp
3433
from jaxtyping import Array, Int32, Key, PyTree, UInt32
3534
import optax
3635

3736
from maxtext.configs import pyconfig
37+
from maxtext.common.train_state_nnx import TrainStateNNX
3838

3939
Batch = Any
4040
Params = PyTree
@@ -157,8 +157,10 @@ def add_diloco_dim(x):
157157
# For NNX, model params (Param variables only) live under abstract_state.model;
158158
# for Linen under abstract_state.params.
159159
if config.pure_nnx:
160-
model_params = abstract_state.model.filter(nnx.Param)
161-
model_params_sharding = state_mesh_shardings.model.filter(nnx.Param)
160+
_, model_params, _ = nnx.split(abstract_state.model, nnx.Param, ...)
161+
model_params = model_params.to_pure_dict()
162+
_, model_params_sharding, _ = nnx.split(state_mesh_shardings.model, nnx.Param, ...)
163+
model_params_sharding = model_params_sharding.to_pure_dict()
162164
else:
163165
model_params = abstract_state.params
164166
model_params_sharding = state_mesh_shardings.params
@@ -216,7 +218,11 @@ def init_diloco_state() -> tuple[DiLoCoTrainState, PyTree]:
216218
# Outer state retains a single copy of the model parameters and optimizer state.
217219
# For NNX, model params (Param variables only) live under state.model;
218220
# for Linen under state.params.
219-
outer_params = state.model.filter(nnx.Param) if config.pure_nnx else state.params
221+
if config.pure_nnx:
222+
_, outer_params, _ = nnx.split(state.model, nnx.Param, ...)
223+
outer_params = outer_params.to_pure_dict()
224+
else:
225+
outer_params = state.params
220226
outer_opt_state = outer_optimizer.init(outer_params)
221227
outer_opt_state_sharding = jax.tree_util.tree_map(lambda x: x.sharding, outer_opt_state)
222228
# For NNX, the step counter lives at state.optimizer.step; for Linen at state.step.
@@ -258,9 +264,11 @@ def synchronize(state):
258264
# state (since last synchronization).
259265
broadcast_outer_params = drjax.broadcast(state.params, mesh=mesh)
260266
# For NNX, model Param vars live under inner_state.model; for Linen under inner_state.params.
261-
inner_model_params = (
262-
nnx.filter_state(state.inner_state.model, nnx.Param) if config.pure_nnx else state.inner_state.params
263-
)
267+
if config.pure_nnx:
268+
_, inner_model_params, _ = nnx.split(state.inner_state.model, nnx.Param, ...)
269+
inner_model_params = inner_model_params.to_pure_dict()
270+
else:
271+
inner_model_params = state.inner_state.params
264272
model_delta = jax.tree.map(lambda x, y: y - x, inner_model_params, broadcast_outer_params)
265273
# Treat the average delta as the outer optimizer's gradient and apply to
266274
# the global (outer) model params.
@@ -273,15 +281,34 @@ def synchronize(state):
273281
if config.pure_nnx:
274282
# For NNX: merge new Param vars back with the non-Param model vars (e.g. RNG state).
275283
def replace_nnx_model_params(s, new_params):
276-
non_param_model = nnx.filter_state(s.model, nnx.Not(nnx.Param))
277-
new_model = nnx.merge_state(non_param_model, new_params)
278-
# Assign via __setitem__ so nested States are stored as plain dicts (matching
279-
# nnx.state()'s pytree structure). The dict-literal constructor keeps them as
280-
# State objects, which makes jax.lax.cond see mismatched pytree structures.
281-
result = type(s)({})
282-
result["model"] = new_model
283-
result["optimizer"] = s["optimizer"]
284-
return result
284+
s_model = s["model"] if hasattr(s, "keys") else s.model
285+
s_opt = s["optimizer"] if hasattr(s, "keys") else s.optimizer
286+
287+
graphdef, _, non_param_state = nnx.split(s_model, nnx.Param, ...)
288+
new_model = nnx.merge(graphdef, new_params, non_param_state)
289+
290+
if type(s_model).__name__ == "State":
291+
new_model = nnx.state(new_model)
292+
elif isinstance(s_model, dict):
293+
new_model = nnx.to_pure_dict(new_model)
294+
295+
if hasattr(s, "keys"):
296+
# Replace "model" leaves by path, keeping s's treedef. Picking by position
297+
# (leaves[N:]) breaks if a key sorts before "model"; reconstructing via
298+
# type(s)({...}) breaks the lax.cond match — nnx.State recursive-wraps.
299+
leaves_with_paths, treedef = jax.tree_util.tree_flatten_with_path(s)
300+
new_model_iter = iter(jax.tree_util.tree_leaves(new_model))
301+
302+
def _is_model_leaf(path):
303+
if not path:
304+
return False
305+
k = path[0]
306+
return getattr(k, "key", None) == "model" or getattr(k, "name", None) == "model"
307+
308+
new_leaves = [next(new_model_iter) if _is_model_leaf(p) else leaf for p, leaf in leaves_with_paths]
309+
return jax.tree_util.tree_unflatten(treedef, new_leaves)
310+
else:
311+
return TrainStateNNX(new_model, s_opt)
285312

286313
new_inner_state = drjax.map_fn(
287314
lambda s: replace_nnx_model_params(s, new_outer_params),

src/maxtext/trainers/post_train/sft/train_sft_native.py

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import tensorflow as tf
2626
import jax
2727

28+
from flax import nnx
2829
from flax.linen import partitioning as nn_partitioning
2930

3031
from maxtext.configs import pyconfig
@@ -75,13 +76,24 @@ def train_loop(config, recorder, state=None):
7576

7677
params_shardings, state_mesh_shardings = sharding.maybe_update_params_sharding_with_opt(config, state_mesh_shardings)
7778

79+
# NNX jits over the GraphDef + a flat nnx.State, so split the TrainStateNNX
80+
# here (mirrors trainers/pre_train/train.py). Linen jits over the module.
81+
if config.pure_nnx:
82+
jit_model, state = nnx.split(state)
83+
else:
84+
jit_model = model
85+
7886
p_train_step, p_eval_step = train_utils.jit_train_and_eval_step(
79-
config, model, mesh, state, state_mesh_shardings, train_step, eval_step, eval_data_iterator, params_shardings
87+
config, jit_model, mesh, state, state_mesh_shardings, train_step, eval_step, eval_data_iterator, params_shardings
8088
)
8189

90+
# Only the Linen step takes a dropout rng; pass it only there so the args
91+
# match the jitted in_shardings (see get_functional_train_with_signature).
92+
rng_args = () if config.pure_nnx else (init_rng,)
93+
8294
with jax.set_mesh(mesh), nn_partitioning.axis_rules(config.logical_axis_rules):
8395
shaped_batch = maxtext_utils.get_shaped_batch(config)
84-
compiled = p_train_step.lower(state, shaped_batch, init_rng).compile()
96+
compiled = p_train_step.lower(state, shaped_batch, *rng_args).compile()
8597
compiled_stats = compiled.memory_analysis()
8698
max_utils.print_compiled_memory_stats(compiled_stats)
8799

@@ -91,7 +103,11 @@ def train_loop(config, recorder, state=None):
91103
metric_logger = MetricLogger(config=config, learning_rate_schedule=learning_rate_schedule)
92104

93105
# Write train config params, num model params, and XLA flags to tensorboard
94-
metric_logger.write_setup_info_to_tensorboard(state.params)
106+
if config.pure_nnx:
107+
_, setup_params, _ = nnx.split(state.model, nnx.Param, ...)
108+
else:
109+
setup_params = state.params
110+
metric_logger.write_setup_info_to_tensorboard(setup_params)
95111

96112
_job_completed_gracefully = False
97113
try:
@@ -103,9 +119,10 @@ def train_loop(config, recorder, state=None):
103119
example_batch = data_loader.load_next_batch()
104120
# pylint: disable=not-callable
105121
nextrng = jax.jit(jax.random.fold_in)(init_rng, step)
122+
step_rng_args = () if config.pure_nnx else (nextrng,)
106123
with maybe_record_goodput(recorder, GoodputEvent.STEP, step):
107124
with jax.set_mesh(mesh), nn_partitioning.axis_rules(config.logical_axis_rules):
108-
state, metrics = p_train_step(state, example_batch, nextrng)
125+
state, metrics = p_train_step(state, example_batch, *step_rng_args)
109126

110127
step_time_delta = datetime.datetime.now() - last_step_completion
111128

@@ -134,7 +151,7 @@ def train_loop(config, recorder, state=None):
134151
if config.eval_steps > 0 and eval_step_count >= config.eval_steps:
135152
break
136153
with jax.set_mesh(mesh), nn_partitioning.axis_rules(config.logical_axis_rules):
137-
eval_metrics = p_eval_step(state, eval_batch, nextrng)
154+
eval_metrics = p_eval_step(state, eval_batch, *step_rng_args)
138155
eval_step_time_delta = datetime.datetime.now() - last_eval_step_completion
139156
last_eval_step_completion = datetime.datetime.now()
140157
metric_logger.buffer_and_write_metrics(

0 commit comments

Comments
 (0)