Dieser hybride RAG-Ansatz vereint das Beste aus GCPs Indizierungs- und Grounding-Fähigkeiten mit AWSs robuster Data Lake-Integration. Ich nutze Amazon S3 effektiv als unseren primären Datenspeicher für Rohdokumente, die dann verarbeitet und in eine Amazon Bedrock Knowledge Base indiziert werden. Gleichzeitig speist eine parallele Pipeline relevante Daten in Vertex AI Search ein.

Die „Brücke“ ist entscheidend: Sicherstellung, dass Vektor-Embeddings und Metadaten harmonisiert werden, oft durch eine gemeinsame, Cloud-agnostische Vektor-Datenbank oder einen hochentwickelten Synchronisationsmechanismus. Dies ermöglicht meinem RAG-Orchestrator, beide Quellen abzufragen und einen umfassenden Kontext für die LLM-Fundierung zu synthetisieren.

Für die Codegenerierung, insbesondere bei komplexen technischen Workflows, bleibt Anthropic's Claude 4.6 Sonnet ein Maßstab. Anstatt direkter API-Aufrufe leite ich Anfragen jedoch über OpenRouter. Dies bietet eine entscheidende Abstraktionsschicht, die Modell-Failover, Kostenoptimierung und vereinfachtes API-Management ermöglicht. Es bedeutet, dass ich, wenn Claude 4.6 Sonnet langsam läuft oder zu teuer wird, nahtlos zu Gemini 2.5 Pro (über OpenRouter) wechseln kann, ohne meinen Anwendungscode zu ändern. LangChain und LlamaIndex bauen dann die agentenbasierte Orchestrierung darauf auf, was ein kontextuelles Verständnis von Codebasen und die dynamische Nutzung von Tools ermöglicht.

Bei der Architekturgestaltung der Komponente „geteilte Vektor-DB oder Synchronisationsdienst“ bewerte ich Vektordatenbank-Optionen sorgfältig basierend auf Latenzanforderungen, Funktionsumfang und Betriebskosten. Für die europäische Datensouveränität bedeutet dies oft die Auswahl von Anbietern mit EU-basierter Infrastruktur oder Self-Hosting.

• Pinecone: Ein vollständig verwalteter Vektordatenbankdienst, bekannt für seine Skalierbarkeit und Leistung. Er bietet Regionen in Europa an, was ihn zu einem starken Kandidaten für verwaltete Lösungen macht, wenn spezifische EU-Regionen für die Datenresidenz verfügbar sind. Ich schätze seine Benutzerfreundlichkeit und API-Konsistenz.
• Weaviate: Dies kann als verwalteter Dienst (Weaviate Cloud) oder selbst gehostet auf Kubernetes betrieben werden. Seine flexiblen Bereitstellungsoptionen machen es attraktiv für die Einhaltung strenger Datenresidenzanforderungen, da ich es in bestimmten EU-VPCs oder GKE/EKS-Clustern bereitstellen kann. Es bietet auch eine robuste API und ein Modul-Ökosystem.
• AlloyDB für PostgreSQL mit pgvector: Für diejenigen, die bereits stark in PostgreSQL oder GCP investiert sind, bietet die Nutzung von AlloyDB mit der pgvector-Erweiterung eine leistungsstarke und integrierte Lösung. Obwohl pgvector die Rohleistung spezialisierter Vektordatenbanken für extrem große Szenarien möglicherweise nicht erreicht, bietet es eine ausgezeichnete Datenlokalität und vereinfachte Verwaltung in einer vertrauten relationalen Datenbankumgebung. Der Betrieb von AlloyDB in europe-west3 gewährleistet die Datenresidenz.

Meine Wahl hängt typischerweise von der für RAG erforderlichen Latenz, den benötigten spezifischen Funktionen (z. B. Filterung, Hybridsuche) und den operativen Präferenzen für verwaltete gegenüber selbst gehosteten Lösungen innerhalb der EU ab.

Compliance ist kein nachträglicher Einfall; sie ist im Design verankert. Für die DSGVO ist die Datenresidenz von größter Bedeutung. Alle RAG-Quellen müssen sich in EU-zentralen Regionen befinden (z. B. eu-central-1 für AWS, europe-west3 für GCP). Das bedeutet, dass S3-Buckets, Bedrock Knowledge Bases und Vertex AI Search-Datenspeicher ausschließlich in diesen Regionen bereitgestellt werden. Über die Residenz hinaus ist die Anonymisierung von PII entscheidend. Bevor Prompts externe APIs wie OpenRouter erreichen, stellt eine PII-Bereinigungsschicht sicher, dass sensible Daten niemals unser kontrolliertes Umfeld verlassen. Dies beinhaltet oft clientseitige Verarbeitung oder einen dedizierten Proxy-Dienst innerhalb unserer kontrollierten Perimeter. Ab heute ist Gemini 3.1 nur über einen globalen Endpunkt und nicht über regionale verfügbar, daher verwenden wir die Version 2.5.

In einer KI-gesteuerten Architektur geht es bei der Modell-Governance nicht nur um die Versionierung. Es geht darum sicherzustellen, dass jedes Modell, von Embedding-Generatoren bis hin zu Code-Generierungs-LLMs, strenge Sicherheitsstandards einhält. Ich implementiere:

