PyTorch¶

Snowflake ML Model Registry는 PyTorch(``torch.nn.Module``에서 파생된 모델)를 사용하여 만든 모델을 지원합니다.

log_model 을 호출할 때 options 사전에서 다음 추가 옵션을 사용할 수 있습니다.

옵션	설명
`target_methods`	모델 오브젝트에서 사용할 수 있는 메서드의 이름 목록입니다. PyTorch 모델에는 기본적으로 `forward` 대상 메서드가 있습니다.
`cuda_version`	GPU로 플랫폼에 배포할 때 사용할 CUDA 런타임 버전으로 기본값은 11.8입니다. 수동으로 `None` 으로 설정하면 GPU가 있는 플랫폼에 모델을 배포할 수 없습니다.
`multiple_inputs`	모델이 여러 텐서 입력을 예상하는지 여부입니다. 기본값은 `False` 입니다. ``True``인 경우 모델은 단일 텐서 대신 텐서 목록을 입력으로 허용합니다.

레지스트리가 대상 메서드의 서명을 알도록 PyTorch 모델을 로깅할 때 sample_input_data 또는 signatures 매개 변수를 지정해야 합니다.

참고

pandas DataFrames를 사용하는 경우(기본적으로 float64 사용), PyTorch 모델 레이어는 dtype 불일치 오류를 방지하기 위해 ``dtype=torch.float64``로 생성됩니다.

예¶

이 예제에서 ``reg``는 ``snowflake.ml.registry.Registry``의 인스턴스라고 가정합니다.

import torch
import torch.nn as nn
from sklearn import datasets, model_selection

# Define a simple neural network for classification
class IrisClassifier(nn.Module):
    def __init__(self, input_dim: int, hidden_dim: int, output_dim: int):
        super().__init__()
        # Use float64 to match pandas DataFrame default dtype
        self.model = nn.Sequential(
            nn.Linear(input_dim, hidden_dim, dtype=torch.float64),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim, dtype=torch.float64),
            nn.ReLU(),
            nn.Linear(hidden_dim, output_dim, dtype=torch.float64),
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.model(x)

# 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)

# Create model
model = IrisClassifier(input_dim=4, hidden_dim=32, output_dim=3)

# Train the model
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
criterion = nn.CrossEntropyLoss()

X_train_tensor = torch.tensor(X_train.values)
y_train_tensor = torch.tensor(y_train.values, dtype=torch.long)

model.train()
for epoch in range(100):
    optimizer.zero_grad()
    outputs = model(X_train_tensor)
    loss = criterion(outputs, y_train_tensor)
    loss.backward()
    optimizer.step()

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

# Make predictions
result_df = model_ref.run(X_test[-10:])

Copy