reference/code/dMoE/training/scripts/moe_convertor.py

import os
import re
from argparse import ArgumentParser
from dataclasses import dataclass
from glob import glob
from typing import Generator

import torch
from safetensors.torch import safe_open
from tqdm import tqdm
from transformers import AutoConfig

from veomni.models import build_tokenizer, save_model_weights


@dataclass
class StateDictIterator:
    filepath: str

    def __iter__(self) -> Generator[tuple[str, torch.Tensor], None, None]:
        if self.filepath.endswith(".safetensors"):
            with safe_open(self.filepath, framework="pt", device="cpu") as f:
                for key in f.keys():
                    yield key, f.get_tensor(key)

        else:
            state_dict = torch.load(self.filepath, map_location="cpu", weights_only=True, mmap=True)
            for key in state_dict.keys():
                yield key, state_dict[key]

def moe_merge(state_dict: dict[str, torch.Tensor], config) -> dict[str, torch.Tensor]:
    new_state_dict: dict[str, torch.Tensor] = dict()
    processed_keys: set[str] = set()        

    num_layers = config.num_hidden_layers
    num_experts = config.num_experts
    first_k_dense_replace = config.first_k_dense_replace

    print(f"Merging {num_layers} layers with {num_experts} experts each")
    proj_types = ["gate_proj", "up_proj", "down_proj"]

    for layer_id in range(first_k_dense_replace, num_layers):
        for proj_type in proj_types:
            expert_weights = []
            current_expert_keys = []                

            for expert_id in range(num_experts):
                expert_key = f"model.layers.{layer_id}.mlp.experts.{expert_id}.{proj_type}.weight"
                assert expert_key in state_dict, f"Missing key: {expert_key}"
                expert_weights.append(state_dict[expert_key])
                current_expert_keys.append(expert_key)

            assert len(expert_weights) == num_experts
            merged_weight = torch.stack(expert_weights, dim=0)
            new_key = f"model.layers.{layer_id}.mlp.experts.{proj_type}"
            new_state_dict[new_key] = merged_weight
            processed_keys.update(current_expert_keys)
            for key in current_expert_keys:
                del state_dict[key]
            print(f"✓ Layer {layer_id}.{proj_type}: {expert_weights[0].shape} -> {merged_weight.shape}")                
            del expert_weights


    for key, tensor in state_dict.items():
        if key not in processed_keys:
            new_state_dict[key] = tensor
    return new_state_dict

def split_moe_experts(
    merged_state_dict: dict[str, torch.Tensor], config
) -> dict[str, torch.Tensor]:
    split_state_dict: dict[str, torch.Tensor] = dict()

    num_experts = getattr(config, "num_experts", None)
    if num_experts is None:
        raise ValueError(
            "Could not find the 'num_experts' attribute in the configuration. "
            "Please ensure the correct model configuration is passed."
        )

    proj_types = ["gate_proj", "up_proj", "down_proj"]

    merged_key_pattern = re.compile(
        r"model\.layers\.(\d+)\.mlp\.experts\.(" + "|".join(proj_types) + r")$"
    )

    for key, merged_tensor in merged_state_dict.items():
        match = merged_key_pattern.match(key)

        if match:
            layer_id = match.group(1)
            proj_type = match.group(2)

            if not (merged_tensor.dim() > 1 and merged_tensor.shape[0] == num_experts):
                raise ValueError(
                    f"Tensor '{key}' has an unexpected shape {merged_tensor.shape}. "
                    f"Its first dimension should be equal to the number of experts ({num_experts})."
                )

            for expert_id in range(num_experts):
                expert_tensor = merged_tensor[expert_id]
                original_key = f"model.layers.{layer_id}.mlp.experts.{expert_id}.{proj_type}.weight"
                split_state_dict[original_key] = expert_tensor

            print(f"Unmerged {key} -> {num_experts} individual expert weights")
        else:
            split_state_dict[key] = merged_tensor

    return split_state_dict


def main(input_path, output_path):
    torch.set_default_dtype(torch.bfloat16)
    os.makedirs(output_path, exist_ok=True)

    config = AutoConfig.from_pretrained(input_path, trust_remote_code=True)
    tokenizer = build_tokenizer(input_path)

    safetensor_files = list(glob(os.path.join(input_path, "*.safetensors")))
    safetensor_files.sort()
    state_dict_iterators = [StateDictIterator(shard_file) for shard_file in safetensor_files]

    state_dict = {}
    for state_dict_iterator in tqdm(state_dict_iterators, desc="Loading checkpoint shards"):
        for name, tensor in state_dict_iterator:
            state_dict[name] = tensor.cpu()

    if args.mode == "merge":
        new_state_dict = moe_merge(state_dict, config)
    elif args.mode == "split":
        new_state_dict = split_moe_experts(state_dict, config)
    else:
        raise ValueError("unsupport mode")
    
    state_dict.clear()
    model_assets = [config, tokenizer]
    save_model_weights(output_path, new_state_dict, model_assets=model_assets)


if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("-i", "--input-path", type=str, required=True)
    parser.add_argument("-o", "--output-path", type=str, required=True)
    parser.add_argument("-m", "--mode", type=str, default="merge", choices=["merge", "split"])
    args = parser.parse_args()
    main(args.input_path, args.output_path)