Crea una pipeline

Agent Platform Pipelines ti consente di orchestrare i tuoi flussi di lavoro di machine learning (ML) in modo serverless. Prima che Agent Platform Pipelines possa orchestrare il flusso di lavoro ML, devi descrivere il flusso di lavoro come una pipeline. Le pipeline ML sono flussi di lavoro ML portatili e scalabili basati su container e servizi Google Cloud .

Questa guida descrive come iniziare a creare pipeline ML.

Quale SDK per le pipeline devo utilizzare?

Agent Platform Pipelines può eseguire pipeline create utilizzando uno dei seguenti SDK:

  • SDK Kubeflow Pipelines 2.0 o versioni successive

  • TensorFlow Extended v0.30.0 o versioni successive

Se utilizzi TensorFlow in un flusso di lavoro ML che elabora terabyte di dati strutturati o dati di testo, ti consigliamo di creare la pipeline utilizzando TFX.

Per altri casi d'uso, ti consigliamo di creare la pipeline utilizzando l'SDK Kubeflow Pipelines. Creando una pipeline con l'SDK Kubeflow Pipelines, puoi implementare il flusso di lavoro creando componenti personalizzati o riutilizzando componenti predefiniti, ad esempio i componenti della pipeline diGoogle Cloud . I componenti della pipeline diGoogle Cloud semplificano l'utilizzo dei servizi di Gemini Enterprise Agent Platform come AutoML nella tua pipeline.

Questa guida descrive come creare pipeline utilizzando l'SDK Kubeflow Pipelines.

Prima di iniziare

Prima di creare ed eseguire le pipeline, segui queste istruzioni per configurare il progetto e l'ambiente di sviluppo Google Cloud .

  1. Per preparare il tuo progetto Google Cloud all'esecuzione delle pipeline ML, segui le istruzioni della guida alla configurazione del progettoGoogle Cloud .

  2. Installa la versione 2 o successiva dell'SDK Kubeflow Pipelines.

    pip install --upgrade "kfp>=2,<3"

  1. Per utilizzare il client Python di Gemini Enterprise Agent Platform nelle tue pipeline, installa le librerie client di Gemini Enterprise Agent Platform v1.7 o successive.

  2. Per utilizzare i servizi di Gemini Enterprise Agent Platform nelle tue pipeline, installa l'SDKGoogle Cloud .

Inizia a creare una pipeline

Per orchestrare il flusso di lavoro ML su Agent Platform Pipelines, devi prima descrivere il flusso di lavoro come pipeline. L'esempio seguente mostra come utilizzare i componenti della pipelineGoogle Cloud con Gemini Enterprise Agent Platform per creare un set di dati, addestrare un modello con AutoML ed eseguire il deployment del modello addestrato per le previsioni.

Prima di eseguire il seguente esempio di codice, devi configurare l'autenticazione.

Come configurare l'autenticazione

