[Python] How to reference all models in a registry folder in deployment.yml for azure ml...

I want a single managed online endpoint that can serve dozens of forecasting models (7 algorithms × N service-centre “hubs”) by routing requests to the correct model at runtime. Locally this already works: the scoring script loads every sub-directory inside a models/ folder and keeps them in a global cache.

We created a scoring script to test it in local and its working fine.

import joblib
import os
import pandas as pd
import json
import importlib.util
import inspect
from pathlib import Path
from ml_service.helpers import forecast_autoreg
ALGOS_SK = {"knn","gbr","lr","arima"}
ALGOS_FN = {"moving_average","weighted_average","simple_exponential"}

models = None

def load_functional_model(model_path: Path, func_name: str):
"""
Load `model_path` and return the callable named `func_name`.
Raises if the function doesn't exist or isn't callable.
"""
spec = importlib.util.spec_from_file_location("functional_model", model_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)

obj = getattr(module, func_name, None)
if obj is None:
raise AttributeError(f"{func_name} not found in {model_path}")
if not callable(obj):
raise TypeError(f"{func_name} exists in {model_path} but is not callable")
return obj

def init():
global models
model_dir = Path("models")
models = {}
for m in model_dir.iterdir():
if m.is_dir():
model_name = m.name
model_pkl_path = m / "model.pkl"
model_py_path = m / "model.py"
if model_pkl_path.exists():
models[model_name] = joblib.load(model_pkl_path)
elif model_py_path.exists():
algo_name = model_name.split("_forecast_model_hub_")[0]
models[model_name] = load_functional_model(model_py_path, algo_name)

def run(raw_request: str):
req = json.loads(raw_request)
hub = str(req["hub"])
algo = req["algo"].lower()
history = req["history"]
horizon = int(req.get("horizon", 6))
n_lags = int(req.get("n_lags", 4))
model_name = f"{algo}_forecast_model_hub_{hub}"
model =models[model_name]

if algo in ALGOS_SK:
try:

preds = forecast_autoreg(model, history[-n_lags:], horizon)
return {"hub": hub, "algo": algo, "forecast": preds}
except KeyError:
raise {"error": f"Error loading model: {model_name}"}
elif algo in ALGOS_FN:

# Build a pandas.Series with a dummy time-index
y_full = pd.Series(
history,
index=pd.MultiIndex.from_product([[0], list(range(len(history) - 1, -1, -1))])
)

# Inspect the function signature and drop the two required positional args
sig = inspect.signature(model)
print(f"Signature: {sig}")
fn_kwargs = {
k: req[k]
for k in sig.parameters
if k not in ("Y", "horizon") and (k in req)
}
print(f"Function kwargs: {fn_kwargs}")

# Perform inference using the functional model
js, _, _ = model(y_full, horizon, **fn_kwargs)
preds = [m["value"] for m in json.loads(js)[0]["monthData"]]

return {"hub": hub, "algo": algo, "forecast": preds}
else:
return {"error": f"Unknown algo: {algo}"}



Here we are fetching the models from local directory. so its working fine.

But if we want to deploy it, then we have to specify the models inside deployment.yml(According to :

https://learn.microsoft.com/en-us/a...oyment-model-specification?view=azureml-api-2
​

. like this :

$schema: https://azuremlschemas.azureedge.net/latest/managedOnlineDeployment.schema.json
name: blue
endpoint_name: my-endpoint
model: azureml:local-multimodel:3
code_configuration:
  code: ../../model-1/onlinescoring/
  scoring_script: score.py
environment:
  conda_file: ../../model-1/environment/conda.yml
  image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest
instance_type: Standard_DS3_v2
instance_count: 1


But I would have to enumerate every single hub-specific model, which defeats the whole point.

I’m looking for something like:

model: azureml://registries/my-registry/models/forecast_models_folder@hub@latest


or any supported pattern that avoids a long manual model_paths: list.

Continue reading...