GCP Vertex-AI Prediction (2024/12/06)

實驗架構圖

• 架構圖
  

• 輔助說明

  • 情境(使用者流程 & 管理員流程 關係示意)
    

    • 使用者觸發Cloud Function(將照片傳到雲端)
    • Cloud Function會將觸發原因(使用者上傳的照片)傳給Manager佈署的AI
    • Manager佈署的AI會回傳結果(照片辨識結果的記憶體位置)給cloud function
    • Cloud Function會回傳結果(處理後的照片辨識結果)給使用者

  • Manager流程(佈署於美國)
    

    • 這個流程屬於"部屬一個集中式的AI"
    • Manager透過Vertex-AI : 部屬AI模型 + Jupyter-Note進行訓練
    • Netbook則是一個Google事先幫Manager建立好的python環境，可以方便的訓練模型。

  • User流程(佈署於台灣)
    

    • 這個流程屬於"部屬分散式的Cloud Function"
    • step1. 使用者將資料放入Storage(In)
    • step2. Storage(In)的資料放入事件，會觸發cloud function
    • step3. function為避免傳送給AI的請求過大，會將資料轉換格式(Base64編碼)
    • step4. function透過EndPoint(可以理解決跟AI模型溝通的介面)，將資料交給AI進行Predict
    • step5. function將Predict完的結果儲存到Storage(Out)

Vertex-AI Building & Testing

• step1. 打開IAM權限
  

• step2. 創建+彙整 資料集

  • 輸入以下CMD創建bucket，把Google提供的訓練集放入
    

    ​​​​​​​​$ export PROJECT_ID=$DEVSHELL_PROJECT_ID
    ​​​​​​​​$ export BUCKET=[取一個bucket名稱]
    ​​​​​​​​$ gsutil mb -p $PROJECT_ID -c standard -l [使用者決定的Region] gs://${BUCKET} 
    ​​​​​​​​$ gsutil -m cp -r gs://car_damage_lab_images/* gs://${BUCKET} 
    

  • 輸入以下CMD過濾出CSV訓練檔案
    

    ​​​​​​​​$ gsutil cp gs://car_damage_lab_metadata/data.csv . 
    ​​​​​​​​$ sed -i -e "s/car_damage_lab_images/${BUCKET}/g" ./data.csv
    ​​​​​​​​$ gsutil cp ./data.csv  gs://${BUCKET}
    

    • 該Bucket內部資料
      

  • 在Vertex創建訓練資料
    

    
    
    

• step3. 開始訓練

  • step1.

    
    

  • step2.

    
    

  • step3.

    
    

  • step4.

    
    

  • 訓練完成(過程大概花費1小時40分鐘)

• step4. 測試模型

  • 
    

  • 點選進入你創建的AI模型

  • 
    
    

Build EndPoint and Storages

• 創建Storages(In/Out)
  

  ​​​​$ export PROJECT_ID=$DEVSHELL_PROJECT_ID 
  ​​​​$ export BUCKET=ai-input-storage
  ​​​​$ gsutil mb -p $PROJECT_ID -c standard -l asia-east1 gs://${BUCKET} 
  ​​​​   Creating gs://ai-input-storage/...
  ​​​​$ export BUCKET=ai-output-storage
  ​​​​$ gsutil mb -p $PROJECT_ID -c standard -l asia-east1 gs://${BUCKET} 
  ​​​​   Creating gs://ai-output-storage/...
  

