Skip to content

Deployment

While OpsML is not an all-in-one platform that will deploy your model in one click (maybe one day 😄), it does provide a helper class and a happy path to deploy your model. This is outlined below.

flowchart LR
    subgraph Client
    user(fa:fa-user DS) -->|create| data(fa:fa-table Data)
    data -->|create|model(fa:fa-brain Model)
    data -->|package in|datacard(DataCard)
    model -->|package in|modelcard(ModelCard)
    datacard -->|associate|modelcard
    end 

    subgraph Server
    datacard -->|insert into|datareg[(DataRegistry)]
    modelcard -->|insert into|modelreg[(ModelRegistry)]
    end

    subgraph CICD
    modelreg --> dir(Directory)
    dir --> |package|docker(DockerFile)
    end


    subgraph API
    loaded(ModelLoader) -->|load|loaded_model(Model)
    end

    docker --> API

    subgraph UI
    vis(visualize)
    end

    user --> vis
    modelreg -->|view in|UI
    datareg -->|view in|UI


    style Client rx:10,ry:10
    style Server rx:10,ry:10
    style CICD rx:10,ry:10
    style API rx:10,ry:10
    style UI rx:10,ry:10


    style user fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style data fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style model fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style datacard fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style modelcard fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style loaded fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style dir fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style docker fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style loaded_model fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style vis fill:#028e6b,stroke:black,stroke-width:2px,color:white,font-weight:bolder

    style datareg fill:#5e0fb7,stroke:black,stroke-width:2px,color:white,font-weight:bolder
    style modelreg fill:#5e0fb7,stroke:black,stroke-width:2px,color:white,font-weight:bolder

Steps:

DS Worflow

  1. DS creates data and model
  2. DS packages data and model into appropriate interfaces and DataCard and ModelCard, respectively.
  3. DataCard and ModelCard are registered and pushed to their respective registries.

CICD Workflow

  1. During CICD, the model is downloaded from the ModelRegistry via the Opsml CLI to a directory
  2. Model directory and API logic are packaged into a docker image

API Workflow

  1. Docker image is deployed to a server.
  2. During startup, the API logic leverages the ModelLoader class to load the model from a local directory.
  3. Model is now ready to be used by the API.

opsml.model.ModelLoader

Helper class for loading models from disk and downloading via opsml-cli