Per configurare l'autenticazione, devi creare una chiave dell'account di servizio e impostare una variabile di ambiente per il percorso della chiave dell'account di servizio.

  1. Crea un service account:

    1. Nella console Google Cloud vai a Crea service account.

      Vai a Crea service account

    2. Nel campo Nome service account, inserisci un nome.
    3. (Facoltativo) Nel campo Descrizione service account, inserisci una descrizione.
    4. Fai clic su Crea.
    5. Fai clic sul campo Seleziona un ruolo. In Tutti i ruoli, seleziona Gemini Enterprise Agent Platform > Utente Agent Platform.

    6. Fai clic su Fine per creare il account di servizio.

      Non chiudere la finestra del browser. Lo utilizzerai nel prossimo passaggio.

  2. Crea una chiave del account di servizio per l'autenticazione:

    1. Nella console Google Cloud , fai clic sull'indirizzo email del account di servizio che hai creato.
    2. Fai clic su Chiavi.
    3. Fai clic su Aggiungi chiave, poi su Crea nuova chiave.
    4. Fai clic su Crea. Un file della chiave JSON viene scaricato sul computer.
    5. Fai clic su Chiudi.
  3. Concedi al nuovo account di servizio l'accesso al account di servizio che utilizzi per eseguire le pipeline.
    1. Fai clic su per tornare all'elenco dei service account.

    2. Fai clic sul nome del account di servizio che utilizzi per eseguire le pipeline. Viene visualizzata la pagina Dettagli service account.

      Se hai seguito le istruzioni della guida alla configurazione del progetto per le pipeline della piattaforma dell'agente Gemini Enterprise, questo è lo stesso account di servizio che hai creato nella sezione Configura un account di servizio con autorizzazioni granulari. In caso contrario, Gemini Enterprise Agent Platform utilizza il account di servizio predefinito di Compute Engine per eseguire le pipeline. Il account di servizio predefinito di Compute Engine ha un nome simile al seguente: PROJECT_NUMBER-compute@developer.gserviceaccount.com

    3. Fai clic sulla scheda Autorizzazioni.
    4. Fai clic su Concedi l'accesso. Viene visualizzato il riquadro Aggiungi entità.
    5. Nella casella Nuove entità, inserisci l'indirizzo email del service account che hai creato in un passaggio precedente.
    6. Nell'elenco a discesa Ruolo, seleziona Service account > Utente service account.
    7. Fai clic su Salva.

  4. Imposta la variabile di ambiente GOOGLE_APPLICATION_CREDENTIALS sul percorso del file JSON contenente la chiave dell'account di servizio. Questa variabile si applica solo alla sessione di Shell corrente, quindi se apri una nuova sessione, imposta di nuovo la variabile.

    Esempio: Linux o macOS

    Sostituisci [PATH] con il percorso del file JSON contenente la chiave dell'account di servizio. 

    export GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    Ad esempio:

    export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

    Esempio: Windows

    Sostituisci [PATH] con il percorso del file JSON che contiene la chiave del account di servizio e [FILE_NAME] con il nome file.

    Con PowerShell:

    $env:GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    Ad esempio:

    $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\[FILE_NAME].json"

    Con il prompt dei comandi:

    set GOOGLE_APPLICATION_CREDENTIALS=[PATH]

Definisci il flusso di lavoro con il pacchetto DSL Kubeflow Pipelines

Il pacchetto kfp.dsl contiene il linguaggio specifico del dominio (DSL) che puoi usare per definire e interagire con le pipeline e i componenti.

I componenti della pipeline di Kubeflow sono funzioni factory che creano i passaggi della pipeline. Ogni componente descrive gli input, gli output e l'implementazione del componente. Ad esempio, nell'esempio di codice seguente, ds_op è un componente.

I componenti vengono utilizzati per creare i passaggi della pipeline. Quando viene eseguita una pipeline, i passaggi si susseguono man mano che i dati da cui dipendono diventano disponibili. Ad esempio, un componente di addestramento potrebbe prendere un file CSV come input e usarlo per addestrare un modello.

import kfp
from google.cloud import aiplatform
from google_cloud_pipeline_components.v1.dataset import ImageDatasetCreateOp
from google_cloud_pipeline_components.v1.automl.training_job import AutoMLImageTrainingJobRunOp
from google_cloud_pipeline_components.v1.endpoint import EndpointCreateOp, ModelDeployOp

project_id = PROJECT_ID
pipeline_root_path = PIPELINE_ROOT

# Define the workflow of the pipeline.
@kfp.dsl.pipeline(
    name="automl-image-training-v2",
    pipeline_root=pipeline_root_path)
def pipeline(project_id: str):
    # The first step of your workflow is a dataset generator.
    # This step takes a Google Cloud Pipeline Component, providing the necessary
    # input arguments, and uses the Python variable `ds_op` to define its
    # output. Note that here the `ds_op` only stores the definition of the
    # output but not the actual returned object from the execution. The value
    # of the object is not accessible at the dsl.pipeline level, and can only be
    # retrieved by providing it as the input to a downstream component.
    ds_op = ImageDatasetCreateOp(
        project=project_id,
        display_name="flowers",
        gcs_source="gs://cloud-samples-data/vision/automl_classification/flowers/all_data_v2.csv",
        import_schema_uri=aiplatform.schema.dataset.ioformat.image.single_label_classification,
    )

    # The second step is a model training component. It takes the dataset
    # outputted from the first step, supplies it as an input argument to the
    # component (see `dataset=ds_op.outputs["dataset"]`), and will put its
    # outputs into `training_job_run_op`.
    training_job_run_op = AutoMLImageTrainingJobRunOp(
        project=project_id,
        display_name="train-iris-automl-mbsdk-1",
        prediction_type="classification",
        model_type="CLOUD",
        dataset=ds_op.outputs["dataset"],
        model_display_name="iris-classification-model-mbsdk",
        training_fraction_split=0.6,
        validation_fraction_split=0.2,
        test_fraction_split=0.2,
        budget_milli_node_hours=8000,
    )

    # The third and fourth step are for deploying the model.
    create_endpoint_op = EndpointCreateOp(
        project=project_id,
        display_name = "create-endpoint",
    )

    model_deploy_op = ModelDeployOp(
        model=training_job_run_op.outputs["model"],
        endpoint=create_endpoint_op.outputs['endpoint'],
        automatic_resources_min_replica_count=1,
        automatic_resources_max_replica_count=1,
    )

