Skip to content

Model Orchastrator

Implemented a FactoryDesign pattern to create a pipeline based on the configuration provided. Implemented pipelines: 1. ModelPipeline 2. TuningPipeline

CustomPipeline

Bases: ABC, Pipeline

Source code in model_forge/model/model_orchastrator.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class CustomPipeline(ABC, Pipeline):
    def transform_without_predictor(self, X):
        """
        Transform the data without using the predictor step.

        Args:
            X (array-like): The input data to be transformed.

        Returns:
            array-like: The transformed data.
        """
        # Add your code here to transform the data
        return self[:-1].transform(X)

    @property
    def stepnames(self):
        return list(self.named_steps.keys())

    @staticmethod
    def create_from_config():
        ...

transform_without_predictor(X)

Transform the data without using the predictor step.

Parameters:

Name Type Description Default
X array - like

The input data to be transformed.

 required

Returns:

Type Description
Source code in model_forge/model/model_orchastrator.py 
152
153
154
155
156
157
158
159
160
161
162
163
def transform_without_predictor(self, X):
    """
    Transform the data without using the predictor step.

    Args:
        X (array-like): The input data to be transformed.

    Returns:
        array-like: The transformed data.
    """
    # Add your code here to transform the data
    return self[:-1].transform(X)

ModelOrchestrator

Bases: Orchestartor

Class representing the model orchestrator.

This class is responsible for creating a model pipeline and managing the features.

Attributes:

Name Type Description
cfg Config

The configuration object.

Methods:

Name Description
create_pipeline

Creates a model pipeline based on the configuration.
features_in

Returns the features specified in the configuration.
Source code in model_forge/model/model_orchastrator.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class ModelOrchestrator(Orchestartor):
    """
    Class representing the model orchestrator.

    This class is responsible for creating a model pipeline and managing the features.

    Attributes:
        cfg (Config): The configuration object.

    Methods:
        create_pipeline: Creates a model pipeline based on the configuration.
        features_in: Returns the features specified in the configuration.

    """

    def create_pipeline(self):
        """
        Creates a model pipeline based on the configuration.

        Returns:
            ModelPipeline: The created model pipeline.

        """
        return ModelPipeline.create_from_config(self.cfg)

    def features_in(self, cfg):
        """
        Returns the features specified in the configuration.

        Args:
            cfg (Config): The configuration object.

        Returns:
            list: The list of features.

        """
        return cfg.features

create_pipeline()

Creates a model pipeline based on the configuration.

Returns:

Name Type Description
ModelPipeline

The created model pipeline.
Source code in model_forge/model/model_orchastrator.py 
56
57
58
59
60
61
62
63
64
def create_pipeline(self):
    """
    Creates a model pipeline based on the configuration.

    Returns:
        ModelPipeline: The created model pipeline.

    """
    return ModelPipeline.create_from_config(self.cfg)

features_in(cfg)

Returns the features specified in the configuration.

Parameters:

Name Type Description Default
cfg Config

The configuration object.

 required

Returns:

Name Type Description
list

The list of features.
Source code in model_forge/model/model_orchastrator.py 
66
67
68
69
70
71
72
73
74
75
76
77
def features_in(self, cfg):
    """
    Returns the features specified in the configuration.

    Args:
        cfg (Config): The configuration object.

    Returns:
        list: The list of features.

    """
    return cfg.features

ModelPipeline

Bases: CustomPipeline

Custom pipeline class for defining a model pipeline.

Source code in model_forge/model/model_orchastrator.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
class ModelPipeline(CustomPipeline):
    """
    Custom pipeline class for defining a model pipeline.
    """

    @classmethod
    def create_from_config(cls, cfg: DictConfig | OmegaConf) -> "ModelPipeline":
        """
        Create a custom model pipeline from the provided configuration.

        Args:
            cfg (DictConfig | OmegaConf): The configuration object.

        Returns:
            ModelPipeline: The created custom model pipeline.
        """
        # First create list of tuples from the modelsteps list
        pipeline_list = []
        for i, step in enumerate(cfg.model.model_steps):
            _step_dict = list(step.values())[0]
            _step_name = list(step.keys())[0]
            pipeline_list.append((_step_name, instantiate(_step_dict)))

        # Create instance of cls
        custom_pipeline = cls(steps=pipeline_list)
        return custom_pipeline

create_from_config(cfg) classmethod

Create a custom model pipeline from the provided configuration.

Parameters:

Name Type Description Default
cfg DictConfig | OmegaConf

The configuration object.

 required

Returns:

Name Type Description
ModelPipeline ModelPipeline

The created custom model pipeline.
Source code in model_forge/model/model_orchastrator.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
@classmethod
def create_from_config(cls, cfg: DictConfig | OmegaConf) -> "ModelPipeline":
    """
    Create a custom model pipeline from the provided configuration.

    Args:
        cfg (DictConfig | OmegaConf): The configuration object.

    Returns:
        ModelPipeline: The created custom model pipeline.
    """
    # First create list of tuples from the modelsteps list
    pipeline_list = []
    for i, step in enumerate(cfg.model.model_steps):
        _step_dict = list(step.values())[0]
        _step_name = list(step.keys())[0]
        pipeline_list.append((_step_name, instantiate(_step_dict)))

    # Create instance of cls
    custom_pipeline = cls(steps=pipeline_list)
    return custom_pipeline

Orchestartor

Bases: ABC

Abstract base class for orchestrating the model pipeline.

Attributes:

Name Type Description
cfg dict

