Como começar a usar a classificação de imagens com a Vertex AI e o BigQuery

Este guia oferece um fluxo de trabalho completo de ponta a ponta para treinar modelos e classificar recursos de imagens usando a plataforma Vertex AI do Google Cloud com o Gemini 2.5 Flash. Você vai aprender a integrar o BigQuery para recuperação de dados, o Cloud Storage para gerenciamento de recursos e a Vertex AI para inferência de machine learning em um ambiente Python Colab.

Configuração

Defina as seguintes variáveis específicas do projeto antes de executar os exemplos de código:

PROJECT_ID = "PROJECT_ID"
REGION = "REGION "  # e.g., "us-central1"
LOCATION = "LOCATION "  # e.g., "us"
CUSTOMER_ID = "CUSTOMER_ID" # required to subscribe to the dataset

Configuração do ambiente

Instale as dependências necessárias e configure a autenticação para acessar os serviços do Google Cloud:

# Install Google Cloud SDK dependencies for AI Platform integration
!pip install google-cloud-aiplatform google-cloud-storage google-cloud-bigquery google-cloud-bigquery-data-exchange -q

# Import core libraries for cloud services and machine learning operations
import json
import os
from google.cloud import bigquery
import vertexai
from vertexai.generative_models import GenerativeModel, Part

# Configure authentication for Google Cloud service access
# Initiates OAuth flow in new browser tab if authentication required
from google.colab import auth

if os.environ.get("VERTEX_PRODUCT") != "COLAB_ENTERPRISE":
  from google.colab import auth
  auth.authenticate_user(project_id=PROJECT_ID)

# Initialize Vertex AI client with project configuration
vertexai.init(project=PROJECT_ID, location=REGION)

print(f"Vertex AI initialized for project: {PROJECT_ID} in region: {REGION}")

Assinar o conjunto de dados do Analytics Hub

Você também precisa assinar o conjunto de dados do Analytics Hub.

from google.cloud import bigquery_data_exchange_v1beta1

ah_client = bigquery_data_exchange_v1beta1.AnalyticsHubServiceClient()

HUB_PROJECT_ID = 'maps-platform-analytics-hub'
DATA_EXCHANGE_ID = f"imagery_insights_exchange_{LOCATION}"
LINKED_DATASET_NAME = f"imagery_insights___preview___{LOCATION}"


# subscribe to the listing (create a linked dataset in your consumer project)
destination_dataset = bigquery_data_exchange_v1beta1.DestinationDataset()
destination_dataset.dataset_reference.dataset_id = LINKED_DATASET_NAME
destination_dataset.dataset_reference.project_id = PROJECT_ID
destination_dataset.location = LOCATION
LISTING_ID=f"imagery_insights_{CUSTOMER_ID.replace('-', '_')}__{LOCATION}"

published_listing = f"projects/{HUB_PROJECT_ID}/locations/{LOCATION}/dataExchanges/{DATA_EXCHANGE_ID}/listings/{LISTING_ID}"

request = bigquery_data_exchange_v1beta1.SubscribeListingRequest(
    destination_dataset=destination_dataset,
    name=published_listing,
)

# request the subscription
ah_client.subscribe_listing(request=request)

Extração de dados com o BigQuery

Execute uma consulta do BigQuery para extrair URIs do Google Cloud Storage da tabela latest_observations. Esses URIs serão transmitidos diretamente ao modelo da Vertex AI para classificação.

# Initialize BigQuery client
bigquery_client = bigquery.Client(project=PROJECT_ID)

# Define SQL query to retrieve observation records from imagery dataset
query = f"""
SELECT
 *
FROM
 `{PROJECT_ID}.imagery_insights___preview___{LOCATION}.latest_observations`
LIMIT 10;
"""

print(f"Executing BigQuery query:\n{query}")

# Submit query job to BigQuery service and await completion
query_job = bigquery_client.query(query)

# Transform query results into structured data format for downstream processing
# Convert BigQuery Row objects to dictionary representations for enhanced accessibility
query_response_data = []
for row in query_job:
   query_response_data.append(dict(row))

# Extract Cloud Storage URIs from result set, filtering null values
gcs_uris = [item.get("gcs_uri") for item in query_response_data if item.get("gcs_uri")]

print(f"BigQuery query returned {len(query_response_data)} records.")
print(f"Extracted {len(gcs_uris)} GCS URIs:")
for uri in gcs_uris:
   print(uri)

Função de classificação de imagens

Essa função auxiliar processa a classificação de imagens usando o modelo Gemini 2.5 Flash da Vertex AI:

def classify_image_with_gemini(gcs_uri: str, prompt: str = "What is in this image?") -> str:
   """
   Performs multimodal image classification using Vertex AI's Gemini 2.5 Flash model.

   Leverages direct Cloud Storage integration to process image assets without local
   download requirements, enabling scalable batch processing workflows.

   Args:
       gcs_uri (str): Fully qualified Google Cloud Storage URI 
                     (format: gs://bucket-name/path/to/image.jpg)
       prompt (str): Natural language instruction for classification task execution

   Returns:
       str: Generated textual description from the generative model, or error message
            if classification pipeline fails

   Raises:
       Exception: Captures service-level errors and returns structured failure response
   """
   try:
       # Instantiate Gemini 2.5 Flash model for inference operations
       model = GenerativeModel("gemini-2.5-flash")

       # Construct multimodal Part object from Cloud Storage reference
       # Note: MIME type may need dynamic inference for mixed image formats
       image_part = Part.from_uri(uri=gcs_uri, mime_type="image/jpeg")

       # Execute multimodal inference request with combined visual and textual inputs
       responses = model.generate_content([image_part, prompt])
       return responses.text
   except Exception as e:
       print(f"Error classifying image from URI {gcs_uri}: {e}")
       return "Classification failed."

Classificação de imagens em lote

Processe todos os URIs extraídos e gere classificações:

classification_results = []

# Execute batch classification pipeline across all extracted GCS URIs
for uri in gcs_uris:
   print(f"\nProcessing: {uri}")

   # Define comprehensive classification prompt for detailed feature extraction
   classification_prompt = "Describe this image in detail, focusing on any objects, signs, or features visible."

   # Invoke Gemini model for multimodal inference on current asset
   result = classify_image_with_gemini(uri, classification_prompt)

   # Aggregate structured results for downstream analytics and reporting
   classification_results.append({"gcs_uri": uri, "classification": result})

   print(f"Classification for {uri}:\n{result}")

Próximas etapas

Com as imagens classificadas, considere estes fluxos de trabalho avançados:

  • Ajuste fino do modelo: use os resultados da classificação para treinar modelos personalizados.
  • Processamento automatizado: configure o Cloud Functions para classificar novas imagens automaticamente.
  • Análise de dados: faça análises estatísticas em padrões de classificação.
  • Integração: conecte os resultados a aplicativos downstream.

Solução de problemas

Problemas comuns e soluções:

  • Erros de autenticação: verifique se os papéis do IAM e a ativação da API estão corretos.
  • Limitação de taxa: implemente a espera exponencial para grandes lotes.
  • Restrições de memória: processe imagens em lotes menores para conjuntos de dados grandes.
  • Erros de formato de URI: verifique se os URIs do GCS seguem o formato gs://bucket-name/path/to/image.

Para mais suporte, consulte a documentação da Vertex AI e do BigQuery.