Inferenza batch per BigQuery

Questa pagina descrive come ottenere inferenze batch utilizzando BigQuery.

1. Prepara gli input

Input di archiviazione BigQuery

Il tuo account di servizio deve disporre delle autorizzazioni BigQuery appropriate. Per concedere al account di servizio il ruolo BigQuery User, utilizza il comando gcloud iam service-accounts add-iam-policy-binding come segue:

    gcloud projects add-iam-policy-binding PROJECT_ID \
        --member="serviceAccount:SERVICE_ACCOUNT_ID@PROJECT_ID.iam.gserviceaccount.com" \
        --role="roles/bigquery.user"

Sostituisci i seguenti valori:

  • PROJECT_ID: l'ID del progetto in cui è stato creato il account di servizio.
  • SERVICE_ACCOUNT_ID: l'ID del account di servizio.

È obbligatoria una colonna request, che deve essere un JSON valido. Questi dati JSON rappresentano l'input per il modello.

I contenuti della colonna request devono corrispondere alla struttura di un GenerateContentRequest.

La tabella di input può avere tipi di dati di colonna diversi da request. Queste colonne possono avere tipi di dati BigQuery, ad eccezione dei seguenti: array, struct, range, datetime e geography. Queste colonne vengono ignorate per la generazione dei contenuti, ma incluse nella tabella di output.

Input di esempio (JSON) 
        
{
  "contents": [
    {
      "role": "user",
      "parts": [
        {
          "text": "Give me a recipe for banana bread."
        }
      ]
    }
  ],
  "system_instruction": {
    "parts": [
      {
        "text": "You are a chef."
      }
    ]
  }
}

2. Invia un job batch

Puoi creare un job batch tramite la console Google Cloud , l'SDK Google Gen AI o l'API REST.

Il job e la tabella devono trovarsi nella stessa regione.

Console

  1. Nella sezione Agent Platform della console Google Cloud , vai alla pagina Inferenza batch.

    Vai a Inferenza batch

  2. Fai clic su Crea.

REST

Per creare un job di inferenza batch, utilizza il metodo projects.locations.batchPredictionJobs.create.

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

  • ENDPOINT_PREFIX: la regione della risorsa del modello seguita da -. Ad esempio: us-central1-. Se utilizzi l'endpoint globale, lascia vuoto.
  • LOCATION: una regione che supporta i modelli Gemini. Se utilizzi l'endpoint globale, inserisci global.
  • PROJECT_ID: il tuo ID progetto.
  • MODEL_PATH: il nome del modello del publisher, ad esempio publishers/google/models/gemini-2.0-flash-001, o il nome dell'endpoint ottimizzato, ad esempio projects/PROJECT_ID/locations/LOCATION/models/MODEL_ID, dove MODEL_ID è l'ID del modello ottimizzato.
  • INPUT_URI: la tabella BigQuery in cui si trova l'input dell'inferenza batch, ad esempio bq://myproject.mydataset.input_table. Il set di dati deve trovarsi nella stessa regione del job di inferenza batch. I set di dati multiregionali non sono supportati.
  • OUTPUT_FORMAT: per l'output in una tabella BigQuery, specifica bigquery. Per l'output in un bucket Cloud Storage, specifica jsonl.
  • DESTINATION: per BigQuery, specifica bigqueryDestination. Per Cloud Storage, specifica gcsDestination.
  • OUTPUT_URI_FIELD_NAME: per BigQuery, specifica outputUri. Per Cloud Storage, specifica outputUriPrefix.
  • OUTPUT_URI: per BigQuery, specifica la località della tabella, ad esempio bq://myproject.mydataset.output_result. La regione del set di dati BigQuery di output deve essere la stessa del job di inferenza batch di Agent Platform. Per Cloud Storage, specifica la posizione del bucket e della directory, ad esempio gs://mybucket/path/to/output.

Metodo HTTP e URL: 

POST https://ENDPOINT_PREFIXaiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs

Corpo JSON della richiesta: 