Source code in opsml/model/loader.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 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
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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
class ModelLoader:
    """Helper class for loading models from disk and downloading via opsml-cli"""

    def __init__(self, path: Path):
        """Initialize ModelLoader

        Args:
            interface:
                ModelInterface for the model
            path:
                Directory path to the model artifacts. This is expected to be
                a local path on disk.
        """

        self.path = path
        self.metadata = self._load_metadata()
        self.interface = self._load_interface()
        self.drift_profile = self._load_drift_profile()

    def _load_drift_profile(self) -> Optional[Union[SpcDriftProfile]]:
        """Load drift profile from disk"""
        if not hasattr(self.metadata, "drift"):
            return None

        drift_type: str = self.metadata.drift["drift_type"]
        drift_profile_path = (self.path / SaveName.DRIFT_PROFILE.value).with_suffix(Suffix.JSON.value)

        # check if path exists
        if not drift_profile_path.exists():
            return None

        # load drift profile json to string
        with drift_profile_path.open("r") as file_:
            if drift_type == DriftType.SPC.value:  # type: ignore
                return SpcDriftProfile.model_validate_json(file_.read())

        raise ValueError(f"Drift type {drift_type} not supported")

    def _load_interface(self) -> ModelInterface:
        """Loads a ModelInterface from disk using metadata

        Args:
            interface:
                ModelInterface to load

        Returns:
            ModelInterface
        """
        from opsml.storage.card_loader import _get_model_interface

        Interface = _get_model_interface(self.metadata.model_interface)  # pylint: disable=invalid-name

        loaded_interface = Interface.model_construct(
            _fields_set={"name", "repository", "version"},
            **{
                "name": self.metadata.model_name,
                "repository": self.metadata.model_repository,
                "version": self.metadata.model_version,
            },
        )

        loaded_interface.model_type = self.metadata.model_type

        if hasattr(self.metadata, "prepocessor_name"):
            loaded_interface.preprocessor_name = self.metadata.preprocessor_name

        if hasattr(self.metadata, "tokenizer_name"):
            loaded_interface.tokenizer_name = self.metadata.tokenizer_name

        if hasattr(self.metadata, "feature_extractor_name"):
            loaded_interface.feature_extractor_name = self.metadata.feature_extractor_name

        return loaded_interface

    @property
    def model(self) -> Any:
        return self.interface.model

    @property
    def onnx_model(self) -> OnnxModel:
        assert self.interface.onnx_model is not None, "OnnxModel not loaded"
        return self.interface.onnx_model

    @property
    def preprocessor(self) -> Any:
        """Quick access to preprocessor from interface"""

        if hasattr(self.interface, "preprocessor"):
            return self.interface.preprocessor

        if hasattr(self.interface, "tokenizer"):
            if self.interface.tokenizer is not None:
                return self.interface.tokenizer

        if hasattr(self.interface, "feature_extractor"):
            if self.interface.feature_extractor is not None:
                return self.interface.feature_extractor

        return None

    def _load_metadata(self) -> ModelMetadata:
        """Load metadata from disk"""
        metadata_path = (self.path / SaveName.MODEL_METADATA.value).with_suffix(Suffix.JSON.value)

        with metadata_path.open("r") as file_:
            return ModelMetadata(**json.load(file_))

    def _load_huggingface_preprocessors(self) -> None:
        """Load huggingface preprocessors from disk"""

        assert isinstance(self.interface, HuggingFaceModel), "HuggingFaceModel interface required"

        if self.preprocessor is not None:
            return

        if hasattr(self.metadata, "tokenizer_name"):
            load_path = (self.path / SaveName.TOKENIZER.value).with_suffix("")
            self.interface.load_tokenizer(load_path)
            return

        if hasattr(self.metadata, "feature_extractor_name"):
            load_path = (self.path / SaveName.FEATURE_EXTRACTOR.value).with_suffix("")
            self.interface.load_feature_extractor(load_path)
            return

        return

    def load_preprocessor(self) -> None:
        """Load preprocessor from disk"""

        if isinstance(self.interface, HuggingFaceModel):
            self._load_huggingface_preprocessors()
            return

        if hasattr(self.metadata, "preprocessor_name"):
            load_path = (self.path / SaveName.PREPROCESSOR.value).with_suffix(self.interface.preprocessor_suffix)
            self.interface.load_preprocessor(load_path)
            return

        return

    def load_model(self, **kwargs: Any) -> None:
        load_path = (self.path / SaveName.TRAINED_MODEL.value).with_suffix(self.interface.model_suffix)
        self.interface.load_model(load_path, **kwargs)

        if isinstance(self.interface, HuggingFaceModel):
            if self.interface.is_pipeline:
                self.interface.to_pipeline()

    def _load_huggingface_onnx_model(self, load_quantized: bool) -> None:
        assert isinstance(self.interface, HuggingFaceModel), "Expected HuggingFaceModel"
        save_name = SaveName.QUANTIZED_MODEL.value if load_quantized else SaveName.ONNX_MODEL.value

        if self.interface.is_pipeline:
            self._load_huggingface_preprocessors()

        load_path = (self.path / save_name).with_suffix(self.interface.model_suffix)
        self.interface.onnx_model = OnnxModel(onnx_version=self.metadata.onnx_version)
        self.interface.load_onnx_model(load_path)

    def load_onnx_model(self, load_quantized: bool = False, onnx_args: Optional[HuggingFaceOnnxArgs] = None) -> None:
        """Load onnx model from disk

        Args:

            ------Note: These args only apply to HuggingFace models------

            load_quantized:
                If True, load quantized model

            onnx_args:
                Additional onnx args needed to load the model

        """
        if isinstance(self.interface, HuggingFaceModel):
            self.interface.onnx_args = onnx_args
            self._load_huggingface_onnx_model(load_quantized)
            return

        load_path = (self.path / SaveName.ONNX_MODEL.value).with_suffix(Suffix.ONNX.value)
        self.interface.onnx_model = OnnxModel(onnx_version=self.metadata.onnx_version)
        self.interface.load_onnx_model(load_path)
        return
preprocessor: Any property

Quick access to preprocessor from interface

__init__(path)

Initialize ModelLoader

Parameters:

Name Type Description Default
interface

ModelInterface for the model

 required
path Path

Directory path to the model artifacts. This is expected to be a local path on disk.

 required
Source code in opsml/model/loader.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(self, path: Path):
    """Initialize ModelLoader

    Args:
        interface:
            ModelInterface for the model
        path:
            Directory path to the model artifacts. This is expected to be
            a local path on disk.
    """

    self.path = path
    self.metadata = self._load_metadata()
    self.interface = self._load_interface()
    self.drift_profile = self._load_drift_profile()
load_onnx_model(load_quantized=False, onnx_args=None)

Load onnx model from disk

Args:

------Note: These args only apply to HuggingFace models------

load_quantized:
    If True, load quantized model

onnx_args:
    Additional onnx args needed to load the model
Source code in opsml/model/loader.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def load_onnx_model(self, load_quantized: bool = False, onnx_args: Optional[HuggingFaceOnnxArgs] = None) -> None:
    """Load onnx model from disk

    Args:

        ------Note: These args only apply to HuggingFace models------

        load_quantized:
            If True, load quantized model

        onnx_args:
            Additional onnx args needed to load the model

    """
    if isinstance(self.interface, HuggingFaceModel):
        self.interface.onnx_args = onnx_args
        self._load_huggingface_onnx_model(load_quantized)
        return

    load_path = (self.path / SaveName.ONNX_MODEL.value).with_suffix(Suffix.ONNX.value)
    self.interface.onnx_model = OnnxModel(onnx_version=self.metadata.onnx_version)
    self.interface.load_onnx_model(load_path)
    return
load_preprocessor()

Load preprocessor from disk

Source code in opsml/model/loader.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
def load_preprocessor(self) -> None:
    """Load preprocessor from disk"""

    if isinstance(self.interface, HuggingFaceModel):
        self._load_huggingface_preprocessors()
        return

    if hasattr(self.metadata, "preprocessor_name"):
        load_path = (self.path / SaveName.PREPROCESSOR.value).with_suffix(self.interface.preprocessor_suffix)
        self.interface.load_preprocessor(load_path)
        return

    return