IA sostenible: un modelo de cinco capas para la optimización de recursos

Si bien las APIs directas y en tiempo real para el 'uso de agua por GPU' aún están surgiendo, los proveedores de la nube son cada vez más transparentes sobre el impacto ambiental. Por ejemplo, Google Cloud permite exportar datos de la huella de carbono a BigQuery. El siguiente script conceptual muestra cómo se podrían usar estos datos para informar las decisiones arquitectónicas.

# conceptual_carbon_reporter.py
# This is a conceptual example. Actual implementation requires enabling
# billing data and Carbon Footprint exports to a BigQuery dataset.

import os

def get_gcp_carbon_footprint_info(project_id: str, region: str) -> dict:
    """
    Conceptual function to retrieve carbon footprint data for a GCP project.
    In a real scenario, this would query a BigQuery table containing exported data
    from the GCP Carbon Footprint report.
    """
    print(f"Querying for carbon intensity in project {project_id} for region {region}...")

    # In a real implementation, you would use the google-cloud-bigquery client
    # to run a SQL query against your exported carbon data.
    # Example Query:
    # """
    # SELECT gcp_specific_co2e_per_kwh FROM `my_project.carbon_reports.gcp_carbon_footprint`
    # WHERE region = @region ORDER BY usage_date DESC LIMIT 1
    # """

    # We'll use simulated data here for illustration.
    simulated_data = {
        "europe-west1": {"carbon_intensity_gCO2eq_per_kWh": 101, "energy_source": "Mixed, includes gas"},
        "europe-north1": {"carbon_intensity_gCO2eq_per_kWh": 64, "energy_source": "Hydro and wind dominant"},
        "europe-west4": {"carbon_intensity_gCO2eq_per_kWh": 122, "energy_source": "Mixed"},
        "westeurope": {"carbon_intensity_gCO2eq_per_kWh": 301, "energy_source": "Mixed, significant fossil fuels"} # Azure Example
    }

    if region in simulated_data:
        print(f"Found estimated carbon data for {region}.")
        return simulated_data[region]
    else:
        print(f"No specific carbon data available for {region}. Check your BigQuery export.")
        return {"carbon_intensity_gCO2eq_per_kWh": 200, "energy_source": "Generic estimate"}

if __name__ == "__main__":
    # Ensure this environment variable is set in your shell.
    gcp_project_id = os.getenv("GCP_PROJECT_ID", "your-gcp-project-id") 
    if gcp_project_id == "your-gcp-project-id":
        print("Warning: GCP_PROJECT_ID environment variable not set.")

    # Choosing a region with lower carbon intensity for a new workload.
    target_region = "europe-north1" # Finland

    carbon_info = get_gcp_carbon_footprint_info(gcp_project_id, target_region)
    print(f"\
Carbon Information for {target_region}:")
    print(f"  - Intensity: {carbon_info['carbon_intensity_gCO2eq_per_kWh']} gCO2eq/kWh")
    print(f"  - Source: {carbon_info['energy_source']}")
    print(f"\
Recommendation: Deploying AI workloads to {target_region} is preferable due to its lower carbon intensity.")

Este tipo de toma de decisiones basada en datos es donde la arquitectura pasa de la teoría a la práctica impactante.

La primera y más impactante decisión es dónde ejecutar su carga de trabajo. Priorizo las regiones con un alto porcentaje de energías renovables. Para GCP, europe-north1 (Finlandia) es una excelente elección. Así es como aprovisiono un clúster GKE allí, optimizado para futuras cargas de trabajo de IA, utilizando Terraform.

# main.tf - Aprovisionamiento de un clúster GKE energéticamente eficiente

terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 5.0"
    }
  }
}

variable "gcp_project_id" {
  description = "El ID del proyecto de GCP para desplegar recursos."
  type        = string
}

variable "gcp_region" {
  description = "La región de GCP, elegida por su baja intensidad de carbono."
  type        = string
  default     = "europe-north1"
}

provider "google" {
  project = var.gcp_project_id
  region  = var.gcp_region
}

resource "google_container_cluster" "ai_cluster" {
  name     = "ai-workload-cluster"
  location = var.gcp_region

  # Comience con un grupo predeterminado pequeño y elimínelo para usar nuestro grupo de IA personalizado
  initial_node_count = 1
  remove_default_node_pool = true

  workload_identity_config {
    workload_pool = "${var.gcp_project_id}.svc.id.goog"
  }

  logging_service    = "logging.googleapis.com/kubernetes"
  monitoring_service = "monitoring.googleapis.com/kubernetes"
}

