MindIE-LLM AclGraph 模型迁移指南¶

1. 核心交付件 (必需)
   ├─ Router (router_my_model.py)
   ├─ Config (config_my_model.py)
   └─ Model (my_model.py)

2. 扩展实现 (可选)
   ├─ InputBuilder (input_builder_my_model.py)
   └─ ToolCallsProcessor (tool_calls_processor_my_model.py)

3. 测试验证
   └─ 加载模型权重并运行推理

mindie_llm/runtime/models/my_model/
├── __init__.py                         # 模块初始化
├── router_my_model.py                  # Router (必需)
├── config_my_model.py                  # Config (必需)
├── my_model.py                         # Model (必需)
├── input_builder_my_model.py           # InputBuilder (可选)
└── tool_calls_processor_my_model.py    # ToolCallsProcessor (可选)

1. BaseRouter.__post_init__()
   └─ 从 config_dict 获取 model_type

2. 访问 router.config 属性
   └─ 调用 _get_config_cls() -> 动态导入 Config 类
   └─ 创建配置实例

3. 访问 router.tokenizer 属性
   └─ 调用 _get_tokenizer() -> 加载 HuggingFace tokenizer

4. 访问 router.input_builder 属性
   └─ 调用 _get_input_builder() -> 创建 InputBuilder

5. 访问 router.model_cls 属性
   └─ 调用 _get_model_cls() -> 动态导入 Model 类

# router_my_model.py
from dataclasses import dataclass
from mindie_llm.runtime.models.base.router import BaseRouter

@dataclass
class MyModelRouter(BaseRouter):
    """MyModel Router"""

    def _get_input_builder(self):
        """
        获取 InputBuilder

        如果模型需要特殊的输入格式（如自定义 chat_template），
        需要重写此方法返回自定义的 InputBuilder。
        默认返回 None，使用框架默认的 InputBuilder。
        """
        return None

    def _get_tool_calls_parser(self):
        """
        获取工具调用解析器名称

        如果模型支持 ToolsCall，需要重写此方法。
        默认不支持工具调用，返回 None。
        """
        return None

# config_my_model.py
from dataclasses import dataclass
from mindie_llm.runtime.config.huggingface_config import HuggingFaceConfig

@dataclass
class MyModelConfig(HuggingFaceConfig):
    """MyModel 配置类，继承自 HuggingFaceConfig，定义模型特有的配置项。"""

    use_qk_norm: bool = True  # 是否使用 QK 归一化，如果模型有额外参数配置可类似添加

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 如果模型有新的配置项，在这里处理

    def _create_rope_scaling(self, rope_scaling_dict, rope_theta, max_position_embeddings):
        """创建 RoPE 缩放配置"""
        return YourRopeScaling.from_dict(
            rope_scaling_dict,
            rope_theta=rope_theta,
            max_position_embeddings=max_position_embeddings
        )

MyModelForCausalLM (顶层，包含 LM Head)
├── MyModelModel (基础模型，包含 Embedding + Layers + Norm)
│   ├── VocabParallelEmbedding (词嵌入层)
│   ├── MyModelLayer × N (Transformer 层)
│   │   ├── MyModelAttention (注意力层)
│   │   │   ├── QKVParallelLinear (QKV 投影)
│   │   │   ├── RotaryEmbedding (旋转位置编码)
│   │   │   ├── Attention (注意力计算)
│   │   │   └── RowParallelLinear (输出投影)
│   │   ├── MyModelMoe 或 MyModelMlp (MoE 或 Dense FFN)
│   │   │   ├── FusedMoE (MoE 专家网络，可选)
│   │   │   │   ├── Gate 投影
│   │   │   │   ├── Up 投影
│   │   │   │   └── Down 投影
│   │   │   ├── Gate (路由网络，MoE)
│   │   │   ├── Shared Experts (共享专家，可选)
│   │   │   └── 或 Dense MLP (Gate + Up + Down 投影)
│   │   ├── RMSNorm (输入归一化)
│   │   └── RMSNorm (注意力后归一化)
│   └── RMSNorm (最终归一化)
└── ParallelLMHead (语言模型头)