• Modellregister: Ein zentraler Katalog für alle Modelle, einschließlich ihrer Quelle, Version und Herkunft der Trainingsdaten.
• Schwachstellen-Scanning: Embedding-Modelle und kundenspezifische, feinabgestimmte LLMs werden auf bekannte Schwachstellen und die Einhaltung von Sicherheitsgrundlagen gescannt.
• Audit-Protokollierung: Jede Interaktion mit einer LLM-API, insbesondere solche, die über OpenRouter geleitet werden, wird mit relevanten Metadaten (Anforderungs-IDs, Token-Zählungen, Zeitstempel) für Compliance- und Kostenanalyse protokolliert. Dies ist entscheidend für die DSGVO, da es einen überprüfbaren Verlauf der Datenverarbeitungsaktivitäten liefert.
• Zugriffskontrolle: Granulare IAM-Richtlinien schränken ein, wer bestimmte Modelle bereitstellen, aktualisieren oder sogar aufrufen kann, und integrieren sich in die Workload Identity Federation für Cross-Cloud-Szenarien.

Ich verwende Terraform, um die grundlegenden Compute-Dienste in GCP und AWS zu definieren und zu verwalten, die unsere RAG- und Codegenerierungskomponenten hosten. Dies gewährleistet Konsistenz und Auditierbarkeit und platziert, was entscheidend ist, Ressourcen in EU-Regionen für die GDPR-Compliance.

# main.tf für Cross-Cloud-Computing

# Google Cloud Provider für Europa konfigurieren
provider "google" {
  project = var.gcp_project_id
  region  = "europe-west3" # Frankfurt, Deutschland
}

# AWS Provider für Europa konfigurieren
provider "aws" {
  region = "eu-central-1" # Frankfurt, Deutschland
}

# --- GCP Cloud Run für den Anwendungsdienst ---
resource "google_cloud_run_service" "main_app_service" {
  name     = "multi-cloud-ai-service"
  location = "europe-west3"
  template {
    spec {
      containers {
        image = "gcr.io/${var.gcp_project_id}/multi-cloud-ai-app:latest"
        env {
          name  = "OPENROUTER_API_KEY"
          value = var.openrouter_api_key
        }
        env {
          name  = "AWS_REGION"
          value = "eu-central-1"
        }
        env {
          name = "GCP_PROJECT_ID"
          value = var.gcp_project_id
        }
        env {
          name = "GCP_REGION"
          value = var.gcp_region
        }
        env {
          name = "GCP_DATASTORE_ID"
          value = var.gcp_datastore_id # Vertex AI Search Datastore-ID
        }
        env {
          name = "AWS_BEDROCK_KB_ID"
          value = var.aws_bedrock_kb_id # Bedrock Knowledge Base-ID
        }
      }
      service_account_name = google_service_account.cloud_run_sa.email
    }
  }
  traffic {
    percent = 100
    latest  = true
  }
}

resource "google_service_account" "cloud_run_sa" {
  account_id   = "cloud-run-ai-sa"
  display_name = "Dienstkonto für Multi-Cloud AI Cloud Run-Dienst"
}

# --- AWS Lambda für potenziellen Bedrock-spezifischen Proxy oder asynchrone Aufgaben ---
resource "aws_lambda_function" "bedrock_proxy_lambda" {
  filename      = "lambda_function_payload.zip"
  function_name = "bedrock-rag-proxy"
  role          = aws_iam_role.lambda_exec_role.arn
  handler       = "lambda_function.handler"
  runtime       = "python3.12"
  memory_size   = 512 # MB
  timeout       = 90 # Sekunden
  source_code_hash = filebase64sha256("lambda_function_payload.zip") # Sicherstellen, dass diese Datei für terraform apply existiert
  vpc_config {
    subnet_ids = [
      aws_subnet.private_subnet_a.id,
      aws_subnet.private_subnet_b.id
    ]
    security_group_ids = [aws_security_group.lambda_sg.id]
  }
  environment {
    variables = {
      BEDROCK_REGION = "eu-central-1"
      # Bei Bedarf weitere AWS-spezifische Umgebungsvariablen hinzufügen
    }
  }
}

resource "aws_iam_role" "lambda_exec_role" {
  name = "lambda-bedrock-exec-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = {
        Service = "lambda.amazonaws.com"
      }
    }]
  })
}

# Richtlinien für Bedrock-Zugriff, VPC-Zugriff usw. anhängen
resource "aws_iam_role_policy_attachment" "lambda_bedrock_policy" {
  role       = aws_iam_role.lambda_exec_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonBedrockFullAccess" # Für die Produktion auf das Prinzip des geringsten Privilegs anpassen
}

# ... VPC, Subnetz, Sicherheitsgruppen-Definitionen für AWS Lambda ...
# Hinweis: Eine vollständige VPC-Einrichtung ist umfangreicher und erfordert eine sorgfältige Planung für die Vernetzung.
resource "aws_vpc" "main_vpc" {
  cidr_block = "10.0.0.0/16"
  instance_tenancy = "default"
  tags = {
    Name = "multi-cloud-ai-vpc"
  }
}

resource "aws_subnet" "private_subnet_a" {
  vpc_id            = aws_vpc.main_vpc.id
  cidr_block        = "10.0.1.0/24"
  availability_zone = "eu-central-1a"
  tags = {
    Name = "private-subnet-a"
  }
}