{
  "displayName": "my-bigquery-batch-inference-job",
  "model": "MODEL_PATH",
  "inputConfig": {
    "instancesFormat": "bigquery",
    "bigquerySource":{
      "inputUri" : "INPUT_URI"
    }
  },
  "outputConfig": {
    "predictionsFormat": "OUTPUT_FORMAT",
    "DESTINATION": {
      "OUTPUT_URI_FIELD_NAME": "OUTPUT_URI"
    }
  }
}

Per inviare la richiesta, scegli una di queste opzioni:

curl

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente: 

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://ENDPOINT_PREFIXaiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs"

PowerShell

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente: 

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://ENDPOINT_PREFIXaiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs" | Select-Object -Expand Content

Dovresti ricevere una risposta JSON simile alla seguente.

La risposta include un identificatore univoco per il job batch. Puoi eseguire il polling dello stato del job batch utilizzando BATCH_JOB_ID. Per saperne di più, consulta Monitora lo stato del job. Nota: i report su service account personalizzato, avanzamento in tempo reale, CMEK e VPCSC non sono supportati.

Python

Installa

pip install --upgrade google-genai

Per saperne di più, consulta la documentazione di riferimento dell'SDK.

Imposta le variabili di ambiente per utilizzare l'SDK Google Gen AI con Vertex AI: 

# Replace the `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION` values
# with appropriate values for your project.
export GOOGLE_CLOUD_PROJECT=GOOGLE_CLOUD_PROJECT
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_ENTERPRISE=True
 

import time

from google import genai
from google.genai.types import CreateBatchJobConfig, HttpOptions, JobState

client = genai.Client(http_options=HttpOptions(api_version="v1"))

# TODO(developer): Update and un-comment below line
# output_uri = f"bq://your-project.your_dataset.your_table"

job = client.batches.create(
    # To use a tuned model, set the model param to your tuned model using the following format:
    # model="projects/{PROJECT_ID}/locations/{LOCATION}/models/{MODEL_ID}
    model="gemini-3.5-flash",
    src="bq://storage-samples.generative_ai.batch_requests_for_multimodal_input",
    config=CreateBatchJobConfig(dest=output_uri),
)
print(f"Job name: {job.name}")
print(f"Job state: {job.state}")
# Example response:
# Job name: projects/.../locations/.../batchPredictionJobs/9876453210000000000
# Job state: JOB_STATE_PENDING

# See the documentation: https://googleapis.github.io/python-genai/genai.html#genai.types.BatchJob
completed_states = {
    JobState.JOB_STATE_SUCCEEDED,
    JobState.JOB_STATE_FAILED,
    JobState.JOB_STATE_CANCELLED,
    JobState.JOB_STATE_PAUSED,
}

while job.state not in completed_states:
    time.sleep(30)
    job = client.batches.get(name=job.name)
    print(f"Job state: {job.state}")
# Example response:
# Job state: JOB_STATE_PENDING
# Job state: JOB_STATE_RUNNING
# Job state: JOB_STATE_RUNNING
# ...
# Job state: JOB_STATE_SUCCEEDED

3. Monitora lo stato e l'avanzamento del job

Dopo l'invio del job, puoi controllare lo stato del job batch utilizzando l'API, l'SDK e la console Google Cloud .

Console

  1. Vai alla pagina Inferenza batch.

    Vai a Inferenza batch

  2. Seleziona il job batch per monitorarne l'avanzamento.

REST

Per monitorare un job di inferenza batch, utilizza il metodo projects.locations.batchPredictionJobs.get e visualizza il campo CompletionStats nella risposta.

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

  • ENDPOINT_PREFIX: la regione della risorsa del modello seguita da -. Ad esempio: us-central1-. Se utilizzi l'endpoint globale, lascia vuoto.
  • LOCATION: una regione che supporta i modelli Gemini. Se utilizzi l'endpoint globale, inserisci global.
  • PROJECT_ID: il tuo ID progetto.
  • BATCH_JOB_ID: l'ID job batch.

Metodo HTTP e URL: 

GET https://ENDPOINT_PREFIXaiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs/BATCH_JOB_ID

