Cette approche RAG hybride fusionne le meilleur des capacités d'indexation et de base de GCP avec l'intégration robuste du lac de données d'AWS. J'utilise efficacement Amazon S3 comme notre principal magasin de données pour les documents bruts, qui sont ensuite traités et indexés dans une base de connaissances Amazon Bedrock. Simultanément, un pipeline parallèle ingère les données pertinentes dans Vertex AI Search.

Le « pont » est critique : assurer que les incorporations vectorielles et les métadonnées sont harmonisées, souvent via une base de données vectorielle partagée et agnostique au cloud ou un mécanisme de synchronisation sophistiqué. Cela permet à mon orchestrateur RAG d'interroger les deux sources et de synthétiser un contexte complet pour le fondement de la LLM.

Pour la génération de code, en particulier pour les flux de travail techniques complexes, Claude 4.6 Sonnet d'Anthropic reste une référence. Mais au lieu d'appels API directs, je route les requêtes via OpenRouter. Cela fournit une couche d'abstraction cruciale, permettant le basculement de modèle, l'optimisation des coûts et la gestion simplifiée des API. Cela signifie que si Claude 4.6 Sonnet est lent ou devient trop cher, je peux passer de manière transparente à Gemini 2.5 Pro (via OpenRouter) sans modifier mon code d'application. LangChain et LlamaIndex construisent ensuite l'orchestration agentique par-dessus, permettant une compréhension contextuelle des bases de code et une utilisation dynamique des outils.

Lors de l'architecture du composant « base de données vectorielle partagée ou service de synchronisation », j'évalue soigneusement les options de base de données vectorielles en fonction des exigences de latence, des ensembles de fonctionnalités et des frais d'exploitation. Pour la souveraineté des données européennes, cela implique souvent de sélectionner des fournisseurs dotés d'une infrastructure basée dans l'UE ou de l'auto-hébergement.

• Pinecone : Un service de base de données vectorielle entièrement géré, connu pour son évolutivité et ses performances. Il propose des régions en Europe, ce qui en fait un solide candidat pour les solutions gérées si des régions spécifiques de l'UE sont disponibles pour la résidence des données. J'apprécie sa facilité d'utilisation et la cohérence de son API.
• Weaviate : Il peut être exécuté en tant que service géré (Weaviate Cloud) ou auto-hébergé sur Kubernetes. Ses options de déploiement flexibles le rendent attrayant pour répondre aux exigences strictes de résidence des données, car je peux le déployer dans des VPC ou des clusters GKE/EKS spécifiques de l'UE. Il fournit également une API robuste et un écosystème de modules.
• AlloyDB pour PostgreSQL avec pgvector : Pour ceux qui sont déjà fortement investis dans PostgreSQL ou GCP, l'utilisation d'AlloyDB avec l'extension pgvector offre une solution puissante et intégrée. Bien que pgvector ne corresponde pas aux performances brutes des bases de données vectorielles spécialisées pour des scénarios à très grande échelle, il offre une excellente localité des données et une gestion simplifiée dans un environnement de base de données relationnelle familier. L'exécution d'AlloyDB dans europe-west3 garantit la résidence des données.

Mon choix dépend généralement de la latence requise pour le RAG, des fonctionnalités spécifiques (par exemple, filtrage, recherche hybride) nécessaires et des préférences opérationnelles pour les solutions gérées par rapport aux solutions auto-hébergées au sein de l'UE.

La conformité n'est pas une réflexion après coup ; elle est intégrée à la conception. Pour le RGPD, la résidence des données est primordiale. Toutes les sources RAG doivent résider dans les régions centrales de l'UE (par exemple, eu-central-1 pour AWS, europe-west3 pour GCP). Cela signifie que les compartiments S3, les bases de connaissances Bedrock et les magasins de données Vertex AI Search sont provisionnés exclusivement dans ces régions. Au-delà de la résidence, l'anonymisation des PII est cruciale. Avant que les invites n'atteignent les API externes comme OpenRouter, une couche de nettoyage des PII garantit que les données sensibles ne quittent jamais notre environnement contrôlé. Cela implique souvent un traitement côté client ou un service proxy dédié au sein de nos périmètres contrôlés. À ce jour, Gemini 3.1 n'est disponible que via un point de terminaison global, et non régional, d'où notre utilisation de la version 2.5.