resource "aws_subnet" "private_subnet_b" {
  vpc_id            = aws_vpc.main_vpc.id
  cidr_block        = "10.0.2.0/24"
  availability_zone = "eu-central-1b"
  tags = {
    Name = "private-subnet-b"
  }
}

resource "aws_security_group" "lambda_sg" {
  name        = "lambda-bedrock-sg"
  description = "Ausgehenden Zugriff für Lambda erlauben"
  vpc_id      = aws_vpc.main_vpc.id

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

variable "gcp_project_id" {
  description = "Ihre GCP-Projekt-ID"
  type        = string
}

variable "gcp_datastore_id" {
  description = "Die ID Ihres Vertex AI Search Datastores"
  type        = string
  sensitive   = false
}

variable "aws_bedrock_kb_id" {
  description = "Die ID Ihrer Amazon Bedrock Knowledge Base"
  type        = string
  sensitive   = false
}

variable "openrouter_api_key" {
  description = "OpenRouter API-Schlüssel"
  type        = string
  sensitive   = true
}

output "cloud_run_service_url" {
  value = google_cloud_run_service.main_app_service.status[0].url
  description = "Die URL des bereitgestellten Cloud Run-Dienstes."
}

output "lambda_function_name" {
  value = aws_lambda_function.bedrock_proxy_lambda.function_name
  description = "Der Name der bereitgestellten AWS Lambda-Funktion."
}

Erwartete Ausgabe (nach terraform apply):

Apply complete! Resources: 7 added, 0 changed, 0 destroyed.
Outputs:
  cloud_run_service_url = "https://multi-cloud-ai-service-...-ew.a.run.app"
  lambda_function_name = "bedrock-rag-proxy"

Unsere Kernanwendungslogik, die auf GCP Cloud Run ausgeführt wird, muss sowohl Vertex AI Search als auch Bedrock Knowledge Bases parallel abfragen, um den bestmöglichen Kontext für unsere LLM zu synthetisieren. Dieser Python-Service-Handler fungiert als RAG-Orchestrator. Beachten Sie, dass Vertex AI Search Teil des Discovery Engine-Dienstes ist, daher verwende ich dafür das Paket google-cloud-discoveryengine.

# rag_orchestrator/main.py (Läuft auf GCP Cloud Run)

import os
import asyncio
from google.cloud import discoveryengine_v1 as discoveryengine # Für Vertex AI Search (Discovery Engine)
import boto3
from langchain_core.documents import Document
from typing import List, Dict

# Clients initialisieren (Umgebungsvariablen für die Konfiguration vorausgesetzt)
# Für Vertex AI Search würden Sie sich normalerweise über Workload Identity authentifizieren,
# wenn Sie auf Cloud Run ausgeführt werden. Das SDK übernimmt dies automatisch.

def get_vertex_ai_search_results(query: str, project_id: str, location: str, data_store_id: str) -> List[Document]:
    """Fragt Vertex AI Search (Discovery Engine) nach relevanten Dokumenten ab."""
    print(f"Vertex AI Search in {location} wird abgefragt für: {query}")
    try:
        client = discoveryengine.SearchServiceClient()

        # Den Serving-Konfigurationspfad für den Datenspeicher erstellen
        serving_config = client.serving_config_path(
            project=project_id,
            location=location, # z.B. europe-west3
            data_store=data_store_id, # Ihre Datenspeicher-ID
            serving_config="default_config", # Standard-Serving-Konfiguration
        )

        request = discoveryengine.SearchRequest(
            serving_config=serving_config,
            query=query,
            page_size=3, # Die Top 3 Ergebnisse anfordern
            query_params=discoveryengine.SearchRequest.QueryParameters(
                # Sie können hier die Abfrageerweiterung oder die Fundierung konfigurieren, wenn in Ihrem Datenspeicher aktiviert.
                # Für 'Grounding with Google Search' stellen Sie sicher, dass es in Ihrem Vertex AI Search-Datenspeicher konfiguriert ist.
                query_expansion_spec=discoveryengine.SearchRequest.QueryExpansionSpec(
                    condition=discoveryengine.SearchRequest.QueryExpansionSpec.Condition.AUTO,
                )
            )
        )

        response = client.search(request)
        results = []
        for result in response.results:
            if result.document and result.document.content:
                # Angenommen 'content' enthält den Haupttext. Anpassen basierend auf Ihrem Datenspeicher-Schema.
                results.append(Document(page_content=result.document.content, metadata={"source": result.document.id}))
        return results
    except Exception as e:
        print(f"Fehler beim Abfragen von Vertex AI Search: {e}")
        # In einem Produktionssystem robuste Fehlerbehandlung und Fallback-Logik implementieren.
        return [Document(page_content=f"Vertex AI Search (mock): Konnte für '{query}' aufgrund eines Fehlers nicht abgerufen werden: {e}")]

def get_bedrock_knowledge_base_results(query: str, kb_id: str, region: str) -> List[Document]:
    """Fragt eine Amazon Bedrock Knowledge Base nach relevanten Dokumenten ab."""
    print(f"Bedrock Knowledge Base '{kb_id}' in {region} wird abgefragt für: {query}")
    boto_session = boto3.Session(region_name=region)
    bedrock_agent_runtime = boto_session.client("bedrock-agent-runtime")

    try:
        response = bedrock_agent_runtime.retrieve(
            knowledgeBaseId=kb_id,
            retrievalQuery={
                'text': query
            },
            retrievalConfiguration={
                'vectorSearchConfiguration': {
                    'numberOfResults': 3
                }
            }
        )
        results = []
        for item in response.get('retrievalResults', []):
            content = item.get('content', {}).get('text', '')
            metadata = item.get('location', {})
            results.append(Document(page_content=content, metadata=metadata))
        return results
    except Exception as e:
        print(f"Fehler beim Abfragen der Bedrock KB: {e}")
        return [Document(page_content=f"Bedrock KB (mock): Konnte für '{query}' aufgrund eines Fehlers nicht abgerufen werden: {e}")]

async def parallel_rag_query(query: str) -> List[Document]:
    """Führt RAG-Abfragen parallel gegen GCP und AWS aus."""
    gcp_project_id = os.environ.get("GCP_PROJECT_ID", "ihr-gcp-projekt") # Sicherstellen, dass dies über env var gesetzt ist
    gcp_region = os.environ.get("GCP_REGION", "europe-west3")
    gcp_datastore_id = os.environ.get("GCP_DATASTORE_ID", "ihre-datastore-id")
    aws_kb_id = os.environ.get("AWS_BEDROCK_KB_ID", "ihre-bedrock-kb-id")
    aws_region = os.environ.get("AWS_REGION", "eu-central-1")

    gcp_results, aws_results = await asyncio.gather(
        asyncio.to_thread(get_vertex_ai_search_results, query, gcp_project_id, gcp_region, gcp_datastore_id),
        asyncio.to_thread(get_bedrock_knowledge_base_results, query, aws_kb_id, aws_region),
    )

    # Ergebnisse kombinieren und Duplikate entfernen für einen umfassenden Kontext
    all_results = gcp_results + aws_results
    return all_results

# Beispielnutzung (z.B. in einem FastAPI- oder Flask-Endpunkt)
async def handle_rag_request(query: str):
    """Simuliert die Bearbeitung einer eingehenden RAG-Anfrage."""
    context_documents = await parallel_rag_query(query)
    # Weitere Verarbeitung mit LangChain/LlamaIndex zur Prompt-Konstruktion
    # und dann an LLM über OpenRouter-Proxy senden
    return context_documents

if __name__ == "__main__":
    # Dieser Teil wäre typischerweise Teil eines Webservers oder einer Funktionsaufrufung.
    # Für lokale Tests sicherstellen, dass Dummy-Umgebungsvariablen gesetzt oder Werte übergeben werden.
    os.environ["GCP_PROJECT_ID"] = os.environ.get("GCP_PROJECT_ID", "dummy-gcp-project")
    os.environ["GCP_DATASTORE_ID"] = os.environ.get("GCP_DATASTORE_ID", "dummy-datastore-id")
    os.environ["AWS_BEDROCK_KB_ID"] = os.environ.get("AWS_BEDROCK_KB_ID", "dummy-kb-id")

    sample_query = "neueste GDPR-Änderungen für die KI-Datenverarbeitung"
    print(f"\nParallele RAG-Abfrage wird ausgeführt für: {sample_query}")
    results = asyncio.run(handle_rag_request(sample_query))
    for i, doc in enumerate(results):
        print(f"- Dokument {i+1}: {doc.page_content[:100]}...")

Erklärung: Dieser rag_orchestrator demonstriert die parallele Abfrage von zwei unterschiedlichen RAG-Quellen. asyncio.to_thread ist der Schlüssel zum effizienten Auslagern blockierender I/O-Aufrufe an boto3 und den Vertex AI Search (Discovery Engine) SDK-Client, wodurch die Haupt-Event-Schleife reaktionsfähig bleibt. Die Ergebnisse werden dann kombiniert und sind bereit für die Integration in einen endgültigen Prompt für einen LLM.

Um mehrere LLM-APIs effizient zu verwalten und Failover zwischen Gemini und Claude zu implementieren, stelle ich einen kleinen Python-Proxy-Dienst bereit. Dieser abstrahiert die Komplexität verschiedener API-Endpunkte und ermöglicht eine dynamische Modellauswahl basierend auf Kosten, Leistung oder Verfügbarkeit. Es ist ein robustes Muster zur Verbesserung der Zuverlässigkeit von LLM-Integrationen und zur Kostenverwaltung.

# openrouter_proxy/app.py (Läuft auf GCP Cloud Run, daneben oder als Teil von main_app_service)

import os
import requests
import json
from typing import Dict, Any, List

class OpenRouterProxy:
    def __init__(self, api_key: str, default_model: str = "anthropic/claude-4.6-sonnet", fallback_model: str = "google/gemini-2.5-flash"):
        self.api_key = api_key
        self.default_model = default_model
        self.fallback_model = fallback_model
        self.base_url = "https://openrouter.ai/api/v1/chat/completions"

    def _make_request(self, model: str, messages: List[Dict], stream: bool = False, **kwargs: Any) -> Dict:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "HTTP-Referer": "https://thecloudarchitect.io/", # Optional: Für OpenRouter-Analysen
            "X-Title": "Multi-Cloud AI-Dienst", # Optional: Für OpenRouter-Analysen
            "Content-Type": "application/json"
        }
        payload = {
            "model": model,
            "messages": messages,
            "stream": stream,
            **kwargs
        }
        try:
            response = requests.post(self.base_url, headers=headers, json=payload, timeout=90)
            response.raise_for_status() # Löst eine Ausnahme für HTTP-Fehler aus
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Fehler beim Aufruf von OpenRouter mit Modell {model}: {e}")
            raise

    def chat_completion(self, messages: List[Dict], preferred_model: str = None, **kwargs: Any) -> Dict:
        chosen_model = preferred_model if preferred_model else self.default_model
        try:
            print(f"Versuch der Chat-Vervollständigung mit Modell: {chosen_model}")
            return self._make_request(chosen_model, messages, **kwargs)
        except Exception as e:
            print(f"Primäres Modell {chosen_model} ist fehlgeschlagen. Rückgriff auf {self.fallback_model}. Fehler: {e}")
            if chosen_model != self.fallback_model: # Verhindert unendlichen Rückgriff, wenn Rückgriff ebenfalls fehlschlägt
                return self._make_request(self.fallback_model, messages, **kwargs)
            else:
                raise # Erneut auslösen, wenn Rückgriff ebenfalls fehlschlägt

