Bắt đầu phân loại hình ảnh bằng Vertex AI và BigQuery

Hướng dẫn này cung cấp quy trình làm việc hoàn chỉnh từ đầu đến cuối để huấn luyện mô hình và phân loại tài sản hình ảnh bằng nền tảng Vertex AI của Google Cloud với Gemini 2.5 Flash. Bạn sẽ tìm hiểu cách tích hợp BigQuery để truy xuất dữ liệu, Cloud Storage để quản lý tài sản và Vertex AI để suy luận học máy trong môi trường Python Colab.

Cấu hình

Đặt các biến dành riêng cho dự án sau đây trước khi chạy các mẫu mã:

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

Thiết lập môi trường

Cài đặt các phần phụ thuộc bắt buộc và định cấu hình quy trình xác thực để truy cập vào các dịch vụ của 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}")

Đăng ký theo dõi tập dữ liệu của Trung tâm phân tích

Bạn cũng phải đăng ký tập dữ liệu 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)

Trích xuất dữ liệu bằng BigQuery

Thực thi một truy vấn BigQuery để trích xuất URI Google Cloud Storage từ bảng latest_observations. Các URI này sẽ được chuyển trực tiếp đến mô hình Vertex AI để phân loại.

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

Chức năng phân loại hình ảnh

Hàm trợ giúp này xử lý việc phân loại hình ảnh bằng mô hình Gemini 2.5 Flash của 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."

Phân loại hàng loạt hình ảnh

Xử lý tất cả các URI đã trích xuất và tạo phân loại:

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

Các bước tiếp theo

Sau khi phân loại hình ảnh, hãy cân nhắc các quy trình nâng cao sau:

  • Tinh chỉnh mô hình: Sử dụng kết quả phân loại để huấn luyện các mô hình tuỳ chỉnh.
  • glide-logo-1482-px-01.pngXử lý tự động: Thiết lập Cloud Functions để tự động phân loại hình ảnh mới.
  • Phân tích dữ liệu: Thực hiện phân tích thống kê về các mẫu phân loại.
  • Tích hợp: Kết nối kết quả với các ứng dụng hạ nguồn.

Khắc phục sự cố

Các vấn đề thường gặp và giải pháp:

  • Lỗi xác thực: Đảm bảo bạn có vai trò IAM phù hợp và đã bật API.
  • Giới hạn tốc độ: Triển khai thuật toán thời gian đợi lũy thừa cho các lô lớn.
  • запросыHạn chế về bộ nhớ: Xử lý hình ảnh theo các lô nhỏ hơn đối với các tập dữ liệu lớn.
  • Lỗi định dạng URI: Xác minh rằng URI GCS tuân theo định dạng gs://bucket-name/path/to/image.

Để được hỗ trợ thêm, hãy tham khảo tài liệu về Vertex AI và tài liệu về BigQuery.