Dans une architecture axée sur l'IA, la gouvernance des modèles ne concerne pas uniquement le versionnement. Il s'agit de s'assurer que chaque modèle, des générateurs d'incorporations aux LLM de génération de code, respecte des normes de sécurité strictes. J'implémente :

• Registre des modèles : Un catalogue central pour tous les modèles, y compris leur source, leur version et la provenance des données d'entraînement.
• Analyse des vulnérabilités : Les modèles d'incorporation et les LLM personnalisés affinés sont analysés pour détecter les vulnérabilités connues et s'assurer de leur conformité aux bases de référence de sécurité.
• Journalisation d'audit : Chaque interaction avec une API LLM, en particulier celles acheminées via OpenRouter, est journalisée avec les métadonnées pertinentes (ID de requête, nombre de jetons, horodatages) pour la conformité et l'analyse des coûts. Ceci est crucial pour le RGPD, car il fournit une piste d'audit des activités de traitement des données.
• Contrôle d'accès : Des politiques IAM granulaires restreignent qui peut déployer, mettre à jour ou même invoquer des modèles spécifiques, en s'intégrant à Workload Identity Federation pour les scénarios inter-cloud.

J'utilise Terraform pour définir et gérer les services de calcul de base dans GCP et AWS qui hébergent nos composants RAG et de génération de code. Cela garantit la cohérence et l'auditabilité, et surtout, place les ressources dans les régions de l'UE pour la conformité au RGPD.

# main.tf pour le calcul inter-cloud

# Configurer le fournisseur Google Cloud pour l'Europe
provider "google" {
  project = var.gcp_project_id
  region  = "europe-west3" # Francfort, Allemagne
}

# Configurer le fournisseur AWS pour l'Europe
provider "aws" {
  region = "eu-central-1" # Francfort, Allemagne
}

# --- GCP Cloud Run pour le service d'application ---
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 # ID du magasin de données Vertex AI Search
        }
        env {
          name = "AWS_BEDROCK_KB_ID"
          value = var.aws_bedrock_kb_id # ID de la base de connaissances Bedrock
        }
      }
      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 = "Compte de service pour le service Cloud Run multi-cloud IA"
}

# --- AWS Lambda pour un proxy Bedrock spécifique ou des tâches asynchrones ---
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 # Mo
  timeout       = 90 # secondes
  source_code_hash = filebase64sha256("lambda_function_payload.zip") # Assurez-vous que ce fichier existe pour terraform apply
  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"
      # Ajouter d'autres variables d'environnement spécifiques à AWS si nécessaire
    }
  }
}

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

# Attacher les politiques pour l'accès Bedrock, l'accès VPC, etc.
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" # Ajuster au moindre privilège pour la production
}

# ... Définitions VPC, sous-réseau, groupe de sécurité pour AWS Lambda ...
# Note : Une configuration VPC complète est plus étendue et nécessite une planification minutieuse du réseau.
resource "aws_vpc" "main_vpc" {
  cidr_block = "10.0.0.0/16"
  instance_tenancy = "default"
  tags = {
    Name = "vpc-ia-multi-cloud"
  }
}

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 = "sous-réseau-privé-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 = "sous-réseau-privé-b"
  }
}

resource "aws_security_group" "lambda_sg" {
  name        = "lambda-bedrock-sg"
  description = "Autoriser l'accès sortant pour Lambda"
  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 = "Votre ID de projet GCP"
  type        = string
}

variable "gcp_datastore_id" {
  description = "L'ID de votre magasin de données Vertex AI Search"
  type        = string
  sensitive   = false
}

variable "aws_bedrock_kb_id" {
  description = "L'ID de votre base de connaissances Amazon Bedrock"
  type        = string
  sensitive   = false
}

variable "openrouter_api_key" {
  description = "Clé API OpenRouter"
  type        = string
  sensitive   = true
}

output "cloud_run_service_url" {
  value = google_cloud_run_service.main_app_service.status[0].url
  description = "L'URL du service Cloud Run déployé."
}

output "lambda_function_name" {
  value = aws_lambda_function.bedrock_proxy_lambda.function_name
  description = "Le nom de la fonction AWS Lambda déployée."
}