resource "google_container_node_pool" "ai_node_pool" {
  name       = "ai-gpu-pool-main"
  location   = var.gcp_region
  cluster    = google_container_cluster.ai_cluster.name
  node_count = 1

  node_config {
    # Un tipo de máquina equilibrado. Para producción, perfile su carga de trabajo.
    machine_type = "n2-standard-8" 
    disk_size_gb = 100
    disk_type    = "pd-balanced"

    # Las VM preemptivas o Spot pueden reducir los costos hasta en un 80% para trabajos de entrenamiento
    # tolerantes a fallos, lo que también reduce la barrera financiera para usar hardware más
    # eficiente energéticamente (pero caro).
    spot = true

    # Para cargas de trabajo de GPU, especifique aceleradores. Asegúrese de que estén disponibles
    # en la región seleccionada y se ajusten a su presupuesto.
    # guest_accelerator {
    #   type  = "nvidia-tesla-t4"
    #   count = 1
    # }

    # Una cuenta de servicio dedicada es una buena práctica para la seguridad.
    # Reemplace con una cuenta de servicio pre-creada.
    service_account = "${var.gcp_project_id}@${var.gcp_project_id}.iam.gserviceaccount.com"
    oauth_scopes = [
      "https://www.googleapis.com/auth/cloud-platform",
    ]
  }

  autoscaling {
    min_node_count = 0 # Escalar a cero para ahorrar costos cuando está inactivo
    max_node_count = 3
  }

  management {
    auto_repair  = true
    auto_upgrade = true
  }
}

output "cluster_name" {
  description = "El nombre del clúster GKE."
  value       = google_container_cluster.ai_cluster.name
}

output "cluster_location" {
  description = "La ubicación del clúster GKE."
  value       = google_container_cluster.ai_cluster.location
}

Con esta configuración, hemos definido una base escalable, rentable y más consciente de la energía para nuestras cargas de trabajo de IA.

Los datos tienen 'gravedad', y moverlos consume energía. Para muchas tareas de inferencia, ejecutar el modelo cerca del usuario en una red de borde es mucho más eficiente que ir y venir a una nube central. Workers AI de Cloudflare nos permite hacer esto con funciones sin servidor. Aquí se explica cómo construir un resumidor de borde en Python.

Primero, instale la CLI de wrangler e inicie sesión:

# Install the Cloudflare CLI
npm install -g wrangler

# Authenticate wrangler with your Cloudflare account
wrangler login

A continuación, cree un nuevo directorio de proyecto. A diferencia del predeterminado, lo configuraremos manualmente para Python.

# Create a project and navigate into it
mkdir ai-edge-summarizer && cd ai-edge-summarizer

# Create our source directory and Python file
mkdir src
touch src/worker.py

# Create a config file for wrangler
touch wrangler.toml

Ahora, configure wrangler.toml para usar un worker de Python.

# wrangler.toml
name = "ai-edge-summarizer"
main = "src/worker.py:AIWorker" # Points to the Python file and the class within it
compatibility_date = "2024-03-20"
compatibility_flags = ["python_workers"]

[ai]
binding = "AI" # This makes the AI service available in our code as env.AI

Finalmente, escriba la lógica del worker de Python. Este código define una clase con un método fetch que el tiempo de ejecución de Cloudflare ejecutará en cada solicitud.

# src/worker.py

import json

# El tiempo de ejecución de Python de Cloudflare Workers proporciona la clase Response y el objeto env.
# No es necesario importarlos desde un paquete.

class AIWorker:
    async def fetch(self, request, env, ctx):
        if request.method != 'POST':
            return Response('Method Not Allowed', status=405)

        try:
            data = await request.json()
            text = data.get('text')
        except json.JSONDecodeError:
            return Response('Invalid JSON in request body', status=400)

        if not text:
            return Response('Missing "text" in request body', status=400)

        # Ejecutar un modelo de resumen directamente en la red de borde de Cloudflare
        try:
            ai_response = await env.AI.run(
                "@cf/facebook/bart-large-cnn",
                {
                    "input_text": text,
                    "max_length": 150,
                    "min_length": 30,
                }
            )
        except Exception as e:
            print(f"Error running Workers AI model: {e}")
            return Response("Failed to run AI model", status=500)

        return Response(
            json.dumps({"summary": ai_response.get("summary")}),
            headers={'content-type': 'application/json'}
        )

Despliegue el worker en la red de Cloudflare:

# Deploy the Cloudflare Worker
wrangler deploy

Este patrón reduce significativamente los costos de transporte de datos y la latencia, impactando directamente la huella energética de su aplicación.

El caso de uso más convincente para la IA en un mundo con recursos limitados es aplicarla al problema mismo. Modelemos una aplicación Python para predecir la calidad del agua a partir de datos de sensores IoT. Este tipo de sistema permite una intervención proactiva, previniendo la contaminación y conservando un recurso crítico.

