L'IA durable : un modèle en cinq couches pour l'optimisation des ressources

Alors que les API directes et en temps réel pour la « consommation d'eau par GPU » sont encore en émergence, les fournisseurs de cloud sont de plus en plus transparents quant à l'impact environnemental. Par exemple, Google Cloud vous permet d'exporter les données d'empreinte carbone vers BigQuery. Le script conceptuel suivant montre comment vous pourriez utiliser ces données pour éclairer vos décisions architecturales.

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

Ce type de prise de décision basée sur les données est là où l'architecture passe de la théorie à une pratique impactante.

La première décision, et la plus impactante, est où exécuter votre charge de travail. Je privilégie les régions avec un pourcentage élevé d'énergies renouvelables. Pour GCP, europe-north1 (Finlande) est un excellent choix. Voici comment je provisionne un cluster GKE là-bas, optimisé pour les futures charges de travail d'IA, en utilisant 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
}

Avec cette configuration, nous avons défini une base évolutive, rentable et plus soucieuse de l'énergie pour nos charges de travail d'IA.

Les données ont une gravité, et les déplacer consomme de l'énergie. Pour de nombreuses tâches d'inférence, exécuter le modèle près de l'utilisateur sur un réseau de périphérie est bien plus efficace qu'un aller-retour vers un cloud central. Workers AI de Cloudflare nous permet de faire cela avec des fonctions sans serveur. Voici comment construire un sommaire en périphérie en Python.

Tout d'abord, installez l'interface de ligne de commande wrangler et connectez-vous :

# Install the Cloudflare CLI
npm install -g wrangler

# Authenticate wrangler with your Cloudflare account
wrangler login

Ensuite, créez un nouveau répertoire de projet. Contrairement à la configuration par défaut, nous le configurerons manuellement pour 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

Maintenant, configurez wrangler.toml pour utiliser 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

Enfin, écrivez la logique du worker Python. Ce code définit une classe avec une méthode fetch que l'environnement d'exécution de Cloudflare exécutera à chaque requête.

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

Déployez le worker sur le réseau Cloudflare :

# Deploy the Cloudflare Worker
wrangler deploy

Ce modèle réduit considérablement les coûts de transport de données et la latence, impactant directement l'empreinte énergétique de votre application.

Le cas d'utilisation le plus convaincant pour l'IA dans un monde aux ressources limitées est de l'appliquer au problème lui-même. Modélisons une application Python pour prédire la qualité de l'eau à partir des données de capteurs IoT. Ce type de système permet une intervention proactive, prévenant la contamination et conservant une ressource critique.

Ce script simule un composant d'application qui collecte des données de capteurs et appelle un point d'accès de modèle déployé pour une prédiction. Le modèle lui-même aurait été entraîné sur notre cluster GKE de l'étape 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}")

Cette application fonctionnelle représente l'aboutissement de la pile — justifiant le coût des ressources des couches inférieures en apportant une valeur tangible, dans ce cas, pour la gestion environnementale.

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

  • Raison : Les identifiants utilisés par Terraform (probablement issus de gcloud auth application-default login) ne disposent pas des permissions IAM nécessaires pour créer des clusters GKE ou des ressources associées (roles/container.admin).
  • Solution : Assurez-vous que votre utilisateur ou compte de service dispose des rôles IAM nécessaires dans le projet GCP cible. Pour GKE, roles/container.admin est puissant mais efficace pour les tests. Pour la production, utilisez des rôles à privilège minimum.

• Erreur : Wrangler Workers AI: Invalid model ID

  • Raison : L'identifiant du modèle (@cf/facebook/bart-large-cnn) est mal orthographié, obsolète ou incorrect.
  • Solution : Vérifiez le catalogue officiel de Cloudflare AI pour la liste actuelle des modèles disponibles et leurs noms exacts. Ceux-ci peuvent changer avec le temps.

• Erreur : Script Python ConnectionRefusedError

  • Raison : Cela se produit si vous fournissez un AI_MODEL_ENDPOINT réel qui est incorrect, hors service ou bloqué par un pare-feu. Le script d'exemple est conçu pour fonctionner en mode simulé si le point d'accès n'est pas entièrement configuré.
  • Solution : Si vous testez un point d'accès réel, vérifiez son URL, assurez-vous qu'il est déployé et en cours d'exécution, et vérifiez que le réseau de votre machine locale autorise les connexions sortantes vers celui-ci.