Résultat attendu (après 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"

Notre logique d'application principale, exécutée sur GCP Cloud Run, doit interroger simultanément Vertex AI Search et les bases de connaissances Bedrock pour synthétiser le meilleur contexte possible pour notre LLM. Ce gestionnaire de services Python agit comme l'orchestrateur RAG. Notez que Vertex AI Search fait partie du service Discovery Engine, j'utilise donc le package google-cloud-discoveryengine pour cela.

# rag_orchestrator/main.py (Exécution sur GCP Cloud Run)

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

# Initialiser les clients (en supposant des variables d'environnement pour la configuration)
# Pour Vertex AI Search, vous vous authentifieriez généralement via Workload Identity
# lors de l'exécution sur Cloud Run. Le SDK gère cela automatiquement.

def get_vertex_ai_search_results(query: str, project_id: str, location: str, data_store_id: str) -> List[Document]:
    """Interroge Vertex AI Search (Discovery Engine) pour les documents pertinents."""
    print(f"Interrogation de Vertex AI Search dans {location} pour : {query}")
    try:
        client = discoveryengine.SearchServiceClient()

        # Construire le chemin de configuration de service pour le magasin de données
        serving_config = client.serving_config_path(
            project=project_id,
            location=location, # ex. europe-west3
            data_store=data_store_id, # Votre ID de magasin de données
            serving_config="default_config", # Configuration de service par défaut
        )

        request = discoveryengine.SearchRequest(
            serving_config=serving_config,
            query=query,
            page_size=3, # Demander les 3 meilleurs résultats
            query_params=discoveryengine.SearchRequest.QueryParameters(
                # Vous pouvez configurer l'expansion de requête ou le fondement ici si activé dans votre magasin de données.
                # Pour 'Grounding with Google Search', assurez-vous qu'il est configuré dans votre magasin de données Vertex AI Search.
                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:
                # En supposant que 'content' contient le texte principal. Ajuster en fonction du schéma de votre magasin de données.
                results.append(Document(page_content=result.document.content, metadata={"source": result.document.id}))
        return results
    except Exception as e:
        print(f"Erreur lors de l'interrogation de Vertex AI Search : {e}")
        # Dans un système de production, implémentez une gestion robuste des erreurs et une logique de repli.
        return [Document(page_content=f"Vertex AI Search (simulé) : Impossible de récupérer pour '{query}' en raison d'une erreur : {e}")]

def get_bedrock_knowledge_base_results(query: str, kb_id: str, region: str) -> List[Document]:
    """Interroge une base de connaissances Amazon Bedrock pour les documents pertinents."""
    print(f"Interrogation de la base de connaissances Bedrock '{kb_id}' dans {region} pour : {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"Erreur lors de l'interrogation de la base de connaissances Bedrock : {e}")
        return [Document(page_content=f"Base de connaissances Bedrock (simulée) : Impossible de récupérer pour '{query}' en raison d'une erreur : {e}")]

async def parallel_rag_query(query: str) -> List[Document]:
    """Exécute des requêtes RAG contre GCP et AWS en parallèle."""
    gcp_project_id = os.environ.get("GCP_PROJECT_ID", "votre-projet-gcp") # Assurez-vous que cela est défini via une variable d'environnement
    gcp_region = os.environ.get("GCP_REGION", "europe-west3")
    gcp_datastore_id = os.environ.get("GCP_DATASTORE_ID", "votre-id-de-magasin-de-données")
    aws_kb_id = os.environ.get("AWS_BEDROCK_KB_ID", "votre-id-de-base-de-connaissances-bedrock")
    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),
    )

    # Combiner et dédupliquer les résultats pour un contexte complet
    all_results = gcp_results + aws_results
    return all_results

# Exemple d'utilisation (par exemple, dans un point de terminaison FastAPI ou Flask)
async def handle_rag_request(query: str):
    """Simule le traitement d'une requête RAG entrante."""
    context_documents = await parallel_rag_query(query)
    # Traiter davantage avec LangChain/LlamaIndex pour la construction de l'invite
    # puis envoyer au LLM via le proxy OpenRouter
    return context_documents

if __name__ == "__main__":
    # Cette partie ferait généralement partie d'un serveur web ou d'une invocation de fonction.
    # Pour les tests locaux, assurez-vous que des variables d'environnement factices sont définies ou que des valeurs sont transmises.
    os.environ["GCP_PROJECT_ID"] = os.environ.get("GCP_PROJECT_ID", "projet-gcp-factice")
    os.environ["GCP_DATASTORE_ID"] = os.environ.get("GCP_DATASTORE_ID", "id-magasin-de-données-factice")
    os.environ["AWS_BEDROCK_KB_ID"] = os.environ.get("AWS_BEDROCK_KB_ID", "id-kb-bedrock-factice")

    sample_query = "derniers changements du RGPD pour le traitement des données IA"
    print(f"\nExécution de la requête RAG parallèle pour : {sample_query}")
    results = asyncio.run(handle_rag_request(sample_query))
    for i, doc in enumerate(results):
        print(f"- Document {i+1}: {doc.page_content[:100]}...")

Explication : Cet rag_orchestrator démontre l'interrogation parallèle de deux sources RAG distinctes. asyncio.to_thread est essentiel pour décharger efficacement les appels d'E/S bloquants vers boto3 et le client SDK Vertex AI Search (Discovery Engine), permettant à la boucle d'événements principale de rester réactive. Les résultats sont ensuite combinés, prêts à être intégrés dans une invite finale pour un LLM.

Pour gérer efficacement plusieurs API LLM et implémenter le basculement entre Gemini et Claude, je déploie un petit service proxy Python. Cela abstraie la complexité des différents points de terminaison API et permet une sélection dynamique de modèles basée sur le coût, les performances ou la disponibilité. C'est un modèle robuste pour améliorer la fiabilité des intégrations LLM et gérer les coûts.

# openrouter_proxy/app.py (Exécution sur GCP Cloud Run, à côté ou dans le cadre de 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/", # Facultatif : Pour les analyses OpenRouter
            "X-Title": "Service IA multi-cloud", # Facultatif : Pour les analyses OpenRouter
            "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ève une exception pour les erreurs HTTP
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Erreur lors de l'appel d'OpenRouter avec le modèle {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"Tentative de complétion de chat avec le modèle : {chosen_model}")
            return self._make_request(chosen_model, messages, **kwargs)
        except Exception as e:
            print(f"Le modèle principal {chosen_model} a échoué. Repli sur {self.fallback_model}. Erreur : {e}")
            if chosen_model != self.fallback_model: # Empêche le repli infini si le repli échoue également
                return self._make_request(self.fallback_model, messages, **kwargs)
            else:
                raise # Relance si le repli échoue également

# Exemple d'utilisation
if __name__ == "__main__":
    openrouter_api_key = os.environ.get("OPENROUTER_API_KEY")
    if not openrouter_api_key:
        # Pour les tests locaux, remplacez par une clé valide ou définissez la variable d'environnement.
        print("La variable d'environnement OPENROUTER_API_KEY n'est pas définie. Utilisation d'une clé factice à des fins d'illustration.")
        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": "Écrivez une fonction Python pour analyser une chaîne JSON en dictionnaire, en gérant les erreurs potentielles."}
    ]

    try:
        response = proxy.chat_completion(messages)
        print("\n--- Réponse du modèle principal ---")
        print(response["choices"][0]["message"]["content"])
    except Exception as e:
        print(f"Échec de l'obtention de la réponse d'un modèle quelconque : {e}")

    # Simuler l'échec du modèle principal pour tester le repli
    print("\n--- Simulation de l'échec du modèle principal et test du repli ---")
    # Dans un scénario réel, vous intégreriez cela avec un disjoncteur ou un contrôle de santé.
    # Pour cet exemple, faisons semblant de demander explicitement un modèle inexistant pour déclencher le repli.
    try:
        # Utilisation d'une clé API factice pour cet exemple, ce qui entraînera probablement un échec.
        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--- Réponse du modèle de repli ---")
        print(response_fallback["choices"][0]["message"]["content"])
    except Exception as e:
        print(f"Échec même avec le repli, en raison d d'un problème initial de clé API ou d'un échec réel du modèle : {e}")

Avant d'envoyer des invites générées par l'utilisateur ou du contenu extrait par RAG à des API LLM externes (même via OpenRouter), je m'assure que les données sensibles sont supprimées ou anonymisées. Il s'agit d'une exigence RGPD essentielle pour maintenir une forteresse de conformité et de confidentialité.

# 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
        # Modèles regex pour les PII courants. Ceci est illustratif ; un système de production
        # utiliserait une bibliothèque dédiée de détection des PII (par exemple, 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", # Modèle de téléphone EU simplifié
            "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]))", # Illustratif, les IBAN sont complexes
            "address": r"\b(\d{1,4}[ ](?:[A-Za-z0-9'-]+[ ]?){1,}[A-Za-z]{2,})\b" # Modèle d'adresse de base
        }

    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():
            # Utilisation de re.sub avec une fonction de remplacement pour une substitution correcte en Python
            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:
                # Créer une copie pour éviter de modifier le dictionnaire de message original si ce n'est pas l'intention.
                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

