Modelli di IA e progresso algoritmico

In qualsiasi sistema MLOps di produzione che costruisco, la governance dei modelli è fondamentale. Non si tratta solo di tracciare la lignaggio, ma anche di garantire conformità e sicurezza.

• Controllo di Versione: Imporre Git per tutto il codice della pipeline e utilizzare il Registro Modelli per il versionamento dei binari del modello e dei loro metadati associati.
• Registrazione degli Audit (Audit Logging): Tutti gli eventi del ciclo di vita — esecuzioni di addestramento, distribuzioni, modifiche di stato, log di predizione — devono essere acquisiti. Sia GCP che Azure forniscono una registrazione estesa e non ripudiabile per questo scopo.
• Controllo degli Accessi: Utilizzare politiche IAM rigorose per definire chi può addestrare, registrare, distribuire o interrogare i modelli. Questo segue il principio del minimo privilegio.
• Scansione dei Modelli: Una pratica in evoluzione ma critica è l'integrazione di scansioni di sicurezza per gli artefatti dei modelli al fine di rilevare vulnerabilità o dati sensibili incorporati. Questa dovrebbe far parte della tua pipeline CI/CD.

Ecco uno sguardo pratico su come un singolo passaggio di addestramento diventa un asset codificato e versionato in una Vertex AI Pipeline usando l'SDK Kubeflow Pipelines (kfp).

# pipeline_components.py
from kfp.v2.dsl import pipeline, component
from kfp.compiler import Compiler

# Define a custom component for model training
@component(packages_to_install=['scikit-learn', 'pandas', 'google-cloud-aiplatform'])
def train_model_component(
    dataset_uri: str,
    model_output_uri: str,
    project_id: str,
    region: str = 'europe-west1'
):
    import pandas as pd
    from sklearn.model_selection import train_test_split
    from sklearn.linear_model import LogisticRegression
    from sklearn.metrics import accuracy_score
    import google.cloud.aiplatform as aiplatform
    import logging

    logging.basicConfig(level=logging.INFO)
    aiplatform.init(project=project_id, location=region)

    logging.info(f"Loading dataset from: {dataset_uri}")
    # In a real scenario, this would involve reading from GCS or BigQuery
    # For simplicity, we'll simulate a dataset
    data = pd.DataFrame({
        'feature1': [i for i in range(100)],
        'feature2': [i * 2 for i in range(100)],
        'target': [0 if i < 50 else 1 for i in range(100)]
    })

    X = data[['feature1', 'feature2']]
    y = data['target']
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    logging.info("Training Logistic Regression model...")
    model = LogisticRegression()
    model.fit(X_train, y_train)
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    logging.info(f"Model trained with accuracy: {accuracy}")

    # Simulate saving model to GCS for later registration
    # In a real scenario, you'd use job.log_artifact or save to GCS
    # For this example, we just 'log' the output URI
    logging.info(f"Model artifact would be saved to: {model_output_uri}/model.pkl")

# This component would then be used inside a larger pipeline definition.
# For example:
# @pipeline(name='simple-mlops-pipeline', pipeline_root='gs://your-bucket/pipeline_root')
# def ml_pipeline(project_id: str):
#    train_task = train_model_component(dataset_uri='gs://my-data/dataset.csv',
#                                       model_output_uri='gs://my-models',
#                                       project_id=project_id)
# Compiler().compile(pipeline_func=ml_pipeline, package_path='simple_mlops_pipeline.json')

Questo semplice componente incapsula la logica di addestramento, rendendola un passaggio riutilizzabile e versionabile in un flusso di lavoro automatizzato più ampio.

Innanzitutto, crea un account di servizio dedicato per le pipeline MLOps per applicare il principio del minimo privilegio.

# Set environment variables. Replace 'your-gcp-project-id' with your actual project ID.
export GCP_PROJECT_ID="your-gcp-project-id"
export GCP_REGION="europe-west1"

# Enable the Vertex AI API
gcloud services enable aiplatform.googleapis.com --project=$GCP_PROJECT_ID