# Beispielnutzung
if __name__ == "__main__":
    openrouter_api_key = os.environ.get("OPENROUTER_API_KEY")
    if not openrouter_api_key:
        # Für lokale Tests durch einen gültigen Schlüssel ersetzen oder Umgebungsvariable setzen.
        print("Umgebungsvariable OPENROUTER_API_KEY nicht gesetzt. Verwende einen Dummy-Schlüssel zur Veranschaulichung.")
        openrouter_api_key = "sk-dummykey123"

    proxy = OpenRouterProxy(api_key=openrouter_api_key,
                            default_model="anthropic/claude-4.6-sonnet",
                            fallback_model="google/gemini-2.5-pro")

    messages = [
        {"role": "user", "content": "Schreibe eine Python-Funktion, um einen JSON-String in ein Dictionary zu parsen und dabei potenzielle Fehler zu behandeln."}
    ]

    try:
        response = proxy.chat_completion(messages)
        print("\n--- Primäre Modellantwort ---")
        print(response["choices"][0]["message"]["content"])
    except Exception as e:
        print(f"Antwort von keinem Modell erhalten: {e}")

    # Simulieren des Ausfalls des primären Modells, um den Fallback zu testen
    print("\n--- Simulieren des Ausfalls des primären Modells und Testen des Fallbacks ---")
    # In einem realen Szenario würden Sie dies mit einem Leistungsschalter oder einer Zustandsprüfung integrieren.
    # Für dieses Beispiel tun wir so, als würden wir explizit ein nicht existierendes Modell anfordern, um den Fallback auszulösen.
    try:
        # Verwendung eines Dummy-API-Schlüssels für dieses Beispiel, was wahrscheinlich zu einem Fehler führt.
        proxy_with_bad_default = OpenRouterProxy(api_key="invalid-key",
                                                 default_model="anthropic/claude-4.6-sonnet",
                                                 fallback_model="google/gemini-2.5-pro")
        response_fallback = proxy_with_bad_default.chat_completion(messages)
        print("\n--- Fallback-Modellantwort ---")
        print(response_fallback["choices"][0]["message"]["content"])
    except Exception as e:
        print(f"Auch mit Fallback fehlgeschlagen, aufgrund eines anfänglichen API-Schlüsselproblems oder eines tatsächlichen Modellfehlers: {e}")

