Nachhaltige KI: Ein Fünf-Schichten-Modell zur Ressourcenoptimierung

Während direkte Echtzeit-APIs für den „Wasserverbrauch pro GPU“ noch in den Kinderschuhen stecken, werden Cloud-Anbieter transparenter hinsichtlich ihrer Umweltauswirkungen. Google Cloud ermöglicht es beispielsweise, Kohlenstoff-Fußabdruck-Daten nach BigQuery zu exportieren. Das folgende Konzeptskript zeigt, wie Sie solche Daten nutzen könnten, um Ihre Architekturentscheidungen zu treffen.

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

Diese Art der datengesteuerten Entscheidungsfindung ist der Punkt, an dem Architektur von der Theorie zu einer wirkungsvollen Praxis wird.

Die erste und wichtigste Entscheidung ist, wo Sie Ihre Workload ausführen. Ich priorisiere Regionen mit einem hohen Anteil an erneuerbaren Energien. Für GCP ist europe-north1 (Finnland) eine ausgezeichnete Wahl. So stelle ich dort einen GKE-Cluster bereit, der für zukünftige KI-Workloads optimiert ist, mithilfe von 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
}

Mit dieser Konfiguration haben wir eine skalierbare, kosteneffiziente und energiebewusstere Grundlage für unsere KI-Workloads geschaffen.

Daten haben eine Gravitation, und deren Bewegung verbraucht Energie. Für viele Inferenzaufgaben ist es weitaus effizienter, das Modell nah am Benutzer in einem Edge-Netzwerk auszuführen, als es zu einer zentralen Cloud hin- und herzusenden. Cloudflare's Workers AI ermöglicht uns dies mit serverlosen Funktionen. So bauen Sie einen Edge-Summarizer in Python.

Installieren Sie zuerst das wrangler CLI und melden Sie sich an:

# Install the Cloudflare CLI
npm install -g wrangler

# Authenticate wrangler with your Cloudflare account
wrangler login

Erstellen Sie als Nächstes ein neues Projektverzeichnis. Anders als beim Standard werden wir es manuell für Python einrichten.

# 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

Konfigurieren Sie nun wrangler.toml, um einen Python-Worker zu verwenden.

# 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

Schreiben Sie schließlich die Python-Worker-Logik. Dieser Code definiert eine Klasse mit einer fetch-Methode, die die Cloudflare-Laufzeit bei jeder Anfrage ausführt.

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

Stellen Sie den Worker im Cloudflare-Netzwerk bereit:

# Deploy the Cloudflare Worker
wrangler deploy

Dieses Muster reduziert die Datentransportkosten und die Latenz erheblich, was sich direkt auf den Energieverbrauch Ihrer Anwendung auswirkt.

Der überzeugendste Anwendungsfall für KI in einer ressourcenbeschränkten Welt ist es, sie auf das Problem selbst anzuwenden. Modellieren wir eine Python-Anwendung zur Vorhersage der Wasserqualität aus IoT-Sensordaten. Solch ein System ermöglicht proaktives Eingreifen, verhindert Kontaminationen und bewahrt eine kritische Ressource.

Dieses Skript simuliert eine Anwendungskomponente, die Sensordaten sammelt und einen bereitgestellten Modell-Endpunkt für eine Vorhersage aufruft. Das Modell selbst wäre auf unserem GKE-Cluster aus Schritt 1 trainiert worden.

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

Diese funktionale Anwendung stellt den Höhepunkt des Stacks dar – sie rechtfertigt die Ressourcenkosten der unteren Schichten, indem sie einen greifbaren Wert liefert, in diesem Fall für das Umweltmanagement.

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

  • Grund: Die von Terraform verwendeten Anmeldeinformationen (wahrscheinlich von gcloud auth application-default login) verfügen nicht über die IAM-Berechtigungen zum Erstellen von GKE-Clustern oder verwandten Ressourcen (roles/container.admin).
  • Lösung: Stellen Sie sicher, dass Ihr Benutzer- oder Dienstkonto über die erforderlichen IAM-Rollen im Ziel-GCP-Projekt verfügt. Für GKE ist roles/container.admin mächtig, aber effektiv für Tests. Für die Produktion verwenden Sie Rollen mit den geringsten Rechten.

• Fehler: Wrangler Workers AI: Invalid model ID

  • Grund: Der Modellidentifikator (@cf/facebook/bart-large-cnn) ist falsch geschrieben, veraltet oder inkorrekt.
  • Lösung: Überprüfen Sie den offiziellen Cloudflare AI-Katalog auf die aktuelle Liste der verfügbaren Modelle und deren genaue Namen. Diese können sich im Laufe der Zeit ändern.

• Fehler: Python-Skript ConnectionRefusedError

  • Grund: Dies tritt auf, wenn Sie einen echten AI_MODEL_ENDPOINT angeben, der falsch, nicht verfügbar oder durch eine Firewall blockiert ist. Das Beispielskript ist so konzipiert, dass es in einem simulierten Modus läuft, wenn der Endpunkt nicht vollständig konfiguriert ist.
  • Lösung: Wenn Sie einen echten Endpunkt testen, überprüfen Sie dessen URL, stellen Sie sicher, dass er bereitgestellt und ausgeführt wird, und prüfen Sie, ob das Netzwerk Ihres lokalen Rechners ausgehende Verbindungen zu ihm zulässt.