Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions src/maxtext/checkpoint_conversion/to_maxtext.py
Original file line number Diff line number Diff line change
Expand Up @@ -319,22 +319,20 @@ def get_maxtext_model_info(config):
# Get abstract model structure (name, shape) without materializing the weights to save memory
abstract_params_tree = maxtext_utils.get_abstract_param(maxtext_model_flax, config)["params"]

abstract_params_flat, _ = jax.tree_util.tree_flatten_with_path(abstract_params_tree)
# Standardize abstract tree for later unflattening
abstract_params_tree = jax.tree.map(
lambda _: 0,
abstract_params_tree,
is_leaf=lambda x: isinstance(x, nn.LogicallyPartitioned),
abstract_params_flat, abstract_params_treedef = jax.tree_util.tree_flatten_with_path(
abstract_params_tree, is_leaf=lambda x: isinstance(x, nn.LogicallyPartitioned)
)
abstract_params_treedef = jax.tree_util.tree_structure(abstract_params_tree)

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

# preprocess state
maxtext_abstract_dict = {}
for mt_target_idx, (path_tuple, abstract_leaf_value) in enumerate(abstract_params_flat):
mt_param_key = "params-" + "-".join(param_key_parts_from_path(path_tuple))
mt_target_shape = abstract_leaf_value.shape
if isinstance(abstract_leaf_value, nn.LogicallyPartitioned):
mt_target_shape = abstract_leaf_value.value.shape
else:
mt_target_shape = abstract_leaf_value.shape
maxtext_abstract_dict[mt_param_key] = (mt_target_idx, mt_target_shape)

return maxtext_abstract_dict, abstract_params_treedef
Expand Down
2 changes: 1 addition & 1 deletion src/maxtext/checkpoint_conversion/utils/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -967,7 +967,7 @@ def extract_nnx_weights(weights_dict: dict) -> dict[str, np.ndarray]:
for path_tuple, leaf_value in leaves_with_paths:
path_keys = param_key_parts_from_path(path_tuple)
# Skip NNX RNG state variables (not model weights)
if "to_nnx__rngs" in path_keys or any(k.endswith("_rngs") for k in path_keys):
if "to_nnx__rngs" in path_keys or any(k == "rngs" or k.endswith("_rngs") for k in path_keys):
continue
maxtext_param_key = "params-" + "-".join(path_keys)
if not isinstance(leaf_value, (jax.Array, np.ndarray)):
Expand Down
21 changes: 17 additions & 4 deletions src/maxtext/common/checkpointing.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.


"""Create an Orbax CheckpointManager with specified (Async or not) Checkpointer."""

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


def create_orbax_checkpoint_manager(
Expand Down Expand Up @@ -1015,7 +1021,8 @@ def maybe_save_checkpoint(checkpoint_manager, state, config, data_iterator, step
actual_step = int(step)
else:
if config.pure_nnx:
actual_step = int(state.optimizer.step) - 1
# Under DiLoCo the step lives on the DiLoCoTrainState; otherwise on the optimizer.
actual_step = int(state.step if config.enable_diloco else state.optimizer.step) - 1
else:
# Linen TrainState has .step attribute
actual_step = int(state.step) - 1
Expand All @@ -1026,7 +1033,13 @@ def maybe_save_checkpoint(checkpoint_manager, state, config, data_iterator, step

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

# Determine if a checkpoint save should be forced, overriding the usual `config.checkpoint_period` logic.
# This occurs if this function was called:
Expand Down
6 changes: 3 additions & 3 deletions src/maxtext/configs/base.yml
Original file line number Diff line number Diff line change
Expand Up @@ -1235,9 +1235,9 @@ position_id_per_seconds: 25
subslice_shape: ""

# NNX
enable_nnx: false
pure_nnx_decoder: false
pure_nnx: false
enable_nnx: true
pure_nnx_decoder: true
pure_nnx: true

################################## Qwen3-Next Specific Configs ##################################
# Kernel size for the 1D convolution in the Gated Delta Net
Expand Down
2 changes: 2 additions & 0 deletions src/maxtext/layers/nnx_decoders.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
from maxtext.layers.quantizations import AqtQuantization as Quant
from maxtext.models import (
deepseek,
deepseek4,
deepseek_batchsplit,
deepseek_batchsplit_fp8,
gemma,
Expand Down Expand Up @@ -1121,6 +1122,7 @@ def get_deepseek():
DecoderBlockType.SIMPLE: [simple_layer.SimpleDecoderLayer],
DecoderBlockType.SIMPLE_MLP: [simple_layer.SimpleMlpDecoderLayer],
DecoderBlockType.DEEPSEEK: get_deepseek(),
DecoderBlockType.DEEPSEEK4: get_scannable(deepseek4.DeepSeek4DecoderLayer, deepseek4.DeepSeek4ScannableBlock),
DecoderBlockType.GPT_OSS: get_scannable(gpt_oss.GptOssDecoderLayer, gpt_oss.GptOssScannableBlock),
DecoderBlockType.QWEN3_NEXT: get_scannable(qwen3.Qwen3NextDecoderLayer, qwen3.Qwen3NextScannableBlock),
DecoderBlockType.QWEN3_5: get_scannable(qwen3_5.Qwen3_5DecoderLayer, qwen3_5.Qwen3_5ScannableBlock),
Expand Down
26 changes: 10 additions & 16 deletions src/maxtext/layers/nnx_wrappers.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
from flax.core import FrozenDict
from flax.core import meta
from flax.nnx import graph
from flax.nnx import tracers as nnx_tracers
from flax.nnx import variablelib
from flax.nnx.bridge import module as bdg_module
from flax.nnx.module import Module
Expand Down Expand Up @@ -183,19 +182,6 @@ def is_linen_initializing() -> bool:
return False


def _refresh_variable_trace_state(module: Module) -> None:
"""Resets stale ``_trace_state`` on Variables to unblock downstream ``nnx.split``.

``nnx.update`` called with JAX tracer values uses ``_unsafe_bypass_check=True``,
which leaves Variables with a stale ``_trace_state`` from the outer Python
context and breaks ``nnx.split`` with "Cannot extract graph node from different
trace level". Resets ``_trace_state`` on any Variable whose ``_can_update`` is False.
"""
for _, v in nnx.graph.iter_graph(module):
if isinstance(v, variablelib.Variable) and not v._can_update: # pylint: disable=protected-access
object.__setattr__(v, "_trace_state", nnx_tracers.TraceState())


class ToNNX(Module):
"""A wrapper to turn any Linen module into an NNX module.

Expand Down Expand Up @@ -542,9 +528,17 @@ def maybe_unbox(x):
f"Found unknown module paths in incoming state:{paths_str}. Intermediate modules have been reconstructed."
)

# Filter out unknown paths so we don't try to assign them to static attributes
filtered_state_flat = {k: v for k, v in new_state_flat.items() if k not in unknown_state_flat}
new_state = nnx.State(nnx.traversals.unflatten_mapping(filtered_state_flat))

# Rebind the module to the current trace via split / update / merge.
# nnx.update directly on the live module can leave stale tracers.
graphdef, full_state = nnx.split(module)
nnx.update(full_state, new_state)
module = nnx.merge(graphdef, full_state)

_fix_for_qwix_quantization(module)
nnx.update(module, new_state)
_refresh_variable_trace_state(module)
method_fn = _get_module_method(module, nnx_method)
out = method_fn(module, *args, **kwargs)
self._update_variables(module)
Expand Down
3 changes: 1 addition & 2 deletions src/maxtext/layers/quantizations.py
Original file line number Diff line number Diff line change
Expand Up @@ -1031,8 +1031,7 @@ def generate_quantizer_set(
fp8_recipe: recipe.Recipe | None = default_recipe,
n_groups: int | None = None,
):
"""Generates a set of quantizers for TransformerEngine."""

"""Route quantizer-set generation through the OVERWRITE_WITH_GRADIENT collection."""
OVERWRITE_WITH_GRADIENT = "_overwrite_with_gradient"
return super().generate_quantizer_set( # pytype: disable=wrong-keyword-args
postfix=postfix,
Expand Down
59 changes: 43 additions & 16 deletions src/maxtext/trainers/diloco/diloco.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,13 @@
import drjax
from flax import nnx
from flax import struct
from flax.training import train_state
import jax
import jax.numpy as jnp
from jaxtyping import Array, Int32, Key, PyTree, UInt32
import optax

from maxtext.configs import pyconfig
from maxtext.common.train_state_nnx import TrainStateNNX

Batch = Any
Params = PyTree
Expand Down Expand Up @@ -157,8 +157,10 @@ def add_diloco_dim(x):
# For NNX, model params (Param variables only) live under abstract_state.model;
# for Linen under abstract_state.params.
if config.pure_nnx:
model_params = abstract_state.model.filter(nnx.Param)
model_params_sharding = state_mesh_shardings.model.filter(nnx.Param)
_, model_params, _ = nnx.split(abstract_state.model, nnx.Param, ...)
model_params = model_params.to_pure_dict()
_, model_params_sharding, _ = nnx.split(state_mesh_shardings.model, nnx.Param, ...)
model_params_sharding = model_params_sharding.to_pure_dict()
else:
model_params = abstract_state.params
model_params_sharding = state_mesh_shardings.params
Expand Down Expand Up @@ -216,7 +218,11 @@ def init_diloco_state() -> tuple[DiLoCoTrainState, PyTree]:
# Outer state retains a single copy of the model parameters and optimizer state.
# For NNX, model params (Param variables only) live under state.model;
# for Linen under state.params.
outer_params = state.model.filter(nnx.Param) if config.pure_nnx else state.params
if config.pure_nnx:
_, outer_params, _ = nnx.split(state.model, nnx.Param, ...)
outer_params = outer_params.to_pure_dict()
else:
outer_params = state.params
outer_opt_state = outer_optimizer.init(outer_params)
outer_opt_state_sharding = jax.tree_util.tree_map(lambda x: x.sharding, outer_opt_state)
# For NNX, the step counter lives at state.optimizer.step; for Linen at state.step.
Expand Down Expand Up @@ -258,9 +264,11 @@ def synchronize(state):
# state (since last synchronization).
broadcast_outer_params = drjax.broadcast(state.params, mesh=mesh)
# For NNX, model Param vars live under inner_state.model; for Linen under inner_state.params.
inner_model_params = (
nnx.filter_state(state.inner_state.model, nnx.Param) if config.pure_nnx else state.inner_state.params
)
if config.pure_nnx:
_, inner_model_params, _ = nnx.split(state.inner_state.model, nnx.Param, ...)
inner_model_params = inner_model_params.to_pure_dict()
else:
inner_model_params = state.inner_state.params
model_delta = jax.tree.map(lambda x, y: y - x, inner_model_params, broadcast_outer_params)
# Treat the average delta as the outer optimizer's gradient and apply to
# the global (outer) model params.
Expand All @@ -273,15 +281,34 @@ def synchronize(state):
if config.pure_nnx:
# For NNX: merge new Param vars back with the non-Param model vars (e.g. RNG state).
def replace_nnx_model_params(s, new_params):
non_param_model = nnx.filter_state(s.model, nnx.Not(nnx.Param))
new_model = nnx.merge_state(non_param_model, new_params)
# Assign via __setitem__ so nested States are stored as plain dicts (matching
# nnx.state()'s pytree structure). The dict-literal constructor keeps them as
# State objects, which makes jax.lax.cond see mismatched pytree structures.
result = type(s)({})
result["model"] = new_model
result["optimizer"] = s["optimizer"]
return result
s_model = s["model"] if hasattr(s, "keys") else s.model
s_opt = s["optimizer"] if hasattr(s, "keys") else s.optimizer

graphdef, _, non_param_state = nnx.split(s_model, nnx.Param, ...)
new_model = nnx.merge(graphdef, new_params, non_param_state)

if type(s_model).__name__ == "State":
new_model = nnx.state(new_model)
elif isinstance(s_model, dict):
new_model = nnx.to_pure_dict(new_model)

if hasattr(s, "keys"):
# Replace "model" leaves by path, keeping s's treedef. Picking by position
# (leaves[N:]) breaks if a key sorts before "model"; reconstructing via
# type(s)({...}) breaks the lax.cond match — nnx.State recursive-wraps.
leaves_with_paths, treedef = jax.tree_util.tree_flatten_with_path(s)
new_model_iter = iter(jax.tree_util.tree_leaves(new_model))

def _is_model_leaf(path):
if not path:
return False
k = path[0]
return getattr(k, "key", None) == "model" or getattr(k, "name", None) == "model"

new_leaves = [next(new_model_iter) if _is_model_leaf(p) else leaf for p, leaf in leaves_with_paths]
return jax.tree_util.tree_unflatten(treedef, new_leaves)
Comment thread
ecnal-cienet marked this conversation as resolved.
else:
return TrainStateNNX(new_model, s_opt)

new_inner_state = drjax.map_fn(
lambda s: replace_nnx_model_params(s, new_outer_params),
Expand Down
20 changes: 17 additions & 3 deletions src/maxtext/trainers/pre_train/train.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,8 @@
def get_first_step(model, state):
if isinstance(model, nn.Module):
return int(state.step)
if hasattr(state, "inner_state"): # DiLoCoTrainState (NNX DiLoCo): step is the optimizer step var
return int(state.step.get_value())
return int(state.optimizer.step.get_value())


Expand Down Expand Up @@ -681,7 +683,8 @@ def training_loop_iteration(

with jax.profiler.StepTraceAnnotation("train", step_num=step):
example_batch = data_loader.load_next_batch(rampup_manager=rampup_manager)
if isinstance(model, nn.Module):
# DiLoCo's inner step takes the rng like the Linen step does.
if isinstance(model, nn.Module) or config.enable_diloco:
# pylint: disable=not-callable
step_rng_args = (jax.jit(jax.random.fold_in)(init_rng, step),)
else:
Expand Down Expand Up @@ -768,10 +771,18 @@ def train_loop(config, recorder, state=None):

if isinstance(model, nn.Module):
jit_model = model
elif config.enable_diloco:
# state is the DiLoCoTrainState; `model` is already the TrainStateNNX graphdef the inner step needs.
jit_model = model
else:
jit_model, state = nnx.split(state)

params_shardings, state_mesh_shardings = sharding.maybe_update_params_sharding_with_opt(config, state_mesh_shardings)
if config.pure_nnx and config.enable_diloco:
# DiLoCoTrainState.params already holds the param shardings the inner step needs;
# the Zero-1 opt overlay doesn't apply through the diloco wrapper.
params_shardings = state_mesh_shardings.params
else:
params_shardings, state_mesh_shardings = sharding.maybe_update_params_sharding_with_opt(config, state_mesh_shardings)

p_train_step, p_eval_step = train_utils.jit_train_and_eval_step(
config,
Expand All @@ -793,7 +804,8 @@ def train_loop(config, recorder, state=None):
elif config.shard_optimizer_over_data:
# NNX: reshard state so params match the data-sharded in_shardings (Zero-1 layout)
state = jax.device_put(state, state_mesh_shardings)
if isinstance(model, nn.Module):
if isinstance(model, nn.Module) or config.enable_diloco:
# The DiLoCo train step takes (state, batch, rng), like the Linen step.
lower_args = (state, shaped_batch, init_rng)
else:
lower_args = (state, shaped_batch)
Expand All @@ -809,6 +821,8 @@ def train_loop(config, recorder, state=None):
# Write train config params, num model params, and XLA flags to tensorboard
if isinstance(model, nn.Module):
setup_params = state.params
elif config.enable_diloco:
setup_params = state.params # DiLoCoTrainState.params: the outer (global) params
else:
_, setup_params, _ = nnx.split(state.model, nnx.Param, ...)
metric_logger_instance.write_setup_info_to_tensorboard(setup_params)
Expand Down
18 changes: 6 additions & 12 deletions src/maxtext/trainers/pre_train/train_compile.py
Original file line number Diff line number Diff line change
Expand Up @@ -261,12 +261,9 @@ def is_oom(argv: Sequence[str]) -> bool:
# Update params_shardings when shard_optimizer_over_data is enabled (Zero-1)
params_shardings, state_mesh_shardings = sharding.maybe_update_params_sharding_with_opt(config, state_mesh_shardings)

# When ZeRO-1 is enabled, we need to use the original params_shardings for input shardings
# but keep the updated state_mesh_shardings for the optimizer state
if config.shard_optimizer_over_data:
input_state_mesh_shardings = state_mesh_shardings.replace(params=params_shardings)
else:
input_state_mesh_shardings = state_mesh_shardings
input_state_mesh_shardings = sharding.build_zero1_input_state_mesh_shardings(
config, state_mesh_shardings, params_shardings
)

# Get data sharding
data_sharding = sharding.get_input_data_sharding(config, topology_mesh)
Expand Down Expand Up @@ -343,12 +340,9 @@ def main(argv: Sequence[str]) -> None:
# Update params_shardings when shard_optimizer_over_data is enabled (Zero-1)
params_shardings, state_mesh_shardings = sharding.maybe_update_params_sharding_with_opt(config, state_mesh_shardings)

# When ZeRO-1 is enabled, we need to use the original params_shardings for input shardings
# but keep the updated state_mesh_shardings for the optimizer state
if config.shard_optimizer_over_data:
input_state_mesh_shardings = state_mesh_shardings.replace(params=params_shardings)
else:
input_state_mesh_shardings = state_mesh_shardings
input_state_mesh_shardings = sharding.build_zero1_input_state_mesh_shardings(
config, state_mesh_shardings, params_shardings
)

# Get data sharding
data_sharding = sharding.get_input_data_sharding(config, topology_mesh)
Expand Down
Loading
Loading