雲端運算結合人工智慧應用與實踐

import boto3
import sagemaker
from sagemaker import get_execution_role
from sagemaker.sklearn import SKLearn
from sagemaker.model import Model
from sagemaker.predictor import Predictor
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional
import json
import logging

# 設定日誌記錄
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SageMakerMLPipeline:
    """
    AWS SageMaker 機器學習流程管理類別
    提供從資料準備到模型部署的完整功能
    """

    def __init__(self, region: str = 'us-west-2'):
        """
        初始化 SageMaker 客戶端

        參數:
            region: AWS 區域
        """
        # 建立 SageMaker session
        self.session = sagemaker.Session()
        # 取得執行角色
        self.role = get_execution_role()
        # 設定預設 S3 bucket
        self.bucket = self.session.default_bucket()
        # 設定區域
        self.region = region
        # 建立 SageMaker 客戶端
        self.sm_client = boto3.client('sagemaker', region_name=region)

        logger.info(f"SageMaker Pipeline 初始化完成")
        logger.info(f"使用 bucket: {self.bucket}")
        logger.info(f"使用角色: {self.role}")

    def upload_data(self, local_path: str, s3_prefix: str) -> str:
        """
        上傳訓練資料到 S3

        參數:
            local_path: 本地資料路徑
            s3_prefix: S3 前綴路徑

        回傳:
            S3 URI
        """
        # 上傳資料到 S3
        s3_uri = self.session.upload_data(
            path=local_path,
            bucket=self.bucket,
            key_prefix=s3_prefix
        )
        logger.info(f"資料已上傳至: {s3_uri}")
        return s3_uri

    def create_sklearn_estimator(self,
                                   entry_point: str,
                                   instance_type: str = 'ml.m5.xlarge',
                                   instance_count: int = 1,
                                   framework_version: str = '1.0-1',
                                   hyperparameters: Dict[str, Any] = None) -> SKLearn:
        """
        建立 Scikit-learn 訓練器

        參數:
            entry_point: 訓練腳本路徑
            instance_type: 訓練實例類型
            instance_count: 訓練實例數量
            framework_version: Scikit-learn 版本
            hyperparameters: 超參數字典

        回傳:
            SKLearn 估計器
        """
        # 建立 SKLearn 估計器
        sklearn_estimator = SKLearn(
            entry_point=entry_point,
            role=self.role,
            instance_type=instance_type,
            instance_count=instance_count,
            framework_version=framework_version,
            py_version='py3',
            hyperparameters=hyperparameters or {},
            sagemaker_session=self.session
        )

        logger.info(f"SKLearn 估計器已建立")
        logger.info(f"實例類型: {instance_type}")
        logger.info(f"超參數: {hyperparameters}")

        return sklearn_estimator

    def train_model(self, estimator: SKLearn, train_data: str,
                    validation_data: str = None, wait: bool = True) -> str:
        """
        訓練模型

        參數:
            estimator: SageMaker 估計器
            train_data: 訓練資料 S3 URI
            validation_data: 驗證資料 S3 URI
            wait: 是否等待訓練完成

        回傳:
            訓練任務名稱
        """
        # 準備輸入資料
        inputs = {'train': train_data}
        if validation_data:
            inputs['validation'] = validation_data

        # 開始訓練
        logger.info("開始模型訓練...")
        estimator.fit(inputs, wait=wait)

        # 取得訓練任務名稱
        training_job_name = estimator.latest_training_job.name
        logger.info(f"訓練任務已完成: {training_job_name}")

        return training_job_name

    def deploy_model(self, estimator: SKLearn,
                     endpoint_name: str,
                     instance_type: str = 'ml.m5.large',
                     instance_count: int = 1) -> Predictor:
        """
        部署模型到端點

        參數:
            estimator: 已訓練的估計器
            endpoint_name: 端點名稱
            instance_type: 推論實例類型
            instance_count: 推論實例數量

        回傳:
            預測器物件
        """
        logger.info(f"部署模型至端點: {endpoint_name}")

        # 部署模型
        predictor = estimator.deploy(
            initial_instance_count=instance_count,
            instance_type=instance_type,
            endpoint_name=endpoint_name
        )

        logger.info(f"端點已建立: {endpoint_name}")

        return predictor

    def invoke_endpoint(self, endpoint_name: str, payload: Any) -> Dict[str, Any]:
        """
        呼叫推論端點

        參數:
            endpoint_name: 端點名稱
            payload: 輸入資料

        回傳:
            預測結果
        """
        # 建立執行時期客戶端
        runtime_client = boto3.client('sagemaker-runtime', region_name=self.region)

        # 序列化輸入資料
        body = json.dumps(payload)

        # 呼叫端點
        response = runtime_client.invoke_endpoint(
            EndpointName=endpoint_name,
            ContentType='application/json',
            Body=body
        )

        # 解析回應
        result = json.loads(response['Body'].read().decode())

        return result

    def create_model_registry(self, model_package_group_name: str,
                               model_package_group_description: str):
        """
        建立模型註冊群組

        參數:
            model_package_group_name: 模型套件群組名稱
            model_package_group_description: 描述
        """
        try:
            response = self.sm_client.create_model_package_group(
                ModelPackageGroupName=model_package_group_name,
                ModelPackageGroupDescription=model_package_group_description
            )
            logger.info(f"模型註冊群組已建立: {model_package_group_name}")
        except Exception as e:
            logger.warning(f"建立模型註冊群組失敗: {e}")

    def register_model(self, model_package_group_name: str,
                        model_url: str,
                        image_uri: str,
                        approval_status: str = 'PendingManualApproval') -> str:
        """
        註冊模型版本

        參數:
            model_package_group_name: 模型套件群組名稱
            model_url: 模型成品 S3 URL
            image_uri: 推論容器映像 URI
            approval_status: 核准狀態

        回傳:
            模型套件 ARN
        """
        # 定義推論規格
        inference_spec = {
            'Containers': [
                {
                    'Image': image_uri,
                    'ModelDataUrl': model_url
                }
            ],
            'SupportedContentTypes': ['application/json'],
            'SupportedResponseMIMETypes': ['application/json']
        }

        # 註冊模型
        response = self.sm_client.create_model_package(
            ModelPackageGroupName=model_package_group_name,
            InferenceSpecification=inference_spec,
            ModelApprovalStatus=approval_status
        )

        model_package_arn = response['ModelPackageArn']
        logger.info(f"模型已註冊: {model_package_arn}")

        return model_package_arn

    def delete_endpoint(self, endpoint_name: str):
        """
        刪除推論端點

        參數:
            endpoint_name: 端點名稱
        """
        try:
            # 刪除端點
            self.sm_client.delete_endpoint(EndpointName=endpoint_name)
            # 刪除端點配置
            self.sm_client.delete_endpoint_config(EndpointConfigName=endpoint_name)
            logger.info(f"端點已刪除: {endpoint_name}")
        except Exception as e:
            logger.error(f"刪除端點失敗: {e}")