• 輸入以下CMD創建AI-EndPoint

  • 創建

    ​​​​​​​​$ gcloud ai endpoints create \
    ​​​​​​​​--region=us-central1 \
    ​​​​​​​​--display-name="[使用者決定名稱]"
    

  • 部屬AI模型

    ​​​​​​​​# 可以使用以下指令查詢endpoint跟AI-model的ID
    ​​​​​​​​
    ​​​​​​​​$ gcloud ai endpoints list 
    ​​​​​​​​--region=[endpoint的region] \
    ​​​​​​​​--filter="display_name=[endpoint的名稱]" \
    ​​​​​​​​--format="table(name)"
    ​​​​​​​​
    ​​​​​​​​$ gcloud ai models list 
    ​​​​​​​​--region=[AI-model的region] \
    ​​​​​​​​--filter="display_name=[AI-model的名稱]" \
    ​​​​​​​​--format="table(name)"
    

    ​​​​​​​​$ gcloud ai endpoints deploy-model [endpoint的ID] \
    ​​​​​​​​--region=[跟AI-model同一個region] \
    ​​​​​​​​--model=[AI-model的ID] \
    ​​​​​​​​--display-name="[使用者決定的名稱]" \
    ​​​​​​​​--machine-type="[使用者決定硬體規格(例如n1-standard-2)"] \
    ​​​​​​​​--min-replica-count=[使用者可以決定最低會有多少個model同時服務(我設定為1)] \
    ​​​​​​​​--max-replica-count=[使用者可以決定最多會有多少個model同時服務(我設定為1)]
    

  • 檢查

    ​​​​​​​​$ gcloud ai endpoints describe [endpoint id] --region=[endpoint region]
    ​​​​​​​​$ gcloud ai models describe [AI-model的ID] --region=[AI-model的region]
    

Build the cloud function

• Cloud function基本配置

  • 當使用者傳照片到Storage(On)時才觸發，所以要選Storage(On)的名稱
    
  • 為了避免使用者傳輸檔案過大 + 使用者一次傳輸大量照片 : 需要額外設置Memory大小(我測試時有幾次因為照片問題導致Memory不夠用)
    
  • Memory不夠用的錯誤Log消息