# Create a service account for your MLOps pipeline
export SERVICE_ACCOUNT_NAME="mlops-pipeline-sa"
export SERVICE_ACCOUNT_EMAIL="${SERVICE_ACCOUNT_NAME}@${GCP_PROJECT_ID}.iam.gserviceaccount.com"

gcloud iam service-accounts create $SERVICE_ACCOUNT_NAME \
    --display-name="Service Account for MLOps Pipelines" \
    --project=$GCP_PROJECT_ID

# Grant necessary roles. For production, use more granular roles than storage.objectAdmin.
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member="serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
    --role="roles/aiplatform.user"

gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member="serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
    --role="roles/storage.objectAdmin"

Questa configurazione una tantum crea l'identità e i permessi che il nostro flusso di lavoro automatizzato utilizzerà.

Questo script simula un'esecuzione di addestramento locale, quindi carica l'artefatto risultante su Cloud Storage e lo registra nel Vertex AI Model Registry. È progettato per essere idempotente, gestendo i casi in cui risorse come i bucket GCS esistono già.

# train_and_register.py
import os
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import joblib

import google.cloud.aiplatform as aiplatform
from google.cloud import storage
from google.api_core import exceptions

# --- Configuration ---
# These are read from environment variables set in your shell.
PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')

if not PROJECT_ID:
    raise ValueError("GCP_PROJECT_ID environment variable not set.")

BUCKET_NAME = f"{PROJECT_ID}-mlops-models-eu"
MODEL_DISPLAY_NAME = "MyFraudDetectionModel"
MODEL_DESCRIPTION = "A simple RandomForestClassifier for fraud detection"

# Initialize Vertex AI SDK
aiplatform.init(project=PROJECT_ID, location=REGION)

# --- Simulate Data ---
# In a real project, you'd fetch data from BigQuery or GCS.
data = pd.DataFrame({
    'transaction_amount': [100, 200, 50, 1000, 75, 500, 120, 800, 30, 600],
    'num_items': [1, 2, 1, 5, 1, 3, 1, 4, 1, 2],
    'is_fraud': [0, 0, 0, 1, 0, 1, 0, 1, 0, 1]
})
X = data[['transaction_amount', 'num_items']]
y = data['is_fraud']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# --- Train Model ---
print("Training RandomForestClassifier...")
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
predictions = model.predict(X_test)
accuracy = accuracy_score(y_test, predictions)
print(f"Model trained with accuracy: {accuracy:.2f}")

# --- Save Model Artifact Locally ---
local_model_path = "model.joblib"
joblib.dump(model, local_model_path)
print(f"Model saved locally to {local_model_path}")

# --- Upload Model to GCS ---
storage_client = storage.Client(project=PROJECT_ID)
bucket_uri = f"gs://{BUCKET_NAME}"
try:
    bucket = storage_client.create_bucket(BUCKET_NAME, location=REGION)
    print(f"Created GCS bucket: {bucket_uri}")
except exceptions.Conflict:
    print(f"GCS bucket {bucket_uri} already exists.")
    bucket = storage_client.get_bucket(BUCKET_NAME)

model_filename = "model.joblib"
artifact_directory = f"{MODEL_DISPLAY_NAME}"
blob = bucket.blob(f"{artifact_directory}/{model_filename}")
blob.upload_from_filename(local_model_path)
gcs_artifact_path = f"{bucket_uri}/{artifact_directory}"
print(f"Model artifact uploaded to GCS: {gcs_artifact_path}")

# --- Register Model in Vertex AI Model Registry ---
# Using a pre-built scikit-learn container for serving.
SERVING_CONTAINER_IMAGE = f"{REGION}-docker.pkg.dev/cloud-aiplatform/prediction/sklearn-cpu.1-0:latest"

# Check if model already exists to upload a new version.
existing_models = aiplatform.Model.list(filter=f'display_name="{MODEL_DISPLAY_NAME}"')
parent_model_resource = None
if existing_models:
    print(f"Model '{MODEL_DISPLAY_NAME}' already exists. Creating a new version.")
    parent_model_resource = existing_models[0].resource_name
else:
    print(f"Creating new model '{MODEL_DISPLAY_NAME}' and registering version 1.")

