recipes/docker-images/glm53-kindling/experimental/snapshot/base_loader.py

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import os
from abc import ABC, abstractmethod

import torch
import torch.nn as nn

import vllm.envs as envs
from vllm.config import ModelConfig, VllmConfig
from vllm.config.load import LoadConfig
from vllm.logger import init_logger
from vllm.model_executor.model_loader.reload import finalize_layerwise_processing
from vllm.model_executor.model_loader.utils import (
    initialize_model,
    process_weights_after_loading,
)
from vllm.platforms import current_platform
from vllm.tracing import instrument
from vllm.utils.mem_utils import format_gib
from vllm.utils.torch_utils import set_default_torch_dtype

logger = init_logger(__name__)


def _optional(module: str, *names: str):
    """`module`, or a stand-in whose `names` do nothing when it is not installed."""
    import importlib
    from types import SimpleNamespace

    try:
        return importlib.import_module(module)
    except ModuleNotFoundError as e:
        if e.name != module:
            raise
        return SimpleNamespace(**{n: (lambda *args, **kwargs: None) for n in names})


class BaseModelLoader(ABC):
    """Base class for model loaders."""

    def __init__(self, load_config: LoadConfig):
        self.load_config = load_config

    @abstractmethod
    def download_model(self, model_config: ModelConfig) -> None:
        """Download a model so that it can be immediately loaded."""
        raise NotImplementedError

    @abstractmethod
    def load_weights(self, model: nn.Module, model_config: ModelConfig) -> None:
        """Load weights into a model. This standalone API allows
        inplace weights loading for an already-initialized model"""
        raise NotImplementedError

    def create_model(
        self, vllm_config: VllmConfig, model_config: ModelConfig, prefix: str = ""
    ) -> nn.Module:
        """Create a model with the given configurations."""
        model = initialize_model(
            vllm_config=vllm_config,
            model_config=model_config,
            prefix=prefix,
        )
        log_online_quantization(vllm_config)
        log_model_inspection(model)
        return model

    @instrument(span_name="Load model")
    def load_model(
        self, vllm_config: VllmConfig, model_config: ModelConfig, prefix: str = ""
    ) -> nn.Module:
        """Load a model with the given configurations."""
        # Local weight snapshot (weight_snapshot.py): rebuild from it when
        # this rank has one, otherwise load normally and write one.
        # fp8.yaml and megamoe.yaml mount these; without them their steps do nothing.
        dense_fp8 = _optional("vllm.model_executor.model_loader.dense_fp8", "convert", "simulate_nvfp4")
        megamoe_vllm = _optional("vllm.model_executor.layers.fused_moe.megamoe_vllm", "install")
        from vllm.model_executor.model_loader.weight_snapshot import Snapshot

        snapshot = Snapshot.for_model(vllm_config, model_config)
        normal = os.environ.get("VLLM_WEIGHT_SNAPSHOT_NORMAL") == "1"  # diagnostic: load, dump, no save
        if snapshot is not None and snapshot.complete() and not normal:
            # The normal path on dummy weights, then every tensor overwritten
            # in place: the object graph is exactly a real load's.
            from vllm.model_executor.model_loader.dummy_loader import DummyModelLoader

            model = self._load_from_checkpoint(
                vllm_config, model_config, prefix,
                weights_from=DummyModelLoader(self.load_config).load_weights,
            )
            dense_fp8.convert(model)
            snapshot.restore(model)
            # What a model's own load_weights does after loading, which the
            # dummy loader skips: the DFlash drafter fuses its KV weights.
            for m in model.modules():
                if hasattr(m, "_build_fused_kv_buffers"):
                    m._build_fused_kv_buffers()
            if os.environ.get("VLLM_WEIGHT_SNAPSHOT_VERIFY") == "1":
                from vllm.model_executor.model_loader.weight_snapshot import verify

                # The reference registers the same layer names; keep the
                # snapshot model's registrations and discard its.
                cc = vllm_config.compilation_config
                kept = {k: (dict(getattr(cc, k)) if isinstance(getattr(cc, k), dict) else list(getattr(cc, k)))
                        for k in ("static_forward_context", "static_all_moe_layers") if hasattr(cc, k)}
                for k in kept:
                    getattr(cc, k).clear()
                reference = self._load_from_checkpoint(vllm_config, model_config, prefix)
                for k, v in kept.items():
                    getattr(cc, k).clear()
                    getattr(cc, k).update(v) if isinstance(v, dict) else getattr(cc, k).extend(v)
                verify(model, reference)
                # Diagnostic only: never serve the model under test.
                logger.warning("Weight snapshot verify done; this worker now idles. Take the stack down.")
                import time

                while True:
                    time.sleep(3600)
            self._dump_state(model, vllm_config, snapshot, "snapshot")
            dense_fp8.simulate_nvfp4(model)
            megamoe_vllm.install(model)
            return model
        model = self._load_from_checkpoint(vllm_config, model_config, prefix)
        dense_fp8.convert(model)
        for m in model.modules():  # the drafter fuses its KV weights from whatever format they are now in
            if hasattr(m, "_build_fused_kv_buffers"):
                m._build_fused_kv_buffers()
        if snapshot is not None:
            self._dump_state(model, vllm_config, snapshot, "normal")
            if not normal:
                snapshot.save(model)
        dense_fp8.simulate_nvfp4(model)
        megamoe_vllm.install(model)
        return model

    @staticmethod
    def _dump_state(model, vllm_config, snapshot, mode):
        out = os.environ.get("VLLM_WEIGHT_SNAPSHOT_DUMP")
        if out and snapshot is not None:
            from vllm.model_executor.model_loader.weight_snapshot import dump_state

            name = os.path.basename(snapshot.path).rsplit("-", 1)[0]
            dump_state(model, vllm_config, os.path.join(out, f"{mode}-{name}.txt"))

    def _load_from_checkpoint(
        self, vllm_config: VllmConfig, model_config: ModelConfig, prefix: str = "", weights_from=None
    ) -> nn.Module:
        device_config = vllm_config.device_config
        load_config = vllm_config.load_config
        load_device = (
            device_config.device if load_config.device is None else load_config.device
        )
        target_device = torch.device(load_device)
        with set_default_torch_dtype(model_config.dtype):
            with target_device:
                model = self.create_model(
                    vllm_config=vllm_config,
                    model_config=model_config,
                    prefix=prefix,
                )

            logger.debug("Loading weights on %s ...", load_device)
            (weights_from or self.load_weights)(model, model_config)

            # Log peak GPU memory after loading weights. This is needed
            # to have test coverage on peak memory for online quantization.
            if current_platform.is_cuda_alike() or current_platform.is_xpu():
                peak_memory = torch.accelerator.max_memory_allocated()
                logger.debug_once(
                    "Peak GPU memory after loading weights: %s GiB",
                    format_gib(peak_memory),
                )

            # Process weights into kernel format. Note that when using online
            # quantization, weights are (typically) quantized as they are loaded.
            if _has_online_quant(model):
                finalize_layerwise_processing(model, model_config)

            process_weights_after_loading(model, model_config, target_device)

        return model.eval()


def log_model_inspection(model: nn.Module) -> None:
    """Log model structure if VLLM_LOG_MODEL_INSPECTION=1."""
    if not envs.VLLM_LOG_MODEL_INSPECTION:
        return

    from vllm.model_inspection import format_model_inspection

    logger.info("vLLM model structure:\n%s", format_model_inspection(model))


def log_online_quantization(vllm_config: VllmConfig) -> None:
    """Log the online-quantized layer count and types, when applicable."""
    from vllm.model_executor.layers.quantization.online.base import (
        OnlineQuantizationConfig,
    )

    quant_config = vllm_config.quant_config
    online_quantization_config = getattr(
        quant_config, "online_quantization_config", None
    )
    if isinstance(online_quantization_config, OnlineQuantizationConfig):
        quant_config = online_quantization_config
    if not isinstance(quant_config, OnlineQuantizationConfig):
        return

    logger.info(
        "Quantized %d layers of types: %s",
        len(quant_config.quantized_layers),
        "; ".join(quant_config.quantized_layer_summaries),
    )


def _has_online_quant(model: nn.Module):
    for module in model.modules():
        quant_method = getattr(module, "quant_method", None)
        if getattr(quant_method, "uses_meta_device", False):
            return True

    return False