Sostituisci quanto segue:

  • PROJECT_ID: il progetto Google Cloud in cui viene eseguita questa pipeline.

  • PIPELINE_ROOT_PATH: specifica un URI Cloud Storage a cui il tuo service account di servizio può accedere. Gli artefatti delle esecuzioni della pipeline vengono archiviati nella radice della pipeline. L'URI Cloud Storage deve iniziare con gs://.

    La radice della pipeline può essere impostata come argomento dell'annotazione @kfp.dsl.pipeline sulla funzione della pipeline o quando chiami create_run_from_job_spec per creare un'esecuzione della pipeline.

Compila la pipeline in un file YAML

Dopo aver definito il flusso di lavoro della pipeline, puoi proseguire con la compilazione della pipeline in formato YAML. Il file YAML include tutte le informazioni per l'esecuzione della pipeline su Agent Platform Pipelines.

from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=pipeline,
    package_path='image_classif_pipeline.yaml'
)

Invia l'esecuzione della pipeline

Dopo aver compilato il flusso di lavoro della pipeline in formato YAML, puoi utilizzare il client Python di Gemini Enterprise Agent Platform per inviare ed eseguire la pipeline.

import google.cloud.aiplatform as aip

# Before initializing, make sure to set the GOOGLE_APPLICATION_CREDENTIALS
# environment variable to the path of your service account.
aip.init(
    project=project_id,
    location=PROJECT_REGION,
)

# Prepare the pipeline job
job = aip.PipelineJob(
    display_name="automl-image-training-v2",
    template_path="image_classif_pipeline.yaml",
    pipeline_root=pipeline_root_path,
    parameter_values={
        'project_id': project_id
    }
)

job.submit()

Sostituisci quanto segue:

  • PROJECT_REGION: la regione in cui viene eseguita la pipeline.

Nell'esempio precedente:

  1. Una pipeline Kubeflow è definita come funzione Python. La funzione è annotata con il decorator @kfp.dsl.pipeline, che specifica il nome e il percorso principale della pipeline. Quest'ultimo indica la posizione in cui sono archiviati gli artefatti della pipeline.
  2. I passaggi del flusso di lavoro della pipeline vengono creati utilizzando i componenti della pipeline diGoogle Cloud . Utilizzando gli output di un componente come input di un altro componente, il flusso di lavoro della pipeline viene definito come grafo. Ad esempio, training_job_run_op dipende dall'output dataset di ds_op.
  3. La pipeline viene compilata con kfp.compiler.Compiler.
  4. L'esecuzione di una pipeline deve essere eseguita su Agent Platform Pipelines utilizzando il client Python di Gemini Enterprise Agent Platform. Quando esegui una pipeline, puoi eseguire l'override del nome e del percorso principale della pipeline. Le esecuzioni della pipeline possono essere raggruppate utilizzando il nome della pipeline. L'override del nome della pipeline è utile per distinguere tra le esecuzioni della pipeline in produzione e quelle sperimentali.

Per scoprire di più sulla creazione di pipeline, consulta la sezione Creazione di pipeline Kubeflow e segui gli esempi e i tutorial.

(Facoltativo) Testa una pipeline in locale

Dopo aver definito le pipeline e i componenti, puoi testare il codice dei componenti eseguendolo nel tuo ambiente di creazione locale. Eseguendo la pipeline o un componente localmente, puoi identificare ed eseguire il debug di potenziali problemi prima di creare un'esecuzione della pipeline in un ambiente remoto, come Agent Platform Pipelines. Per saperne di più sull'esecuzione locale di pipeline e componenti, consulta Esecuzione locale nella documentazione di KFP.

Questa pagina mostra come definire ed eseguire una pipeline composta da due attività.

Configura l'ambiente locale

  1. (Facoltativo) Installa Docker.

  2. Utilizza il seguente esempio di codice per definire una pipeline iniziale:

    from kfp import dsl