# Exemple d'utilisation
if __name__ == "__main__":
    scrubber = PIIScrubber(replace_with_hash=True)
    sample_messages = [
        {"role": "user", "content": "Mon email est mark@thecloudarchitect.io et mon téléphone est +352 176 12345678. La commande a été passée avec la carte 4111222233334444."},
        {"role": "system", "content": "Bonjour, comment puis-je vous aider ?"}
    ]

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

    scrubbed_messages = scrubber.scrub_messages(sample_messages)
    print("\n--- Messages nettoyés ---")
    for msg in scrubbed_messages:
        print(msg)

    # Exemple avec le hachage désactivé
    scrubber_redacted = PIIScrubber(replace_with_hash=False)
    sample_text = "Mon IBAN est LU89370400440532013000."
    scrubbed_text = scrubber_redacted.scrub_text(sample_text)
    print(f"\n--- Texte nettoyé (redacté) : {scrubbed_text} ---")

Explication : Cette classe PIIScrubber utilise des expressions régulières pour détecter et masquer ou hacher les types de PII courants dans le texte. Bien qu'illustratif, un système de qualité production s'intégrerait à des services DLP cloud natifs (comme Google Cloud DLP ou AWS Macie) ou à des bibliothèques spécialisées pour une détection de PII plus robuste et configurable sur divers types de données. L'essentiel est d'appliquer ce nettoyage avant que les données ne quittent votre zone UE de confiance. Les expressions régulières sont un point de départ ; une solution complète nécessite un raffinement et des tests continus par rapport aux données réelles.