Bevor benutzergenerierte Prompts oder RAG-extrahierte Inhalte an externe LLM-APIs (sogar über OpenRouter) gesendet werden, stelle ich sicher, dass sensible Daten entfernt oder anonymisiert werden. Dies ist eine kritische GDPR-Anforderung zur Aufrechterhaltung einer Compliance- und Datenschutz-Festung.

# data_privacy/pii_scrubber.py

import re
import hashlib
from typing import Dict, Any, List

class PIIScrubber:
    def __init__(self, replace_with_hash: bool = False):
        self.replace_with_hash = replace_with_hash
        # Regex-Muster für gängige PII. Dies ist illustrativ; ein Produktionssystem
        # würde eine spezielle PII-Erkennungsbibliothek verwenden (z.B. Presidio, Google DLP, AWS Macie).
        self.pii_patterns = {
            "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
            "phone_number_eu": r"\b(?:\+|00)[1-9](?:[\s.-]?\d{1,}){7,14}\b", # Vereinfachtes EU-Telefonmuster
            "credit_card": r"\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14}|6(?:011|5[0-9]{2})[0-9]{12}|3[47][0-9]{13}|3(?:0[0-5]|[68][0-9])[0-9]{11}|(?:2131|1800|35\d{3})\d{11})\b",
            "iban": r"\b[A-Z]{2}[0-9]{2}(?:[ ]?[0-9]{4}){4}(?:[ ]?[0-9]{1,2})?(\b|(?![0-9A-Za-z]))", # Illustrativ, IBANs sind komplex
            "address": r"\b(\d{1,4}[ ](?:[A-Za-z0-9'-]+[ ]?){1,}[A-Za-z]{2,})\b" # Grundlegendes Adressmuster
        }

    def _hash_value(self, value: str) -> str:
        return hashlib.sha256(value.encode('utf-8')).hexdigest()

    def scrub_text(self, text: str) -> str:
        scrubbed_text = text
        for pii_type, pattern in self.pii_patterns.items():
            # re.sub mit einer Replacer-Funktion für die korrekte Ersetzung in Python verwenden
            def replacer(match):
                original_value = match.group(0)
                if self.replace_with_hash:
                    return f"[{pii_type.upper()}_HASH:{self._hash_value(original_value)[:8]}]"
                else:
                    return f"[{pii_type.upper()}_REDACTED]"
            scrubbed_text = re.sub(pattern, replacer, scrubbed_text)
        return scrubbed_text

    def scrub_messages(self, messages: List[Dict]) -> List[Dict]:
        scrubbed_messages = []
        for message in messages:
            if message["role"] == "user" and "content" in message:
                # Eine Kopie erstellen, um das Original-Nachrichten-Dict nicht direkt zu ändern, falls dies nicht beabsichtigt ist.
                scrubbed_message = message.copy()
                scrubbed_message["content"] = self.scrub_text(scrubbed_message["content"])
                scrubbed_messages.append(scrubbed_message)
            else:
                scrubbed_messages.append(message)
        return scrubbed_messages

