Azure AI服务：Azure ML Studio工作流

import jsonfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Any  @dataclassclass Workspace:    """Azure ML 工作区：顶层资源容器"""    name: str    subscription_id: str    resource_group: str    location: str    compute_targets: Dict[str, "Compute"] = field(default_factory=dict)    datastores: Dict[str, "DataAsset"] = field(default_factory=dict)    models: Dict[str, List["Model"]] = field(default_factory=dict)    endpoints: Dict[str, "Endpoint"] = field(default_factory=dict)     def register_compute(self, compute: "Compute") -> None:        self.compute_targets[compute.name] = compute     def register_data(self, asset: "DataAsset") -> None:        self.datastores[asset.name] = asset  @dataclassclass Compute:    """计算资源：训练和推理集群"""    name: str    compute_type: str    vm_size: str    max_nodes: int = 4    min_nodes: int = 0    status: str = "idle"     def start(self) -> str:        self.status = "running"        return f"Compute '{self.name}' started ({self.vm_size})"     def stop(self) -> str:        self.status = "idle"        return f"Compute '{self.name}' stopped"  @dataclassclass DataAsset:    """数据资产：存储与版本管理"""    name: str    version: str    path: str    data_type: str    description: str = ""  @dataclassclass Environment:    """训练环境：可重现的依赖配置"""    name: str    version: str    python_version: str    conda_dependencies: List[str] = field(default_factory=list)    docker_image: Optional[str] = None     def is_ready(self) -> bool:        return bool(self.conda_dependencies or self.docker_image)  @dataclassclass Model:    """模型：版本化注册"""    name: str    version: str    framework: str    path: str    metrics: Dict[str, float] = field(default_factory=dict)    description: str = ""    tags: Dict[str, str] = field(default_factory=dict)  class ModelRegistry:    """模型注册表：版本管理"""     def __init__(self, workspace: Workspace):        self.workspace = workspace     def register(self, model: Model) -> str:        versions = self.workspace.models.setdefault(model.name, [])        if any(m.version == model.version for m in versions):            raise ValueError(f"Model {model.name} v{model.version} already exists")        versions.append(model)        versions.sort(key=lambda m: list(map(int, m.version.split("."))))        return f"Registered {model.name}@{model.version}"     def latest(self, name: str) -> Model:        return self.workspace.models[name][-1]  @dataclassclass Endpoint:    """端点：实时与批量推理"""    name: str    endpoint_type: str    model_name: str    model_version: str    status: str = "inactive"     def deploy(self) -> str:        self.status = "active"        return f"Endpoint '{self.name}' ({self.endpoint_type}) deployed"  class PipelineStep:    """管道步骤基类"""     def __init__(self, name: str):        self.name = name        self.inputs: List[str] = []        self.outputs: List[str] = []        self.status: str = "pending"     def run(self, context: Dict[str, Any]) -> str:        self.status = "completed"        return f"Step '{self.name}' completed"  class DataPrepStep(PipelineStep):    def __init__(self):        super().__init__("data_preparation")     def run(self, context: Dict[str, Any]) -> str:        asset: DataAsset = context["data_asset"]        self.outputs.append(f"{asset.name}:{asset.version}")        self.status = "completed"        return f"Prepared data {asset.name}@{asset.version} ({asset.data_type})"  class TrainStep(PipelineStep):    def __init__(self, epochs: int = 10):        super().__init__("model_training")        self.epochs = epochs     def run(self, context: Dict[str, Any]) -> str:        compute: Compute = context["compute"]        env: Environment = context["environment"]        compute.start()        self.status = "completed"        return f"Trained on {compute.name} ({compute.vm_size}) in {env.name}, {self.epochs} epochs"  class EvaluateStep(PipelineStep):    def __init__(self):        super().__init__("model_evaluation")     def run(self, context: Dict[str, Any]) -> str:        metrics = {"accuracy": 0.94, "f1": 0.91, "auc": 0.96}        context["metrics"] = metrics        self.status = "completed"        return f"Evaluated: {metrics}"  class RegisterStep(PipelineStep):    def __init__(self, registry: ModelRegistry):        super().__init__("model_registration")        self.registry = registry     def run(self, context: Dict[str, Any]) -> str:        model = Model(            name="customer-churn",            version=context.get("next_version", "1.0.0"),            framework="scikit-learn",            path="outputs/model.pkl",            metrics=context.get("metrics", {}),            tags={"stage": "production"},        )        self.registry.register(model)        context["model"] = model        self.status = "completed"        return f"Registered model {model.name}@{model.version}"  class DeployStep(PipelineStep):    def __init__(self, workspace: Workspace):        super().__init__("endpoint_deployment")        self.workspace = workspace     def run(self, context: Dict[str, Any]) -> str:        model: Model = context["model"]        endpoint = Endpoint(            name="churn-realtime",            endpoint_type="real-time",            model_name=model.name,            model_version=model.version,        )        endpoint.deploy()        self.workspace.endpoints[endpoint.name] = endpoint        context["endpoint"] = endpoint        self.status = "completed"        return f"Deployed {endpoint.name} serving {model.name}@{model.version}"  class Pipeline:    """ML 管道：编排步骤"""     def __init__(self, name: str):        self.name = name        self.steps: List[PipelineStep] = []     def add_step(self, step: PipelineStep) -> "Pipeline":        self.steps.append(step)        return self     def execute(self, context: Dict[str, Any]) -> List[str]:        logs = [f"Pipeline '{self.name}' started with {len(self.steps)} steps"]        for step in self.steps:            logs.append(step.run(context))        logs.append(f"Pipeline '{self.name}' completed successfully")        return logs  def demo_azure_ml_workflow() -> Dict[str, Any]:    """演示完整 Azure ML 工作流"""    ws = Workspace(        name="ml-prod-ws",        subscription_id="00000000-0000-0000-0000-000000000000",        resource_group="rg-ml-prod",        location="eastus",    )    compute = Compute("cpu-cluster", "cpu-cluster", "STANDARD_DS3_V2", max_nodes=6)    ws.register_compute(compute)    asset = DataAsset("customer-data", "1", "azureml://datastores/workspaceblobstore/paths/data/", "uri_folder")    ws.register_data(asset)    env = Environment("sklearn-env", "5", "3.10", ["scikit-learn==1.3", "numpy==1.25"], "mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04")    registry = ModelRegistry(ws)    pipeline = Pipeline("churn-training-pipeline")    pipeline.add_step(DataPrepStep()).add_step(TrainStep(epochs=15))    pipeline.add_step(EvaluateStep()).add_step(RegisterStep(registry)).add_step(DeployStep(ws))    context = {"data_asset": asset, "compute": compute, "environment": env, "next_version": "1.0.0"}    logs = pipeline.execute(context)    return {"logs": logs, "model": registry.latest("customer-churn"), "endpoint": ws.endpoints["churn-realtime"]}  if __name__ == "__main__":    result = demo_azure_ml_workflow()    print("\n".join(result["logs"]))    print(f"Model: {result['model'].name}@{result['model'].version} metrics={result['model'].metrics}")    print(f"Endpoint: {result['endpoint'].name} status={result['endpoint'].status}")