Configuration parameters for the orchestrator.

Methods:

Name Description
create_pipeline

Abstract method for creating the model pipeline.
features_in

Abstract method for processing the input features.
Source code in model_forge/model/model_orchastrator.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class Orchestartor(ABC):
    """
    Abstract base class for orchestrating the model pipeline.

    Attributes:
        cfg (dict): Configuration parameters for the orchestrator.

    Methods:
        create_pipeline(): Abstract method for creating the model pipeline.
        features_in(cfg): Abstract method for processing the input features.

    """

    def __init__(self, cfg) -> None:
        self.cfg = cfg

    @abstractmethod
    def create_pipeline(self):
        ...

    @abstractmethod
    def features_in(self, cfg):
        ...

TuningOrchestrator

Bases: Orchestartor

Class representing the orchestrator for model tuning.

Source code in model_forge/model/model_orchastrator.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class TuningOrchestrator(Orchestartor):

    """
    Class representing the orchestrator for model tuning.
    """

    def __init__(self, cfg, trial) -> None:
        super().__init__(cfg)
        self.trial = trial

    def create_pipeline(self):
        """
        Creates a tuning pipeline based on the configuration and trial.
        """
        return TuningPipeline.create_from_config(self.cfg, self.trial)

    def features_in(self, cfg):
        """
        Returns the features specified in the given configuration.
        """
        return cfg.features

create_pipeline()

Creates a tuning pipeline based on the configuration and trial.

Source code in model_forge/model/model_orchastrator.py 
90
91
92
93
94
def create_pipeline(self):
    """
    Creates a tuning pipeline based on the configuration and trial.
    """
    return TuningPipeline.create_from_config(self.cfg, self.trial)

features_in(cfg)

Returns the features specified in the given configuration.

Source code in model_forge/model/model_orchastrator.py 
 96
 97
 98
 99
100
def features_in(self, cfg):
    """
    Returns the features specified in the given configuration.
    """
    return cfg.features

TuningPipeline

Bases: CustomPipeline

Source code in model_forge/model/model_orchastrator.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
class TuningPipeline(CustomPipeline):
    @classmethod
    def create_from_config(cls, cfg: DictConfig | OmegaConf, trial) -> "TuningPipeline":
        """
        Create a custom model pipeline from the provided configuration.

        Args:
            cfg (DictConfig | OmegaConf): The configuration object.

        Returns:
            TuningPipeline: The created custom model pipeline.
        """
        # First create list of tuples from the modelsteps list
        params = cls.create_tuning_params(cfg, trial)
        pipeline_list = []
        for i, step in enumerate(cfg.model.model_steps):
            _step_dict = next(iter(step.items()))

            pipeline_list.append(
                (str(i), instantiate(_step_dict[1], **(params[_step_dict[0]])))
            )

        # Create instance of cls
        custom_pipeline = cls(steps=pipeline_list)
        return custom_pipeline

    @classmethod
    def create_tuning_params(cls, cfg, trial):
        """
        Create tuning parameters based on the provided configuration.

        Args:
            cfg (object): The configuration object containing hyperparameters.

        Returns:
            dict: A dictionary containing the tuning parameters for each step.

        """
        params = {}
        for step in cfg.hyperparameters:
            params_steps = {}
            for parameters in cfg.hyperparameters[step]:
                parameter_trial = instantiate(parameters)
                params_steps[
                    parameter_trial.parameter_name
                ] = parameter_trial.create_range(trial)
            params[step] = params_steps
        return params

create_from_config(cfg, trial) classmethod

Create a custom model pipeline from the provided configuration.

Parameters:

Name Type Description Default
cfg DictConfig | OmegaConf

The configuration object.

 required

Returns:

Name Type Description
TuningPipeline TuningPipeline

The created custom model pipeline.
Source code in model_forge/model/model_orchastrator.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@classmethod
def create_from_config(cls, cfg: DictConfig | OmegaConf, trial) -> "TuningPipeline":
    """
    Create a custom model pipeline from the provided configuration.

    Args:
        cfg (DictConfig | OmegaConf): The configuration object.

    Returns:
        TuningPipeline: The created custom model pipeline.
    """
    # First create list of tuples from the modelsteps list
    params = cls.create_tuning_params(cfg, trial)
    pipeline_list = []
    for i, step in enumerate(cfg.model.model_steps):
        _step_dict = next(iter(step.items()))

        pipeline_list.append(
            (str(i), instantiate(_step_dict[1], **(params[_step_dict[0]])))
        )

    # Create instance of cls
    custom_pipeline = cls(steps=pipeline_list)
    return custom_pipeline

create_tuning_params(cfg, trial) classmethod

Create tuning parameters based on the provided configuration.

Parameters:

Name Type Description Default
cfg object

The configuration object containing hyperparameters.

 required

Returns:

Name Type Description
dict

A dictionary containing the tuning parameters for each step.
Source code in model_forge/model/model_orchastrator.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
@classmethod
def create_tuning_params(cls, cfg, trial):
    """
    Create tuning parameters based on the provided configuration.

    Args:
        cfg (object): The configuration object containing hyperparameters.

    Returns:
        dict: A dictionary containing the tuning parameters for each step.

    """
    params = {}
    for step in cfg.hyperparameters:
        params_steps = {}
        for parameters in cfg.hyperparameters[step]:
            parameter_trial = instantiate(parameters)
            params_steps[
                parameter_trial.parameter_name
            ] = parameter_trial.create_range(trial)
        params[step] = params_steps
    return params