Le suivi des coûts et de la latence entre les clouds est essentiel pour le FinOps et l'optimisation des performances. J'intègre OpenTelemetry pour le traçage, puis j'envoie les métriques à CloudWatch (AWS) et Cloud Monitoring (GCP). Cela nous donne une vue unifiée tout en respectant les outils natifs.

# 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
# Le client Python OpenTelemetry pour Google Cloud Monitoring fait partie de 'opentelemetry-exporter-google-cloud'
# from opentelemetry.exporter.google_cloud import CloudMonitoringMetricsExporter
# Pour AWS, utilisez généralement boto3 pour CloudWatch ou l'exportateur OpenTelemetry pour les métriques X-Ray/CloudWatch.
import boto3

import time
import random
from typing import Dict

# Configurer le traceur OpenTelemetry
resource = Resource.create({
    "service.name": "service-ia-multi-cloud",
    "service.version": "1.0.0",
    "cloud.provider": "gcp", # Peut être défini dynamiquement en fonction du contexte de déploiement
    "cloud.region": os.environ.get("GCP_REGION", "europe-west3"),
})

provider = TracerProvider(resource=resource)
span_processor = BatchSpanProcessor(ConsoleSpanExporter()) # Pour la sortie console pendant le développement
provider.add_span_processor(span_processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# En supposant un taux de conversion de 1 $ \approx 0,92 € pour les calculs de prix illustratifs.
USD_TO_EUR_RATE = 0.92

def record_llm_call_metrics(model_name: str, tokens_used: int, latency_ms: float, cost_usd: float):
    """Enregistre les métriques d'appel LLM pour GCP Cloud Monitoring et AWS CloudWatch (conceptuellement)."""
    cost_eur = cost_usd * USD_TO_EUR_RATE

    # --- GCP Cloud Monitoring (via l'exportateur OpenTelemetry) ---
    # Dans une configuration réelle, j'initialiserais CloudMonitoringMetricsExporter et l'utiliserais pour envoyer des métriques personnalisées.
    # Pour que cela fonctionne, assurez-vous que les informations d'identification GCP sont configurées avec l'étendue d'écriture de Cloud Monitoring.
    print(f"[GCP Cloud Monitoring] Exportation des métriques pour {model_name} : jetons={tokens_used}, latence={latency_ms:.2f}ms, coût=€{cost_eur:.4f} (${cost_usd:.4f})")

    # --- AWS CloudWatch (via Boto3) ---
    # J'utilise boto3 pour envoyer des métriques personnalisées à CloudWatch.
    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' # Utilisez 'Count' pour la devise à moins qu'une unité de 'Currency' spécifique ne soit prise en charge et souhaitée.
                }
            ]
        )
        print(f"[AWS CloudWatch] Métriques exportées pour {model_name}")
    except Exception as e:
        print(f"[AWS CloudWatch] Erreur lors de l'exportation des métriques : {e}")

