Modelos de IA y progreso algorítmico

En cualquier sistema MLOps de producción que construyo, la gobernanza de modelos es primordial. No se trata solo de rastrear el linaje, sino también de hacer cumplir el cumplimiento y la seguridad.

• Control de Versiones: Aplicar Git para todo el código del pipeline y usar el Registro de Modelos para versionar los binarios del modelo y sus metadatos asociados.
• Registro de Auditoría (Audit Logging): Todos los eventos del ciclo de vida —ejecuciones de entrenamiento, despliegues, cambios de etapa, registros de predicción— deben ser capturados. Tanto GCP como Azure proporcionan un registro extenso y no repudiable para este propósito.
• Control de Acceso: Usar políticas estrictas de IAM para definir quién puede entrenar, registrar, desplegar o consultar modelos. Esto sigue el principio de menor privilegio.
• Escaneo de Modelos (Model Scanning): Una práctica en evolución pero crítica es integrar escaneos de seguridad para los artefactos del modelo para detectar vulnerabilidades o datos sensibles incrustados. Esto debería ser parte de su pipeline de CI/CD.

Aquí hay una mirada práctica a cómo un solo paso de entrenamiento se convierte en un activo codificado y versionado en un Pipeline de Vertex AI usando el SDK de 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')

Este componente simple encapsula la lógica de entrenamiento, convirtiéndolo en un paso reutilizable y versionable en un flujo de trabajo automatizado más grande.

Primero, cree una cuenta de servicio dedicada para los pipelines de MLOps para hacer cumplir el principio de menor 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"

Esta configuración única crea la identidad y los permisos que utilizará nuestro flujo de trabajo automatizado.

Este script simula una ejecución de entrenamiento local, luego sube el artefacto resultante a Cloud Storage y lo registra en el Registro de Modelos de Vertex AI. Está diseñado para ser idempotente, manejando casos en los que recursos como los buckets de GCS ya existen.

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

Observe el parámetro parent_model: esto es crucial para crear nuevas versiones de un modelo existente, formando el núcleo del linaje de su modelo.

Una vez registrado, desplegar el modelo en un punto final gestionado lo hace disponible para predicciones en tiempo real.

# 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}'.")

Este script abstrae la infraestructura subyacente, proporcionándole un servicio de inferencia escalable y altamente disponible. El parámetro traffic_split es cómo implemento despliegues canary y pruebas A/B en producción.

Con el modelo desplegado, ahora puede enviarle solicitudes de inferencia.

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

Preste mucha atención al formato de entrada. Cualquier falta de coincidencia entre las características enviadas aquí y las características utilizadas en el entrenamiento es una fuente común de error.

El monitoreo de modelos es innegociable para la IA en producción. Si bien una configuración completa va más allá del alcance de este artículo, es crucial comprender el concepto. Usted define qué monitorear (por ejemplo, desviación de datos, desviación de conceptos) y proporciona un conjunto de datos de referencia (sus datos de entrenamiento) para la comparación. Vertex AI luego registra automáticamente las solicitudes de predicción y le alerta cuando el rendimiento se desvía.

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

Este ciclo de retroalimentación continua es lo que hace que la Cuarta Capa de IA sea robusta y resiliente. Para una inmersión profunda, remito a mis clientes a la documentación oficial de Monitoreo de Modelos de Vertex AI.

# 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. Error: 403 Permiso denegado

   • Causa: Casi siempre un problema de IAM. La cuenta de servicio (o su usuario) carece de los permisos necesarios como roles/aiplatform.user o roles/storage.objectAdmin.
   • Solución: Verifique los roles otorgados al principal que ejecuta el código. Asegúrese de que la API aiplatform.googleapis.com esté habilitada en su proyecto.

2. Error: BucketAlreadyOwnedByYou durante la configuración

   • Causa: El bucket de GCS que intenta crear ya existe de una ejecución anterior.
   • Solución: Esto es esperado en scripts idempotentes. Mi ejemplo train_and_register.py incluye un bloque try-except para manejar esto con elegancia.

3. Error: The model's serving container failed to start durante el despliegue

   • Causa: Una falta de coincidencia entre el artefacto del modelo y el contenedor de servicio. Los problemas comunes incluyen un artifact_uri incorrecto (debe apuntar al directorio que contiene el modelo), usar el contenedor incorrecto para su framework de ML (por ejemplo, modelo de TensorFlow con un contenedor de Scikit-learn), o que el machine_type sea demasiado pequeño.
   • Solución: Verifique los registros del contenedor para el punto final en Google Cloud Logging. Los mensajes de error allí suelen ser específicos y le indicarán la causa raíz.

Utilizo un script de orquestación simple como este para probar todo el 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 es Fundamental: Trate los modelos como software, con un ciclo de vida automatizado, reproducible y gobernable.
• Utilice Plataformas Gestionadas: Aproveche los servicios nativos de la nube como Vertex AI o Azure Machine Learning por sus registros de modelos, pipelines y monitoreo integrados.
• Versionar y Monitorear Todo: Cada modelo, configuración y despliegue debe ser rastreado. El monitoreo continuo de la desviación (drift) es innegociable.
• Seguridad desde el Inicio: Implemente IAM estricto e incorpore el escaneo de seguridad en sus pipelines de despliegue.

Para madurar su Cuarta Capa de IA, recomiendo centrarse en estas áreas a continuación:

1. CI/CD Automatizado para ML: Integre estos scripts en Cloud Build, GitHub Actions o Azure DevOps para crear un verdadero flujo de trabajo GitOps para sus modelos.
2. Monitoreo Avanzado de Modelos: Vaya más allá de la detección básica de desviación. Explore las herramientas de explicabilidad (XAI) y construya disparadores de reentrenamiento automatizados. Comience con Vertex AI Explainable AI.
3. Optimización de Costos y FinOps: Analice el costo de su cómputo de entrenamiento e inferencia. Utilice tipos de máquinas apropiados y explore opciones como instancias reservadas para cargas de trabajo predecibles.
4. Integración de Almacén de Características (Feature Store): Para operaciones a escala empresarial, integrar un almacén de características es el siguiente paso lógico para asegurar la coherencia y eliminar esfuerzos redundantes de ingeniería de datos.

Al adoptar estos principios de MLOps, convierte la a menudo opaca Cuarta Capa de IA en un motor transparente, eficiente y potente para la innovación de su organización.