uploaded_model = aiplatform.Model.upload(
    display_name=MODEL_DISPLAY_NAME,
    artifact_uri=gcs_artifact_path,
    serving_container_image_uri=SERVING_CONTAINER_IMAGE,
    description=MODEL_DESCRIPTION,
    parent_model=parent_model_resource,
    sync=True
)

print(f"Model '{uploaded_model.display_name}' version '{uploaded_model.version_id}' registered.")
print(f"Model resource name: {uploaded_model.resource_name}")

Si noti il parametro parent_model — questo è cruciale per creare nuove versioni di un modello esistente, costituendo il nucleo della tua lignaggio del modello.

Una volta registrato, la distribuzione del modello su un endpoint gestito lo rende disponibile per predizioni in tempo reale.

# deploy_model.py
import os
import google.cloud.aiplatform as aiplatform

PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')
MODEL_DISPLAY_NAME = "MyFraudDetectionModel"
ENDPOINT_DISPLAY_NAME = "fraud-detection-endpoint"

aiplatform.init(project=PROJECT_ID, location=REGION)

# Retrieve the latest version of the model from the registry.
models = aiplatform.Model.list(filter=f'display_name="{MODEL_DISPLAY_NAME}"', order_by="create_time desc")
if not models:
    raise ValueError(f"Model '{MODEL_DISPLAY_NAME}' not found in registry.")

model_to_deploy = models[0]
print(f"Found model '{model_to_deploy.display_name}' version '{model_to_deploy.version_id}'.")

# Create or get existing endpoint.
endpoints = aiplatform.Endpoint.list(filter=f'display_name="{ENDPOINT_DISPLAY_NAME}"')
if endpoints:
    endpoint = endpoints[0]
    print(f"Using existing endpoint: {endpoint.display_name}")
else:
    print(f"Creating new endpoint: {ENDPOINT_DISPLAY_NAME}")
    endpoint = aiplatform.Endpoint.create(
        display_name=ENDPOINT_DISPLAY_NAME,
        project=PROJECT_ID,
        location=REGION
    )

# Deploy the model. I start with min_replica_count=1 for cost efficiency.
TRAFFIC_SPLIT = {"0": 100} # 100% traffic to the new model version.

endpoint.deploy(
    model=model_to_deploy,
    deployed_model_display_name=f"{MODEL_DISPLAY_NAME}-v{model_to_deploy.version_id}",
    machine_type="n1-standard-2", # Choose a machine type appropriate for your workload.
    min_replica_count=1,
    max_replica_count=2,
    traffic_split=TRAFFIC_SPLIT,
    sync=True
)

print(f"Model '{model_to_deploy.display_name}' version '{model_to_deploy.version_id}' deployed to endpoint '{endpoint.display_name}'.")

Questo script astrae l'infrastruttura sottostante, fornendoti un servizio di inferenza scalabile e ad alta disponibilità. Il parametro traffic_split è il modo in cui implemento distribuzioni canary e test A/B in produzione.

Con il modello distribuito, puoi ora inviargli richieste di inferenza.

# predict_online.py
import os
import google.cloud.aiplatform as aiplatform

PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')
ENDPOINT_DISPLAY_NAME = "fraud-detection-endpoint"

aiplatform.init(project=PROJECT_ID, location=REGION)

# Get the endpoint object.
endpoints = aiplatform.Endpoint.list(filter=f'display_name="{ENDPOINT_DISPLAY_NAME}"')
if not endpoints:
    raise ValueError(f"Endpoint '{ENDPOINT_DISPLAY_NAME}' not found.")
endpoint = endpoints[0]

# Prepare instances for prediction. The format must match what the model expects.
instances = [
    {"transaction_amount": 150.0, "num_items": 1},
    {"transaction_amount": 900.0, "num_items": 4},
]

print(f"Sending prediction request to endpoint: {endpoint.display_name}")
prediction_response = endpoint.predict(instances=instances)

print("Prediction response:")
for prediction in prediction_response.predictions:
    print(f"  {prediction}")