# Define a component to add two numbers.
@dsl.component
def add(a: int, b: int) -> int:
    return a + b

# Define a simple pipeline using the component.
@dsl.pipeline
def addition_pipeline(x: int, y: int, z: int) -> int:
    task1 = add(a=x, b=y)
    task2 = add(a=task1.output, b=z)
    return task2.output

Richiama un'esecuzione locale

Inizializza una sessione locale utilizzando la funzione local.init(). Quando utilizzi local.init(), l'SDK KFP esegue localmente le pipeline e i componenti quando li chiami.

Quando utilizzi local.init(), devi specificare un tipo di runner. Il tipo di runner indica come KFP deve eseguire ogni attività.

Utilizza l'esempio seguente per specificare il tipo di runner DockerRunner per l'esecuzione di ogni task in un container. Per ulteriori informazioni sui runner locali supportati da KFP, consulta Runner locali nella documentazione di KFP.

from kfp import local

local.init(runner=local.DockerRunner())

pipeline_task = addition_pipeline(x=1, y=2, z=3)

Utilizza il seguente codice per visualizzare l'output dell'attività della pipeline dopo l'esecuzione locale:

print(f'Result: {pipeline_task.output}')

Creazione di pipeline Kubeflow

Per creare una pipeline, segui questa procedura.

  1. Progetta la pipeline come una serie di componenti. Per favorire la riusabilità, ogni componente deve avere una singola responsabilità. Quando possibile, progetta la pipeline in modo da riutilizzare componenti collaudati, come i componenti della pipeline diGoogle Cloud .

    Scopri di più sulla progettazione delle pipeline.

  2. Crea tutti i componenti personalizzati necessari per implementare il flusso di lavoro ML utilizzando l'SDK Kubeflow Pipelines. I componenti sono insiemi di codice autonomi che eseguono un passaggio del flusso di lavoro ML. Utilizza le seguenti opzioni per creare i componenti della pipeline.

    • Pacchettizza il codice del componente come immagine container. Questa opzione ti consente di includere nella pipeline codice scritto in qualsiasi linguaggio che possa essere pacchettizzato come immagine container.

      Scopri come creare un componente di Kubeflow Pipelines.

    • Implementa il codice del componente come funzione Python autonoma e utilizza l'SDK Kubeflow Pipelines per pacchettizzare la funzione come componente. Questa opzione semplifica la creazione di componenti basati su Python.

      Scopri come creare componenti basati su funzioni Python.

  3. Crea la pipeline come funzione Python.

    Scopri di più sulla definizione della pipeline come funzione Python.

  4. Utilizza il compilatore dell'SDK Kubeflow Pipelines per compilare la pipeline.

    from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=PIPELINE_FUNCTION,
    package_path=PIPELINE_PACKAGE_PATH)

    Sostituisci quanto segue:

    • PIPELINE_FUNCTION: il nome della funzione della pipeline.
    • PIPELINE_PACKAGE_PATH: il percorso in cui archiviare la pipeline compilata.

  5. Esegui la pipeline utilizzando la console Google Cloud o Python.

Accesso alle risorse Google Cloud in una pipeline

Se non specifichi un account di servizio quando esegui una pipeline, Agent Platform Pipelines utilizza il service account predefinito di Compute Engine per eseguire la pipeline. Agent Platform Pipelines utilizza anche il account di servizio di un'esecuzione della pipeline per autorizzare la pipeline ad accedere alle risorseGoogle Cloud . Il account di servizio predefinito di Compute Engine ha il ruolo Project Editor per impostazione predefinita. In questo modo, le tue pipeline potrebbero ottenere un accesso eccessivo alle risorse Google Cloud nel tuo progetto Google Cloud .

Ti consigliamo di creare un account di servizio per eseguire le pipeline e poi concedere a questo account autorizzazioni granulari per le risorse Google Cloud necessarie per eseguire la pipeline.

Scopri di più sull'utilizzo di Identity and Access Management per creare un service account e gestire l'accesso concesso a un service account.

Mantieni aggiornate le pipeline

I client SDK e le immagini container che utilizzi per creare ed eseguire le pipeline vengono aggiornati periodicamente alle nuove versioni per applicare patch alle vulnerabilità di sicurezza e aggiungere nuove funzionalità. Per mantenere le pipeline aggiornate con l'ultima versione, ti consigliamo di:

Passaggi successivi