Stockage de modèles personnalisés dans le registre de modèles de Snowflake¶

Définir le contexte du modèle¶

Les modèles nécessitent souvent un ou plusieurs fichiers statiques, tels que des fichiers de configuration ou des poids de modèle, en plus du code. Les modèles personnalisés doivent fournir des informations sur tous ces artefacts afin que le registre sache qu’il doit les assembler avec le modèle. Les artefacts peuvent être déclarés à l’aide de la classe ModelContext comme suit.

from snowflake.ml.model import custom_model

mc = custom_model.ModelContext(
    artifacts={
        'config': 'local_model_dir/config.json'
    },
)

Les chemins vers les fichiers d’artefacts dans le contexte du modèle sont relatifs au répertoire actuel dans l’environnement à partir duquel le modèle est enregistré. Le registre des modèles de Snowflake utilise ces informations pour s’assurer que tout le code et toutes les données nécessaires sont déployés dans l’entrepôt où votre modèle sera exécuté. Au moment de l’exécution, le modèle peut trouver ces artefacts en appelant self.context.path('config') (la valeur 'config' est la même que la clé du dictionnaire transmise à ModelContext).

Outre les fichiers statiques, un modèle peut composer d’autres modèles ou pipelines d’un type pris en charge (par exemple, un modèle snowflake.ml.modeling.pipeline ou scikit-learn). Le registre sait déjà comment enregistrer ces types d’objets, vous pouvez donc passer les objets Python directement dans le contexte du modèle à l’aide de model_refs, comme indiqué ici. Vous n’avez pas besoin de mettre en paquet ces objets vous-même. Cela peut être utile pour l’agrégation bootstrap (bagging) ou pour le prétraitement ou le post-traitement.

mc = custom_model.ModelContext(
    artifacts={
        'config': 'local_model_dir/config.json'
    },
    models={
        'm1': model1,
        'm2': model2,
        'feature_preproc': preproc
    }
)

Le registre de modèle sérialise ces références de modèle lors de l’enregistrement du modèle, et les réhydrate au moment de l’exécution. Votre modèle peut récupérer des références aux modèles subordonnés en utilisant, par exemple, self.context.model_ref('m1'). Si le modèle expose une méthode predict, votre code peut l’appeler directement à partir de la référence du modèle récupéré, par exemple avec self.context.model_ref('m1').predict().

En résumé, le contexte d’un modèle personnalisé déclare les objets Python à sérialiser ainsi que les artefacts du modèle, qui sont des fichiers locaux utilisés par le modèle et qui doivent être stockés dans Snowflake avec le code. Votre classe de modèle utilise le contexte pour localiser le code et les artefacts ; cela fonctionne que votre modèle soit exécuté localement ou dans Snowflake.

Écrire la classe de modèle personnalisée¶

Pour indiquer au registre des modèles comment enregistrer et déployer votre modèle personnalisé, héritez de snowflake.ml.model.custom_model.CustomModel.

Les modèles peuvent exposer plusieurs méthodes d’inférence (par exemple, les modèles scikit-learn exposent les méthodes predict et predict_proba). Pour déclarer des fonctions d’inférence dans votre modèle personnalisé, définissez-les comme des méthodes publiques de votre sous-classe et décorez-les avec @custom_model.inference_api. Ce décorateur indique qu’une méthode fait partie de l’API publique du modèle, ce qui permet de l’appeler depuis Python ou SQL via le registre du modèle. Les méthodes décorées par inference_api doivent accepter et renvoyer un DataFrame pandas. Un nombre quelconque de méthodes peut être décoré avec inference_api.

Vous trouverez ci-dessous un squelette de classe de modèle personnalisé avec une API publique.

Notez l’utilisation de context.path pour ouvrir le fichier de biais et de self.context.model_ref pour obtenir des références aux classes de modèles subordonnées afin que leurs méthodes predict puissent être appelées.

from snowflake.ml.model import custom_model
import pandas as pd

class ExamplePipelineModel(custom_model.CustomModel):

    @custom_model.inference_api
    def predict(self, input: pd.DataFrame) -> pd.DataFrame:
        ...
        return pd.DataFrame(...)

