Premiers pas avec la classification d'images à l'aide de Vertex AI et BigQuery

Ce guide fournit un workflow complet de bout en bout pour entraîner des modèles et classer des composants d'imagerie à l'aide de la plate-forme Vertex AI de Google Cloud avec Gemini 2.5 Flash. Vous apprendrez à intégrer BigQuery pour récupérer des données, Cloud Storage pour gérer les assets et Vertex AI pour l'inférence du machine learning dans un environnement Python Colab.

Configuration

Définissez les variables spécifiques au projet suivantes avant d'exécuter les exemples de code :

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

Configuration de l'environnement

Installez les dépendances requises et configurez l'authentification pour accéder aux services 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}")

S'abonner à l'ensemble de données Analytics Hub

Vous devez également vous abonner à l'ensemble de données 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)

Extraction de données avec BigQuery

Exécutez une requête BigQuery pour extraire les URI Google Cloud Storage de la table latest_observations. Ces URI seront transmis directement au modèle Vertex AI pour classification.

# 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)

Fonction de classification d'images

Cette fonction d'assistance gère la classification des images à l'aide du modèle Gemini 2.5 Flash de 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."

Classification d'images par lots

Traitez tous les URI extraits et générez des classifications :

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}")

Étapes suivantes

Une fois vos images classées, voici quelques workflows avancés à envisager :

  • Affinage de modèle : utilisez les résultats de classification pour entraîner des modèles personnalisés.
  • Traitement automatisé : configurez Cloud Functions pour classer automatiquement les nouvelles images.
  • Analyse des données : effectuez une analyse statistique des modèles de classification.
  • Intégration : connectez les résultats aux applications en aval.

Dépannage

Problèmes courants et solutions :

  • Erreurs d'authentification : assurez-vous que les rôles IAM sont appropriés et que l'API est activée.
  • Limitation du débit : implémentez un intervalle exponentiel entre les tentatives pour les grands lots.
  • Contraintes de mémoire : traitez les images par petits lots pour les grands ensembles de données.
  • Erreurs de format d'URI : vérifiez que les URI GCS suivent le format gs://bucket-name/path/to/image.

Pour obtenir de l'aide supplémentaire, consultez la documentation Vertex AI et la documentation BigQuery.