# Beispielnutzung
if __name__ == "__main__":
    scrubber = PIIScrubber(replace_with_hash=True)
    sample_messages = [
        {"role": "user", "content": "Meine E-Mail ist mark@thecloudarchitect.io und mein Telefon ist +352 176 12345678. Die Bestellung wurde mit Karte 4111222233334444 aufgegeben."},
        {"role": "system", "content": "Hallo, wie kann ich Ihnen helfen?"}
    ]

    print("\n--- Originalnachrichten ---")
    for msg in sample_messages:
        print(msg)

    scrubbed_messages = scrubber.scrub_messages(sample_messages)
    print("\n--- Bereinigte Nachrichten ---")
    for msg in scrubbed_messages:
        print(msg)

    # Beispiel mit deaktiviertem Hashing
    scrubber_redacted = PIIScrubber(replace_with_hash=False)
    sample_text = "Meine IBAN ist LU89370400440532013000."
    scrubbed_text = scrubber_redacted.scrub_text(sample_text)
    print(f"\n--- Bereinigter Text (geschwärzt): {scrubbed_text} ---")

Erklärung: Diese PIIScrubber-Klasse verwendet reguläre Ausdrücke, um gängige PII-Typen innerhalb von Text zu erkennen und zu redigieren oder zu hashen. Obwohl dies illustrativ ist, würde ein produktionsreifes System mit Cloud-nativen DLP-Diensten (wie Google Cloud DLP oder AWS Macie) oder spezialisierten Bibliotheken für eine robustere und konfigurierbarere PII-Erkennung über verschiedene Datentypen hinweg integriert werden. Der Schlüssel ist, diese Bereinigung bevor Daten Ihre vertrauenswürdige EU-Zone verlassen, anzuwenden. Reguläre Ausdrücke sind ein Ausgangspunkt; eine umfassende Lösung erfordert kontinuierliche Verfeinerung und Tests anhand realer Daten.

Das Verfolgen von Kosten und Latenz über Clouds hinweg ist entscheidend für FinOps und Leistungsoptimierung. Ich integriere OpenTelemetry für Tracing und pushe dann Metriken sowohl an CloudWatch (AWS) als auch an Cloud Monitoring (GCP). Dies gibt uns eine einheitliche Ansicht, während die nativen Tools respektiert werden.

# observability/metrics_exporter.py

import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.trace import Status, StatusCode
# Der OpenTelemetry Python-Client für Google Cloud Monitoring ist Teil von 'opentelemetry-exporter-google-cloud'
# from opentelemetry.exporter.google_cloud import CloudMonitoringMetricsExporter
# Für AWS wird typischerweise boto3 für CloudWatch oder der OpenTelemetry-Exporter für X-Ray/CloudWatch-Metriken verwendet.
import boto3

import time
import random
from typing import Dict

# OpenTelemetry Tracer konfigurieren
resource = Resource.create({
    "service.name": "multi-cloud-ai-service",
    "service.version": "1.0.0",
    "cloud.provider": "gcp", # Kann dynamisch basierend auf dem Bereitstellungskontext festgelegt werden
    "cloud.region": os.environ.get("GCP_REGION", "europe-west3"),
})

