AI sostenibile: un modello a cinque strati per l'ottimizzazione delle risorse

Mentre API dirette e in tempo reale per "consumo d'acqua per GPU" sono ancora emergenti, i fornitori di cloud stanno diventando più trasparenti riguardo all'impatto ambientale. Ad esempio, Google Cloud ti consente di esportare i dati sull'impronta di carbonio in BigQuery. Lo script concettuale seguente mostra come potresti utilizzare tali dati per informare le tue decisioni architettoniche.

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

Questo tipo di processo decisionale basato sui dati è dove l'architettura si muove dalla teoria alla pratica di impatto.

La prima e più significativa decisione è dove eseguire il tuo carico di lavoro. Do la priorità alle regioni con un'alta percentuale di energie rinnovabili. Per GCP, europe-north1 (Finlandia) è una scelta eccellente. Ecco come effettuo il provisioning di un cluster GKE lì, ottimizzato per futuri carichi di lavoro AI, usando Terraform.

# main.tf - Provisioning an energy-efficient GKE cluster

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

variable "gcp_project_id" {
  description = "The GCP project ID to deploy resources into."
  type        = string
}

variable "gcp_region" {
  description = "The GCP region, chosen for its low carbon intensity."
  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

  # Start with a small default pool and remove it to use our custom AI pool
  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 {
    # A balanced machine type. For production, profile your workload.
    machine_type = "n2-standard-8" 
    disk_size_gb = 100
    disk_type    = "pd-balanced"

    # Preemptible or Spot VMs can slash costs by up to 80% for fault-tolerant
    # training jobs, which also reduces the financial barrier to using more
    # energy-efficient (but expensive) hardware.
    spot = true

    # For GPU workloads, specify accelerators. Ensure they are available
    # in the selected region and fit your budget.
    # guest_accelerator {
    #   type  = "nvidia-tesla-t4"
    #   count = 1
    # }

    # A dedicated service account is best practice for security.
    # Replace with a pre-created service account.
    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 # Scale to zero to save costs when idle
    max_node_count = 3
  }

  management {
    auto_repair  = true
    auto_upgrade = true
  }
}

output "cluster_name" {
  description = "The name of the GKE cluster."
  value       = google_container_cluster.ai_cluster.name
}

output "cluster_location" {
  description = "The location of the GKE cluster."
  value       = google_container_cluster.ai_cluster.location
}

Con questa configurazione, abbiamo definito una base scalabile, conveniente e più attenta all'energia per i nostri carichi di lavoro di IA.

I dati hanno una gravità e il loro spostamento consuma energia. Per molte attività di inferenza, eseguire il modello vicino all'utente su una rete edge è molto più efficiente che fare un "round-trip" verso un cloud centrale. Cloudflare Workers AI ci permette di farlo con funzioni serverless. Ecco come costruire un riassuntore edge in Python.

Innanzitutto, installa la CLI wrangler ed effettua il login:

# Install the Cloudflare CLI
npm install -g wrangler

# Authenticate wrangler with your Cloudflare account
wrangler login

Successivamente, crea una nuova directory di progetto. A differenza dell'impostazione predefinita, la configureremo manualmente per 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

Ora, configura wrangler.toml per utilizzare un worker 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

Infine, scrivi la logica del worker Python. Questo codice definisce una classe con un metodo fetch che il runtime di Cloudflare eseguirà su ogni richiesta.

# src/worker.py

import json

# The Cloudflare Workers Python runtime provides the `Response` class and `env` object.
# They do not need to be imported from a package.

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)

        # Run a summarization model directly on the Cloudflare edge network
        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'}
        )

Distribuisci il worker alla rete Cloudflare:

# Deploy the Cloudflare Worker
wrangler deploy

Questo pattern riduce significativamente i costi di trasporto dati e la latenza, influenzando direttamente l'impronta energetica della tua applicazione.

Il caso d'uso più convincente per l'IA in un mondo con risorse limitate è quello di applicarla al problema stesso. Modelliamo un'applicazione Python per prevedere la qualità dell'acqua dai dati dei sensori IoT. Questo tipo di sistema consente un intervento proattivo, prevenendo la contaminazione e conservando una risorsa critica.

Questo script simula un componente applicativo che raccoglie dati da sensori e chiama un endpoint di modello distribuito per una previsione. Il modello stesso sarebbe stato addestrato sul nostro cluster GKE del Passo 1.

# water_quality_monitor.py

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

class WaterQualityPredictor:
    """Simulates a client for a deployed water quality prediction model."""

    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:
        """
        Calls the AI model endpoint. For this example, we simulate the call
        and the model's logic if the endpoint is not real.
        """
        print(f"Sending sensor data for prediction: {json.dumps(sensor_data)}")

        # In a real system, you would use a library like 'requests' or 'httpx'
        # to make a POST request to 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:
    """Simulates reading from an IoT sensor array."""
    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__":
    # In a real deployment, this would be a secured endpoint, e.g., from an Azure ML deployment.
    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}")

Questa applicazione funzionale rappresenta il culmine dello stack, giustificando il costo delle risorse degli strati inferiori fornendo un valore tangibile, in questo caso, per la gestione ambientale.

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

  • Motivo: Le credenziali utilizzate da Terraform (probabilmente da gcloud auth application-default login) non dispongono delle autorizzazioni IAM per creare cluster GKE o risorse correlate (roles/container.admin).
  • Soluzione: Assicurati che il tuo utente o account di servizio abbia i ruoli IAM necessari nel progetto GCP di destinazione. Per GKE, roles/container.admin è potente ma efficace per i test. Per la produzione, utilizza ruoli a privilegio minimo.

• Errore: Wrangler Workers AI: Invalid model ID

  • Motivo: L'identificatore del modello (@cf/facebook/bart-large-cnn) è errato, obsoleto o non valido.
  • Soluzione: Controlla il catalogo ufficiale di Cloudflare AI per l'elenco corrente dei modelli disponibili e i loro nomi esatti. Questi possono cambiare nel tempo.

• Errore: Script Python ConnectionRefusedError

  • Motivo: Si verifica se si fornisce un AI_MODEL_ENDPOINT reale che non è corretto, non è attivo o è bloccato da un firewall. Lo script di esempio è progettato per funzionare in modalità simulata se l'endpoint non è completamente configurato.
  • Soluzione: Se stai testando un endpoint reale, verifica il suo URL, assicurati che sia distribuito e in esecuzione e controlla che la rete del tuo computer locale consenta connessioni in uscita verso di esso.