Per inviare la richiesta, scegli una di queste opzioni:

curl

Esegui questo comando: 

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://ENDPOINT_PREFIXaiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs/BATCH_JOB_ID"

PowerShell

Esegui questo comando: 

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://ENDPOINT_PREFIXaiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs/BATCH_JOB_ID" | Select-Object -Expand Content

Dovresti ricevere una risposta JSON simile alla seguente.

Python

Installa

pip install --upgrade google-genai

Per saperne di più, consulta la documentazione di riferimento dell'SDK.

Imposta le variabili di ambiente per utilizzare l'SDK Google Gen AI con Vertex AI: 

# Replace the `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION` values
# with appropriate values for your project.
export GOOGLE_CLOUD_PROJECT=GOOGLE_CLOUD_PROJECT
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_ENTERPRISE=True
 

    from google import genai
    from google.genai.types import HttpOptions

    client = genai.Client(http_options=HttpOptions(api_version="v1"))

    # Get the batch job
# Eg. batch_job_name = "projects/123456789012/locations/.../batchPredictionJobs/1234567890123456789"
    batch_job = client.batches.get(name=batch_job_name)

    print(f"Job state: {batch_job.state}")
    # Example response:
    # Job state: JOB_STATE_PENDING
    # Job state: JOB_STATE_RUNNING
    # Job state: JOB_STATE_SUCCEEDED

Lo stato di un determinato job batch può essere uno dei seguenti:

  • JOB_STATE_PENDING: accoda per capacità. Il job può rimanere nello stato queue per un massimo di 72 ore prima di passare allo stato running.
  • JOB_STATE_RUNNING: il file di input è stato convalidato correttamente e il batch è in esecuzione.
  • JOB_STATE_SUCCEEDED: il batch è stato completato e i risultati sono pronti
  • JOB_STATE_FAILED: il file di input non ha superato il processo di convalida o non è stato possibile completarlo entro la finestra temporale di 24 ore dopo l'attivazione dello stato RUNNING.
  • JOB_STATE_CANCELLING: il batch è in fase di annullamento.
  • JOB_STATE_CANCELLED: il batch è stato annullato.

4. Recupera l'output batch

Al termine di un'attività di inferenza batch, l'output viene archiviato nella tabella BigQuery specificata nella richiesta.

Per le righe riuscite, le risposte del modello vengono archiviate nella colonna response. In caso contrario, i dettagli dell'errore vengono archiviati nella colonna status per un'ulteriore ispezione.

Esempio di output

Esempio di operazione riuscita

{
  "candidates": [
    {
      "content": {
        "role": "model",
        "parts": [
          {
            "text": "In a medium bowl, whisk together the flour, baking soda, baking powder."
          }
        ]
      },
      "finishReason": "STOP",
      "safetyRatings": [
        {
          "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
          "probability": "NEGLIGIBLE",
          "probabilityScore": 0.14057204,
          "severity": "HARM_SEVERITY_NEGLIGIBLE",
          "severityScore": 0.14270912
        }
      ]
    }
  ],
  "usageMetadata": {
    "promptTokenCount": 8,
    "candidatesTokenCount": 396,
    "totalTokenCount": 404
  }
}

Esempio di operazione non riuscita

  • Richiesta

    {"contents":[{"parts":{"text":"Explain how AI works in a few words."},"role":"tester"}]}

  • Risposta

    Bad Request: {"error": {"code": 400, "message": "Please use a valid role: user, model.", "status": "INVALID_ARGUMENT"}}

Riprendi un job batch incompleto

Se un job di inferenza batch non va a buon fine, viene annullato o scade, puoi riprenderlo creando un nuovo job batch. Il nuovo job elabora solo le richieste incomplete o non riuscite del job precedente, unendo i nuovi risultati a quelli delle richieste completate in precedenza. Ad esempio, se il job batch incompleto era stato completato al 90%, il nuovo job elabora il restante 10% delle richieste.

Per riprendere un job batch, creane uno nuovo fornendo il file Cloud Storage o l'output della tabella BigQuery come input per il nuovo job.