# Copyright © 2026 DeepGrove AI. import math from dataclasses import dataclass from functools import lru_cache, partial from typing import Any, List, Optional import mlx.core as mx import mlx.nn as nn # The native extension uses the MLX 0.32.0 C++ ABI. _maple_native = None if mx.__version__ == "0.32.0": try: from mlx_lm_maple import _maple_native except ImportError: try: from mlx_lm import _maple_native except ImportError: pass # Absolute imports so this file also works standalone when shipped inside a # checkpoint and loaded via the config's `model_file` (trust_remote_code). from mlx_lm.models.activations import swiglu from mlx_lm.models.base import ( BaseModelArgs, create_attention_mask, scaled_dot_product_attention, ) from mlx_lm.models.cache import KVCache as _KVCache from mlx_lm.models.cache import RotatingKVCache as _RotatingKVCache from mlx_lm.models.rope_utils import initialize_rope from mlx_lm.models.switch_layers import QuantizedSwitchLinear, SwitchLinear # SwiGLU clamp for the MoE experts only (the dense MapleMLP is unclamped); # part of the trained forward pass, not an optional guard. MLP_CLAMP = 7.0 @partial(mx.compile, shapeless=True) def clamped_swiglu(gate, x): # Python floats, not 0-d arrays, so bf16 activations stay bf16. return nn.silu(mx.minimum(gate, MLP_CLAMP)) * mx.clip(x, -MLP_CLAMP, MLP_CLAMP) def _matches(fast, reference): """Compare output shapes, dtypes, and values. Validate input layouts first.""" try: got, want = fast(), reference() mx.eval(got, want) except Exception: return False return len(got) == len(want) and all( g.shape == w.shape and g.dtype == w.dtype and bool(mx.array_equal(g, w)) for g, w in zip(got, want) ) def _exact_result(fast, reference): """Return the fast output after an exact single-array probe, otherwise None.""" try: got = fast() except Exception: return None return got if _matches(lambda: (got,), lambda: (reference(),)) else None class MapleRMSNorm(nn.Module): """RMSNorm with the weight multiply in float32. The reference rounds only the finished product; mx.fast.rms_norm rounds the normalized activation first (~1% per element). Float32 inputs to the same kernel reproduce the reference bit-for-bit. """ def __init__(self, dims: int, eps: float = 1e-6): super().__init__() self.weight = mx.ones((dims,)) self.eps = eps def __call__(self, x: mx.array) -> mx.array: return mx.fast.rms_norm( x.astype(mx.float32), self.weight.astype(mx.float32), self.eps ).astype(x.dtype) @lru_cache(maxsize=None) def _make_add_rms_norm_kernel(eps, aggregate=False): """Residual/RMSNorm with optional ordered, BF16-rounded expert reduction.""" residual = "float v = (float)x[j] + (float)r[j];" store_h = "h_out[j] = vb;" store_hn = "hn_out[j] = (T_)((float)w[j] * (hb[i] * scale));" if aggregate: residual = """ float agg = 0.0f; { #pragma clang fp contract(off) for (uint e = 0; e < 8; ++e) { agg = agg + (float)r[e * N + j] * scores[e]; } } float v = (float)x[j] + (float)(T_)agg; """ guard = "if (j / (N / 8u) == threadgroup_position_in_grid.x)" store_h = f"{guard} {{ {store_h} }}" store_hn = f"{guard} {{ {store_hn} }}" source = """ uint tid = thread_position_in_threadgroup.x; constexpr uint N = DIM; constexpr uint PT = 4u; float hb[PT]; float ss = 0.0f; for (uint i = 0; i < PT; ++i) { uint j = tid * PT + i; RESIDUAL T_ vb = (T_)v; // one rounding, same as a bf16 add STORE_H hb[i] = (float)vb; // norm sees the rounded stream ss += hb[i] * hb[i]; } ss = simd_sum(ss); // Match MLX 0.32.0 rms_single_row: four values per lane, // then the same two-level SIMD reduction and precise reciprocal root. threadgroup float sums[32]; threadgroup float inv_mean[1]; uint sg = tid / 32u; uint lane = tid % 32u; if (sg == 0u) sums[lane] = 0.0f; threadgroup_barrier(mem_flags::mem_threadgroup); if (lane == 0u) sums[sg] = ss; threadgroup_barrier(mem_flags::mem_threadgroup); if (sg == 0u) { float tot = simd_sum(sums[lane]); if (lane == 0u) inv_mean[0] = metal::precise::rsqrt(tot / (float)N + EPS_); } threadgroup_barrier(mem_flags::mem_threadgroup); float scale = inv_mean[0]; for (uint i = 0; i < PT; ++i) { uint j = tid * PT + i; STORE_HN } """ source = ( source.replace("RESIDUAL", residual) .replace("STORE_HN", store_hn) .replace("STORE_H", store_h) .replace("EPS_", f"{eps:.10e}f") ) tag = f"{eps:.3e}".replace(".", "_").replace("-", "m").replace("+", "p") return mx.fast.metal_kernel( name=f"maple_{'aggregate_' if aggregate else ''}add_rms_norm_{tag}", input_names=["x", "r", "scores", "w"] if aggregate else ["x", "r", "w"], output_names=["h_out", "hn_out"], source=source, ) def _add_rms_norm(h, r, w, eps): return _make_add_rms_norm_kernel(eps)( inputs=[h.reshape(-1), r.reshape(-1), w], template=[("T_", h.dtype), ("DIM", h.shape[-1])], grid=(h.shape[-1] // 4, 1, 1), threadgroup=(h.shape[-1] // 4, 1, 1), output_shapes=[h.shape, h.shape], output_dtypes=[h.dtype, h.dtype], ) def _add_rms_norm_ok(dim, dtype, w, eps): if mx.__version__ != "0.32.0" or dim % 128 or dim > 4096 or dtype != w.dtype: return False x = mx.random.normal((1, 1, dim), key=mx.random.key(0)).astype(dtype) r = mx.random.normal((1, 1, dim), key=mx.random.key(1)).astype(dtype) return _matches( lambda: _add_rms_norm(x, r, w, eps), lambda: ( x + r, mx.fast.rms_norm( (x + r).astype(mx.float32), w.astype(mx.float32), eps ).astype(dtype), ), ) @mx.compile def _aggregate_add_rms_norm(h, r, scores, w, eps): return _make_add_rms_norm_kernel(eps, aggregate=True)( inputs=[h.reshape(-1), r.reshape(-1), scores.reshape(-1), w], template=[("T_", h.dtype), ("DIM", h.shape[-1])], grid=(8 * h.shape[-1] // 4, 1, 1), threadgroup=(h.shape[-1] // 4, 1, 1), output_shapes=[h.shape, h.shape], output_dtypes=[h.dtype, h.dtype], ) def _aggregate_add_rms_norm_ok(dim, dtype, w, eps): """Require bitwise agreement with the unfused operations on live weights.""" if mx.__version__ != "0.32.0" or dim % 128 or dim > 4096 or dtype != w.dtype: return False try: for seed in range(4): h = mx.random.normal((1, 1, dim), key=mx.random.key(seed)).astype(dtype) y = mx.random.normal((1, 1, 8, dim), key=mx.random.key(10 + seed)).astype( dtype ) scores = mx.softmax( mx.random.normal((1, 1, 8), key=mx.random.key(20 + seed)) ) got = _aggregate_add_rms_norm(h, y, scores, w, eps) residual = h + aggregate_expert_outputs(y, scores) want = ( residual, mx.fast.rms_norm( residual.astype(mx.float32), w.astype(mx.float32), eps ).astype(dtype), ) mx.eval(got, want) if not all(bool(mx.array_equal(a, b)) for a, b in zip(got, want)): return False except Exception: return False return True # Inlined rather than imported from switch_layers: those helpers are private # (underscore-prefixed), and this file must keep loading against whatever # mlx-lm a user has installed when it ships inside a checkpoint. def _gather_sort(x, indices): *_, M = indices.shape indices = indices.flatten() order = mx.argsort(indices) inv_order = mx.argsort(order) return x.flatten(0, -3)[order // M], indices[order], inv_order def _scatter_unsort(x, inv_order, shape=None): x = x[inv_order] if shape is not None: x = mx.unflatten(x, 0, shape) return x @dataclass class ModelArgs(BaseModelArgs): model_type: str = "maple" hidden_size: int = 2048 intermediate_size: int = 5120 moe_intermediate_size: int = 512 num_hidden_layers: int = 24 num_attention_heads: int = 16 num_key_value_heads: int = 4 head_dim: int = 128 num_experts: int = 256 num_experts_per_tok: int = 8 first_k_dense_replace: int = 0 rms_norm_eps: float = 1e-6 rope_theta: float = 10000.0 rope_scaling: Optional[dict] = None partial_rotary_factor: float = 0.5 max_position_embeddings: int = 140000 vocab_size: int = 151936 sliding_window: int = 512 layer_types: Optional[List[str]] = None use_qk_norm: bool = True use_bias: bool = False tie_word_embeddings: bool = False # FlashHead metadata written by `mlx_lm.ternary --flash-head`. The exact # lm_head is the default; opt in to the approximate fast head with # mlx_lm.load(..., model_config={"use_flash_head": True}). flash_head: Optional[dict] = None use_flash_head: bool = False # Populated from the checkpoint's config; sanitize() reads group_size from # it to expand row-scale (`row_alpha`) ternary tensors. quantization: Optional[dict] = None def __post_init__(self): # Single source of truth for per-layer attention types: attention # (RoPE/NoPE), masks, and caches all read this resolved list. if not self.layer_types: self.layer_types = ["full_attention"] * self.num_hidden_layers _rope_inv_freq_kernel = mx.fast.metal_kernel( name="maple_rope_inv_freq", input_names=["log_base"], output_names=["out"], source=""" uint i = thread_position_in_grid.x; if (i < HALF) out[i] = metal::exp2(-(float(i) / float(HALF)) * log_base[0]); """, ) def _make_qk_norm_rope_kernel(): """Fused per-head RMSNorm + partial RoPE for single-token decode. One dispatch replaces q_norm, k_norm and two rope calls. One simdgroup per head: normalize head_dim values, scale by the head's norm weight, and rotate the first ROPE_DIM dims (non-traditional pairing i, i+R/2) at the given position. NoPE layers pass ROPE_DIM=0. """ source = r""" uint head = thread_position_in_grid.y; uint lane = thread_position_in_grid.x; constexpr int per_lane = HEAD_DIM / 32; const device T_* xh = x + head * HEAD_DIM; device T_* oh = out + head * HEAD_DIM; if (head >= (uint)NQK) { // v heads: passthrough copy for (int i = 0; i < per_lane; ++i) { int j = lane * per_lane + i; oh[j] = xh[j]; } return; } const device T_* wh = head < (uint)NQ ? qw : kw; float ss = 0.0f; for (int i = 0; i < per_lane; ++i) { float v = (float)xh[lane * per_lane + i]; ss += v * v; } ss = simd_sum(ss); float pos = pos_eps[0]; float eps = pos_eps[1]; float scale = metal::precise::rsqrt(ss / HEAD_DIM + eps); for (int i = 0; i < per_lane; ++i) { int j = lane * per_lane + i; float v = (float)(T_)((float)wh[j] * ((float)xh[j] * scale)); if (ROPE_DIM > 0 && j < ROPE_DIM) { constexpr int rhalf = ROPE_DIM > 0 ? ROPE_DIM / 2 : 1; int p = j < rhalf ? j : j - rhalf; float inv = metal::exp2(-(float(p) / float(rhalf)) * pos_eps[2]); float theta = pos * inv; float c = metal::fast::cos(theta); float s = metal::fast::sin(theta); int j2 = j < rhalf ? j + rhalf : j - rhalf; float u = (float)(T_)((float)wh[j2] * ((float)xh[j2] * scale)); v = j < rhalf ? (v * c - u * s) : (u * s + v * c); } oh[j] = (T_)v; } """ return mx.fast.metal_kernel( name="maple_qk_norm_rope", input_names=["x", "qw", "kw", "pos_eps"], output_names=["out"], source=source, ) _qk_norm_rope_kernel = _make_qk_norm_rope_kernel() # Affine dot-product evaluation order follows MLX v0.32.0 quantized.h # (Copyright Apple Inc., MIT license). Keep its fp32 accumulation and bf16 # rounding boundaries, including those within the fused SwiGLU activation. _ROW_QDOT_HEADER = r""" float maple_qdot(const device uchar* w, const thread float* x, float scale, float bias, float sum) { float accum = 0; for (int i=0;i<4;++i) { accum += (x[4*i]*(w[i]&0x03) + x[4*i+1]*(w[i]&0x0c) + x[4*i+2]*(w[i]&0x30) + x[4*i+3]*(w[i]&0xc0)); } return scale*accum+sum*bias; } """ _row_qmv_kernel = mx.fast.metal_kernel( name="maple_row_qmv_exact", input_names=["x", "weight", "scales", "biases", "ids"], output_names=["out"], header=_ROW_QDOT_HEADER, source=r""" uint lid=thread_index_in_simdgroup,sg=simdgroup_index_in_threadgroup; uint slot=threadgroup_position_in_grid.y; uint expert=GATHER?ids[slot]:0; uint row0=threadgroup_position_in_grid.x*8+sg*4; const device uchar* w=(const device uchar*)weight+expert*N*(K/4); const device bfloat16_t* sc=scales+expert*N*(GROUPED?K/128:1); const device bfloat16_t* bi=biases+expert*N*(GROUPED?K/128:1); const device bfloat16_t* xv=x+(SPLIT?slot*K:0); float result[4]={0}; for(int k=0;k(bfloat16_t)(-7.0f)?u:(bfloat16_t)(-7.0f); u=u<(bfloat16_t)7.0f?u:(bfloat16_t)7.0f; auto t=1/(1+metal::exp(metal::abs(g))); bfloat16_t sig=(g<0)?t:1-t; bfloat16_t silu=g*sig; out[slot*512+first+r]=silu*u; } } """, ) def _row_quantized_metadata(p): """Use compact row metadata with snapshots, live group metadata without.""" if ( mx.__version__ != "0.32.0" or type(p) not in (nn.QuantizedLinear, QuantizedSwitchLinear) or p.bits != 2 or p.group_size != 128 or p.mode != "affine" or p.weight.dtype != mx.uint32 or p.scales.dtype != mx.bfloat16 or p.get("biases") is None or p.biases.dtype != mx.bfloat16 ): return None native = hasattr(_maple_native, "ArraySnapshot") sources = (p.weight, p.scales, p.biases) if native else None state = p.get("_maple_row_state") if ( native and state is not None and state.get("sources") is not None and state["sources"].matches(sources) ): return state if state["supported"] else None k, n = p.scales.shape[-1] * 128, p.weight.shape[-2] if ( k not in (512, 2048) or n % 8 or p.weight.shape[-1] != k // 16 or p.scales.shape != (*p.weight.shape[:-1], k // 128) or p.biases.shape != p.scales.shape ): return None if not native: if ( state is None or state.get("sources") is not None or (state.get("k"), state.get("n")) != (k, n) ): state = dict(k=k, n=n, qmv_ok=None, zero=mx.array([0], mx.uint32), sources=None) p["_maple_row_state"] = state # Live arrays need no snapshot: no derived weight values are cached. state.update(scales=p.scales, biases=p.biases) return state state = { "sources": _maple_native.ArraySnapshot(sources), "supported": False, "qmv_ok": None, } p["_maple_row_state"] = state sc, bi = p.scales[..., :1], p.biases[..., :1] same = mx.all(p.scales.view(mx.uint16) == sc.view(mx.uint16)) & mx.all( p.biases.view(mx.uint16) == bi.view(mx.uint16) ) mx.eval(same) if not bool(same): return None sc, bi = mx.contiguous(sc[..., 0]), mx.contiguous(bi[..., 0]) zero = mx.array([0], mx.uint32) mx.eval(sc, bi, zero) state.update(supported=True, scales=sc, biases=bi, zero=zero, k=k, n=n) return state @mx.compile def _row_qmv_arrays(x, weight, scales, biases, ids, gather, split): """Compile dispatch construction; changing arrays stay explicit inputs.""" k, n = weight.shape[-1] * 16, weight.shape[-2] shape = (8, 1, n) if gather else (*x.shape[:-1], n) return _row_qmv_kernel( inputs=[x, weight, scales, biases, ids], template=[ ("K", k), ("N", n), ("GATHER", gather), ("SPLIT", split), ("GROUPED", scales.ndim == weight.ndim), ], grid=(n // 4 * 32, 8 if gather else 1, 1), threadgroup=(64, 1, 1), output_shapes=[shape], output_dtypes=[mx.bfloat16], )[0] @mx.compile def _row_experts_arrays(x, indices, uw, us, ub, dw, ds, db): ids = indices.reshape(-1).astype(mx.uint32) y = _row_up_kernel( inputs=[x, uw, us, ub, ids], template=[("GROUPED", us.ndim == uw.ndim)], grid=(4096, 8, 1), threadgroup=(64, 1, 1), output_shapes=[(8, 1, 512)], output_dtypes=[mx.bfloat16], )[0] return _row_qmv_arrays(y, dw, ds, db, ids, True, True).reshape(1, 1, 8, 2048) def _decode_projection(x, p): """Exact, runtime-probed row-metadata GEMV for a single bf16 vector.""" if x.ndim != 3 or x.shape[:2] != (1, 1) or x.dtype != mx.bfloat16: return p(x) if type(p) is not nn.QuantizedLinear: return p(x) state = _row_quantized_metadata(p) if state is None or x.shape[-1] != state["k"] or state["qmv_ok"] is False: return p(x) def fast(): y = _row_qmv_arrays( x, p.weight, state["scales"], state["biases"], state["zero"], False, False ) return y + p.bias if "bias" in p else y if state["qmv_ok"] is None: got = _exact_result(fast, lambda: p(x)) state["qmv_ok"] = got is not None return got if state["qmv_ok"] else p(x) return fast() class MapleAttention(nn.Module): def __init__(self, args: ModelArgs, layer_idx: int): super().__init__() self.num_attention_heads = args.num_attention_heads self.num_key_value_heads = args.num_key_value_heads self.head_dim = args.head_dim or args.hidden_size // args.num_attention_heads self.scale = self.head_dim**-0.5 self.use_qk_norm = args.use_qk_norm # q/k/v are stored fused (one matmul per step); sanitize() concatenates # the checkpoint's split projections. self.qkv_proj = nn.Linear( args.hidden_size, (args.num_attention_heads + 2 * args.num_key_value_heads) * self.head_dim, bias=args.use_bias, ) self.o_proj = nn.Linear( args.num_attention_heads * self.head_dim, args.hidden_size, bias=args.use_bias, ) if args.use_qk_norm: self.q_norm = MapleRMSNorm(self.head_dim, eps=args.rms_norm_eps) self.k_norm = MapleRMSNorm(self.head_dim, eps=args.rms_norm_eps) self._eps = args.rms_norm_eps self._rope_base = args.rope_theta self._qk_w = None self._qk_sources = None self._inv_freq = None self._fused_qk = None # None = unprobed, then True/False self._native_qkv = None # Maple applies RoPE only on sliding-window layers; full-attention # layers use no positional encoding (NoPE). self.use_rope = args.layer_types[layer_idx] == "sliding_attention" if self.use_rope: rope_dim = int(self.head_dim * args.partial_rotary_factor) self.rope = initialize_rope( rope_dim, args.rope_theta, traditional=False, scaling_config=args.rope_scaling, max_position_embeddings=args.max_position_embeddings, ) def _ensure_qk_state(self): sources = (self.q_norm.weight, self.k_norm.weight) if self._qk_sources is None or not self._qk_sources.matches(sources): n_q = self.num_attention_heads n_kv = self.num_key_value_heads self._qk_w = mx.contiguous( mx.concatenate( [ mx.broadcast_to(self.q_norm.weight[None], (n_q, self.head_dim)), mx.broadcast_to( self.k_norm.weight[None], (n_kv, self.head_dim) ), ] ) ) if self.use_rope: half = self.rope.dims // 2 self._inv_freq = _rope_inv_freq_kernel( inputs=[mx.array([math.log2(self._rope_base)], mx.float32)], template=[("HALF", half)], grid=(half, 1, 1), threadgroup=(32, 1, 1), output_shapes=[(half,)], output_dtypes=[mx.float32], )[0] else: self._inv_freq = mx.ones((1,), dtype=mx.float32) if hasattr(_maple_native, "ArraySnapshot"): mx.eval(self._qk_w, self._inv_freq) self._qk_sources = _maple_native.ArraySnapshot(sources) def _qk_fused(self, qkv, offset): """Normalize/rotate Q and K; copy any trailing V heads unchanged.""" # cache.offset is a Python int for a plain cache but an mx.array for # the batched caches; coerce before constructing the kernel input. pos_eps = mx.array( [float(offset), self._eps, math.log2(self._rope_base)], dtype=mx.float32 ) return _qk_norm_rope_kernel( inputs=[qkv, self.q_norm.weight, self.k_norm.weight, pos_eps], template=[ ("T_", qkv.dtype), ("HEAD_DIM", self.head_dim), ("ROPE_DIM", self.rope.dims if self.use_rope else 0), ("NQ", self.num_attention_heads), ("NQK", self.num_attention_heads + self.num_key_value_heads), ], grid=(32, qkv.shape[0], 1), threadgroup=(32, 1, 1), output_shapes=[qkv.shape], output_dtypes=[qkv.dtype], )[0] def _qk_supported(self, dtype): return ( mx.__version__ == "0.32.0" and self.head_dim == 128 and dtype == mx.bfloat16 and self.q_norm.weight.dtype == dtype and self.k_norm.weight.dtype == dtype and ( not self.use_rope or ( type(self.rope) is nn.RoPE and self.rope.scale == 1.0 and not self.rope.traditional ) ) ) def _probe_native_qkv(self): if not ( _maple_native is not None and getattr(_maple_native, "arithmetic_version", None) == 3 and self.num_attention_heads == 16 and self.num_key_value_heads == 4 and self._qk_supported(mx.bfloat16) and (not self.use_rope or self.rope.dims == 64) ): return False try: self._ensure_qk_state() for offset in (7, 613): x = mx.random.normal((24, 128), key=mx.random.key(offset)).astype( mx.bfloat16 ) cache = mx.zeros((2048 + 8 * 16 * 128,), mx.bfloat16) got = _maple_native.prepare_qkv( x, self._qk_w, self._inv_freq, cache, offset, 7, 64 if self.use_rope else 0, self._eps, ) ref = mx.concatenate([self._qk_reference(x[:20], offset), x[20:]]) wanted = mx.zeros((1, 8, 16, 128), cache.dtype) wanted[:, :, 7:8, :] = ref[16:].reshape(1, 8, 1, 128) mx.eval(got, ref, wanted) if not ( bool(mx.array_equal(got[:2048], ref[:16].reshape(-1))) and bool(mx.array_equal(got[2048:], wanted.reshape(-1))) ): return False except Exception: return False return True def _qk_reference(self, qk, offset): """The same result from stock ops: fallback, and the yardstick the fused kernel is checked against.""" n_q = self.num_attention_heads q = self.q_norm(qk[None, :n_q, None, :]) k = self.k_norm(qk[None, n_q:, None, :]) if self.use_rope: q = self.rope(q, offset=offset) k = self.rope(k, offset=offset) return mx.concatenate([q, k], axis=1).reshape(qk.shape) def __call__( self, x: mx.array, mask: Optional[mx.array] = None, cache: Optional[Any] = None, ) -> mx.array: B, L, _ = x.shape qkv = _decode_projection(x, self.qkv_proj) if B == L == 1 else self.qkv_proj(x) if ( B == 1 and L == 1 and self.use_qk_norm and self._qk_supported(qkv.dtype) and cache is not None and hasattr(cache, "native_compatible") ): if self._native_qkv is None: self._native_qkv = self._probe_native_qkv() if self._native_qkv and ( cache._native_buffer is not None or cache.native_compatible() ): queries, keys, values = cache.update_native(qkv, self) output = scaled_dot_product_attention( queries, keys, values, cache=cache, scale=self.scale, mask=mask ) return _decode_projection( output.transpose(0, 2, 1, 3).reshape(1, 1, -1), self.o_proj ) if B == 1 and L == 1 and self.use_qk_norm and self._qk_supported(qkv.dtype): n_q = self.num_attention_heads n_kv = self.num_key_value_heads qk_size = (n_q + n_kv) * self.head_dim qk = qkv.reshape(-1)[:qk_size].reshape(n_q + n_kv, self.head_dim) if self._fused_qk is None: # A nonzero position, so a broken rotation cannot pass. self._fused_qk = _matches( lambda: (self._qk_fused(qk, 7),), lambda: (self._qk_reference(qk, 7),), ) offset = cache.offset if cache is not None else 0 out = (self._qk_fused if self._fused_qk else self._qk_reference)(qk, offset) queries = out[:n_q].reshape(1, n_q, 1, self.head_dim) keys = out[n_q:].reshape(1, n_kv, 1, self.head_dim) values = qkv.reshape(-1)[qk_size:].reshape(1, n_kv, 1, self.head_dim) else: q_size = self.num_attention_heads * self.head_dim kv_size = self.num_key_value_heads * self.head_dim q, k, v = mx.split(qkv, [q_size, q_size + kv_size], axis=-1) queries = q.reshape(B, L, self.num_attention_heads, self.head_dim) keys = k.reshape(B, L, self.num_key_value_heads, self.head_dim) values = v.reshape(B, L, self.num_key_value_heads, self.head_dim) if self.use_qk_norm: queries = self.q_norm(queries) keys = self.k_norm(keys) queries = queries.transpose(0, 2, 1, 3) keys = keys.transpose(0, 2, 1, 3) values = values.transpose(0, 2, 1, 3) if self.use_rope: offset = cache.offset if cache is not None else 0 queries = self.rope(queries, offset=offset) keys = self.rope(keys, offset=offset) if cache is not None: keys, values = cache.update_and_fetch(keys, values) output = scaled_dot_product_attention( queries, keys, values, cache=cache, scale=self.scale, mask=mask ) output = output.transpose(0, 2, 1, 3).reshape(B, L, -1) return ( _decode_projection(output, self.o_proj) if B == L == 1 else self.o_proj(output) ) class MapleMLP(nn.Module): def __init__(self, args: ModelArgs, intermediate_size: Optional[int] = None): super().__init__() intermediate_size = intermediate_size or args.intermediate_size self.gate_proj = nn.Linear( args.hidden_size, intermediate_size, bias=args.use_bias ) self.up_proj = nn.Linear( args.hidden_size, intermediate_size, bias=args.use_bias ) self.down_proj = nn.Linear( intermediate_size, args.hidden_size, bias=args.use_bias ) def __call__(self, x) -> mx.array: # Dense / shared-expert MLP: no clamp; only the MoE experts clamp. # Unused at first_k_dense_replace=0 with no shared experts, but keep # it faithful. return self.down_proj(swiglu(self.gate_proj(x), self.up_proj(x))) @mx.compile def group_expert_select(gates, top_k): # Maple routes with a plain softmax over all experts followed by top-k # selection and renormalization, computed in float32. scores = mx.softmax(gates.astype(mx.float32), axis=-1) inds = mx.argpartition(scores, kth=-top_k, axis=-1)[..., -top_k:] scores = mx.take_along_axis(scores, inds, axis=-1) scores = scores / (scores.sum(axis=-1, keepdims=True) + 1e-20) return inds, scores class MapleGate(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.top_k = args.num_experts_per_tok self.num_experts = args.num_experts self.hidden_size = args.hidden_size # Kept as a raw parameter (not nn.Linear) so quantization never # touches it. The matmul accumulates in float32 and selection runs on # float32 scores. self.weight = mx.zeros((args.num_experts, args.hidden_size)) def __call__(self, x): # Preserve FP32 routing: rounding near-tied logits to BF16 changes picks. gates = x.astype(mx.float32) @ self.weight.astype(mx.float32).T inds, scores = group_expert_select(gates, self.top_k) if self.top_k == 8: inds = inds.astype(mx.uint32 if x.size == self.hidden_size else mx.int32) return inds, scores @partial(mx.compile, shapeless=True) def aggregate_expert_outputs(expert_outputs, scores): # Combined in float32, rounded once at the end (reference `moe_infer`). return ( (expert_outputs.astype(mx.float32) * scores[..., None]) .sum(axis=-2) .astype(expert_outputs.dtype) ) class MapleSwitchGLU(nn.Module): """SwitchGLU with the up and gate projections fused into one gather matmul; sanitize() concatenates the checkpoint's split tensors.""" def __init__(self, input_dims, hidden_dims, num_experts, bias=False): super().__init__() self.up_gate_proj = SwitchLinear( input_dims, 2 * hidden_dims, num_experts, bias=bias ) self.down_proj = SwitchLinear(hidden_dims, input_dims, num_experts, bias=bias) self._decode_row_pair = None self._decode_row_ok = None def _decode_row_experts(self, x, indices, up, down): return _row_experts_arrays( x, indices, self.up_gate_proj.weight, up["scales"], up["biases"], self.down_proj.weight, down["scales"], down["biases"], ) def __call__(self, x, indices): p, d = self.up_gate_proj, self.down_proj if ( x.dtype == mx.bfloat16 and x.shape == (1, 1, 2048) and indices.shape == (1, 1, 8) and indices.dtype in (mx.int32, mx.uint32) and type(p) is QuantizedSwitchLinear and type(d) is QuantizedSwitchLinear and p.weight.shape[-2:] == (1024, 128) and d.weight.shape[-2:] == (2048, 32) and p.weight.shape[0] == d.weight.shape[0] and "bias" not in p and "bias" not in d ): up, down = _row_quantized_metadata(p), _row_quantized_metadata(d) if up is not None and down is not None: pair = (up, down) if self._decode_row_pair is None or any( a is not b for a, b in zip(pair, self._decode_row_pair) ): self._decode_row_pair, self._decode_row_ok = pair, None if self._decode_row_ok is None: got = _exact_result( lambda: self._decode_row_experts(x, indices, up, down), lambda: self._call(x, indices), ) self._decode_row_ok = got is not None if self._decode_row_ok: return got elif self._decode_row_ok: return self._decode_row_experts(x, indices, up, down) return self._call(x, indices) def _call(self, x, indices): x = mx.expand_dims(x, (-2, -3)) do_sort = indices.size >= 64 idx = indices inv_order = None if do_sort: x, idx, inv_order = _gather_sort(x, indices) x_up, x_gate = mx.split( self.up_gate_proj(x, idx, sorted_indices=do_sort), 2, axis=-1 ) x = self.down_proj(clamped_swiglu(x_gate, x_up), idx, sorted_indices=do_sort) if do_sort: x = _scatter_unsort(x, inv_order, indices.shape) return x.squeeze(-2) class MapleSparseMoeBlock(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.gate = MapleGate(args) self.switch_mlp = MapleSwitchGLU( args.hidden_size, args.moe_intermediate_size, args.num_experts, bias=args.use_bias, ) def __call__(self, x): inds, scores = self.gate(x) y = self.switch_mlp(x, inds) return aggregate_expert_outputs(y, scores) class MapleDecoderLayer(nn.Module): def __init__(self, args: ModelArgs, layer_idx: int): super().__init__() self.self_attn = MapleAttention(args, layer_idx) self.mlp = ( MapleSparseMoeBlock(args) if layer_idx >= args.first_k_dense_replace else MapleMLP(args) ) self.input_layernorm = MapleRMSNorm(args.hidden_size, eps=args.rms_norm_eps) self.post_attention_layernorm = MapleRMSNorm( args.hidden_size, eps=args.rms_norm_eps ) def __call__( self, x: mx.array, mask: Optional[mx.array] = None, cache: Optional[Any] = None, ) -> mx.array: r = self.self_attn(self.input_layernorm(x), mask, cache) h = x + r r = self.mlp(self.post_attention_layernorm(h)) return h + r # Match MLX 0.32.0 GEMV and softmax/reduction arithmetic (Apple Inc., MIT). _router_gemv_kernel = mx.fast.metal_kernel( name="maple_router_gemv", input_names=["x", "mat"], output_names=["logits"], source=""" constexpr int K = 2048; constexpr int TM = 4; constexpr int TN = 4; uint lane = thread_position_in_threadgroup.x % 32; uint sg = thread_position_in_threadgroup.x / 32; uint row = threadgroup_position_in_grid.x * 16 + sg * TM; float result[TM] = {0}; for (int base = 0; base < K; base += 128) { float v[TN]; #pragma clang loop unroll(full) for (int n = 0; n < TN; n++) v[n] = float(x[base + lane * TN + n]); #pragma clang loop unroll(full) for (int m = 0; m < TM; m++) { #pragma clang loop unroll(full) for (int n = 0; n < TN; n++) { float w = float(mat[(row + m) * K + base + lane * TN + n]); result[m] += w * v[n]; } } } #pragma clang loop unroll(full) for (int m = 0; m < TM; m++) { #pragma clang loop unroll(full) for (ushort offset = 16; offset >= 1; offset >>= 1) result[m] += simd_shuffle_down(result[m], offset); if (lane == 0) logits[row + m] = result[m]; } """, ) _router_select_kernel = mx.fast.metal_kernel( name="maple_router_select", input_names=["logits"], output_names=["indices", "scores"], source=""" #pragma clang fp contract(off) uint tid = thread_position_in_threadgroup.x; uint lane = tid % 32; uint sg = tid / 32; threadgroup float maxima[32], sums[32], probs[256]; float vals[4]; if (sg == 0) { maxima[lane] = -INFINITY; sums[lane] = 0.0f; } float vmax = tid < 64 ? -MAXFLOAT : -INFINITY; for (int j = 0; j < 4; j++) { vals[j] = tid < 64 ? logits[tid * 4 + j] : -INFINITY; vmax = vmax < vals[j] ? vals[j] : vmax; } threadgroup_barrier(mem_flags::mem_threadgroup); vmax = simd_max(vmax); if (lane == 0) maxima[sg] = vmax; threadgroup_barrier(mem_flags::mem_threadgroup); if (sg == 0) { vmax = simd_max(maxima[lane]); if (lane == 0) maxima[0] = vmax; } threadgroup_barrier(mem_flags::mem_threadgroup); vmax = maxima[0]; float sum = 0.0f; for (int j = 0; j < 4; j++) { vals[j] = tid < 64 ? metal::fast::exp(vals[j] - vmax) : 0.0f; sum += vals[j]; } sum = simd_sum(sum); if (lane == 0) sums[sg] = sum; threadgroup_barrier(mem_flags::mem_threadgroup); if (sg == 0) { sum = simd_sum(sums[lane]); if (lane == 0) sums[0] = sum; } threadgroup_barrier(mem_flags::mem_threadgroup); float inv = 1.0f / sums[0]; if (tid < 64) for (int j = 0; j < 4; j++) probs[tid * 4 + j] = vals[j] * inv; threadgroup_barrier(mem_flags::mem_threadgroup); if (sg == 0) { float candidates[8]; uint selected[8]; for (int j = 0; j < 8; j++) { float p = probs[lane * 8 + j]; candidates[j] = metal::isnan(p) ? INFINITY : p; } for (int pick = 7; pick >= 0; pick--) { float best = -INFINITY; for (int j = 0; j < 8; j++) best = metal::max(best, candidates[j]); best = simd_max(best); int winner = -1; for (int j = 0; j < 8; j++) if (candidates[j] == best) winner = int(lane * 8 + j); winner = simd_max(winner); selected[pick] = uint(winner); for (int j = 0; j < 8; j++) if (lane * 8 + j == uint(winner)) candidates[j] = -INFINITY; } float denom = 0.0f; for (int j = 0; j < 8; j++) denom = probs[selected[j]] + denom; denom = denom + 1e-20f; if (lane < 8) { indices[lane] = selected[lane]; scores[lane] = probs[selected[lane]] / denom; } } """, ) @mx.compile def _norm_router_arrays(h, r, norm_weight, router_weight, eps, portable=False): h, hn = _add_rms_norm(h, r, norm_weight, eps) if portable: logits = hn.astype(mx.float32) @ router_weight.astype(mx.float32).T else: logits = _router_gemv_kernel( inputs=[hn, router_weight], grid=(2048, 1, 1), threadgroup=(128, 1, 1), output_shapes=[(1, 1, 256)], output_dtypes=[mx.float32], )[0] indices, scores = _router_select_kernel( inputs=[logits], grid=(128, 1, 1), threadgroup=(128, 1, 1), output_shapes=[(1, 1, 8), (1, 1, 8)], output_dtypes=[mx.uint32, mx.float32], ) return h, hn, logits, indices, scores def _decode_norm_router(h, r, norm, gate): def fallback(): hh, hn = _add_rms_norm(h, r, norm.weight, norm.eps) indices, scores = gate(hn) return hh, hn, indices, scores if ( mx.__version__ != "0.32.0" or type(norm) is not MapleRMSNorm or type(gate) is not MapleGate or h.shape != (1, 1, 2048) or r.shape != h.shape or h.dtype != mx.bfloat16 or r.dtype != mx.bfloat16 or norm.weight.shape != (2048,) or norm.weight.dtype != mx.bfloat16 or gate.weight.shape != (256, 2048) or gate.weight.dtype != mx.bfloat16 or gate.top_k != 8 or not math.isfinite(norm.eps) or norm.eps <= 0 ): return fallback() native = hasattr(_maple_native, "ArraySnapshot") sources = (norm.weight, gate.weight) state = gate.get("_maple_norm_router_state") if ( state is None or state["eps"] != norm.eps or native != (state["sources"] is not None) or (native and not state["sources"].matches(sources)) ): state = { "sources": _maple_native.ArraySnapshot(sources) if native else None, "eps": norm.eps, "ok": None, } gate["_maple_norm_router_state"] = state if state["ok"] is False: return fallback() def fast(): return _norm_router_arrays(h, r, norm.weight, gate.weight, norm.eps, not native) if state["ok"] is None: state["ok"] = False try: got = fast() hh = h + r hn = norm(hh) logits = hn.astype(mx.float32) @ gate.weight.astype(mx.float32).T indices, scores = group_expert_select(logits, gate.top_k) ref = (hh, hn, logits, indices.astype(mx.uint32), scores) mx.eval(got, ref) state["ok"] = all( a.dtype == b.dtype and a.shape == b.shape and bool( mx.array_equal( a.view(mx.uint16 if a.dtype == mx.bfloat16 else mx.uint32), b.view(mx.uint16 if b.dtype == mx.bfloat16 else mx.uint32), ) ) for a, b in zip(got, ref) ) except Exception: pass if not state["ok"]: return fallback() else: got = fast() return got[0], got[1], got[3], got[4] class MapleModel(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.args = args self.word_embeddings = nn.Embedding(args.vocab_size, args.hidden_size) self.layers = [ MapleDecoderLayer(args, layer_idx=i) for i in range(args.num_hidden_layers) ] self.norm = MapleRMSNorm(args.hidden_size, eps=args.rms_norm_eps) self.layer_types = args.layer_types self.window_size = args.sliding_window self.swa_idx = ( self.layer_types.index("sliding_attention") if "sliding_attention" in self.layer_types else None ) self.ga_idx = ( self.layer_types.index("full_attention") if "full_attention" in self.layer_types else None ) self._fused_add_norm = None # None = unprobed, then True/False self._zero = None self._fused_aggregate_norm = None def _decode_fused(self, h, cache, full_mask, swa_mask): """Decode loop with residual adds folded into the norms. Carries (h, r) instead of adding r back each step, so every add+norm pair is one dispatch. Identical arithmetic: the kernel rounds the sum once (as the bf16 add did) and norms the rounded stream with an fp32 weight multiply. """ if self._zero is None: self._zero = mx.zeros(h.shape, h.dtype) mx.eval(self._zero) r = self._zero # x + 0 is exact in bf16 if self._fused_aggregate_norm is None: self._fused_aggregate_norm = ( h.dtype == mx.bfloat16 and h.shape == (1, 1, 2048) and bool(self.layers) and all( isinstance(l.mlp, MapleSparseMoeBlock) and l.mlp.gate.top_k == 8 and l.input_layernorm.weight.dtype == h.dtype and l.post_attention_layernorm.weight.dtype == h.dtype for l in self.layers ) and _aggregate_add_rms_norm_ok( h.shape[-1], h.dtype, self.norm.weight, self.norm.eps ) ) if self._fused_aggregate_norm: ln = self.layers[0].input_layernorm h, hn = _add_rms_norm(h, r, ln.weight, ln.eps) for i, (layer, c, layer_type) in enumerate( zip(self.layers, cache, self.layer_types) ): mask = full_mask if layer_type == "full_attention" else swa_mask r = layer.self_attn(hn, mask, c) ln = layer.post_attention_layernorm h, hn, inds, scores = _decode_norm_router(h, r, ln, layer.mlp.gate) y = layer.mlp.switch_mlp(hn, inds) ln = ( self.layers[i + 1].input_layernorm if i + 1 < len(self.layers) else self.norm ) h, hn = _aggregate_add_rms_norm(h, y, scores, ln.weight, ln.eps) return hn for layer, c, layer_type in zip(self.layers, cache, self.layer_types): mask = full_mask if layer_type == "full_attention" else swa_mask ln = layer.input_layernorm h, hn = _add_rms_norm(h, r, ln.weight, ln.eps) r = layer.self_attn(hn, mask, c) ln = layer.post_attention_layernorm h, hn = _add_rms_norm(h, r, ln.weight, ln.eps) r = layer.mlp(hn) return _add_rms_norm(h, r, self.norm.weight, self.norm.eps)[1] def __call__( self, inputs: mx.array, cache: Optional[Any] = None, ): h = self.word_embeddings(inputs) if cache is None: cache = [None] * len(self.layers) full_mask = None swa_mask = None if self.ga_idx is not None: full_mask = create_attention_mask(h, cache[self.ga_idx]) if self.swa_idx is not None: swa_mask = create_attention_mask( h, cache[self.swa_idx], window_size=self.window_size ) if h.size == h.shape[-1]: if self._fused_add_norm is None: self._fused_add_norm = _add_rms_norm_ok( h.shape[-1], h.dtype, self.norm.weight, self.norm.eps ) if self._fused_add_norm: return self._decode_fused(h, cache, full_mask, swa_mask) for layer, c, layer_type in zip(self.layers, cache, self.layer_types): mask = full_mask if layer_type == "full_attention" else swa_mask h = layer(h, mask, c) if inputs.shape[-1] > 2048: # Bound large prefill graphs. A repeated 8K prefill otherwise # produced inconsistent results with identical weights on MLX # 0.32.0/0.32.2. Layer evaluation preserves the operations and # context; standard generation already chunks prefill at 2048. mx.eval(h) return self.norm(h) _flash_scatter_kernel = mx.fast.metal_kernel( name="maple_flash_scatter", input_names=["top", "token_map", "logits", "force_ids", "force_logits"], output_names=["out"], source=""" uint i = thread_position_in_grid.x; if (i < COUNT) { int token = token_map[top[i / CLUSTER] * CLUSTER + i % CLUSTER]; bool forced = false; for (uint j = 0; j < FORCED; ++j) forced |= token == force_ids[j]; if (!forced) out[token] = logits[i]; } if (i < FORCED) { bool duplicate = false; for (uint j = 0; j < i; ++j) duplicate |= force_ids[j] == force_ids[i]; if (!duplicate) out[force_ids[i]] = force_logits[i]; } """, ) def _flash_scatter(top, token_map, logits, force_ids, force_logits, vocab_size): return _flash_scatter_kernel( inputs=[top, token_map, logits, force_ids, force_logits], template=[ ("COUNT", logits.size), ("CLUSTER", token_map.shape[1]), ("FORCED", force_ids.size), ], grid=(max(logits.size, force_ids.size), 1, 1), threadgroup=(256, 1, 1), output_shapes=[(1, 1, vocab_size)], output_dtypes=[logits.dtype], init_value=-float("inf"), )[0] # Exact bf16 top-k set: two radix histograms, with MLX's stable tie order. # One threadgroup owns selection and compaction; no cross-group synchronization. _head_select_kernel = mx.fast.metal_kernel( name="maple_head_select", input_names=["x"], output_names=["out"], header=r""" inline uint head_key(bfloat16_t x) { uint raw=as_type(x), magnitude=raw&0x7fffu; if(magnitude>0x7f80u) return 0xffffu; if(magnitude<0x80u) return 0x8000u; return (raw&0x8000u)?((~raw)&0xffffu):(raw^0x8000u); } inline uint head_prefix(uint value,uint tid,threadgroup uint* groups) { uint lane=tid%32u,sg=tid/32u; uint prefix=simd_prefix_exclusive_sum(value); if(lane==31u) groups[sg]=prefix+value; threadgroup_barrier(mem_flags::mem_threadgroup); for(uint w=0;w>8; uint matches=uint((simd_vote::vote_t)simd_ballot(iN-K) {cut[0]=tid;cut[1]=prefix;} threadgroup_barrier(mem_flags::mem_threadgroup); if(tid<256u) atomic_store_explicit(hist+tid,0u,memory_order_relaxed); threadgroup_barrier(mem_flags::mem_threadgroup); for(uint j=0;j>8)==cut[0]) atomic_fetch_add_explicit(hist+(keys[j]&255u),1u,memory_order_relaxed); } threadgroup_barrier(mem_flags::mem_threadgroup); count=tid<256u?atomic_load_explicit(hist+tid,memory_order_relaxed):0u; prefix=head_prefix(count,tid,groups); uint rank=N-K-cut[1]; if(tid<256u && prefix<=rank && prefix+count>rank) { cut[2]=(cut[0]<<8)|tid; cut[3]=N-(cut[1]+prefix+count); cut[4]=count; } threadgroup_barrier(mem_flags::mem_threadgroup); uint greater=0u,equal=0u; for(uint j=0;jcut[2]; equal+=keys[j]==cut[2]; } uint greater_prefix=head_prefix(greater,tid,groups); uint equal_prefix=head_prefix(equal,tid,groups); uint skip=cut[4]-(K-cut[3]); for(uint j=0;j=N) break; if(keys[j]>cut[2]) out[greater_prefix++]=i; else if(keys[j]==cut[2]) { if(equal_prefix>=skip) out[cut[3]+equal_prefix-skip]=i; equal_prefix++; } } """, ) def _head_select(x, k): return _head_select_kernel( inputs=[x], template=[("N", x.size), ("K", k)], grid=(1024, 1, 1), threadgroup=(1024, 1, 1), output_shapes=[(1, k)], output_dtypes=[mx.uint32], )[0] class FlashHead(nn.Module): """Two-phase approximate lm_head for single-stream decode. Phase one scores quantized cluster centroids of the vocabulary; phase two computes exact logits only for the tokens of the top ``n_probes`` clusters (plus a fixed set of forced control tokens such as EOS). All other logits are -inf, so greedy decoding is exact whenever the true argmax lies in the probed clusters. Prefill and batched calls use the exact lm_head. Reference: FlashHead — Efficient Drop-in Replacement for the Classification Head in Language Model Inference. """ def __init__(self, args: ModelArgs): super().__init__() meta = args.flash_head if not meta.get("scaled_centroids"): raise ValueError( "FlashHead metadata predates scaled centroids; regenerate with " "`python -m mlx_lm.ternary --flash-head-only`." ) n_clusters = meta["n_clusters"] cluster_size = meta["cluster_size"] # Default matches the converter's `--probes` default; every generated # checkpoint records the value explicitly. self.n_probes = min(meta.get("n_probes", 512), n_clusters) self.head_group_size = meta.get("head_group_size", 64) self.head_bits = meta.get("head_bits", 4) # The converter stores scaled centroids for a single scoring matmul. self.centroids = nn.QuantizedLinear( args.hidden_size, n_clusters, bias=False, group_size=meta.get("group_size", 64), bits=meta.get("bits", 4), ) self.token_map = mx.zeros((n_clusters, cluster_size), dtype=mx.int32) # Cluster-ordered copy of the quantized lm_head: subset logits are one # gather_qmm over the probed 32-row blocks, with no per-step gather. # It is a row-permutation of lm_head by token_map and nothing more, so # it is derived rather than stored: Model.sanitize rebuilds it at load. hidden = args.hidden_size self.head = { "weight": mx.zeros( (n_clusters, cluster_size, hidden * self.head_bits // 32), dtype=mx.uint32, ), "scales": mx.zeros( (n_clusters, cluster_size, hidden // self.head_group_size), dtype=mx.bfloat16, ), "biases": mx.zeros( (n_clusters, cluster_size, hidden // self.head_group_size), dtype=mx.bfloat16, ), } self._force_ids = mx.array(meta.get("force_tokens", []), dtype=mx.int32) self._scatter = None self._select = None def _scatter_reference(self, top, logits, force_logits, vocab_size): oids = self.token_map[top[0]].reshape(-1) if self._force_ids.size: oids = mx.concatenate([oids, self._force_ids]) logits = mx.concatenate([logits, force_logits]) full = mx.full((1, 1, vocab_size), float("-inf"), dtype=logits.dtype) full[0, 0, oids] = logits return full def __call__(self, h: mx.array, lm_head: nn.Module) -> mx.array: hv = h[:, -1, :] scores = self.centroids(hv) reference = lambda: mx.argpartition(scores, kth=-self.n_probes, axis=-1)[ ..., -self.n_probes : ] if ( mx.__version__ == "0.32.0" and mx.default_device() == mx.gpu and scores.dtype == mx.bfloat16 and scores.ndim == 2 and scores.shape[0] == 1 and 0 < self.n_probes <= scores.size <= 8192 ): if self._select is None: self._select = _matches( lambda: (mx.sort(_head_select(scores, self.n_probes)),), lambda: (mx.sort(reference()),), ) top = _head_select(scores, self.n_probes) if self._select else reference() else: top = reference() logits = mx.gather_qmm( hv.reshape(1, 1, 1, 1, -1), self.head["weight"], self.head["scales"], self.head["biases"], rhs_indices=top[:, None, :], transpose=True, group_size=self.head_group_size, bits=self.head_bits, ).reshape(-1) force_logits = logits[:0] if self._force_ids.size: force_logits = mx.gather_qmm( hv.reshape(1, 1, -1), lm_head.weight.reshape(lm_head.weight.shape[0], 1, -1), lm_head.scales.reshape(lm_head.weight.shape[0], 1, -1), lm_head.biases.reshape(lm_head.weight.shape[0], 1, -1), rhs_indices=self._force_ids, transpose=True, group_size=lm_head.group_size, bits=lm_head.bits, mode=getattr(lm_head, "mode", "affine"), ).reshape(-1) vocab_size = lm_head.weight.shape[0] if mx.__version__ == "0.32.0" and self._force_ids.size <= 8: fast = lambda: _flash_scatter( top, self.token_map, logits, self._force_ids, force_logits, vocab_size ) if self._scatter is None: self._scatter = _matches( lambda: (fast(),), lambda: ( self._scatter_reference(top, logits, force_logits, vocab_size), ), ) if self._scatter: return fast() return self._scatter_reference(top, logits, force_logits, vocab_size) class _FusedKVBase: """Native decode buffer; stock caches handle growth, prefill, and rotation.""" # Defaults also support MLX's from_state(), which skips __init__. _native_buffer = None _native_capacity = 0 _keys = _values = None def native_compatible(self): if self._keys is None: return self._values is None return all( a is not None and a.dtype == mx.bfloat16 and a.ndim == 4 and a.shape[:2] == (1, 4) and a.shape[3] == 128 for a in (self._keys, self._values) ) def _native_kv(self): return self._native_buffer[2048:].reshape(1, 8, self._native_capacity, 128) def update_native(self, qkv, attn): rotating = isinstance(self, _RotatingKVCache) capacity = self._native_capacity if ( self._native_buffer is None or (self.offset >= capacity and (not rotating or capacity < self.max_size)) or (rotating and capacity > self.max_size) ): # Use stock arithmetic and cache allocation at transitions. Packing # Q/K/V into one output lets subsequent writes obey MLX donation. qkv = qkv.reshape(24, 128) qk = attn._qk_reference(qkv[:20], self.offset) self.update_and_fetch( qk[16:].reshape(1, 4, 1, 128), qkv[20:].reshape(1, 4, 1, 128) ) self._native_capacity = self._keys.shape[2] self._native_buffer = mx.concatenate( [qk[:16].reshape(-1), self._keys.reshape(-1), self._values.reshape(-1)] ) self._keys = self._values = None else: attn._ensure_qk_state() if rotating and self._idx == self.max_size: self._idx = self.keep self._native_buffer = _maple_native.prepare_qkv( qkv, attn._qk_w, attn._inv_freq, self._native_buffer, self.offset, self._idx if rotating else self.offset, 64 if attn.use_rope else 0, attn._eps, ) self.offset += 1 if rotating: self._idx += 1 length = min(self.offset, self._native_capacity) queries = self._native_buffer[:2048].reshape(1, 16, 1, 128) kv = self._native_kv() return queries, kv[:, :4, :length, :], kv[:, 4:, :length, :] @property def keys(self): return ( self._native_kv()[:, :4] if self._native_buffer is not None else self._keys ) @keys.setter def keys(self, value): self._unfuse() self._keys = value @property def values(self): return ( self._native_kv()[:, 4:] if self._native_buffer is not None else self._values ) @values.setter def values(self, value): self._unfuse() self._values = value def _unfuse(self): if self._native_buffer is not None: kv = self._native_kv() self._keys, self._values = kv[:, :4], kv[:, 4:] self._native_buffer = None def update_and_fetch(self, keys, values): self._unfuse() keys, values = super().update_and_fetch(keys, values) # Stock ring updates can return the mutable cache array wrappers. # Separate views keep retained outputs stable across later assignments. return keys[:], values[:] # Stock names keep saved caches readable by MLX's public cache loader. class KVCache(_FusedKVBase, _KVCache): pass class RotatingKVCache(_FusedKVBase, _RotatingKVCache): pass class Model(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.args = args self.model_type = args.model_type self.model = MapleModel(args) if not args.tie_word_embeddings: self.lm_head = nn.Linear(args.hidden_size, args.vocab_size, bias=False) if args.flash_head and args.use_flash_head and not args.tie_word_embeddings: self.lm_head_flash = FlashHead(args) else: self.lm_head_flash = None def __call__( self, inputs: mx.array, cache=None, ): out = self.model(inputs, cache) if self.args.tie_word_embeddings: return self.model.word_embeddings.as_linear(out) if ( self.lm_head_flash is not None and out.shape[0] == 1 and out.shape[1] == 1 and isinstance(self.lm_head, nn.QuantizedLinear) and getattr(self.lm_head, "mode", "affine") == "affine" ): return self.lm_head_flash(out, self.lm_head) return self.lm_head(out) def sanitize(self, weights): if self.args.tie_word_embeddings: # Drop the head entirely (weight + quantization scales/biases). weights = {k: v for k, v in weights.items() if not k.startswith("lm_head.")} # FlashHead disabled (e.g. model_config={"flash_head": None}): drop its # tensors so checkpoints that carry them still load. if self.lm_head_flash is None: weights = { k: v for k, v in weights.items() if not k.startswith("lm_head_flash.") } else: # Folded into the centroid rows at generation time; older shards # still carry the tensor. weights.pop("lm_head_flash.cluster_scale", None) # `lm_head_flash.head.*` is lm_head permuted by token_map (see # mlx_lm.ternary.generate_flash_head), so it is pure redundancy on # disk. Checkpoints may ship it or omit it; reconcile both here. if "lm_head_flash.head.weight" not in weights: token_map = weights["lm_head_flash.token_map"] order = token_map.reshape(-1) for k in ("weight", "scales", "biases"): weights[f"lm_head_flash.head.{k}"] = weights[f"lm_head.{k}"][ order ].reshape(*token_map.shape, -1) # Ternary tensors carry one scale per output row, so checkpoints store # it once as `row_alpha` and omit biases entirely (bias == -scale). # Expand here so everything downstream — fusion below, and mlx's own # quantized kernels — sees the per-group layout. Checkpoints written # with `--group-scales` have no row_alpha and pass straight through. row_alpha_keys = [k for k in weights if k.endswith(".row_alpha")] if row_alpha_keys: group_size = (self.args.quantization or {}).get("group_size", 128) for key in row_alpha_keys: alpha = weights.pop(key) prefix = key[: -len(".row_alpha")] packed = weights.get(f"{prefix}.weight") if packed is None: continue # 2-bit packing stores 16 codes per uint32 word. n_groups = (packed.shape[-1] * 16) // group_size scales = mx.contiguous( mx.broadcast_to(alpha[..., None], (*alpha.shape, n_groups)) ) weights[f"{prefix}.scales"] = scales weights[f"{prefix}.biases"] = -scales # Stack per-expert weights from the Hugging Face layout into the # SwitchGLU layout. Already-converted checkpoints pass through. for l in range(self.args.num_hidden_layers): prefix = f"model.layers.{l}" for m in ["gate_proj", "down_proj", "up_proj"]: for k in ["weight", "scales", "biases", "bias"]: if f"{prefix}.mlp.experts.0.{m}.{k}" in weights: to_join = [ weights.pop(f"{prefix}.mlp.experts.{e}.{m}.{k}") for e in range(self.args.num_experts) ] weights[f"{prefix}.mlp.switch_mlp.{m}.{k}"] = mx.stack(to_join) # Fuse split projections: q/k/v -> qkv_proj (rows), MoE up/gate -> # up_gate_proj (per-expert rows). Row-wise quantized tensors # (weight/scales/biases) concatenate losslessly along the output # axis. for suffix in ["weight", "scales", "biases", "bias"]: qkv = [ f"{prefix}.self_attn.{p}.{suffix}" for p in ("q_proj", "k_proj", "v_proj") ] if qkv[0] in weights: weights[f"{prefix}.self_attn.qkv_proj.{suffix}"] = mx.concatenate( [weights.pop(k) for k in qkv], axis=0 ) up = f"{prefix}.mlp.switch_mlp.up_proj.{suffix}" gate = f"{prefix}.mlp.switch_mlp.gate_proj.{suffix}" if up in weights: weights[f"{prefix}.mlp.switch_mlp.up_gate_proj.{suffix}"] = ( mx.concatenate([weights.pop(up), weights.pop(gate)], axis=1) ) return weights def make_cache(self): native = ( getattr(_maple_native, "arithmetic_version", None) == 3 and self.args.num_attention_heads == 16 and self.args.num_key_value_heads == 4 and self.args.head_dim == 128 ) caches = [] for layer_type in self.model.layer_types: if layer_type == "sliding_attention": caches.append( RotatingKVCache(max_size=self.args.sliding_window) if native else _RotatingKVCache(max_size=self.args.sliding_window) ) else: caches.append(KVCache() if native else _KVCache()) return caches @property def layers(self): return self.model.layers @property def quant_predicate(self): def predicate(path, _): if path.endswith("lm_head") or "word_embeddings" in path: return {"group_size": 64, "bits": 4} return True return predicate