1212# See the License for the specific language governing permissions and
1313# limitations under the License.
1414
15- from megatron .bridge .models .conversion .param_mapping import (
16- MegatronParamMapping ,
17- ReplicatedMapping ,
18- )
15+ import re
16+
17+ import torch
18+
19+ from megatron .bridge .models .conversion .param_mapping import MegatronParamMapping , ReplicatedMapping
1920
2021
2122class AmaxMapping (ReplicatedMapping ):
2223 """Amax mapping for quantization."""
2324
24- def __init__ (self , megatron_param : str , hf_param : str ):
25+ def __init__ (self , megatron_param : str , hf_param : str | dict [ str , str ] ):
2526 """Initialize the Amax mapping."""
2627 super ().__init__ (megatron_param , hf_param )
2728 self .allow_hf_name_mismatch = True
@@ -77,8 +78,161 @@ def resolve(self, captures: tuple[str, ...]):
7778 return new_mapping
7879
7980
81+ class MoeAmaxFanoutMapping (AmaxMapping ):
82+ """Shared MoE amax mapping that fans out to per-expert HF quantizers.
83+
84+ Megatron grouped-MoE layers use one quantizer for each rank's local expert
85+ block, while HF names carry an expert wildcard. This mapping gathers those
86+ per-EP-rank amax values and expands the HF expert wildcard during export.
87+ """
88+
89+ _EXPERT_WILDCARD_RE = re .compile (r"(\.experts\.)(\*)(\.)" )
90+
91+ def __init__ (
92+ self ,
93+ megatron_param : str ,
94+ hf_patterns : list [str ],
95+ num_experts : int | None = None ,
96+ ) -> None :
97+ assert hf_patterns , "hf_patterns must be non-empty"
98+ self .hf_patterns = hf_patterns
99+ self .num_experts = num_experts
100+ super ().__init__ (megatron_param , {})
101+
102+ def _validate_patterns (self ) -> None :
103+ """Allow one extra HF wildcard for the expert index."""
104+ return
105+
106+ @property
107+ def is_expert (self ) -> bool :
108+ """Use normal TP handling; EP fanout is handled explicitly here."""
109+ return False
110+
111+ def hf_to_megatron (self , hf_weights , megatron_module ):
112+ """Grouped-MoE amax fanout is export-only."""
113+ return None
114+
115+ def _get_num_experts (self , megatron_module : object | None ) -> int | None :
116+ if self .num_experts is not None :
117+ return self .num_experts
118+
119+ config = getattr (megatron_module , "config" , None )
120+ for source in (config , megatron_module ):
121+ if source is None :
122+ continue
123+ for attr in ("num_moe_experts" , "num_experts" , "n_routed_experts" ):
124+ value = getattr (source , attr , None )
125+ if value is not None :
126+ return value
127+ return None
128+
129+ @classmethod
130+ def _resolve_pattern (
131+ cls ,
132+ pattern : str ,
133+ captures : tuple [str , ...],
134+ max_captures : int ,
135+ ) -> str :
136+ resolved = pattern
137+ capture_index = 0
138+ while "**" in resolved and capture_index < len (captures ) and capture_index < max_captures :
139+ resolved = resolved .replace ("**" , captures [capture_index ], 1 )
140+ capture_index += 1
141+ while "*" in resolved and capture_index < len (captures ) and capture_index < max_captures :
142+ resolved = resolved .replace ("*" , captures [capture_index ], 1 )
143+ capture_index += 1
144+ return resolved
145+
146+ def _get_num_experts_for_rank (self , megatron_module : object | None ) -> int | None :
147+ if self .num_experts is not None :
148+ return self .num_experts
149+
150+ if megatron_module is None :
151+ return self .broadcast_obj_from_pp_rank (None , cache_key = f"{ self .megatron_param } :num_experts" )
152+
153+ num_experts = self ._get_num_experts (megatron_module )
154+ return self .broadcast_obj_from_pp_rank (num_experts , cache_key = f"{ self .megatron_param } :num_experts" )
155+
156+ def _gather_amax_by_ep_rank (self , weight : torch .Tensor ) -> list [torch .Tensor ]:
157+ if self .ep_size == 1 :
158+ return [weight ]
159+
160+ gathered_weights = [torch .empty_like (weight ) for _ in range (self .ep_size )]
161+ torch .distributed .all_gather (gathered_weights , weight , group = self .ep_group )
162+ return gathered_weights
163+
164+ def megatron_to_hf (
165+ self ,
166+ megatron_weights : torch .Tensor | None ,
167+ megatron_module : object | None ,
168+ ) -> dict [str , torch .Tensor ]:
169+ base = super ().megatron_to_hf (megatron_weights , megatron_module )
170+ if not base :
171+ return {}
172+ weight = next (iter (base .values ()))
173+
174+ num_experts = self ._get_num_experts_for_rank (megatron_module )
175+ if num_experts is None or num_experts <= 0 :
176+ raise RuntimeError (
177+ f"Could not determine num_experts for { self .megatron_param } . "
178+ "Expected megatron_module.config.num_moe_experts or num_experts."
179+ )
180+ if num_experts % self .ep_size != 0 :
181+ raise RuntimeError (
182+ f"num_experts ({ num_experts } ) must be divisible by EP size ({ self .ep_size } ) for { self .megatron_param } ."
183+ )
184+
185+ result = {}
186+ weights_by_ep_rank = self ._gather_amax_by_ep_rank (weight )
187+ experts_per_rank = num_experts // self .ep_size
188+ for pattern in self .hf_patterns :
189+ if self ._EXPERT_WILDCARD_RE .search (pattern ):
190+ for ep_rank , ep_weight in enumerate (weights_by_ep_rank ):
191+ expert_start = ep_rank * experts_per_rank
192+ expert_stop = expert_start + experts_per_rank
193+ for expert_idx in range (expert_start , expert_stop ):
194+ hf_name = self ._EXPERT_WILDCARD_RE .sub (
195+ rf"\g<1>{ expert_idx } \g<3>" ,
196+ pattern ,
197+ count = 1 ,
198+ )
199+ result [hf_name ] = ep_weight
200+ else :
201+ result [pattern ] = weight
202+ return result
203+
204+ def resolve (self , captures : tuple [str , ...]) -> "MoeAmaxFanoutMapping" :
205+ """Resolve layer wildcards while preserving the HF expert wildcard."""
206+ megatron_wildcards = self ._count_wildcard_groups (self .megatron_param )
207+ resolved_megatron_param = self ._resolve_pattern (
208+ self .megatron_param ,
209+ captures ,
210+ megatron_wildcards ,
211+ )
212+ resolved_hf_patterns = [
213+ self ._resolve_pattern (pattern , captures , megatron_wildcards ) for pattern in self .hf_patterns
214+ ]
215+ new_mapping = type (self )(
216+ resolved_megatron_param ,
217+ resolved_hf_patterns ,
218+ self .num_experts ,
219+ )
220+ new_mapping .allow_hf_name_mismatch = self .allow_hf_name_mismatch
221+ return new_mapping
222+
223+
224+ def _convert_hf_weight_names (hf_param : str | dict [str , str ], mapped_name : str ) -> list [str ]:
225+ if isinstance (hf_param , dict ):
226+ return [
227+ value .removesuffix (".weight" ) + mapped_name for value in hf_param .values () if value .endswith (".weight" )
228+ ]
229+ if isinstance (hf_param , str ) and hf_param .endswith (".weight" ):
230+ return [hf_param .removesuffix (".weight" ) + mapped_name ]
231+ return []
232+
233+
80234def convert_to_amax_map (
81- mappings : list [MegatronParamMapping ], mapped_name = ".weight_quantizer._amax"
235+ mappings : list [MegatronParamMapping ], mapped_name : str = ".weight_quantizer._amax"
82236) -> list [MegatronParamMapping ]:
83237 """Convert weight mappings to amax mappings for quantization.
84238
@@ -93,26 +247,40 @@ def convert_to_amax_map(
93247 List of new MegatronParamMapping objects for amax parameters
94248
95249 Note:
96- Only mappings with parameter names ending in '.weight' are converted.
97- Other mappings are ignored.
250+ Mappings ending in '.weight' become regular amax mappings. MoE expert
251+ mappings ending in '.weight*' become fanout mappings because Megatron
252+ stores a shared expert amax while HF stores per-expert amax names.
98253 """
99254 extended_mapping = []
100255
101256 for mapping in mappings :
257+ if mapping .megatron_param .endswith (".weight*" ):
258+ new_megatron_param = mapping .megatron_param [: - len (".weight*" )] + mapped_name
259+ hf_patterns = _convert_hf_weight_names (mapping .hf_param , mapped_name )
260+
261+ if hf_patterns :
262+ extended_mapping .append (
263+ MoeAmaxFanoutMapping (
264+ megatron_param = new_megatron_param ,
265+ hf_patterns = hf_patterns ,
266+ )
267+ )
268+ continue
269+
102270 if not mapping .megatron_param .endswith (".weight" ):
103271 continue
104272
105- new_megatron_param = mapping .megatron_param .replace (".weight" , mapped_name )
273+ new_megatron_param = mapping .megatron_param .removesuffix (".weight" ) + mapped_name
106274
107275 if isinstance (mapping .hf_param , dict ):
108276 # For dict-based hf_param (e.g., QKVMapping, GatedMLPMapping)
109277 new_hf_param = {
110- key : (value .replace (".weight" , mapped_name ) if value .endswith (".weight" ) else value )
278+ key : (value .removesuffix (".weight" ) + mapped_name if value .endswith (".weight" ) else value )
111279 for key , value in mapping .hf_param .items ()
112280 }
113281 elif isinstance (mapping .hf_param , str ):
114282 if mapping .hf_param .endswith (".weight" ):
115- new_hf_param = mapping .hf_param .replace (".weight" , mapped_name )
283+ new_hf_param = mapping .hf_param .removesuffix (".weight" ) + mapped_name
116284 else :
117285 continue
118286 else :
0 commit comments