import torch import torch.nn as nn import torch.nn.functional as F from transformers.models.lfm2_moe.modeling_lfm2_moe import ( Lfm2MoeForCausalLM, Lfm2MoeSparseMoeBlock, Lfm2MoeTopKRouter, Lfm2MoeExperts, Lfm2MoeModel, ) from transformers.cache_utils import DynamicCache from transformers.masking_utils import create_causal_mask, create_recurrent_attention_mask from transformers.modeling_outputs import MoeModelOutputWithPast from .configuration_lfm2_moe_custom import Lfm2MoeCustomConfig # =================================================================== # HOTFIX: Installed transformers is missing "linear_attention" key # in causal_mask_mapping. Replace the base forward at runtime. # =================================================================== _original_lfm_forward = Lfm2MoeModel.forward def _lfm_forward_fixed( self, input_ids: torch.LongTensor | None = None, attention_mask: torch.Tensor | None = None, position_ids: torch.LongTensor | None = None, past_key_values=None, inputs_embeds: torch.FloatTensor | None = None, use_cache: bool | None = None, **kwargs, ): if (input_ids is None) ^ (inputs_embeds is not None): raise ValueError("You must specify exactly one of input_ids or inputs_embeds") if inputs_embeds is None: inputs_embeds = self.embed_tokens(input_ids) if use_cache and past_key_values is None: past_key_values = DynamicCache(config=self.config) if position_ids is None: past_seen_tokens = past_key_values.get_seq_length() if past_key_values is not None else 0 position_ids = torch.arange(inputs_embeds.shape[1], device=inputs_embeds.device) + past_seen_tokens position_ids = position_ids.unsqueeze(0) if not isinstance(causal_mask_mapping := attention_mask, dict): mask_kwargs = { "config": self.config, "inputs_embeds": inputs_embeds, "attention_mask": attention_mask, "past_key_values": past_key_values, "position_ids": position_ids, } causal_mask_mapping = { "full_attention": create_causal_mask(**mask_kwargs), "conv": create_recurrent_attention_mask(**mask_kwargs), "linear_attention": None, } hidden_states = inputs_embeds position_embeddings = self.pos_emb(hidden_states, position_ids=position_ids) for i, decoder_layer in enumerate(self.layers[: self.config.num_hidden_layers]): hidden_states = decoder_layer( hidden_states, attention_mask=causal_mask_mapping[self.config.layer_types[i]], position_ids=position_ids, past_key_values=past_key_values, position_embeddings=position_embeddings, **kwargs, ) hidden_states = self.embedding_norm(hidden_states) return MoeModelOutputWithPast( last_hidden_state=hidden_states, past_key_values=past_key_values, ) Lfm2MoeModel.forward = _lfm_forward_fixed # =================================================================== class Lfm2MoeDynamicTopKRouter(Lfm2MoeTopKRouter): def __init__(self, config: Lfm2MoeCustomConfig, layer_idx: int = -1): nn.Module.__init__(self) self.num_experts = config.num_experts self.norm_topk_prob = config.norm_topk_prob self.hidden_dim = config.hidden_size self.weight = nn.Parameter(torch.zeros(self.num_experts, self.hidden_dim)) self.routed_scaling_factor = config.routed_scaling_factor self.use_expert_bias = config.use_expert_bias self.budget = config.budget self.min_dynamic_k = 4 self.max_dynamic_k = min(4, self.num_experts) self.layer_idx = layer_idx # ---- EXPERT ACTIVATION STATISTICS ---- self.collect_stats = False self._stats_sum = 0.0 # sum of dynamic_k values self._stats_sum_sq = 0.0 # sum of squared dynamic_k values self._stats_count = 0 # total tokens seen # -------------------------------------- @staticmethod def inverse_simpson_ratio(probs, dim=-1, eps=1e-12): """ISR = 1 / sum(p_i^2). Measures effective number of choices.""" return 1.0 / (probs.pow(2).sum(dim=dim) + eps) def forward(self, hidden_states, expert_bias=None): router_logits = F.linear(hidden_states, self.weight) # ---- DYNAMIC TOP-K via ISR ---- std = router_logits.std(dim=-1, keepdim=True) # norm_logits = router_logits / (std + 1e-12) norm_logits = router_logits probs = F.softmax(norm_logits, dim=-1) isr = self.inverse_simpson_ratio(probs, dim=-1) dynamic_k = torch.ceil(isr * self.budget).long().clamp( self.min_dynamic_k, self.max_dynamic_k ) # ------------------------------- # ---- RECORD STATS (zero overhead when disabled) ---- if self.collect_stats: # dynamic_k shape: [batch, seq] or [batch*seq] k_vals = dynamic_k.detach().cpu().float() self._stats_sum += k_vals.sum().item() self._stats_sum_sq += (k_vals ** 2).sum().item() self._stats_count += k_vals.numel() # ---------------------------------------------------- routing_weights = router_logits.sigmoid() batch_max_k = int(dynamic_k.max().item()) batch_max_k = max(batch_max_k, self.min_dynamic_k) batch_max_k = min(batch_max_k, self.max_dynamic_k) if self.use_expert_bias: scores_for_routing = routing_weights + expert_bias _, selected_experts = torch.topk(scores_for_routing, k=batch_max_k, dim=-1) gathered_weights = torch.gather( routing_weights, dim=1, index=selected_experts ).type_as(router_logits) else: gathered_weights, selected_experts = torch.topk( routing_weights, k=batch_max_k, dim=-1 ) positions = torch.arange(batch_max_k, device=hidden_states.device).unsqueeze(0) valid_mask = positions < dynamic_k.unsqueeze(1) dummy_idx = self.num_experts selected_experts = torch.where( valid_mask, selected_experts, torch.full_like(selected_experts, dummy_idx) ) gathered_weights = torch.where( valid_mask, gathered_weights, torch.zeros_like(gathered_weights) ) if self.norm_topk_prob: sum_weights = gathered_weights.sum(dim=-1, keepdim=True) gathered_weights = gathered_weights / (sum_weights + 1e-6) routing_weights = gathered_weights * self.routed_scaling_factor return router_logits, routing_weights, selected_experts class Lfm2MoeDynamicExperts(Lfm2MoeExperts): """ Same as Lfm2MoeExperts but skips dummy expert indices (== num_experts) so that masked-out top-k slots do not trigger wasted MLP compute. """ def __init__(self, config): super().__init__(config) def forward(self, hidden_states, top_k_index, top_k_weights): final_hidden_states = torch.zeros_like(hidden_states) dummy = self.num_experts with torch.no_grad(): expert_mask = torch.nn.functional.one_hot( top_k_index, num_classes=self.num_experts + 1 ) expert_mask = expert_mask.permute(2, 1, 0) expert_hit = torch.greater(expert_mask.sum(dim=(-1, -2)), 0).nonzero() for expert_idx in expert_hit: expert_idx = expert_idx[0] if expert_idx == dummy: continue top_k_pos, token_idx = torch.where(expert_mask[expert_idx]) current_state = hidden_states[token_idx] gate, up = nn.functional.linear( current_state, self.gate_up_proj[expert_idx] ).chunk(2, dim=-1) current_hidden_states = self.act_fn(gate) * up current_hidden_states = nn.functional.linear( current_hidden_states, self.down_proj[expert_idx] ) current_hidden_states = current_hidden_states * top_k_weights[ token_idx, top_k_pos, None ] final_hidden_states.index_add_( 0, token_idx, current_hidden_states.to(final_hidden_states.dtype) ) return final_hidden_states class Lfm2MoeCustomSparseMoeBlock(Lfm2MoeSparseMoeBlock): def __init__(self, config: Lfm2MoeCustomConfig, layer_idx: int): nn.Module.__init__(self) self.experts = Lfm2MoeDynamicExperts(config) self.gate = Lfm2MoeDynamicTopKRouter(config, layer_idx=layer_idx) self.use_expert_bias = config.use_expert_bias if self.use_expert_bias: self.register_buffer( "expert_bias", torch.zeros(config.num_experts, dtype=torch.float32) ) class Lfm2MoeCustomForCausalLM(Lfm2MoeForCausalLM): config_class = Lfm2MoeCustomConfig def __init__(self, config: Lfm2MoeCustomConfig): super().__init__(config) for layer_idx, layer in enumerate(self.model.layers): if layer_idx >= config.num_dense_layers: layer.feed_forward = Lfm2MoeCustomSparseMoeBlock(config, layer_idx) # ------------------------------------------------------------------ # EXPERT ACTIVATION STATISTICS API # ------------------------------------------------------------------ def enable_expert_stats(self): """Start collecting per-token expert activation counts.""" for layer in self.model.layers: if hasattr(layer, "feed_forward") and hasattr(layer.feed_forward, "gate"): layer.feed_forward.gate.collect_stats = True def disable_expert_stats(self): """Stop collecting expert activation counts.""" for layer in self.model.layers: if hasattr(layer, "feed_forward") and hasattr(layer.feed_forward, "gate"): layer.feed_forward.gate.collect_stats = False def reset_expert_stats(self): """Clear all accumulated statistics.""" for layer in self.model.layers: if hasattr(layer, "feed_forward") and hasattr(layer.feed_forward, "gate"): gate = layer.feed_forward.gate gate._stats_sum = 0.0 gate._stats_sum_sq = 0.0 gate._stats_count = 0 def get_expert_activation_stats(self): """Return mean/std of experts activated per token, globally and per layer. Returns: dict with keys: - global: {mean_experts_per_token, std_experts_per_token, total_tokens_processed} - per_layer: {layer_0: {mean, std, count}, ...} """ per_layer = {} global_sum = 0.0 global_sum_sq = 0.0 global_count = 0 for layer_idx, layer in enumerate(self.model.layers): if hasattr(layer, "feed_forward") and hasattr(layer.feed_forward, "gate"): gate = layer.feed_forward.gate if gate._stats_count > 0: mean = gate._stats_sum / gate._stats_count mean_sq = gate._stats_sum_sq / gate._stats_count # population std std = max(0.0, mean_sq - mean ** 2) ** 0.5 per_layer[f"layer_{layer_idx}"] = { "mean": round(mean, 4), "std": round(std, 4), "count": int(gate._stats_count), } global_sum += gate._stats_sum global_sum_sq += gate._stats_sum_sq global_count += gate._stats_count global_mean = global_sum / global_count if global_count > 0 else 0.0 global_mean_sq = global_sum_sq / global_count if global_count > 0 else 0.0 global_std = max(0.0, global_mean_sq - global_mean ** 2) ** 0.5 if global_count > 0 else 0.0 return { "global": { "mean_experts_per_token": round(global_mean, 4), "std_experts_per_token": round(global_std, 4), "total_tokens_processed": int(global_count), }, "per_layer": per_layer, }