En réunissant toutes les pièces du puzzle, vous obtenez un modèle personnalisé entièrement fonctionnel.

class ExamplePipelineModel(custom_model.CustomModel):
    def __init__(self, context: custom_model.ModelContext) -> None:
        super().__init__(context)
        v = open(context.path('config')).read()
        self.bias = json.loads(v)['bias']

    @custom_model.inference_api
    def predict(self, input: pd.DataFrame) -> pd.DataFrame:
        features = self.context.model_ref('feature_preproc').transform(input)
        model_output = self.context.model_ref('m2').predict(
            self.context.model_ref('m1').predict(features)
        )
        return pd.DataFrame({
            'output': model_output + self.bias})

Utilisation du modèle personnalisé¶

Vous pouvez tester votre nouveau modèle personnalisé (pipeline) en l’exécutant localement comme suit :

my_model_pipeline = ExamplePipelineModel(mc)
output_df = my_model_pipeline.predict(input_df)

Ou l’enregistrer dans le registre et le déployer sur Snowflake. Comme indiqué dans l’exemple de code suivant, fournissez conda_dependencies (ou pip_requirements) pour spécifier les bibliothèques dont la classe de modèle a besoin.

Fournissez sample_input_data (un DataFrame pandas) pour déduire la signature d’entrée du modèle. Vous pouvez également fournir une signature du modèle.

reg = Registry(session=sp_session, database_name="ML", schema_name="REGISTRY")
mv = reg.log_model(my_model_pipeline,
            model_name="my_custom_model_pipeline",
            version_name="v1",
            conda_dependencies=["scikit-learn"],
            comment="My Custom ML Model Pipeline",
            sample_input_data=train_features)
output_df = mv.run(input_df)

Inférence de fonction de table¶

À partir de Snowpark ML 1.5.4, vous pouvez enregistrer des modèles avec des méthodes d’inférence qui renvoient plusieurs colonnes. Pour ce faire, enregistrez votre modèle avec l’option {"function_type": "TABLE_FUNCTION"} et utilisez le décorateur @inference_api comme ci-dessus. Dans l’exemple suivant, la méthode décorée renvoie un DataFrame pandas qui comprend deux colonnes de sortie.

class ExampleTableFunctionModel(custom_model.CustomModel):

  @custom_model.inference_api
  def predict(self, input: pd.DataFrame) -> pd.DataFrame:
      output_df = pandas.DataFrame([{"OUTPUT1": input["INPUT1"] + 1, input["INPUT2"] + 1}])
      return output_df

my_model = ExampleTableFunctionModel()
reg = Registry(session=sp_session, database_name="ML", schema_name="REGISTRY")
mv = reg.log_model(my_model,
            model_name="my_custom_table_function_model",
            version_name="v1",
            options={"function_type": "TABLE_FUNCTION"},
            sample_input_data=train_features
            )
output_df = mv.run(input_df)

Si le modèle inclut plusieurs méthodes d’inférence, utilisez l’option method_options pour enregistrer le modèle, indiquant lesquels sont FUNCTION et qui sont TABLE_FUNCTION :

reg = Registry(session=sp_session, database_name="ML", schema_name="REGISTRY")
mv = reg.log_model(my_model,
            model_name="my_custom_table_function_model",
            version_name="v1",
            options={
              "method_options": {                                 ###
                "METHOD1": {"function_type": "TABLE_FUNCTION"},   ###
                "METHOD2": {"function_type": "FUNCTION"}          ###
              }
            },
            sample_input_data=train_features
            )

La méthode d’inférence de la fonction de table de modèle enregistrée peut également être appelée en SQL comme suit :

SELECT OUTPUT1, OUTPUT2
  FROM input_table,
      table(
          MY_MODEL!PREDICT(input_table.INPUT1, input_table.INPUT2)
      )

Pour en savoir plus sur les méthodes d’inférence de modèle partitionné où la méthode d’inférence prend une partition de données en entrée et génère plusieurs lignes et colonnes, voir Modèles personnalisés partitionnés.