def perform_llm_call(model: str, prompt: str) -> Dict:
    """Simule un appel LLM et enregistre ses métriques de performance."""
    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"Exécution d'un appel LLM avec {model} pour l'invite : {prompt[:50]}...")
        start_time = time.time()
        time.sleep(random.uniform(0.5, 2.0)) # Simuler la latence réseau et le traitement
        end_time = time.time()

        latency_ms = (end_time - start_time) * 1000
        tokens = random.randint(50, 500)
        # Coût approximatif en USD : Claude 3.5 Sonnet pourrait être d'environ 0,003 $/1K jetons d'entrée, Gemini 2.5 Flash d'environ 0,00035 $/1K jetons.
        # (Ceux-ci sont illustratifs ; vérifiez par rapport à la documentation actuelle du fournisseur et aux tarifs d'OpenRouter).
        cost_per_token_usd = 0.000003 # par exemple, 0,003 $ par 1K jetons, donc 0,000003 $ par jeton
        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": "Sortie LLM simulée"}

if __name__ == "__main__":
    print("\n--- Démarrage de l'exemple d'observabilité ---")
    # Définir des variables d'environnement factices si exécution locale et non configurées ailleurs.
    os.environ["GCP_REGION"] = os.environ.get("GCP_REGION", "europe-west3")
    os.environ["AWS_REGION"] = os.environ.get("AWS_REGION", "eu-central-1")

    # Simuler une série d'appels LLM
    perform_llm_call("anthropic/claude-3.5-sonnet", "Résumer les recherches récentes sur la sécurité de l'IA")
    perform_llm_call("google/gemini-2.5-flash", "Générer un slogan marketing pour un blog sur l'architecture cloud")
    print("--- Exemple d'observabilité terminé ---")

Explication : Ce script illustre comment OpenTelemetry peut être utilisé pour instrumenter les appels LLM. Il montre comment enregistrer des métriques personnalisées comme tokens_used, latency_ms et cost_eur et les envoyer à Google Cloud Monitoring et AWS CloudWatch. Cela offre la visibilité nécessaire aux équipes FinOps et aux ingénieurs pour optimiser les coûts et les performances de leur pile d'IA multi-cloud. Pour les métriques de coût réelles, je configure généralement des tableaux de bord de surveillance des coûts spécifiques dans chaque cloud, recoupés avec les données de facturation d'OpenRouter. J'ai noté le taux de conversion de 1 $ \approx 0,92 € pour les calculs de prix illustratifs.

Après le déploiement, j'utilise une série de commandes pour m'assurer que tout est connecté et fonctionne comme prévu.

# 1. Vérifier le déploiement et l'état du service GCP Cloud Run
gcloud run services describe multi-cloud-ai-service --platform managed --region europe-west3 --format='value(status.url)'
# Résultat attendu :
# https://multi-cloud-ai-service-xxxxxxxx-ew.a.run.app

