使用 Google Cloud Pipeline Components

使用 Google Cloud Pipeline Components (GCPC) 時,您可以使用下列 Gemini Enterprise Agent Platform 和 Google Cloud 功能,保護元件和構件。

為元件指定服務帳戶

使用元件時,您可以選擇指定服務帳戶。元件會啟動,並以這個服務帳戶的權限運作。舉例來說,您可以使用下列程式碼指定 ModelDeploy 元件的服務帳戶:

model_deploy_op = ModelDeployOp(model=training_job_run_op.outputs["model"],
    endpoint=endpoint_op.outputs["endpoint"],
    automatic_resources_min_replica_count=1,
    automatic_resources_max_replica_count=1,
    service_account="SERVICE_ACCOUNT_ID@PROJECT_ID.iam.gserviceaccount.com")

更改下列內容:

  • SERVICE_ACCOUNT_ID:服務帳戶的 ID。
  • PROJECT_ID:專案 ID。

進一步瞭解如何使用自訂服務帳戶,以及設定服務帳戶,以便與 Gemini Enterprise Agent Platform 管道搭配使用。

使用 VPC Service Controls 防範資料竊取

VPC Service Controls 可協助您降低 Gemini Enterprise Agent Platform Pipelines 資料竊取的風險。使用 VPC Service Controls 建立 service perimeter 時,Gemini Enterprise Agent Platform Pipelines 和 Google Cloud Pipeline Components 建立的資源和資料會自動受到保護。舉例來說,使用 VPC Service Controls 保護管道時,下列構件無法離開 service perimeter:

  • AutoML 模型的訓練資料
  • 您建立的模型
  • 批次預測要求的結果

進一步瞭解搭配使用 VPC Service Controls 與 Gemini Enterprise Agent Platform。

設定虛擬私有雲網路對等互連

您可以提供額外參數,設定 Google Cloud Pipeline Components 與虛擬私有雲對接。舉例來說,您可以使用下列程式碼,為 EndpointCreate 元件指定虛擬私有雲網路:

endpoint_create_op = EndpointCreateOp(
    project="PROJECT_ID",
    location="REGION",
    display_name="endpoint-display-name",
    network="NETWORK")

更改下列內容:

  • PROJECT_ID:專案 ID。
  • REGION:使用 Gemini Enterprise Agent Platform 的區域。
  • NETWORK:虛擬私有雲網路,例如 "projects/12345/global/networks/myVPC"。

進一步瞭解 Gemini Enterprise Agent Platform 中的虛擬私有雲網路對接。

使用客戶自行管理的加密金鑰 (CMEK)

根據預設, Google Cloud 會使用 Google 代管的加密金鑰,自動加密靜態資料。如果對保護資料的金鑰有特定的法規遵循或監管要求,可以對資源使用客戶自行管理的加密金鑰 (CMEK)。開始使用客戶自行管理的加密金鑰前,請先瞭解 Gemini Enterprise Agent Platform 的 CMEK 優點,以及目前支援 CMEK 的資源。

使用 CMEK 設定元件

在 Cloud Key Management Service 建立金鑰環和金鑰,並授予 Gemini Enterprise Agent Platform 金鑰加密和解密權限後,即可指定金鑰做為建立參數之一,建立支援 CMEK 的新元件。舉例來說,您可以使用下列程式碼,為 ModelBatchPredict 元件指定索引鍵:

model_batch_predict_op = ModelBatchPredictOp(project="PROJECT_ID",
    model=model_upload_op.outputs["model"],
    encryption_spec_key_name="projects/PROJECT_ID/locations/LOCATION_ID/keyRings/KEY_RING_NAME/cryptoKeys/KEY_NAME")

更改下列內容:

  • PROJECT_ID: Google Cloud 專案 ID。
  • LOCATION_ID:有效的位置或區域 ID,例如 us-central1。
  • KEY_RING_NAME:CMEK 的金鑰環名稱,如要進一步瞭解金鑰環,請參閱「Cloud KMS 資源」。
  • KEY_NAME:CMEK 金鑰名稱。

注意: Google Cloud 非 Gemini Enterprise Agent Platform 的元件可能需要額外權限。例如,BigQuery 元件可能需要加密和解密權限。此外,CMEK 金鑰的位置必須與元件的位置相同。舉例來說,如果 BigQuery 元件從位於美國多區域位置的資料集載入資料,CMEK 金鑰也必須位於美國多區域位置。

在元件中使用或產生構件

Google Cloud SDK 定義了一組 機器學習中繼資料構件類型,可做為元件輸入和輸出內容。部分 Google Cloud 管道元件會使用這些構件做為輸入內容,或產生這些構件做為輸出內容。

