Home / GCP Vertex AI / Computer Vision
← Back to GCP Vertex AI

Computer Vision with Vertex AI

Building vision models for image classification, object detection, and custom visual recognition at enterprise scale with minimal ML expertise.

Overview

Vertex Vision AutoML enables quick model development without custom training code. For advanced needs, use custom training with TensorFlow or PyTorch on Vertex Training infrastructure.

Quick Start: AutoML Image Classification

Python - AutoML Setup
from google.cloud import aiplatform
import json

# Initialize Vertex AI
aiplatform.init(project="your-project-id", location="us-central1")

# Create dataset
dataset = aiplatform.ImageDataset.create(
    display_name="product-images",
    gcs_source="gs://your-bucket/images/train.csv",  # CSV with img_path, label
)

# Training pipeline for image classification
job = aiplatform.AutoMLImageTrainingJob(
    display_name="product-classifier-v1",
    prediction_type="classification",
    multi_label=False,  # Single-label classification
    model_type="CLOUD",  # Cloud-hosted model
)

model = job.run(
    dataset=dataset,
    budget_milli_node_hours=8000,  # Training budget
    disable_early_stopping=False,
)

Custom Training Pipeline

Python - Custom Training
from google.cloud import aiplatform
import tensorflow as tf
from tensorflow.keras import layers, models

# Define custom model
def create_model():
    model = models.Sequential([
        layers.Conv2D(32, (3, 3), activation='relu', input_shape=(224, 224, 3)),
        layers.MaxPooling2D((2, 2)),
        layers.Conv2D(64, (3, 3), activation='relu'),
        layers.MaxPooling2D((2, 2)),
        layers.Conv2D(64, (3, 3), activation='relu'),
        layers.Flatten(),
        layers.Dense(64, activation='relu'),
        layers.Dense(10, activation='softmax')
    ])
    return model

# Training function (saved as train.py)
def train_and_save():
    # Load data
    (x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()
    
    # Normalize
    x_train, x_test = x_train / 255.0, x_test / 255.0
    
    # Build and compile
    model = create_model()
    model.compile(
        optimizer='adam',
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )
    
    # Train
    model.fit(
        x_train, y_train,
        epochs=10,
        batch_size=32,
        validation_data=(x_test, y_test)
    )
    
    # Save to GCS
    model.save('gs://your-bucket/models/cifar10')

# Create custom training job
aiplatform.init(project="your-project-id")

job = aiplatform.CustomTrainingJob(
    display_name="custom-image-classifier",
    script_path="train.py",  # Your training script
    container_uri="gcr.io/cloud-aiplatform/training/tf-cpu.2-12",
    requirements=["tensorflow==2.12"],
)

model = job.run(
    replica_count=1,
    machine_type="n1-standard-4",
)

Object Detection Model

Python - Object Detection Training
from google.cloud import aiplatform

# Create object detection dataset
dataset = aiplatform.ImageDataset.create(
    display_name="product-detection",
    gcs_source="gs://your-bucket/detection/annotations.csv",
    import_schema_uri=aiplatform.schema.dataset.ioformat.image_bounding_box.single_label_classification,
)

# Training job for object detection
job = aiplatform.AutoMLImageTrainingJob(
    display_name="product-detector",
    prediction_type="object_detection",
    model_type="CLOUD",
)

model = job.run(
    dataset=dataset,
    budget_milli_node_hours=20000,  # More budget for detection
)

Batch Prediction

Python - Large-Scale Inference
from google.cloud import aiplatform

# Get deployed model
model = aiplatform.Model("your-model-resource-id")

# Create batch prediction job
batch_job = model.batch_predict(
    job_display_name="product-images-batch",
    gcs_source="gs://your-bucket/images/predict/*.jpg",
    gcs_destination_prefix="gs://your-bucket/predictions/",
)

# Wait for completion
batch_job.wait()

# Process results
print(f"Output location: {batch_job.output_info.gcs_output_directory}")

Real-Time Predictions

Python - Endpoint Deployment
from google.cloud import aiplatform
import base64

# Deploy model to endpoint
endpoint = model.deploy(
    machine_type="n1-standard-4",
    accelerator_type="NVIDIA_TESLA_K80",
    accelerator_count=1,
    traffic_percentage=100,
)

# Make predictions
import json
from PIL import Image
from io import BytesIO

# Load image
with open("product.jpg", "rb") as img_file:
    image_base64 = base64.b64encode(img_file.read()).decode()

# Prepare prediction request
instances = [{
    "data": image_base64
}]

predictions = endpoint.predict(instances=instances)

for pred in predictions:
    print(f"Top classes: {pred['displayNames'][:3]}")
    print(f"Confidences: {pred['confidences'][:3]}")

Model Evaluation & Monitoring

Python - Model Performance Tracking
from google.cloud import aiplatform
from sklearn.metrics import classification_report, confusion_matrix

# Get model evaluation
model = aiplatform.Model("your-model-resource-id")
model_eval = model.get_model_evaluation()

print("Model Evaluation Metrics:")
print(f"Accuracy: {model_eval.metrics['accuracy']}")
print(f"Precision: {model_eval.metrics['precision']}")
print(f"Recall: {model_eval.metrics['recall']}")

# Setup monitoring
job = aiplatform.ModelMonitoringJob.create(
    display_name="model-monitoring",
    objective_config=aiplatform.ModelMonitoringObjectiveConfig(
        skew_detection_config=aiplatform.ModelMonitoringAlertConfig(
            data_drift_threshold=0.1,
            attribution_score_skew_threshold=0.05,
        ),
        prediction_drift_config=aiplatform.ModelMonitoringAlertConfig(
            data_drift_threshold=0.15,
        ),
    ),
)

Advanced: Custom Model with TensorFlow

Python - Production Model Code
import tensorflow as tf
from google.cloud import storage
import os

class ProductClassifier(tf.keras.Model):
    def __init__(self, num_classes):
        super().__init__()
        self.base = tf.keras.applications.EfficientNetB0(
            input_shape=(224, 224, 3),
            include_top=False,
            weights='imagenet'
        )
        self.global_avg = tf.keras.layers.GlobalAveragePooling2D()
        self.dropout = tf.keras.layers.Dropout(0.2)
        self.dense = tf.keras.layers.Dense(num_classes, activation='softmax')
    
    def call(self, inputs, training=None):
        x = self.base(inputs, training=False)
        x = self.global_avg(x)
        x = self.dropout(x, training=training)
        return self.dense(x)

# Training wrapper for Vertex
def train_and_export(args):
    # Load data
    train_dataset = tf.data.Dataset.from_tensor_slices(...)
    train_dataset = train_dataset.shuffle(1000).batch(32)
    
    # Create model
    model = ProductClassifier(num_classes=10)
    model.compile(
        optimizer=tf.keras.optimizers.Adam(1e-4),
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )
    
    # Train
    model.fit(train_dataset, epochs=10)
    
    # Save to GCS
    bucket = storage.Client().bucket(args.bucket)
    model.save('/tmp/model')
    for file in os.listdir('/tmp/model'):
        blob = bucket.blob(f"models/{args.model_name}/{file}")
        blob.upload_from_filename(f"/tmp/model/{file}")

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--bucket")
    parser.add_argument("--model-name")
    args = parser.parse_args()
    train_and_export(args)

Best Practices