class MyModelAttention(nn.Module):
    def __init__(self, config, prefix, quant_config=None):
        super().__init__()
        self.qkv_proj = QKVParallelLinear(..., prefix=[f"{prefix}.q_proj", f"{prefix}.k_proj", f"{prefix}.v_proj"])
        self.o_proj = RowParallelLinear(...)
        self.rope_emb = get_rope(...)
        self.attn = Attention(...)

    def forward(self, positions, hidden_states):
        ...

class MyModelMoe(nn.Module):
    def __init__(self, config, prefix, quant_config=None):
        super().__init__()
        # MoE: 专家网络 + 路由网络 + 可选的共享专家
        self.experts = FusedMoE(..., prefix=f"{prefix}.experts")
        self.gate = ReplicatedLinear(..., prefix=f"{prefix}.gate")
        self.shared_experts = MyModelMLP(...)  # 可选

    def forward(self, hidden_states):
        ...

class MyModelLayer(nn.Module):
    def __init__(self, config, prefix, layer_idx, quant_config=None):
        super().__init__()
        self.input_layernorm = RMSNorm(...)
        self.self_attn = MyModelAttention(..., prefix=f"{prefix}.self_attn")
        self.post_attention_layernorm = RMSNorm(...)
        # MoE 模型 or Dense 模型二选一
        self.mlp = MyModelMoe(..., prefix=f"{prefix}.mlp") # MOE
        self.mlp = MyModelMlp(..., prefix=f"{prefix}.mlp") # DENSE

    def forward(self, positions, hidden_states):
        ...

class MyModelModel(nn.Module):
    def __init__(self, config, prefix, quant_config=None):
        super().__init__()
        self.embed_tokens = VocabParallelEmbedding(...)
        self.layers = nn.ModuleList([
            MyModelLayer(..., prefix=f"{prefix}.layers.{i}")
            for i in range(config.num_hidden_layers)
        ])
        self.norm = RMSNorm(...)

    def forward(self, input_ids, positions):
        ...

class MyModelForCausalLM(BaseModelForCausalLM):
    def __init__(self, mindie_llm_config):
        super().__init__(mindie_llm_config)
        self.model = MyModelModel(..., prefix="model")
        self.lm_head = ParallelLMHead(..., prefix="lm_head")

    def forward(self, input_ids, positions, ...):
        return self.model(input_ids, positions)

    def compute_logits(self, hidden_states):
        ...

# input_builder_my_model.py
from mindie_llm.runtime.models.base.input_builder import InputBuilder

class MyModelInputBuilder(InputBuilder):
    """MyModel InputBuilder"""

    def __init__(self, tokenizer, **kwargs):
        super().__init__(tokenizer, **kwargs)

    def _apply_chat_template(self, conversation, tools_msg=None, **kwargs):
        """应用 Chat Template"""
        if not hasattr(self.tokenizer, "apply_chat_template"):
            raise RuntimeError("Tokenizer 不支持 apply_chat_template")
        return self.tokenizer.apply_chat_template(conversation, **kwargs)

# tool_calls_processor_my_model.py
import re
from mindie_llm.runtime.models.base.tool_calls_processor import (
    ToolCallsProcessorWithXml, ToolCallsProcessorManager
)

@ToolCallsProcessorManager.register_module(module_names=["my_model"])
class ToolCallsProcessorMyModel(ToolCallsProcessorWithXml):
    """MyModel ToolCallsProcessor"""

    def __init__(self, tokenizer=None):
        super().__init__(tokenizer)
        self._tool_calls_regex = re.compile(r'<tool_call\s*({.*?})\s*/>', re.DOTALL)

    @property
    def tool_call_start_token(self) -> str:
        return "<tool_call"

    @property
    def tool_call_end_token(self) -> str:
        return "/>"

    @property
    def tool_call_start_token_id(self) -> int:
        return self.tokenizer.convert_tokens_to_ids("<tool_call")

    @property
    def tool_call_end_token_id(self) -> int:
        return self.tokenizer.convert_tokens_to_ids("/>")

    @property
    def tool_call_regex(self):
        return self._tool_calls_regex