# 訓練腳本範例（train.py）
def create_training_script():
    """
    生成 SageMaker 訓練腳本
    """
    training_script = '''
import argparse
import os
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score
import joblib

def parse_args():
    """解析命令列參數"""
    parser = argparse.ArgumentParser()

    # 超參數
    parser.add_argument('--n-estimators', type=int, default=100)
    parser.add_argument('--max-depth', type=int, default=10)
    parser.add_argument('--min-samples-split', type=int, default=2)

    # SageMaker 特定參數
    parser.add_argument('--model-dir', type=str, default=os.environ.get('SM_MODEL_DIR'))
    parser.add_argument('--train', type=str, default=os.environ.get('SM_CHANNEL_TRAIN'))
    parser.add_argument('--validation', type=str, default=os.environ.get('SM_CHANNEL_VALIDATION'))

    return parser.parse_args()

def train():
    """訓練模型"""
    args = parse_args()

    # 載入訓練資料
    train_data = pd.read_csv(os.path.join(args.train, 'train.csv'))
    X_train = train_data.drop('target', axis=1)
    y_train = train_data['target']

    # 建立模型
    model = RandomForestClassifier(
        n_estimators=args.n_estimators,
        max_depth=args.max_depth,
        min_samples_split=args.min_samples_split,
        random_state=42,
        n_jobs=-1
    )

    # 交叉驗證
    cv_scores = cross_val_score(model, X_train, y_train, cv=5)
    print(f"交叉驗證分數: {cv_scores.mean():.4f} (+/- {cv_scores.std() * 2:.4f})")

    # 訓練最終模型
    model.fit(X_train, y_train)

    # 儲存模型
    model_path = os.path.join(args.model_dir, 'model.joblib')
    joblib.dump(model, model_path)
    print(f"模型已儲存至: {model_path}")

if __name__ == '__main__':
    train()
'''
    return training_script

# 使用範例
if __name__ == "__main__":
    # 建立 SageMaker Pipeline
    pipeline = SageMakerMLPipeline(region='us-west-2')

    # 設定超參數
    hyperparameters = {
        'n-estimators': 200,
        'max-depth': 15,
        'min-samples-split': 5
    }

    # 建立估計器
    estimator = pipeline.create_sklearn_estimator(
        entry_point='train.py',
        instance_type='ml.m5.xlarge',
        hyperparameters=hyperparameters
    )

    # 假設訓練資料已上傳
    train_data_s3 = f"s3://{pipeline.bucket}/data/train"

    # 訓練模型
    # training_job = pipeline.train_model(estimator, train_data_s3)

    # 部署模型
    # predictor = pipeline.deploy_model(
    #     estimator,
    #     endpoint_name='my-sklearn-model',
    #     instance_type='ml.m5.large'
    # )

from google.cloud import aiplatform
from google.cloud import storage
from typing import Dict, Any, List, Optional
import pandas as pd
import logging