# 2. Tester le point de terminaison Cloud Run (remplacez par votre URL réelle)
# Assurez-vous que votre application expose un point de terminaison '/rag-query' pour cet exemple.
curl -X POST -H "Content-Type: application/json" \n     -d '{"query": "Quelle est la dernière avancée dans les modèles de transformeurs ?"}' \n     "$(gcloud run services describe multi-cloud-ai-service --platform managed --region europe-west3 --format='value(status.url)')/rag-query"
# Résultat attendu (simplifié, la réponse réelle dépendra de la logique de votre application) :
# {"context_documents": [...], "llm_response": "..."}

# 3. Vérifier l'existence de la fonction AWS Lambda
aws lambda get-function --function-name bedrock-rag-proxy --region eu-central-1
# Résultat attendu (partiel) :
# {
#     "Configuration": {
#         "FunctionName": "bedrock-rag-proxy",
#         "Runtime": "python3.12",
#         "Handler": "lambda_function.handler",
#         ...
#     }
# }

# 4. Vérifier la connectivité de l'API OpenRouter (exemple avec curl)
# Remplacez $OPENROUTER_API_KEY par votre clé API réelle.
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": "Bonjour"}] }'
# Résultat attendu (partiel) :
# {"choices": [{"message": {"content": "Bonjour ! Comment puis-je vous aider aujourd'hui ?"}, ...}]}

1. Erreur : Autorisation IAM inter-cloud refusée

# Exemple de message d'erreur (provenant des journaux GCP lors de l'appel d'AWS Bedrock)
google.auth.exceptions.RefreshError: ('Failed to retrieve access token: {"error":"invalid_grant", "error_description":"Invalid AWS credential for Workload Identity Federation"}', '...')

**Solution :** Cela signifie généralement que votre configuration Workload Identity Federation est incorrecte. Vérifiez la politique IAM de votre compte de service GCP, la politique de confiance du rôle IAM AWS (en particulier le principal `Federated` et les conditions `StringEquals` pour `google.subject` et `google.aud`), et la politique attachée au rôle AWS pour vous assurer qu'elle accorde les autorisations `bedrock` appropriées. Assurez-vous également que le compte de service GCP est correctement lié au compte de service Cloud Run, comme spécifié dans la [documentation de Google Cloud sur Workload Identity Federation](https://cloud.google.com/iam/docs/manage-workload-identity-federation).

1. Erreur : Fuite de PII détectée

# Cette erreur pourrait ne pas être une erreur système, mais une alerte d'un système DLP (par exemple, Cloud DLP, AWS Macie)
# ou une constatation d'audit de sécurité.

**Solution :** Si des PII sont détectées en aval de votre couche de nettoyage, examinez les modèles de votre `PIIScrubber` et assurez-vous qu'ils sont suffisamment complets pour vos données. Envisagez d'intégrer un service DLP plus robuste et natif du cloud. N'oubliez pas que le nettoyage basé sur des expressions régulières n'est jamais fiable à 100 % ; les services DLP dédiés offrent une précision plus élevée grâce à leurs capacités de détection contextuelles. Validez votre processus de nettoyage avec des audits de données réguliers et des tests d'intrusion.

• Le RAG hybride est crucial pour la performance et la conformité : Combinez Vertex AI Search de GCP avec les bases de connaissances Amazon Bedrock pour tirer parti de leurs forces respectives et répondre aux exigences de résidence des données.
• OpenRouter fournit une abstraction LLM critique : Utilisez OpenRouter pour le basculement entre des modèles comme Claude Sonnet et Gemini 2.5 Flash/Pro, optimisant les coûts et la disponibilité sans modifications du code d'application.
• La conformité au RGPD nécessite un nettoyage des PII et la résidence dans l'UE : Implémentez une anonymisation robuste des PII avant d'envoyer des données à des API externes et provisionnez toutes les sources de données RAG exclusivement dans les régions centrales de l'UE.
• Workload Identity Federation est essentiel pour un accès sécurisé entre les clouds : Permettez aux services GCP d'interagir en toute sécurité avec les ressources AWS sans gérer de clés d'accès à longue durée de vie.
• L'observabilité complète est non négociable : Instrumentez vos systèmes d'IA multi-cloud avec OpenTelemetry pour suivre les coûts et la latence sur AWS CloudWatch et GCP Cloud Monitoring.