本頁說明如何使用及產生這些構件。

使用機器學習構件

在元件 YAML 中使用構件

構件的中繼資料可做為元件的輸入內容。如要準備做為輸入內容的構件,您必須擷取構件,並將其放入元件 YAML 檔案中。

舉例來說,ModelUploadOp 元件會產生 google.VertexModel 構件,可由 ModelDeployOp 元件使用。在元件 YAML 檔案中使用下列程式碼,從輸入內容 (參考資料) 擷取 Gemini Enterprise Agent Platform Model 資源:

"model": "',"{{$.inputs.artifacts['model'].metadata['resourceName']}}", '"'

如需構件中繼資料的完整結構定義,請參閱 Kubeflow GitHub 存放區中的 artifact_types.py 檔案。

在輕量型 Python 元件中使用構件

from kfp.dsl import Artifact, Input

@dsl.component
def classification_model_eval_metrics(
    project: str,
    location: str,  # "us-central1",
    model: Input[Artifact],
) :
   # Consumes the `resourceName` metadata
   model_resource_path = model.metadata["resourceName"]

如需使用 Vertex ML Metadata 構件類型的範例,請參閱「使用表格型資料和 Agent Platform AutoML 訓練分類模型」。

建立機器學習構件

下列程式碼範例說明如何建立 Vertex ML Metadata 構件,讓 GCPC 元件可接受做為輸入內容。

使用匯入工具節點

以下範例會建立 Importer 節點,將新的構件項目註冊至 Vertex ML Metadata。匯入器節點會將構件的 URI 和中繼資料視為原始資料,並封裝成構件。

from google_cloud_pipeline_components import v1
from google_cloud_pipeline_components.types import artifact_types
from kfp.components import importer_node
from kfp import dsl

@dsl.pipeline(name=_PIPELINE_NAME)
def pipeline():
  # Using importer and UnmanagedContainerModel artifact for model upload
  # component.
  importer_spec = importer_node.importer(
      artifact_uri='gs://managed-pipeline-gcpc-e2e-test/automl-tabular/model',
      artifact_class=artifact_types.UnmanagedContainerModel,
      metadata={
          'containerSpec': {
              'imageUri':
                  'us-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server:prod'
          }
      })

  # Consuming the UnmanagedContainerModel artifact for the previous step
  model_upload_with_artifact_op = v1.model.ModelUploadOp(
      project=_GCP_PROJECT_ID,
      location=_GCP_REGION,
      display_name=_MODEL_DISPLAY_NAME,
      unmanaged_container_model=importer_spec.outputs['artifact'])

使用以 Python 函式為基礎的元件

以下範例說明如何直接從 Python 元件輸出 Vertex ML Metadata 構件。

from google_cloud_pipeline_components import v1
from kfp.components import importer_node
from kfp import dsl

@dsl.component(
    base_image='python:3.9',
    packages_to_install=['google-cloud-aiplatform'],
)
# Note currently KFP SDK doesn't support outputting artifacts in `google` namespace.
# Use the base type dsl.Artifact instead.
def return_unmanaged_model(model: dsl.Output[dsl.Artifact]):
  model.metadata['containerSpec'] = {
      'imageUri':
          'us-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server:prod'
  }
  model.uri = f'gs://automl-tabular-pipeline/automl-tabular/model'

@dsl.pipeline(name=_PIPELINE_NAME)
def pipeline():

  unmanaged_model_op = return_unmanaged_model()

  # Consuming the UnmanagedContainerModel artifact for the previous step
  model_upload_with_artifact_op = v1.model.ModelUploadOp(
      project=_GCP_PROJECT_ID,
      location=_GCP_REGION,
      display_name=_MODEL_DISPLAY_NAME,
      unmanaged_container_model=unmanaged_model_op.outputs['model'])

使用自己的容器式元件

以下範例說明如何使用 artifact_types.py 公用程式類別,從容器型元件產生 VertexBatchPredictionJob 構件做為輸出內容。

bp_job_artifact = VertexBatchPredictionJob(
    'batchpredictionjob', vertex_uri_prefix + get_job_response.name,
    get_job_response.name, get_job_response.output_info.bigquery_output_table,
    get_job_response.output_info.bigquery_output_dataset,
    get_job_response.output_info.gcs_output_directory)

    output_artifacts = executor_input_json.get('outputs', {}).get('artifacts', {})
    executor_output['artifacts'] = bp_job_artifact.to_executor_output_artifact(output_artifacts[bp_job_artifact.name])