diff --git a/megatron/core/transformer/spec_utils.py b/megatron/core/transformer/spec_utils.py index b3de8541734..160cad10ed7 100644 --- a/megatron/core/transformer/spec_utils.py +++ b/megatron/core/transformer/spec_utils.py @@ -1,8 +1,11 @@ # Copyright (c) 2023, NVIDIA CORPORATION. All rights reserved. +import logging import types from dataclasses import dataclass, field -from typing import Tuple, Union +from typing import Any, Tuple, Union + +logger = logging.getLogger(__name__) @dataclass @@ -24,7 +27,16 @@ class ModuleSpec: module: Union[Tuple, type] params: dict = field(default_factory=lambda: {}) - submodules: type = None + submodules: object = None + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + """Builds an instance of the module from the spec. + + Args: + *args: Positional arguments to be passed to the module init. + **kwargs: Keyword arguments to be passed to the module init. + """ + return build_module(self, *args, **kwargs) def import_module(module_path: Tuple[str]): @@ -37,12 +49,13 @@ def import_module(module_path: Tuple[str]): try: module = __import__(base_path, globals(), locals(), [name]) except ImportError as e: - print(f"couldn't import module due to {e}") + logger.error(f"couldn't import module due to {e}") return None return vars(module)[name] def get_module(spec_or_module: Union[ModuleSpec, type], **additional_kwargs): + """Returns or imports the provided module.""" # If a module clas is already provided return it as is if isinstance(spec_or_module, (type, types.FunctionType)): return spec_or_module @@ -56,6 +69,13 @@ def get_module(spec_or_module: Union[ModuleSpec, type], **additional_kwargs): def build_module(spec_or_module: Union[ModuleSpec, type], *args, **kwargs): + """Builds an instance of the module from the spec. + + Args: + spec_or_module: The module spec or module class to build. + *args: Positional arguments to be passed to the module init. + **kwargs: Keyword arguments to be passed to the module init. + """ # If the passed `spec_or_module` is # a `Function`, then return it as it is # NOTE: to support an already initialized module add the following condition diff --git a/megatron/core/typed_torch.py b/megatron/core/typed_torch.py new file mode 100644 index 00000000000..dd14eabd1c1 --- /dev/null +++ b/megatron/core/typed_torch.py @@ -0,0 +1,34 @@ +# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""Utilities for improved type hinting with torch interfaces.""" + +from collections.abc import Callable +from typing import Generic, ParamSpec, Protocol, TypeVar + +import torch + +P = ParamSpec('P') +R_co = TypeVar('R_co', covariant=True) + + +class _Module(Generic[P, R_co], Protocol): + """Protocol allowing us to unwrap `forward`.""" + + def forward(self, *args: P.args, **kwargs: P.kwargs) -> R_co: + """Forward method of the matching torch.nn.Module.""" + ... + + +def apply_module(m: _Module[P, R_co], *, check_subclass: bool = True) -> Callable[P, R_co]: + """Returns the provided module unchanged, but with correct type hints. + + Args: + m: An instance of a subclass of `torch.nn.Module`. + check_subclass: If `True`, checks that `m` is a subclass of + `torch.nn.Module` and raises a `TypeError` if not. + + Returns: + That module unchanged, but with correct type hints. + """ + if check_subclass and not issubclass(type(m), torch.nn.Module): + raise TypeError(f'{type(m)} is not a subclass of torch.nn.Module') + return m # type: ignore diff --git a/tests/unit_tests/dist_checkpointing/models/test_bert_model.py b/tests/unit_tests/dist_checkpointing/models/test_bert_model.py index 27f01447851..81b01c8f886 100644 --- a/tests/unit_tests/dist_checkpointing/models/test_bert_model.py +++ b/tests/unit_tests/dist_checkpointing/models/test_bert_model.py @@ -24,13 +24,11 @@ def initialize_bert_model( - seed, layer_spec_fn=bert_layer_with_transformer_engine_spec, vocab_size=128, **config_kwargs + seed, layer_spec=bert_layer_with_transformer_engine_spec, vocab_size=128, **config_kwargs ): torch.manual_seed(seed) model_parallel_cuda_manual_seed(seed) - layer_spec = layer_spec_fn() if callable(layer_spec_fn) else layer_spec_fn - default_config_kwargs = dict( num_layers=8, hidden_size=16,