# 設定日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class VertexAIMLPipeline:
    """
    Google Cloud Vertex AI 機器學習流程管理類別
    """

    def __init__(self, project_id: str, location: str = 'us-central1'):
        """
        初始化 Vertex AI 客戶端

        參數:
            project_id: GCP 專案 ID
            location: GCP 區域
        """
        self.project_id = project_id
        self.location = location

        # 初始化 Vertex AI SDK
        aiplatform.init(
            project=project_id,
            location=location
        )

        # 建立 Storage 客戶端
        self.storage_client = storage.Client(project=project_id)

        logger.info(f"Vertex AI Pipeline 初始化完成")
        logger.info(f"專案: {project_id}, 區域: {location}")

    def upload_to_gcs(self, local_path: str, bucket_name: str,
                      blob_name: str) -> str:
        """
        上傳檔案到 Google Cloud Storage

        參數:
            local_path: 本地檔案路徑
            bucket_name: GCS bucket 名稱
            blob_name: 目標 blob 名稱

        回傳:
            GCS URI
        """
        bucket = self.storage_client.bucket(bucket_name)
        blob = bucket.blob(blob_name)

        # 上傳檔案
        blob.upload_from_filename(local_path)

        gcs_uri = f"gs://{bucket_name}/{blob_name}"
        logger.info(f"檔案已上傳至: {gcs_uri}")

        return gcs_uri

    def create_dataset(self, display_name: str, gcs_source: str,
                        dataset_type: str = 'tabular') -> aiplatform.TabularDataset:
        """
        建立 Vertex AI 資料集

        參數:
            display_name: 資料集顯示名稱
            gcs_source: GCS 來源 URI
            dataset_type: 資料集類型

        回傳:
            資料集物件
        """
        if dataset_type == 'tabular':
            # 建立表格資料集
            dataset = aiplatform.TabularDataset.create(
                display_name=display_name,
                gcs_source=gcs_source
            )
        elif dataset_type == 'image':
            # 建立影像資料集
            dataset = aiplatform.ImageDataset.create(
                display_name=display_name,
                gcs_source=gcs_source
            )
        else:
            raise ValueError(f"不支援的資料集類型: {dataset_type}")

        logger.info(f"資料集已建立: {dataset.display_name}")
        logger.info(f"資料集 ID: {dataset.resource_name}")

        return dataset

    def train_automl_model(self, dataset: aiplatform.TabularDataset,
                           display_name: str,
                           target_column: str,
                           optimization_objective: str = 'minimize-log-loss',
                           budget_milli_node_hours: int = 1000) -> aiplatform.Model:
        """
        使用 AutoML 訓練表格模型

        參數:
            dataset: Vertex AI 資料集
            display_name: 模型顯示名稱
            target_column: 目標欄位名稱
            optimization_objective: 最佳化目標
            budget_milli_node_hours: 訓練預算（毫節點小時）

        回傳:
            訓練後的模型
        """
        logger.info(f"開始 AutoML 訓練: {display_name}")
        logger.info(f"目標欄位: {target_column}")
        logger.info(f"最佳化目標: {optimization_objective}")

        # 建立 AutoML 訓練任務
        job = aiplatform.AutoMLTabularTrainingJob(
            display_name=f"{display_name}_training",
            optimization_prediction_type='classification',
            optimization_objective=optimization_objective
        )

        # 執行訓練
        model = job.run(
            dataset=dataset,
            target_column=target_column,
            budget_milli_node_hours=budget_milli_node_hours,
            model_display_name=display_name
        )

        logger.info(f"模型訓練完成: {model.display_name}")
        logger.info(f"模型 ID: {model.resource_name}")

        return model

    def train_custom_model(self, display_name: str,
                           script_path: str,
                           container_uri: str,
                           model_serving_container_uri: str,
                           args: List[str] = None,
                           replica_count: int = 1,
                           machine_type: str = 'n1-standard-4',
                           accelerator_type: str = None,
                           accelerator_count: int = 0) -> aiplatform.Model:
        """
        執行自訂訓練任務

        參數:
            display_name: 任務顯示名稱
            script_path: 訓練腳本路徑
            container_uri: 訓練容器映像 URI
            model_serving_container_uri: 推論容器映像 URI
            args: 腳本參數
            replica_count: 副本數量
            machine_type: 機器類型
            accelerator_type: 加速器類型
            accelerator_count: 加速器數量

        回傳:
            訓練後的模型
        """
        logger.info(f"開始自訂訓練: {display_name}")

        # 建立自訂訓練任務
        job = aiplatform.CustomTrainingJob(
            display_name=display_name,
            script_path=script_path,
            container_uri=container_uri,
            model_serving_container_image_uri=model_serving_container_uri
        )

        # 執行訓練
        model = job.run(
            args=args or [],
            replica_count=replica_count,
            machine_type=machine_type,
            accelerator_type=accelerator_type,
            accelerator_count=accelerator_count
        )

        logger.info(f"自訂訓練完成: {model.display_name}")

        return model

    def deploy_model(self, model: aiplatform.Model,
                     endpoint_display_name: str,
                     machine_type: str = 'n1-standard-2',
                     min_replica_count: int = 1,
                     max_replica_count: int = 3) -> aiplatform.Endpoint:
        """
        部署模型到端點

        參數:
            model: Vertex AI 模型
            endpoint_display_name: 端點顯示名稱
            machine_type: 機器類型
            min_replica_count: 最小副本數
            max_replica_count: 最大副本數

        回傳:
            端點物件
        """
        logger.info(f"部署模型至端點: {endpoint_display_name}")

        # 建立端點
        endpoint = aiplatform.Endpoint.create(
            display_name=endpoint_display_name
        )

        # 部署模型到端點
        model.deploy(
            endpoint=endpoint,
            deployed_model_display_name=model.display_name,
            machine_type=machine_type,
            min_replica_count=min_replica_count,
            max_replica_count=max_replica_count,
            traffic_split={"0": 100}  # 100% 流量到此模型
        )

        logger.info(f"模型已部署至端點: {endpoint.display_name}")
        logger.info(f"端點 ID: {endpoint.resource_name}")

        return endpoint

    def predict(self, endpoint: aiplatform.Endpoint,
                instances: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        執行線上預測

        參數:
            endpoint: Vertex AI 端點
            instances: 輸入實例列表

        回傳:
            預測結果列表
        """
        # 執行預測
        predictions = endpoint.predict(instances=instances)

        return predictions.predictions

    def batch_predict(self, model: aiplatform.Model,
                      gcs_source: str,
                      gcs_destination_prefix: str,
                      machine_type: str = 'n1-standard-2',
                      starting_replica_count: int = 1) -> aiplatform.BatchPredictionJob:
        """
        執行批次預測

        參數:
            model: Vertex AI 模型
            gcs_source: 輸入資料 GCS URI
            gcs_destination_prefix: 輸出資料 GCS 前綴
            machine_type: 機器類型
            starting_replica_count: 起始副本數

        回傳:
            批次預測任務
        """
        logger.info(f"開始批次預測")
        logger.info(f"輸入: {gcs_source}")
        logger.info(f"輸出: {gcs_destination_prefix}")

        # 建立批次預測任務
        batch_prediction_job = model.batch_predict(
            job_display_name=f"batch_predict_{model.display_name}",
            gcs_source=gcs_source,
            gcs_destination_prefix=gcs_destination_prefix,
            machine_type=machine_type,
            starting_replica_count=starting_replica_count
        )

        logger.info(f"批次預測任務已完成")

        return batch_prediction_job

# 使用範例
if __name__ == "__main__":
    # 初始化 Vertex AI Pipeline
    pipeline = VertexAIMLPipeline(
        project_id='my-gcp-project',
        location='us-central1'
    )

    # 上傳資料
    # gcs_uri = pipeline.upload_to_gcs(
    #     local_path='data.csv',
    #     bucket_name='my-bucket',
    #     blob_name='datasets/data.csv'
    # )

    # 建立資料集
    # dataset = pipeline.create_dataset(
    #     display_name='my-tabular-dataset',
    #     gcs_source=gcs_uri
    # )

    # 訓練 AutoML 模型
    # model = pipeline.train_automl_model(
    #     dataset=dataset,
    #     display_name='my-automl-model',
    #     target_column='target'
    # )

    # 部署模型
    # endpoint = pipeline.deploy_model(
    #     model=model,
    #     endpoint_display_name='my-endpoint'
    # )

from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineEndpoint,
    ManagedOnlineDeployment,
    Model,
    Environment,
    CodeConfiguration
)
from azure.identity import DefaultAzureCredential
from azure.ai.ml import command
from azure.ai.ml import Input, Output
import logging

# 設定日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AzureMLPipeline:
    """
    Azure Machine Learning 機器學習流程管理類別
    """

    def __init__(self, subscription_id: str, resource_group: str,
                 workspace_name: str):
        """
        初始化 Azure ML 客戶端

        參數:
            subscription_id: Azure 訂閱 ID
            resource_group: 資源群組名稱
            workspace_name: 工作區名稱
        """
        # 取得認證
        credential = DefaultAzureCredential()

        # 建立 ML 客戶端
        self.ml_client = MLClient(
            credential=credential,
            subscription_id=subscription_id,
            resource_group_name=resource_group,
            workspace_name=workspace_name
        )

        self.workspace_name = workspace_name

        logger.info(f"Azure ML Pipeline 初始化完成")
        logger.info(f"工作區: {workspace_name}")

    def create_compute_cluster(self, compute_name: str,
                                vm_size: str = 'Standard_DS3_v2',
                                min_instances: int = 0,
                                max_instances: int = 4):
        """
        建立計算叢集

        參數:
            compute_name: 計算叢集名稱
            vm_size: VM 大小
            min_instances: 最小實例數
            max_instances: 最大實例數
        """
        from azure.ai.ml.entities import AmlCompute

        # 檢查計算叢集是否已存在
        try:
            compute = self.ml_client.compute.get(compute_name)
            logger.info(f"計算叢集已存在: {compute_name}")
        except Exception:
            # 建立新的計算叢集
            compute = AmlCompute(
                name=compute_name,
                type="amlcompute",
                size=vm_size,
                min_instances=min_instances,
                max_instances=max_instances
            )
            self.ml_client.compute.begin_create_or_update(compute).result()
            logger.info(f"計算叢集已建立: {compute_name}")

    def register_data(self, name: str, path: str,
                      description: str = '') -> str:
        """
        註冊資料資產

        參數:
            name: 資料資產名稱
            path: 資料路徑（本地或 Azure 儲存體）
            description: 描述

        回傳:
            資料資產 ID
        """
        from azure.ai.ml.entities import Data
        from azure.ai.ml.constants import AssetTypes

        # 建立資料資產
        data_asset = Data(
            path=path,
            type=AssetTypes.URI_FILE,
            description=description,
            name=name
        )

        # 註冊資料資產
        registered_data = self.ml_client.data.create_or_update(data_asset)

        logger.info(f"資料資產已註冊: {registered_data.name}")
        logger.info(f"版本: {registered_data.version}")

        return registered_data.id

    def create_training_job(self, display_name: str,
                             code_path: str,
                             command_str: str,
                             environment_name: str,
                             compute_name: str,
                             inputs: dict = None,
                             outputs: dict = None):
        """
        建立訓練任務

        參數:
            display_name: 任務顯示名稱
            code_path: 程式碼路徑
            command_str: 執行命令
            environment_name: 環境名稱
            compute_name: 計算叢集名稱
            inputs: 輸入參數
            outputs: 輸出參數

        回傳:
            訓練任務
        """
        # 建立命令任務
        job = command(
            display_name=display_name,
            code=code_path,
            command=command_str,
            environment=environment_name,
            compute=compute_name,
            inputs=inputs or {},
            outputs=outputs or {}
        )

        # 提交任務
        returned_job = self.ml_client.jobs.create_or_update(job)

        logger.info(f"訓練任務已提交: {returned_job.name}")
        logger.info(f"狀態: {returned_job.status}")

        return returned_job

    def register_model(self, name: str, path: str,
                        description: str = '',
                        tags: dict = None) -> Model:
        """
        註冊模型

        參數:
            name: 模型名稱
            path: 模型路徑
            description: 描述
            tags: 標籤

        回傳:
            註冊的模型
        """
        # 建立模型物件
        model = Model(
            path=path,
            name=name,
            description=description,
            tags=tags or {}
        )

        # 註冊模型
        registered_model = self.ml_client.models.create_or_update(model)

        logger.info(f"模型已註冊: {registered_model.name}")
        logger.info(f"版本: {registered_model.version}")

        return registered_model

    def create_online_endpoint(self, endpoint_name: str,
                                description: str = '',
                                auth_mode: str = 'key'):
        """
        建立線上端點

        參數:
            endpoint_name: 端點名稱
            description: 描述
            auth_mode: 認證模式

        回傳:
            端點物件
        """
        # 建立端點定義
        endpoint = ManagedOnlineEndpoint(
            name=endpoint_name,
            description=description,
            auth_mode=auth_mode
        )

        # 建立端點
        self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()

        logger.info(f"線上端點已建立: {endpoint_name}")

        return endpoint

    def deploy_to_endpoint(self, endpoint_name: str,
                           deployment_name: str,
                           model_name: str,
                           environment_name: str,
                           code_path: str,
                           scoring_script: str,
                           instance_type: str = 'Standard_DS3_v2',
                           instance_count: int = 1):
        """
        部署模型到端點

        參數:
            endpoint_name: 端點名稱
            deployment_name: 部署名稱
            model_name: 模型名稱
            environment_name: 環境名稱
            code_path: 評分程式碼路徑
            scoring_script: 評分腳本名稱
            instance_type: 實例類型
            instance_count: 實例數量
        """
        # 取得模型
        model = self.ml_client.models.get(model_name, label='latest')

        # 建立部署定義
        deployment = ManagedOnlineDeployment(
            name=deployment_name,
            endpoint_name=endpoint_name,
            model=model,
            environment=environment_name,
            code_configuration=CodeConfiguration(
                code=code_path,
                scoring_script=scoring_script
            ),
            instance_type=instance_type,
            instance_count=instance_count
        )

        # 執行部署
        self.ml_client.online_deployments.begin_create_or_update(
            deployment
        ).result()

        # 設定流量
        endpoint = self.ml_client.online_endpoints.get(endpoint_name)
        endpoint.traffic = {deployment_name: 100}
        self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()

        logger.info(f"模型已部署: {deployment_name}")

    def invoke_endpoint(self, endpoint_name: str, request_file: str):
        """
        呼叫線上端點

        參數:
            endpoint_name: 端點名稱
            request_file: 請求檔案路徑

        回傳:
            預測結果
        """
        # 呼叫端點
        response = self.ml_client.online_endpoints.invoke(
            endpoint_name=endpoint_name,
            request_file=request_file
        )

        return response

# 使用範例
if __name__ == "__main__":
    # 初始化 Azure ML Pipeline
    pipeline = AzureMLPipeline(
        subscription_id='your-subscription-id',
        resource_group='your-resource-group',
        workspace_name='your-workspace'
    )

    # 建立計算叢集
    # pipeline.create_compute_cluster(
    #     compute_name='cpu-cluster',
    #     vm_size='Standard_DS3_v2',
    #     max_instances=4
    # )

    # 建立訓練任務
    # job = pipeline.create_training_job(
    #     display_name='sklearn-training',
    #     code_path='./src',
    #     command_str='python train.py --data ${{inputs.data}}',
    #     environment_name='sklearn-env',
    #     compute_name='cpu-cluster'
    # )

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from typing import Dict, Any, List, Tuple
import logging
from datetime import datetime, timedelta

# 設定日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CloudAIOpsSystem:
    """
    雲端 AIOps 系統
    提供異常檢測、根因分析和自動化維運功能
    """

    def __init__(self):
        """
        初始化 AIOps 系統
        """
        # 異常檢測模型
        self.anomaly_detector = IsolationForest(
            contamination=0.1,
            random_state=42,
            n_estimators=100
        )
        # 特徵縮放器
        self.scaler = StandardScaler()
        # 是否已訓練
        self.is_trained = False
        # 告警閾值
        self.alert_threshold = -0.5
        # 歷史資料
        self.history = []

    def preprocess_metrics(self, metrics: pd.DataFrame) -> np.ndarray:
        """
        預處理效能指標

        參數:
            metrics: 效能指標 DataFrame

        回傳:
            縮放後的特徵陣列
        """
        # 選擇數值欄位
        numeric_cols = metrics.select_dtypes(include=[np.number]).columns
        features = metrics[numeric_cols].values

        # 處理缺失值
        features = np.nan_to_num(features, nan=0.0)

        # 特徵縮放
        if self.is_trained:
            features_scaled = self.scaler.transform(features)
        else:
            features_scaled = self.scaler.fit_transform(features)

        return features_scaled

    def train(self, historical_metrics: pd.DataFrame):
        """
        使用歷史資料訓練異常檢測模型

        參數:
            historical_metrics: 歷史效能指標
        """
        logger.info("開始訓練異常檢測模型...")

        # 預處理資料
        features = self.preprocess_metrics(historical_metrics)

        # 訓練模型
        self.anomaly_detector.fit(features)
        self.is_trained = True

        logger.info(f"模型訓練完成，使用 {len(features)} 筆資料")

    def detect_anomalies(self, current_metrics: pd.DataFrame) -> List[Dict[str, Any]]:
        """
        檢測當前指標中的異常

        參數:
            current_metrics: 當前效能指標

        回傳:
            異常檢測結果列表
        """
        if not self.is_trained:
            raise RuntimeError("模型尚未訓練，請先呼叫 train() 方法")

        # 預處理資料
        features = self.preprocess_metrics(current_metrics)

        # 計算異常分數
        scores = self.anomaly_detector.decision_function(features)

        # 預測異常
        predictions = self.anomaly_detector.predict(features)

        # 整理結果
        results = []
        for i in range(len(predictions)):
            result = {
                'index': i,
                'timestamp': current_metrics.index[i] if hasattr(current_metrics, 'index') else i,
                'anomaly_score': float(scores[i]),
                'is_anomaly': predictions[i] == -1,
                'severity': self._calculate_severity(scores[i])
            }

            # 如果是異常，加入詳細資訊
            if result['is_anomaly']:
                result['anomaly_features'] = self._identify_anomaly_features(
                    current_metrics.iloc[i],
                    features[i]
                )

            results.append(result)

        # 記錄檢測到的異常數量
        anomaly_count = sum(1 for r in results if r['is_anomaly'])
        if anomaly_count > 0:
            logger.warning(f"檢測到 {anomaly_count} 個異常")

        return results

    def _calculate_severity(self, score: float) -> str:
        """
        根據異常分數計算嚴重程度

        參數:
            score: 異常分數

        回傳:
            嚴重程度（low/medium/high/critical）
        """
        if score > 0:
            return 'normal'
        elif score > -0.3:
            return 'low'
        elif score > -0.5:
            return 'medium'
        elif score > -0.7:
            return 'high'
        else:
            return 'critical'

    def _identify_anomaly_features(self, row: pd.Series,
                                    scaled_features: np.ndarray) -> List[str]:
        """
        識別導致異常的特徵

        參數:
            row: 原始資料行
            scaled_features: 縮放後的特徵

        回傳:
            異常特徵列表
        """
        anomaly_features = []

        # 檢查每個特徵的偏離程度
        for i, (col, value) in enumerate(row.items()):
            if isinstance(value, (int, float)):
                # 如果縮放後的值超過 2 個標準差，視為異常特徵
                if abs(scaled_features[i]) > 2:
                    anomaly_features.append(f"{col}: {value}")

        return anomaly_features

    def root_cause_analysis(self, anomalies: List[Dict[str, Any]],
                            metrics: pd.DataFrame) -> Dict[str, Any]:
        """
        執行根因分析

        參數:
            anomalies: 異常檢測結果
            metrics: 效能指標

        回傳:
            根因分析結果
        """
        # 過濾出真正的異常
        true_anomalies = [a for a in anomalies if a['is_anomaly']]

        if not true_anomalies:
            return {'status': 'no_anomalies', 'root_causes': []}

        # 統計異常特徵出現頻率
        feature_counts = {}
        for anomaly in true_anomalies:
            for feature in anomaly.get('anomaly_features', []):
                feature_name = feature.split(':')[0]
                feature_counts[feature_name] = feature_counts.get(feature_name, 0) + 1

        # 排序找出最可能的根因
        sorted_features = sorted(
            feature_counts.items(),
            key=lambda x: x[1],
            reverse=True
        )

        # 生成根因分析報告
        root_causes = []
        for feature, count in sorted_features[:5]:  # 取前 5 個
            confidence = count / len(true_anomalies) * 100
            root_causes.append({
                'feature': feature,
                'occurrence_count': count,
                'confidence': f"{confidence:.1f}%"
            })

        return {
            'status': 'analysis_complete',
            'total_anomalies': len(true_anomalies),
            'root_causes': root_causes,
            'recommendations': self._generate_recommendations(root_causes)
        }

    def _generate_recommendations(self, root_causes: List[Dict[str, Any]]) -> List[str]:
        """
        根據根因生成維運建議

        參數:
            root_causes: 根因列表

        回傳:
            維運建議列表
        """
        recommendations = []

        for cause in root_causes:
            feature = cause['feature'].lower()

            if 'cpu' in feature:
                recommendations.append("建議檢查 CPU 使用率，考慮水平擴展或最佳化應用程式")
            elif 'memory' in feature:
                recommendations.append("建議檢查記憶體使用情況，可能存在記憶體洩漏")
            elif 'disk' in feature:
                recommendations.append("建議檢查磁碟空間和 I/O 效能，考慮清理或擴充儲存")
            elif 'network' in feature:
                recommendations.append("建議檢查網路延遲和頻寬，可能存在網路擁塞")
            elif 'error' in feature or 'exception' in feature:
                recommendations.append("建議檢查錯誤日誌，可能存在應用程式錯誤")
            elif 'latency' in feature or 'response' in feature:
                recommendations.append("建議檢查服務回應時間，可能需要效能最佳化")

        return recommendations if recommendations else ["建議進行深入調查以確定問題根因"]

class AutoRemediation:
    """
    自動修復系統
    根據異常類型執行自動化修復動作
    """

    def __init__(self, cloud_provider: str = 'aws'):
        """
        初始化自動修復系統

        參數:
            cloud_provider: 雲端服務提供商
        """
        self.cloud_provider = cloud_provider
        self.remediation_history = []

    def execute_remediation(self, anomaly: Dict[str, Any],
                             action: str) -> Dict[str, Any]:
        """
        執行修復動作

        參數:
            anomaly: 異常資訊
            action: 修復動作類型

        回傳:
            執行結果
        """
        result = {
            'timestamp': datetime.now().isoformat(),
            'anomaly': anomaly,
            'action': action,
            'status': 'pending'
        }

        try:
            if action == 'scale_out':
                result['details'] = self._scale_out_instances()
            elif action == 'restart_service':
                result['details'] = self._restart_service()
            elif action == 'clear_cache':
                result['details'] = self._clear_cache()
            elif action == 'rollback_deployment':
                result['details'] = self._rollback_deployment()
            else:
                result['details'] = {'message': f'未知的修復動作: {action}'}
                result['status'] = 'failed'
                return result

            result['status'] = 'success'

        except Exception as e:
            result['status'] = 'failed'
            result['error'] = str(e)
            logger.error(f"修復動作執行失敗: {e}")

        # 記錄歷史
        self.remediation_history.append(result)

        return result

    def _scale_out_instances(self) -> Dict[str, Any]:
        """
        執行水平擴展
        """
        logger.info("執行水平擴展...")
        # 實際實作會呼叫雲端 API
        return {
            'action': 'scale_out',
            'new_instance_count': 'N/A (模擬)',
            'message': '已觸發自動擴展'
        }

    def _restart_service(self) -> Dict[str, Any]:
        """
        重啟服務
        """
        logger.info("重啟服務...")
        # 實際實作會呼叫雲端 API 或容器編排系統
        return {
            'action': 'restart_service',
            'message': '服務重啟已觸發'
        }

    def _clear_cache(self) -> Dict[str, Any]:
        """
        清除快取
        """
        logger.info("清除快取...")
        # 實際實作會呼叫快取服務 API
        return {
            'action': 'clear_cache',
            'message': '快取清除已完成'
        }

    def _rollback_deployment(self) -> Dict[str, Any]:
        """
        回滾部署
        """
        logger.info("回滾部署...")
        # 實際實作會呼叫 CI/CD 系統 API
        return {
            'action': 'rollback_deployment',
            'message': '部署回滾已觸發'
        }

# 使用範例
if __name__ == "__main__":
    # 建立模擬的歷史資料
    np.random.seed(42)
    n_samples = 1000

    historical_data = pd.DataFrame({
        'cpu_usage': np.random.normal(50, 10, n_samples),
        'memory_usage': np.random.normal(60, 15, n_samples),
        'disk_io': np.random.normal(100, 20, n_samples),
        'network_in': np.random.normal(500, 100, n_samples),
        'network_out': np.random.normal(300, 80, n_samples),
        'response_time': np.random.normal(200, 50, n_samples),
        'error_rate': np.random.exponential(0.01, n_samples)
    })

    # 建立 AIOps 系統
    aiops = CloudAIOpsSystem()

    # 訓練模型
    aiops.train(historical_data)

    # 建立包含異常的測試資料
    test_data = pd.DataFrame({
        'cpu_usage': [95, 52, 48, 99, 51],  # 第 0 和第 3 個是異常
        'memory_usage': [85, 58, 62, 90, 59],
        'disk_io': [180, 95, 105, 200, 98],
        'network_in': [480, 520, 490, 510, 500],
        'network_out': [310, 290, 305, 300, 295],
        'response_time': [450, 195, 210, 500, 185],
        'error_rate': [0.15, 0.01, 0.02, 0.20, 0.01]
    })

    # 檢測異常
    anomalies = aiops.detect_anomalies(test_data)

    # 輸出結果
    for result in anomalies:
        if result['is_anomaly']:
            print(f"異常檢測 - 索引: {result['index']}")
            print(f"  嚴重程度: {result['severity']}")
            print(f"  異常分數: {result['anomaly_score']:.4f}")
            print(f"  異常特徵: {result['anomaly_features']}")
            print()

    # 執行根因分析
    rca_result = aiops.root_cause_analysis(anomalies, test_data)
    print("根因分析結果:")
    print(f"  總異常數: {rca_result['total_anomalies']}")
    print(f"  可能的根因:")
    for cause in rca_result['root_causes']:
        print(f"    - {cause['feature']}: {cause['confidence']} 信心度")
    print(f"  建議:")
    for rec in rca_result['recommendations']:
        print(f"    - {rec}")

from google.cloud import vision
from google.cloud import language_v1
from google.cloud import speech
from google.cloud import translate_v2 as translate
from google.cloud import texttospeech
from typing import Dict, Any, List
import io
import logging

# 設定日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CloudAIServices:
    """
    雲端 AI 服務整合類別
    提供視覺、語言、語音等 AI 服務的統一介面
    """

    def __init__(self):
        """
        初始化各項 AI 服務客戶端
        """
        # 視覺 AI 客戶端
        self.vision_client = vision.ImageAnnotatorClient()
        # 自然語言處理客戶端
        self.language_client = language_v1.LanguageServiceClient()
        # 語音辨識客戶端
        self.speech_client = speech.SpeechClient()
        # 翻譯客戶端
        self.translate_client = translate.Client()
        # 語音合成客戶端
        self.tts_client = texttospeech.TextToSpeechClient()

        logger.info("雲端 AI 服務客戶端初始化完成")

    def analyze_image(self, image_path: str,
                      features: List[str] = None) -> Dict[str, Any]:
        """
        分析圖片內容

        參數:
            image_path: 圖片路徑（本地或 GCS URI）
            features: 要分析的特徵列表

        回傳:
            分析結果
        """
        # 載入圖片
        if image_path.startswith('gs://'):
            # 從 GCS 載入
            image = vision.Image()
            image.source.image_uri = image_path
        else:
            # 從本地載入
            with io.open(image_path, 'rb') as image_file:
                content = image_file.read()
            image = vision.Image(content=content)

        # 設定要分析的特徵
        if features is None:
            features = ['labels', 'objects', 'faces', 'text']

        results = {}

        # 標籤偵測
        if 'labels' in features:
            response = self.vision_client.label_detection(image=image)
            results['labels'] = [
                {'description': label.description, 'score': label.score}
                for label in response.label_annotations
            ]

        # 物件偵測
        if 'objects' in features:
            response = self.vision_client.object_localization(image=image)
            results['objects'] = [
                {
                    'name': obj.name,
                    'score': obj.score,
                    'bounding_box': [
                        {'x': vertex.x, 'y': vertex.y}
                        for vertex in obj.bounding_poly.normalized_vertices
                    ]
                }
                for obj in response.localized_object_annotations
            ]

        # 臉部偵測
        if 'faces' in features:
            response = self.vision_client.face_detection(image=image)
            results['faces'] = [
                {
                    'joy': face.joy_likelihood.name,
                    'sorrow': face.sorrow_likelihood.name,
                    'anger': face.anger_likelihood.name,
                    'surprise': face.surprise_likelihood.name
                }
                for face in response.face_annotations
            ]

        # 文字辨識（OCR）
        if 'text' in features:
            response = self.vision_client.text_detection(image=image)
            if response.text_annotations:
                results['text'] = response.text_annotations[0].description
            else:
                results['text'] = ''

        logger.info(f"圖片分析完成: {image_path}")

        return results

    def analyze_sentiment(self, text: str) -> Dict[str, Any]:
        """
        分析文字情感

        參數:
            text: 要分析的文字

        回傳:
            情感分析結果
        """
        # 建立文件物件
        document = language_v1.Document(
            content=text,
            type_=language_v1.Document.Type.PLAIN_TEXT
        )

        # 執行情感分析
        response = self.language_client.analyze_sentiment(
            request={'document': document}
        )

        # 整理結果
        result = {
            'document_sentiment': {
                'score': response.document_sentiment.score,
                'magnitude': response.document_sentiment.magnitude
            },
            'sentences': [
                {
                    'text': sentence.text.content,
                    'sentiment': {
                        'score': sentence.sentiment.score,
                        'magnitude': sentence.sentiment.magnitude
                    }
                }
                for sentence in response.sentences
            ]
        }

        # 解釋情感分數
        score = response.document_sentiment.score
        if score > 0.25:
            result['overall_sentiment'] = 'positive'
        elif score < -0.25:
            result['overall_sentiment'] = 'negative'
        else:
            result['overall_sentiment'] = 'neutral'

        logger.info(f"情感分析完成: {result['overall_sentiment']}")

        return result

    def extract_entities(self, text: str) -> List[Dict[str, Any]]:
        """
        提取文字中的實體

        參數:
            text: 要分析的文字

        回傳:
            實體列表
        """
        # 建立文件物件
        document = language_v1.Document(
            content=text,
            type_=language_v1.Document.Type.PLAIN_TEXT
        )

        # 執行實體分析
        response = self.language_client.analyze_entities(
            request={'document': document}
        )

        # 整理結果
        entities = [
            {
                'name': entity.name,
                'type': language_v1.Entity.Type(entity.type_).name,
                'salience': entity.salience,
                'mentions': [
                    {
                        'text': mention.text.content,
                        'type': language_v1.EntityMention.Type(mention.type_).name
                    }
                    for mention in entity.mentions
                ]
            }
            for entity in response.entities
        ]

        logger.info(f"實體提取完成，找到 {len(entities)} 個實體")

        return entities

    def transcribe_audio(self, audio_path: str,
                          language_code: str = 'zh-TW',
                          sample_rate: int = 16000) -> Dict[str, Any]:
        """
        語音轉文字

        參數:
            audio_path: 音訊檔案路徑
            language_code: 語言代碼
            sample_rate: 取樣率

        回傳:
            轉錄結果
        """
        # 載入音訊
        if audio_path.startswith('gs://'):
            audio = speech.RecognitionAudio(uri=audio_path)
        else:
            with io.open(audio_path, 'rb') as audio_file:
                content = audio_file.read()
            audio = speech.RecognitionAudio(content=content)

        # 設定組態
        config = speech.RecognitionConfig(
            encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
            sample_rate_hertz=sample_rate,
            language_code=language_code,
            enable_automatic_punctuation=True,
            enable_word_time_offsets=True
        )

        # 執行語音辨識
        response = self.speech_client.recognize(config=config, audio=audio)

        # 整理結果
        results = []
        for result in response.results:
            alternative = result.alternatives[0]
            results.append({
                'transcript': alternative.transcript,
                'confidence': alternative.confidence,
                'words': [
                    {
                        'word': word.word,
                        'start_time': word.start_time.total_seconds(),
                        'end_time': word.end_time.total_seconds()
                    }
                    for word in alternative.words
                ]
            })

        logger.info(f"語音轉錄完成: {audio_path}")

        return {'results': results}

    def translate_text(self, text: str,
                        target_language: str,
                        source_language: str = None) -> Dict[str, Any]:
        """
        翻譯文字

        參數:
            text: 要翻譯的文字
            target_language: 目標語言代碼
            source_language: 來源語言代碼（可選，會自動偵測）

        回傳:
            翻譯結果
        """
        # 執行翻譯
        if source_language:
            result = self.translate_client.translate(
                text,
                target_language=target_language,
                source_language=source_language
            )
        else:
            result = self.translate_client.translate(
                text,
                target_language=target_language
            )

        translation_result = {
            'input': text,
            'translated_text': result['translatedText'],
            'detected_language': result.get('detectedSourceLanguage', source_language),
            'target_language': target_language
        }

        logger.info(f"翻譯完成: {result.get('detectedSourceLanguage', source_language)} -> {target_language}")

        return translation_result

    def synthesize_speech(self, text: str,
                           output_path: str,
                           language_code: str = 'zh-TW',
                           voice_name: str = 'zh-TW-Standard-A',
                           audio_encoding: str = 'MP3') -> str:
        """
        文字轉語音

        參數:
            text: 要轉換的文字
            output_path: 輸出檔案路徑
            language_code: 語言代碼
            voice_name: 語音名稱
            audio_encoding: 音訊編碼格式

        回傳:
            輸出檔案路徑
        """
        # 設定輸入文字
        synthesis_input = texttospeech.SynthesisInput(text=text)

        # 設定語音參數
        voice = texttospeech.VoiceSelectionParams(
            language_code=language_code,
            name=voice_name
        )

        # 設定音訊組態
        audio_config = texttospeech.AudioConfig(
            audio_encoding=getattr(texttospeech.AudioEncoding, audio_encoding)
        )

        # 執行語音合成
        response = self.tts_client.synthesize_speech(
            input=synthesis_input,
            voice=voice,
            audio_config=audio_config
        )

        # 儲存音訊檔案
        with open(output_path, 'wb') as out:
            out.write(response.audio_content)

        logger.info(f"語音合成完成: {output_path}")

        return output_path

# 使用範例
if __name__ == "__main__":
    # 建立 AI 服務客戶端
    ai_services = CloudAIServices()

    # 情感分析範例
    text = "這個產品真的很棒，使用體驗非常好，我會推薦給朋友！"
    sentiment = ai_services.analyze_sentiment(text)
    print(f"情感分析結果: {sentiment['overall_sentiment']}")
    print(f"情感分數: {sentiment['document_sentiment']['score']:.2f}")

    # 實體提取範例
    text = "蘋果公司在舊金山發表了新款 iPhone，執行長 Tim Cook 親自介紹產品特色。"
    entities = ai_services.extract_entities(text)
    print("\n提取的實體:")
    for entity in entities:
        print(f"  - {entity['name']} ({entity['type']})")

    # 翻譯範例
    text = "雲端運算與人工智慧的結合正在改變世界。"
    translation = ai_services.translate_text(text, target_language='en')
    print(f"\n翻譯結果: {translation['translated_text']}")