Presta molta attenzione al formato di input. Qualsiasi disallineamento tra le feature inviate qui e quelle utilizzate nell'addestramento è una fonte comune di errore.

Il monitoraggio dei modelli è irrinunciabile per l'IA in produzione. Sebbene una configurazione completa vada oltre lo scopo di questo articolo, è fondamentale comprenderne il concetto. Si definisce cosa monitorare (ad esempio, data skew, concept drift) e si fornisce un dataset di riferimento (i dati di addestramento) per il confronto. Vertex AI registra quindi automaticamente le richieste di predizione e ti avvisa quando le prestazioni deviano.

# configure_monitoring.py (Conceptual)
import os
import google.cloud.aiplatform as aiplatform

PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')
ENDPOINT_DISPLAY_NAME = "fraud-detection-endpoint"

aiplatform.init(project=PROJECT_ID, location=REGION)

endpoints = aiplatform.Endpoint.list(filter=f'display_name="{ENDPOINT_DISPLAY_NAME}"')
if not endpoints:
    raise ValueError(f"Endpoint '{ENDPOINT_DISPLAY_NAME}' not found.")
endpoint = endpoints[0]

# This is a conceptual representation. A real implementation requires:
# 1. A BigQuery table with training data for the baseline schema and distribution.
# 2. Skew and drift detection configurations with specified thresholds.
# 3. Alerting configurations (e.g., email notifications).

# from google.cloud.aiplatform import model_monitoring
# from google.cloud.aiplatform.model_monitoring import
#     SkewDetectionConfig, DriftDetectionConfig, ObjectiveConfig

# skew_config = SkewDetectionConfig(data_source=..., skew_thresholds=...)
# objective_config = ObjectiveConfig(skew_detection_config=skew_config, ...)
# endpoint.update_monitoring(objective_configs=objective_config, ...)

print("Conceptual monitoring configuration initiated.")
print("For a full implementation, refer to the Vertex AI Model Monitoring documentation.")

Questo ciclo di feedback continuo è ciò che rende il Quarto Strato di IA robusto e resiliente. Per un approfondimento, indirizzo i miei clienti alla documentazione ufficiale di Vertex AI Model Monitoring.

# Verify registered models
gcloud ai models list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$MODEL_DISPLAY_NAME"

# Verify deployed endpoints
gcloud ai endpoints list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$ENDPOINT_DISPLAY_NAME"