Este script simula un componente de aplicación que recopila datos de sensores y llama a un endpoint de modelo desplegado para una predicción. El modelo mismo se habría entrenado en nuestro clúster GKE del Paso 1.

# water_quality_monitor.py

import os
import json
import random
from datetime import datetime, timezone

class WaterQualityPredictor:
    """Simula un cliente para un modelo de predicción de calidad del agua desplegado."""

    def __init__(self, endpoint_url: str, api_key: str):
        if not endpoint_url or 'example.com' in endpoint_url:
            print("Warning: Using a placeholder endpoint. Predictions will be simulated.")
            self.endpoint_url = None
        else:
            self.endpoint_url = endpoint_url
        self.api_key = api_key
        print(f"Predictor initialized for endpoint: {endpoint_url}")

    def predict(self, sensor_data: dict) -> str:
        """
        Llama al endpoint del modelo de IA. Para este ejemplo, simulamos la llamada
        y la lógica del modelo si el endpoint no es real.
        """
        print(f"Sending sensor data for prediction: {json.dumps(sensor_data)}")

        # En un sistema real, usarías una librería como 'requests' o 'httpx'
        # para hacer una solicitud POST a self.endpoint_url.
        # e.g., response = httpx.post(self.endpoint_url, json=sensor_data, headers=...)

        if not self.endpoint_url:
            # Simulate the model logic locally for demonstration
            ph = sensor_data.get("ph", 7.0)
            turbidity = sensor_data.get("turbidity_ntu", 1.0)

            if not (6.5 <= ph <= 8.5) or turbidity > 5.0:
                return "POOR_QUALITY: IMMEDIATE_ACTION_REQUIRED"
            elif turbidity > 3.0:
                return "MODERATE_QUALITY: MONITORING_ADVISED"
            else:
                return "GOOD_QUALITY: STABLE"

        # This part would execute if a real endpoint was provided
        # try:
        #     # result = response.json()
        #     # return result.get('prediction', 'UNKNOWN')
        # except Exception as e:
        #     return f"ERROR: {e}"
        return "SIMULATION_MODE_ONLY"

def collect_iot_sensor_data() -> dict:
    """Simula la lectura de un conjunto de sensores IoT."""
    return {
        "sensor_id": "WQ-1138",
        "timestamp_utc": datetime.now(timezone.utc).isoformat(), 
        "ph": round(random.uniform(6.0, 9.0), 2), 
        "turbidity_ntu": round(random.uniform(0.5, 8.0), 2), 
        "temperature_c": round(random.uniform(15.0, 25.0), 2), 
    }

if __name__ == "__main__":
    # En un despliegue real, este sería un endpoint seguro, por ejemplo, de un despliegue de Azure ML.
    model_endpoint = os.getenv("AI_MODEL_ENDPOINT", "https://water-quality-predictor.westeurope.azurecontainerapps.io/api/predict") 
    api_key = os.getenv("AI_SERVICE_API_KEY", "super-secret-key") 
    predictor = WaterQualityPredictor(model_endpoint, api_key) 

    print("\
--- Real-Time Water Quality Monitoring Simulation ---")
    for i in range(3):
        readings = collect_iot_sensor_data()
        prediction = predictor.predict(readings) 
        print(f"  -> AI Prediction #{i+1}: {prediction}")

Esta aplicación funcional representa la culminación de la pila, justificando el costo de recursos de las capas inferiores al proporcionar un valor tangible, en este caso, para la gestión ambiental.

• Error: Terraform googleapi: Error 400: The user does not have access

  • Razón: Las credenciales que usa Terraform (probablemente de gcloud auth application-default login) carecen de los permisos IAM para crear clústeres GKE o recursos relacionados (roles/container.admin).
  • Solución: Asegúrese de que su usuario o cuenta de servicio tenga los roles IAM necesarios en el proyecto GCP de destino. Para GKE, roles/container.admin es potente pero eficaz para pruebas. Para producción, use roles de mínimo privilegio.

• Error: Wrangler Workers AI: Invalid model ID

  • Razón: El identificador del modelo (@cf/facebook/bart-large-cnn) está mal escrito, obsoleto o es incorrecto.
  • Solución: Consulte el catálogo oficial de Cloudflare AI para obtener la lista actual de modelos disponibles y sus nombres exactos. Estos pueden cambiar con el tiempo.

• Error: Script Python ConnectionRefusedError

  • Razón: Esto ocurre si proporciona un AI_MODEL_ENDPOINT real que es incorrecto, está inactivo o bloqueado por un firewall. El script de ejemplo está diseñado para ejecutarse en modo simulado si el endpoint no está completamente configurado.
  • Solución: Si está probando un endpoint real, verifique su URL, asegúrese de que esté desplegado y en funcionamiento, y compruebe que la red de su máquina local permite las conexiones salientes a él.