provider = TracerProvider(resource=resource)
span_processor = BatchSpanProcessor(ConsoleSpanExporter()) # Für Konsolenausgabe während der Entwicklung
provider.add_span_processor(span_processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# Unter Annahme eines Umrechnungskurses von 1 $ \approx 0,92 € für illustrative Preisberechnungen.
USD_TO_EUR_RATE = 0.92

def record_llm_call_metrics(model_name: str, tokens_used: int, latency_ms: float, cost_usd: float):
    """Zeichnet LLM-Aufrufmetriken sowohl in GCP Cloud Monitoring als auch in AWS CloudWatch (konzeptionell) auf."""
    cost_eur = cost_usd * USD_TO_EUR_RATE

    # --- GCP Cloud Monitoring (über OpenTelemetry Exporter) ---
    # In einem realen Setup würde ich den CloudMonitoringMetricsExporter initialisieren und verwenden, um benutzerdefinierte Metriken zu pushen.
    # Damit dies funktioniert, stellen Sie sicher, dass die GCP-Anmeldeinformationen mit dem Cloud Monitoring-Schreibbereich eingerichtet sind.
    print(f"[GCP Cloud Monitoring] Metriken werden exportiert für {model_name}: Tokens={tokens_used}, Latenz={latency_ms:.2f}ms, Kosten=€{cost_eur:.4f} (${cost_usd:.4f})")

    # --- AWS CloudWatch (über Boto3) ---
    # Ich verwende boto3, um benutzerdefinierte Metriken an CloudWatch zu pushen.
    try:
        boto_session = boto3.Session(region_name=os.environ.get("AWS_REGION", "eu-central-1"))
        cloudwatch = boto_session.client("cloudwatch")
        cloudwatch.put_metric_data(
            Namespace='MultiCloudAI/LLMCalls',
            MetricData=[
                {
                    'MetricName': 'TokensUsed',
                    'Dimensions': [{'Name': 'ModelName', 'Value': model_name}],
                    'Value': tokens_used,
                    'Unit': 'Count'
                },
                {
                    'MetricName': 'Latency',
                    'Dimensions': [{'Name': 'ModelName', 'Value': model_name}],
                    'Value': latency_ms,
                    'Unit': 'Milliseconds'
                },
                {
                    'MetricName': 'CostEUR',
                    'Dimensions': [{'Name': 'ModelName', 'Value': model_name}],
                    'Value': cost_eur,
                    'Unit': 'Count' # Verwenden Sie 'Count' für Währungen, es sei denn, eine spezifische 'Currency'-Einheit wird unterstützt und ist gewünscht.
                }
            ]
        )
        print(f"[AWS CloudWatch] Metriken exportiert für {model_name}")
    except Exception as e:
        print(f"[AWS CloudWatch] Fehler beim Exportieren von Metriken: {e}")

def perform_llm_call(model: str, prompt: str) -> Dict:
    """Simuliert einen LLM-Aufruf und zeichnet dessen Leistungsmetriken auf."""
    with tracer.start_as_current_span(f"llm-call-{model}") as span:
        span.set_attribute("llm.model_name", model)
        span.set_attribute("llm.prompt_length", len(prompt))

        print(f"Führe LLM-Aufruf mit {model} für Prompt aus: {prompt[:50]}...")
        start_time = time.time()
        time.sleep(random.uniform(0.5, 2.0)) # Simuliere Netzwerklatenz und Verarbeitung
        end_time = time.time()

        latency_ms = (end_time - start_time) * 1000
        tokens = random.randint(50, 500)
        # Ungefähre Kosten in USD: Claude 3.5 Sonnet könnte ~$0.003/1K Token Input sein, Gemini 2.5 Flash ~$0.00035/1K Token
        # (Diese sind illustrativ; überprüfen Sie die aktuellen Herstellerdokumente und die Preise von OpenRouter).
        cost_per_token_usd = 0.000003 # z.B. $0.003 pro 1K Token, also $0.000003 pro Token
        cost_usd = tokens * cost_per_token_usd

        span.set_attribute("llm.tokens_used", tokens)
        span.set_attribute("llm.latency_ms", latency_ms)
        span.set_attribute("llm.cost_usd", cost_usd)
        span.set_status(Status(StatusCode.OK))

        record_llm_call_metrics(model, tokens, latency_ms, cost_usd)

        return {"model": model, "tokens": tokens, "latency_ms": latency_ms, "cost_eur": cost_usd * USD_TO_EUR_RATE, "response": "Simulierte LLM-Ausgabe"}

if __name__ == "__main__":
    print("\n--- Starte Observability-Beispiel ---")
    # Setzen Sie Dummy-Umgebungsvariablen, falls lokal ausgeführt und nicht anderweitig konfiguriert.
    os.environ["GCP_REGION"] = os.environ.get("GCP_REGION", "europe-west3")
    os.environ["AWS_REGION"] = os.environ.get("AWS_REGION", "eu-central-1")

    # Simuliere eine Reihe von LLM-Aufrufen
    perform_llm_call("anthropic/claude-3.5-sonnet", "Zusammenfassung der neuesten KI-Sicherheitsforschung")
    perform_llm_call("google/gemini-2.5-flash", "Generieren Sie einen Marketingslogan für einen Cloud-Architektur-Blog")
    print("--- Observability-Beispiel beendet ---")

Erklärung: Dieses Skript veranschaulicht, wie OpenTelemetry zur Instrumentierung von LLM-Aufrufen verwendet werden kann. Es zeigt, wie benutzerdefinierte Metriken wie tokens_used, latency_ms und cost_eur aufgezeichnet und sowohl an Google Cloud Monitoring als auch an AWS CloudWatch gesendet werden können. Dies bietet die notwendige Transparenz für FinOps-Teams und Ingenieure, um Kosten und Leistung ihres Multi-Cloud-KI-Stacks effektiv zu optimieren. Für tatsächliche Kostenmetriken richte ich typischerweise spezifische Kostenüberwachungs-Dashboards in jeder Cloud ein, die mit den Abrechnungsdaten von OpenRouter abgeglichen werden. Ich habe den Umrechnungskurs von 1 $ \approx 0,92 € für illustrative Preisberechnungen angegeben.

Nach der Bereitstellung verwende ich eine Reihe von Befehlen, um sicherzustellen, dass alles verbunden ist und wie erwartet funktioniert.

# 1. Bereitstellung und Status des GCP Cloud Run-Dienstes überprüfen
gcloud run services describe multi-cloud-ai-service --platform managed --region europe-west3 --format='value(status.url)'
# Erwartete Ausgabe:
# https://multi-cloud-ai-service-xxxxxxxx-ew.a.run.app

# 2. Den Cloud Run-Endpunkt testen (ersetzen Sie ihn durch Ihre tatsächliche URL)
# Stellen Sie sicher, dass Ihre Anwendung für dieses Beispiel einen '/rag-query'-Endpunkt bereitstellt.
curl -X POST -H "Content-Type: application/json" \n     -d '{"query": "Was ist der neueste Fortschritt bei Transformatorenmodellen?"}' \n     "$(gcloud run services describe multi-cloud-ai-service --platform managed --region europe-west3 --format='value(status.url)')/rag-query"
# Erwartete Ausgabe (vereinfacht, die tatsächliche Antwort hängt von Ihrer Anwendungslogik ab):
# {"context_documents": [...], "llm_response": "..."}

# 3. Vorhandensein der AWS Lambda-Funktion überprüfen
aws lambda get-function --function-name bedrock-rag-proxy --region eu-central-1
# Erwartete Ausgabe (teilweise):
# {
#     "Configuration": {
#         "FunctionName": "bedrock-rag-proxy",
#         "Runtime": "python3.12",
#         "Handler": "lambda_function.handler",
#         ...
#     }
# }

# 4. OpenRouter API-Konnektivität prüfen (Beispiel mit curl)
# Ersetzen Sie $OPENROUTER_API_KEY durch Ihren tatsächlichen API-Schlüssel.
curl -X POST https://openrouter.ai/api/v1/chat/completions \n  -H "Authorization: Bearer $OPENROUTER_API_KEY" \n  -H "Content-Type: application/json" \n  -d '{ "model": "anthropic/claude-3.5-sonnet", "messages": [{"role": "user", "content": "Hallo"}] }'
# Erwartete Ausgabe (teilweise):
# {"choices": [{"message": {"content": "Hallo! Wie kann ich Ihnen heute helfen?"}, ...}]}

1. Fehler: Cross-Cloud IAM-Berechtigung verweigert

# Fehlermeldungsbeispiel (aus GCP-Protokollen beim Aufruf von AWS Bedrock)
google.auth.exceptions.RefreshError: ('Failed to retrieve access token: {"error":"invalid_grant", "error_description":"Invalid AWS credential for Workload Identity Federation"}', '...')

**Lösung:** Dies bedeutet typischerweise, dass Ihre Workload Identity Federation-Einrichtung falsch ist. Überprüfen Sie die IAM-Richtlinie Ihres GCP-Dienstkontos, die Vertrauensrichtlinie der AWS-IAM-Rolle (insbesondere den `Federated`-Prinzipal und die `StringEquals`-Bedingungen für `google.subject` und `google.aud`) und die an die AWS-Rolle angehängte Richtlinie, um sicherzustellen, dass sie die entsprechenden `bedrock`-Berechtigungen gewährt. Stellen Sie außerdem sicher, dass das GCP-Dienstkonto korrekt mit dem Cloud Run-Dienstkonto verknüpft ist, wie in der [Google Cloud Workload Identity Federation-Dokumentation](https://cloud.google.com/iam/docs/manage-workload-identity-federation) beschrieben.

1. Fehler: PII-Leckage erkannt

# Dieser Fehler ist möglicherweise kein Systemfehler, sondern eine Warnung von einem DLP-System (z.B. Cloud DLP, AWS Macie)
# oder ein Ergebnis einer Sicherheitsprüfung.

**Lösung:** Wenn PII stromabwärts Ihrer Bereinigungsschicht erkannt werden, überprüfen Sie die Muster Ihres `PIIScrubber` und stellen Sie sicher, dass sie für Ihre Daten ausreichend umfassend sind. Erwägen Sie die Integration eines robusteren, Cloud-nativen DLP-Dienstes. Denken Sie daran, dass die Regex-basierte Bereinigung niemals 100% narrensicher ist; dedizierte DLP-Dienste bieten eine höhere Genauigkeit aufgrund ihrer kontextsensitiven Erkennungsfähigkeiten. Validieren Sie Ihren Bereinigungsprozess durch regelmäßige Datenprüfungen und Penetrationstests.

• Hybrid RAG ist entscheidend für Leistung und Compliance: Kombinieren Sie GCP Vertex AI Search mit Amazon Bedrock Knowledge Bases, um deren jeweilige Stärken zu nutzen und die Anforderungen an die Datenresidenz zu erfüllen.
• OpenRouter bietet kritische LLM-Abstraktion: Verwenden Sie OpenRouter für den Failover zwischen Modellen wie Claude Sonnet und Gemini 2.5 Flash/Pro, um Kosten und Verfügbarkeit ohne Änderungen am Anwendungscode zu optimieren.
• GDPR-Compliance erfordert PII-Bereinigung und EU-Residenz: Implementieren Sie eine robuste PII-Anonymisierung, bevor Sie Daten an externe APIs senden, und stellen Sie alle RAG-Datenquellen ausschließlich in EU-zentralen Cloud-Regionen bereit.
• Workload Identity Federation ist entscheidend für den sicheren Cross-Cloud-Zugriff: Ermöglichen Sie GCP-Diensten, sicher mit AWS-Ressourcen zu interagieren, ohne langlebige Zugriffsschlüssel verwalten zu müssen.
• Umfassende Beobachtbarkeit ist nicht verhandelbar: Instrumentieren Sie Ihre Multi-Cloud-KI-Systeme mit OpenTelemetry, um Kosten und Latenz über AWS CloudWatch und GCP Cloud Monitoring hinweg zu verfolgen.