# Describe an endpoint to see its deployed models
# First, get the endpoint ID
ENDPOINT_ID=$(gcloud ai endpoints list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$ENDPOINT_DISPLAY_NAME" --format="value(name)")
gcloud ai endpoints describe $ENDPOINT_ID --project=$GCP_PROJECT_ID --region=$GCP_REGION

1. Errore: 403 Permesso negato

   • Causa: Quasi sempre un problema IAM. L'account di servizio (o il tuo utente) non dispone dei permessi necessari come roles/aiplatform.user o roles/storage.objectAdmin.
   • Soluzione: Verificare i ruoli concessi al principale che esegue il codice. Assicurarsi che l'API aiplatform.googleapis.com sia abilitata nel tuo progetto.

2. Errore: BucketAlreadyOwnedByYou durante la configurazione

   • Causa: Il bucket GCS che stai cercando di creare esiste già da un'esecuzione precedente.
   • Soluzione: Questo è previsto negli script idempotenti. Il mio esempio train_and_register.py include un blocco try-except per gestirlo con eleganza.

3. Errore: Il container di serving del modello non è riuscito ad avviarsi durante il deploy

   • Causa: Una mancata corrispondenza tra l'artefatto del modello e il container di serving. I problemi comuni includono un artifact_uri errato (deve puntare alla directory contenente il modello), l'utilizzo del container sbagliato per il tuo framework ML (ad esempio, modello TensorFlow con un container Scikit-learn), o il machine_type essere troppo piccolo.
   • Soluzione: Controlla i log del container per l'endpoint in Google Cloud Logging. I messaggi di errore lì sono solitamente specifici e ti indirizzeranno alla causa principale.

Utilizzo un semplice script di orchestrazione come questo per testare l'intera pipeline localmente.

#!/bin/bash

# Exit immediately if a command exits with a non-zero status.
set -e

# --- Configuration ---
# Make sure this is set to your actual GCP project ID.
export GCP_PROJECT_ID="your-gcp-project-id"
export GCP_REGION="europe-west1"
export MODEL_DISPLAY_NAME="MyFraudDetectionModel"
export ENDPOINT_DISPLAY_NAME="fraud-detection-endpoint"

if [[ -z "$GCP_PROJECT_ID" || "$GCP_PROJECT_ID" == "your-gcp-project-id" ]]; then
    echo "Error: GCP_PROJECT_ID is not set. Please edit the script and set it to your GCP project ID."
    exit 1
fi

# Step 1: Initial setup of service account is manual (see guide).

# Step 2: Train and Register Model
echo "\n--- Running Model Training and Registration ---"
python3.12 train_and_register.py

# Step 3: Deploy Model
echo "\n--- Running Model Deployment ---"
python3.12 deploy_model.py

# Step 4: Perform Online Prediction
echo "\n--- Running Online Prediction ---"
python3.12 predict_online.py

_endpoint_id=$(gcloud ai endpoints list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$ENDPOINT_DISPLAY_NAME" --format="value(name)")
_model_id=$(gcloud ai models list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$MODEL_DISPLAY_NAME" --format="value(name)" | head -n 1)

echo "\n--- MLOps Pipeline Execution Complete ---"

# Cleanup (optional - uncomment to undeploy and delete resources after testing)
# echo "\n--- Cleaning up deployed model and endpoint ---"
# DEPLOYED_MODEL_ID=$(gcloud ai endpoints describe ${_endpoint_id} --project=$GCP_PROJECT_ID --region=$GCP_REGION --format='value(deployedModels[0].id)')
# gcloud ai endpoints undeploy-model ${_endpoint_id} --deployed-model-id=${DEPLOYED_MODEL_ID} --project=$GCP_PROJECT_ID --region=$GCP_REGION --quiet
# gcloud ai endpoints delete ${_endpoint_id} --project=$GCP_PROJECT_ID --region=$GCP_REGION --quiet
# echo "--- Cleaning up registered model --- Succeeded"
# gcloud ai models delete ${_model_id} --project=$GCP_PROJECT_ID --region=$GCP_REGION --quiet

• MLOps è Fondamentale: Tratta i modelli come software, con un ciclo di vita automatizzato, riproducibile e governabile.
• Utilizza Piattaforme Gestite: Sfrutta i servizi cloud-native come Vertex AI o Azure Machine Learning per i loro registri modelli integrati, pipeline e monitoraggio.
• Versiona e Monitora Tutto: Ogni modello, configurazione e distribuzione deve essere tracciato. Il monitoraggio continuo del drift è irrinunciabile.
• Sicuro Fin dall'Inizio: Implementa IAM rigorose e integra la scansione di sicurezza nelle tue pipeline di distribuzione.

Per far maturare il tuo Quarto Strato di IA, raccomando di concentrarti su queste aree:

1. CI/CD automatizzato per ML: Integra questi script in Cloud Build, GitHub Actions o Azure DevOps per creare un vero flusso di lavoro GitOps per i tuoi modelli.
2. Monitoraggio Avanzato dei Modelli: Vai oltre la rilevazione di base del drift. Esplora gli strumenti di spiegabilità (XAI) e costruisci trigger di riaddestramento automatizzati. Inizia con Vertex AI Explainable AI.
3. Ottimizzazione dei Costi e FinOps: Analizza il costo del tuo calcolo di addestramento e inferenza. Utilizza tipi di macchine appropriati ed esplora opzioni come le istanze riservate per carichi di lavoro prevedibili.
4. Integrazione del Feature Store: Per operazioni su scala aziendale, l'integrazione di un feature store è il prossimo passo logico per garantire coerenza ed eliminare sforzi di ingegneria dei dati ridondanti.

Adottando questi principi MLOps, trasformi il spesso opaco Quarto Strato di IA in un motore trasparente, efficiente e potente per l'innovazione della tua organizzazione.