• function內部程式碼配置

  • 語言與版本
    

  • main

    ​​​​​​​​import base64
    ​​​​​​​​import json
    ​​​​​​​​import traceback
    ​​​​​​​​from google.cloud import aiplatform_v1 as ai
    ​​​​​​​​from google.cloud import storage
    ​​​​​​​​from google.protobuf.json_format import MessageToDict
    
    
    ​​​​​​​​def image_to_base64(image_data):
    ​​​​​​​​    """
    ​​​​​​​​    將圖片二進制數據轉換為 Base64 編碼
    ​​​​​​​​    """
    ​​​​​​​​    try:
    ​​​​​​​​        base64_string = base64.b64encode(image_data).decode("utf-8")
    ​​​​​​​​        return base64_string
    ​​​​​​​​    except Exception as e:
    ​​​​​​​​        print(f"Error encoding image to Base64: {e}")
    ​​​​​​​​        return None
    
    
    
    ​​​​​​​​def process_prediction(predictions):
    ​​​​​​​​    """
    ​​​​​​​​    將 predictions 結果轉換為 JSON 格式
    ​​​​​​​​    """
    ​​​​​​​​    try:
    ​​​​​​​​        # 如果是字典型資料，直接處理為 JSON 格式
    ​​​​​​​​        if isinstance(predictions, dict):
    ​​​​​​​​            return json.dumps(predictions, indent=2)
    ​​​​​​​​        # 如果是列表型資料，將每個元素處理為字典後序列化
    ​​​​​​​​        elif isinstance(predictions, list):
    ​​​​​​​​            return json.dumps([dict(p) if isinstance(p, dict) else str(p) for p in predictions], indent=2)
    ​​​​​​​​        else:
    ​​​​​​​​            # 無法識別的類型，轉換為字串
    ​​​​​​​​            return json.dumps({"error": "Unsupported prediction format", "data": str(predictions)}, indent=2)
    ​​​​​​​​    except Exception as e:
    ​​​​​​​​        print(f"Error processing predictions: {e}")
    ​​​​​​​​        traceback.print_exc()
    ​​​​​​​​        return json.dumps({"error": "Failed to process predictions", "details": str(e)}, indent=2)
    
    
    
    ​​​​​​​​def save_to_gcs(bucket_name, destination_blob_name, content):
    ​​​​​​​​    """
    ​​​​​​​​    將文字內容保存到指定的 GCS bucket
    ​​​​​​​​    """
    ​​​​​​​​    try:
    ​​​​​​​​        storage_client = storage.Client()
    ​​​​​​​​        bucket = storage_client.bucket(bucket_name)
    ​​​​​​​​        blob = bucket.blob(destination_blob_name)
    ​​​​​​​​        blob.upload_from_string(content)
    ​​​​​​​​        print(f"Saved result to GCS: {bucket_name}/{destination_blob_name}")
    ​​​​​​​​    except Exception as e:
    ​​​​​​​​        print(f"Error saving to GCS: {e}")
    
    
    
    ​​​​​​​​def prediction_job(event, context):
    ​​​​​​​​    project_id       = "marktest1"    # 替換為你的 GCP 專案 ID
    ​​​​​​​​    location         = "us-central1"  # 替換為你的地區
    ​​​​​​​​    api_endpoint     = f"{location}-aiplatform.googleapis.com"
    ​​​​​​​​    endpoint_id      = "7167559071308972032" # 替換為你的 endpoint ID
    ​​​​​​​​    model_id         = "6680690924483248128" # 替換為你的 AI-model ID
    ​​​​​​​​    model_name       = f"projects/{project_id}/locations/{location}/models/{model_id}"
    ​​​​​​​​    src_bucket_name  = event["bucket"]                     # 來源 Bucket 名稱
    ​​​​​​​​    src_file_name    = event["name"]                       # 來源 Bucket 的檔案名稱
    ​​​​​​​​    dest_bucket_name = "ai-output-storage"                 # 替換為目標 Bucket 名稱
    ​​​​​​​​    dest_file_name   = f"predictions/{src_file_name}.txt"  # 保存結果的檔案名稱
    
    ​​​​​​​​    client_options = {"api_endpoint": api_endpoint}
    ​​​​​​​​    client = ai.PredictionServiceClient(client_options=client_options)
    
    ​​​​​​​​    try:
    ​​​​​​​​        # 從 GCS 下載圖片
    ​​​​​​​​        storage_client = storage.Client()
    ​​​​​​​​        bucket = storage_client.bucket(src_bucket_name)
    ​​​​​​​​        blob = bucket.blob(src_file_name)
    ​​​​​​​​        image_data = blob.download_as_bytes()
    
    ​​​​​​​​        # 將圖片轉換為 Base64
    ​​​​​​​​        encoded_image = image_to_base64(image_data)
    ​​​​​​​​        if not encoded_image:
    ​​​​​​​​            raise ValueError("Failed to encode image to Base64")
    ​​​​​​​​        print("Finished Img Convert")
    
    ​​​​​​​​        # 構造 Vertex AI 請求
    ​​​​​​​​        instance = {
    ​​​​​​​​            "content": encoded_image  # 使用 Base64 編碼
    ​​​​​​​​        }
    ​​​​​​​​        endpoint = f"projects/{project_id}/locations/{location}/endpoints/{endpoint_id}"
    
    ​​​​​​​​        # 預測
    ​​​​​​​​        response = client.predict(endpoint=endpoint, instances=[instance])
    
    ​​​​​​​​        # 處理 predictions 結果
    ​​​​​​​​        predictions = MessageToDict(response._pb)
    ​​​​​​​​        print(predictions)
    ​​​​​​​​        print("Finished prediction")
    
    ​​​​​​​​        # 處理 predictions 為可序列化的格式
    ​​​​​​​​        processed_predictions = process_prediction(predictions)
    ​​​​​​​​        print(processed_predictions , "\n")
    
    ​​​​​​​​        # 將結果保存到 GCS
    ​​​​​​​​        save_to_gcs(dest_bucket_name, dest_file_name, processed_predictions)
    ​​​​​​​​        print("Finished saving predictions")
    
    ​​​​​​​​    except Exception as e:
    ​​​​​​​​        print(f"Error in prediction job: {e}")
    ​​​​​​​​        traceback.print_exc()
    

  • requirement

    ​​​​​​​​# Function dependencies, for example:
    ​​​​​​​​# package>=version
    
    ​​​​​​​​google-cloud-aiplatform==1.34.0
    ​​​​​​​​google-cloud-storage==2.12.0
    ​​​​​​​​protobuf>=3.20.0,<4.0.0
    

實驗結果圖

• 使用者傳輸照片
  

• Storage(Out)的結果

  • Storage內部的確產生對應的preediction
    

  • 查看prediction結果
    
    

結論

• 簡單的建立Vertex AI以及對應的訓練方式

• Cloud Function是如何呼叫AI執行prediction的程式碼實現

• 要注意Vertex的造價，真的很高。

Reference

• Vertex 創建參考
• Vertex SDK 參考
• Vertex HTTPs request format 參考
• Vertex Google 技術文件
• Python Google SDK src code