Keras

O Snowflake ML Model Registry oferece suporte a modelos do Keras 3 (keras.Model com Keras versão >= 3.0.0). O Keras 3 é uma estrutura multiback-end que oferece suporte a TensorFlow, PyTorch e JAX como back-ends.

Nota

Para o manipulador do Keras versão < 3.0.0, use the TensorFlow.

As seguintes opções adicionais podem ser usadas no dicionário options quando você chamar log_model:

Opção

Descrição

target_methods

Uma lista dos nomes dos métodos disponíveis no objeto modelo. Os modelos do Keras têm predict como o método de destino padrão.

cuda_version

A versão do tempo de execução CUDA a ser usada ao implantar em uma plataforma com GPU; o padrão é 11.8. Se definido manualmente como None, o modelo não poderá ser implementado em uma plataforma com GPU.

Você deve especificar o parâmetro sample_input_data ou signatures ao registrar um modelo do Keras para que o registro reconheça as assinaturas dos métodos de destino.

Nota

Os modelos do Keras só podem ter um método de destino.

Exemplos

Estes exemplos pressupõem que reg seja uma instância de snowflake.ml.registry.Registry.

Modelo sequencial

O exemplo a seguir demonstra o treinamento de um modelo sequencial do Keras 3, seu registro no Snowflake ML Model Registry e a execução da inferência.

import keras
from sklearn import datasets, model_selection

# Load dataset
iris = datasets.load_iris(as_frame=True)
X = iris.data
y = iris.target

# Rename columns for valid Snowflake identifiers
X.columns = [col.replace(' ', '_').replace('(', '').replace(')', '') for col in X.columns]

X_train, X_test, y_train, y_test = model_selection.train_test_split(X, y, test_size=0.2)

# Build Keras sequential model
model = keras.Sequential([
    keras.layers.Dense(64, activation='relu'),
    keras.layers.Dense(32, activation='relu'),
    keras.layers.Dense(3, activation='softmax')
])

model.compile(
    optimizer='adam',
    loss='sparse_categorical_crossentropy',
    metrics=['accuracy']
)

# Train the model
model.fit(X_train, y_train, epochs=50, verbose=0)

# Log the model
model_ref = reg.log_model(
    model=model,
    model_name="my_keras_classifier",
    version_name="v1",
    sample_input_data=X_test,
)

# Make predictions
result_df = model_ref.run(X_test[-10:], function_name="predict")
Copy

Modelo da API Funcional

O exemplo a seguir demonstra a criação de um modelo usando a API Keras Functional.

import keras
import numpy as np
import pandas as pd

# Create sample data
n_samples, n_features = 100, 10
X = pd.DataFrame(
    np.random.rand(n_samples, n_features),
    columns=[f"feature_{i}" for i in range(n_features)]
)
y = np.random.randint(0, 2, n_samples).astype(np.float32)

# Build model using Functional API
inputs = keras.Input(shape=(n_features,))
x = keras.layers.Dense(32, activation='relu')(inputs)
x = keras.layers.Dense(16, activation='relu')(x)
outputs = keras.layers.Dense(1, activation='sigmoid')(x)
model = keras.Model(inputs=inputs, outputs=outputs)

model.compile(
    optimizer=keras.optimizers.SGD(learning_rate=0.01),
    loss=keras.losses.MeanSquaredError()
)

# Train the model
model.fit(X, y, epochs=10, verbose=0)

# Log the model
model_ref = reg.log_model(
    model=model,
    model_name="my_functional_model",
    version_name="v1",
    sample_input_data=X,
)

# Make predictions
result_df = model_ref.run(X[-10:], function_name="predict")
Copy

Modelo de subclasse personalizada

O exemplo a seguir demonstra a criação de um modelo personalizado por meio da definição da subclasse keras.Model.

import keras
import numpy as np
import pandas as pd

# Define custom model with serialization support
@keras.saving.register_keras_serializable()
class BinaryClassifier(keras.Model):
    def __init__(self, hidden_units: int, output_units: int) -> None:
        super().__init__()
        self.dense1 = keras.layers.Dense(hidden_units, activation="relu")
        self.dense2 = keras.layers.Dense(output_units, activation="sigmoid")

    def call(self, inputs):
        x = self.dense1(inputs)
        return self.dense2(x)

    def get_config(self):
        base_config = super().get_config()
        config = {
            "dense1": keras.saving.serialize_keras_object(self.dense1),
            "dense2": keras.saving.serialize_keras_object(self.dense2),
        }
        return {**base_config, **config}

    @classmethod
    def from_config(cls, config):
        dense1_config = config.pop("dense1")
        dense1 = keras.saving.deserialize_keras_object(dense1_config)
        dense2_config = config.pop("dense2")
        dense2 = keras.saving.deserialize_keras_object(dense2_config)
        obj = cls(1, 1)
        obj.dense1 = dense1
        obj.dense2 = dense2
        return obj

# Create sample data
n_samples, n_features = 100, 10
X = pd.DataFrame(
    np.random.rand(n_samples, n_features),
    columns=[f"feature_{i}" for i in range(n_features)]
)
y = np.random.randint(0, 2, n_samples).astype(np.float32)

# Create and train model
model = BinaryClassifier(hidden_units=32, output_units=1)
model.compile(
    optimizer=keras.optimizers.SGD(learning_rate=0.01),
    loss=keras.losses.MeanSquaredError()
)
model.fit(X, y, epochs=10, verbose=0)

# Log the model
model_ref = reg.log_model(
    model=model,
    model_name="my_custom_classifier",
    version_name="v1",
    sample_input_data=X,
)

# Make predictions
result_df = model_ref.run(X[-10:], function_name="predict")
Copy