can you load ml models from azureml registry to an Azure function app

Question

Is it possible to load registered ml models from azureml model registry to an azure function app and carry out inference or do these models need to be downloaded inside the function app first? any examples or guides?

Answer 1

model.predict > predict is not directly available when you load the model from Azure ML workspace, As you might face the error below:-

'Model' object has no attribute 'predict'

Thus its necessary to download the model via joblib or other library in the memory and then perform the inference. For authentication you can directly use Service Principal authentication in your function app code:-

My Python Function v1, init.py:-

import azure.functions as func
from azureml.core import Workspace
from azureml.core.authentication import ServicePrincipalAuthentication
import joblib
import numpy as np

# Service Principal Authentication details
tenant_id = 'xxxxxxxxxx8592395'
service_principal_id = 'c0c95xxxxxxxxx5cb'
service_principal_password = 'zxxxxxxxxwxj9b6k'
subscription_id = '0xxxxxxxa7'
resource_group = 'vaxxxx8'
workspace_name = 'xxx'
model_name = 'Modelsilicon'

# Authenticate using Service Principal
svc_pr = ServicePrincipalAuthentication(
    tenant_id=tenant_id,
    service_principal_id=service_principal_id,
    service_principal_password=service_principal_password
)

# Load your Azure ML Workspace using Service Principal Authentication
try:
    ws = Workspace.get(
        name=workspace_name,
        auth=svc_pr,
        subscription_id=subscription_id,
        resource_group=resource_group
    )

    # Function to load model from Azure ML Workspace
    def load_model(model_name="Modelsilicon"):
        try:
            # Load the model from Azure ML
            model = ws.models[model_name]
            model_path = model.download(exist_ok=True)  # Download the model
            loaded_model = joblib.load(model_path)
            return loaded_model
        except Exception as e:
            return None

    # Load the sample model for demonstration
    loaded_sample_model = load_model(model_name)

    def main(req: func.HttpRequest) -> func.HttpResponse:
        try:
            req_body = req.get_json()

            # Perform inference using the loaded sample model
            if loaded_sample_model:
                data = np.array(req_body['data']).reshape(-1, 1)  # Assuming data is a list or array
                prediction = loaded_sample_model.predict(data)
                return func.HttpResponse(f"Sample Model Prediction: {prediction}", status_code=200)
            else:
                return func.HttpResponse("Sample model not loaded.", status_code=500)
        except Exception as e:
            return func.HttpResponse(f"Error: {str(e)}", status_code=500)

except Exception as e:
    print(f"Failed to authenticate or load workspace: {str(e)}")
    raise

Request Body:-

{
    "data": [1.5, 2.3, 3.7, 4.8]
}

Output:-

Azure Function Output:-

But if you do not want to download the model, You can create one scoring script and dependencies then run your inference.

scoring.py:-

import joblib
import numpy as np
from azureml.core.model import Model

def init():
    global loaded_model
    # Retrieve the path to the model file
    model_path = Model.get_model_path('your_registered_model_name')  # Replace with your registered model name
    # Load the model
    loaded_model = joblib.load(model_path)

def run(raw_data):
    try:
        data = np.array(json.loads(raw_data)['data']).reshape(-1, 1)  # Assuming data is in JSON format
        prediction = loaded_model.predict(data)
        return prediction.tolist()  # Return prediction as a list
    except Exception as e:
        error = str(e)
        return error

And then use this scoring script for your inference, Refer below:-

from azureml.core import Workspace
from azureml.core.authentication import ServicePrincipalAuthentication
from azureml.core.model import Model
from azureml.core.webservice import Webservice
from azureml.core.conda_dependencies import CondaDependencies
from azureml.core.environment import Environment

# Service Principal Authentication details
tenant_id = 'YOUR_TENANT_ID'
service_principal_id = 'YOUR_SERVICE_PRINCIPAL_ID'
service_principal_password = 'YOUR_SERVICE_PRINCIPAL_PASSWORD'
subscription_id = 'YOUR_SUBSCRIPTION_ID'
resource_group = 'YOUR_RESOURCE_GROUP'
workspace_name = 'YOUR_WORKSPACE_NAME'
model_name = 'YOUR_MODEL_NAME'

# Authenticate using Service Principal
svc_pr = ServicePrincipalAuthentication(
    tenant_id=tenant_id,
    service_principal_id=service_principal_id,
    service_principal_password=service_principal_password
)

# Load your Azure ML Workspace using Service Principal Authentication
try:
    ws = Workspace.get(
        name=workspace_name,
        auth=svc_pr,
        subscription_id=subscription_id,
        resource_group=resource_group
    )

    # Fetch the registered model directly from Azure ML Workspace
    model = Model(ws, name=model_name)

    # Define the environment and dependencies for deployment
    env = Environment(name='myenv')
    conda_deps = CondaDependencies.create(pip_packages=['joblib', 'numpy'])
    env.python.conda_dependencies = conda_deps

    # Define inference configuration
    inference_config = InferenceConfig(entry_script='score.py', environment=env)

    # Deploy the model as a web service
    service_name = 'my-service-name'
    deployment_config = AciWebservice.deploy_configuration(cpu_cores=1, memory_gb=1)
    service = Model.deploy(
        workspace=ws,
        name=service_name,
        models=[model],
        inference_config=inference_config,
        deployment_config=deployment_config
    )

    service.wait_for_deployment(show_output=True)

    # Get the deployed service URL
    service_url = service.scoring_uri

    def main(req: func.HttpRequest) -> func.HttpResponse:
        try:
            req_body = req.get_json()

            # Send HTTP request to the deployed service for inference
            response = requests.post(service_url, json=req_body)
            prediction = response.json()

            return func.HttpResponse(f"Model Prediction: {prediction}", status_code=200)
        except Exception as e:
            return func.HttpResponse(f"Error: {str(e)}", status_code=500)

except Exception as e:
    print(f"Failed to authenticate or load workspace: {str(e)}")
    raise