Azure ML SDK2下Sweep Job多工件输出与MLflow部署问题

以下针对你遇到的5个问题逐一给出实操方案：

问题1：Sweep Job是否确实输出最优模型？ #

是的。Azure ML Sweep Job默认会自动将最优子任务的所有输出 artifacts（包括模型、日志、参数等）作为整个Sweep Job的顶级输出。

• 你可以在Azure ML Studio的Sweep Job详情页中，通过「Best child run」入口查看最优子任务的具体输出；
• 若需自定义最优指标的判断逻辑（如最小化损失、最大化准确率），需在Sweep配置中明确指定：

from azure.ai.ml.sweep import Objective

sweep_job = command_job.sweep(
    sampling_algorithm="random",
    primary_metric="validation_loss",
    objective=Objective("minimize"),  # 指定指标优化方向
    max_total_trials=20
)

• 代码中可通过SDK获取最优子任务的输出路径：

ml_client = MLClient(DefaultAzureCredential(), sub_id, rg_name, ws_name)
sweep_job = ml_client.jobs.get("your-sweep-job-id")
best_child = ml_client.jobs.get(sweep_job.properties.best_child_run_id)
best_model_uri = best_child.outputs.trained_model.path

问题2：使用pyfunc.log_model时输出类型选mlflow_model还是custom_model？ #

选mlflow_model。

• mlflow.pyfunc.log_model()本身就是MLflow官方定义的自定义模型记录方式，生成的模型符合MLflow标准格式，Azure ML会自动识别为mlflow_model类型；
• 组件定义中需对应声明输出类型为mlflow_model：

outputs:
  trained_ensemble:
    type: mlflow_model

• 脚本中记录模型的示例：

mlflow.pyfunc.log_model(
    artifact_path="scaler_autoencoder_wrapper",
    python_model=ScalerAutoencoderWrapper(
        autoencoder=autoencoder, encoder=encoder, scaler=scaler
    ),
    artifacts={
        "autoencoder": "./outputs/autoencoder.h5",
        "encoder": "./outputs/encoder.h5",
        "scaler": "./outputs/scaler.pkl"
    }
)

问题3：设置路径触发yaml.representer.RepresenterError #

这个错误是因为传递了PyYAML无法序列化的对象（如Path类实例），解决方法：

• 所有路径参数统一转为字符串格式，用str(Path对象)替代直接传递Path对象；
• 组件输入输出的路径配置仅传递字符串格式的本地路径或Azure ML URI（如azureml://datastores/xxx/paths/xxx）。

示例修正：

# 错误写法
output_dir = Path("./outputs/model_artifacts")
# 正确写法
output_dir = str(Path("./outputs/model_artifacts"))

问题4：复制模型目录时提示目录不存在，不清楚模型保存路径 #

Azure ML任务中，./outputs目录是默认的artifact上传目录，所有保存在此目录下的文件/文件夹都会被自动上传到工作区存储。

• 训练脚本中，建议将所有模型工件统一保存到./outputs下的子目录，比如./outputs/autoencoder_artifacts；
• 可通过MLflow API获取当前运行的artifact根路径：

artifact_root = mlflow.get_artifact_uri()
# 输出示例：azureml://datastores/workspaceblobstore/paths/ExperimentRun/dcid.xxx/outputs/

• 若需复制最优子任务的模型目录，可通过之前获取的best_model_uri，用Azure ML SDK的ml_client.datastores.download()方法下载到本地或流水线后续组件中。

问题5：仅能传递.h5文件和StandardScaler对象到评估组件，寻求更优方案 #

最优方案是将所有需传递的工件打包为一个MLflow自定义模型，以mlflow_model类型作为流水线组件的输出/输入传递，避免分散传递多个文件：

1. 完善ScalerAutoencoderWrapper类，确保其包含所有需共享的工件，并实现MLflow PyFunc要求的load_context和predict方法：

import mlflow.pyfunc

class ScalerAutoencoderWrapper(mlflow.pyfunc.PythonModel):
    def load_context(self, context):
        # 从artifacts加载所有工件
        self.autoencoder = load_model(context.artifacts["autoencoder"])
        self.encoder = load_model(context.artifacts["encoder"])
        self.scaler = joblib.load(context.artifacts["scaler"])
    
    def predict(self, context, model_input):
        # 实现自定义预测逻辑
        scaled_input = self.scaler.transform(model_input)
        encoded = self.encoder.predict(scaled_input)
        return encoded

2. 训练脚本中用mlflow.pyfunc.log_model()将所有工件打包记录；
3. 评估组件中直接加载整个MLflow模型，即可访问所有内部工件：

import mlflow.pyfunc

model = mlflow.pyfunc.load_model(model_uri=input_model_path)
# 访问内部组件
scaler = model._model_impl.scaler
encoder = model._model_impl.encoder

这种方式既简化了流水线组件间的参数传递，又能统一管理所有模型相关工件，便于后续部署。

内容的提问来源于stack exchange，提问作者Marvin