This hybrid RAG approach merges the best of GCP's indexing and grounding capabilities with AWS's robust data lake integration. I effectively use Amazon S3 as our primary data store for raw documents, which are then processed and indexed into an Amazon Bedrock Knowledge Base. Simultaneously, a parallel pipeline ingests relevant data into Vertex AI Search.

The 'bridge' is critical: ensuring vector embeddings and metadata are harmonized, often through a shared, cloud-agnostic vector database or a sophisticated synchronization mechanism. This allows my RAG orchestrator to query both sources and synthesize a comprehensive context for LLM grounding.

For code generation, especially for complex technical workflows, Anthropic's Claude 4.6 Sonnet remains a benchmark. But instead of direct API calls, I route requests through OpenRouter. This provides a crucial abstraction layer, enabling model failover, cost optimization, and simplified API management. It means if Claude 4.6 Sonnet is performing slowly or becomes too expensive, I can seamlessly switch to Gemini 2.5 Pro (via OpenRouter) without changing my application code. LangChain and LlamaIndex then build the agentic orchestration on top of this, allowing for contextual understanding of codebases and dynamic tool use.

When architecting the "shared vector DB or sync service" component, I carefully evaluate vector database options based on latency requirements, feature sets, and operational overhead. For European data sovereignty, this often means selecting providers with EU-based infrastructure or self-hosting.

• Pinecone: A fully managed vector database service known for its scalability and performance. It offers regions in Europe, making it a strong contender for managed solutions if specific EU regions are available for data residency. I value its ease of use and API consistency.
• Weaviate: This can be run as a managed service (Weaviate Cloud) or self-hosted on Kubernetes. Its flexible deployment options make it attractive for meeting strict data residency requirements, as I can deploy it in specific EU VPCs or GKE/EKS clusters. It also provides a robust API and module ecosystem.
• AlloyDB for PostgreSQL with pgvector: For those already heavily invested in PostgreSQL or GCP, leveraging AlloyDB with the pgvector extension provides a powerful and integrated solution. While pgvector might not match the raw performance of specialized vector databases for extremely high-scale scenarios, it offers excellent data locality and simplified management within a familiar relational database environment. Running AlloyDB in europe-west3 ensures data residency.

My choice typically comes down to the required latency for RAG, the specific features (e.g., filtering, hybrid search) needed, and the operational preferences for managed versus self-hosted solutions within the EU.

Compliance is not an afterthought; it's baked into the design. For GDPR, data residency is paramount. All RAG sources must reside in EU-central regions (e.g., eu-central-1 for AWS, europe-west3 for GCP). This means S3 buckets, Bedrock Knowledge Bases, and Vertex AI Search data stores are provisioned exclusively in these regions. Beyond residency, PII anonymization is crucial. Before prompts hit external APIs like OpenRouter, a PII scrubbing layer ensures sensitive data never leaves our controlled environment. This often involves client-side processing or a dedicated proxy service within our controlled perimeters. As of today, Gemini 3.1 is only available via a global endpoint, not regional ones, hence our use of the 2.5 version.

In an AI-driven architecture, model governance isn't just about versioning. It's about ensuring every model, from embedding generators to code generation LLMs, adheres to strict security standards. I implement:

• Model Registry: A central catalog for all models, including their source, version, and training data provenance.
• Vulnerability Scanning: Embedding models and custom fine-tuned LLMs are scanned for known vulnerabilities and adherence to security baselines.
• Audit Logging: Every interaction with an LLM API, especially those routed through OpenRouter, is logged with relevant metadata (request IDs, token counts, timestamps) for compliance and cost analysis. This is crucial for GDPR, as it provides an auditable trail of data processing activities.
• Access Control: Granular IAM policies restrict who can deploy, update, or even invoke specific models, integrating with Workload Identity Federation for cross-cloud scenarios.

I use Terraform to define and manage the core compute services in GCP and AWS that host our RAG and code generation components. This ensures consistency and auditability, and critically, places resources in EU regions for GDPR compliance.

# main.tf for cross-cloud compute

# Configure Google Cloud Provider for Europe
provider "google" {
  project = var.gcp_project_id
  region  = "europe-west3" # Frankfurt, Germany
}

# Configure AWS Provider for Europe
provider "aws" {
  region = "eu-central-1" # Frankfurt, Germany
}

# --- GCP Cloud Run for Application Service ---
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 data store 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 = "Service Account for Multi-Cloud AI Cloud Run service"
}

# --- AWS Lambda for potential Bedrock-specific proxy or async tasks ---
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 # seconds
  source_code_hash = filebase64sha256("lambda_function_payload.zip") # Ensure this file exists for 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"
      # Add other AWS specific environment variables as needed
    }
  }
}

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

# Attach policies for Bedrock access, VPC access, 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" # Adjust to least privilege for production
}

# ... VPC, Subnet, Security Group definitions for AWS Lambda ...
# Note: A full VPC setup is more extensive and requires careful planning for networking.
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 = "Allow outbound access for 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 = "Your GCP project ID"
  type        = string
}

variable "gcp_datastore_id" {
  description = "The ID of your Vertex AI Search Data Store"
  type        = string
  sensitive   = false
}

variable "aws_bedrock_kb_id" {
  description = "The ID of your Amazon Bedrock Knowledge Base"
  type        = string
  sensitive   = false
}

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

output "cloud_run_service_url" {
  value = google_cloud_run_service.main_app_service.status[0].url
  description = "The URL of the deployed Cloud Run service."
}

output "lambda_function_name" {
  value = aws_lambda_function.bedrock_proxy_lambda.function_name
  description = "The name of the deployed AWS Lambda function."
}

Expected Output (after 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"

Our core application logic, running on GCP Cloud Run, needs to query both Vertex AI Search and Bedrock Knowledge Bases in parallel to synthesize the best possible context for our LLM. This Python service handler acts as the RAG Orchestrator. Note that Vertex AI Search is part of the Discovery Engine service, so I use the google-cloud-discoveryengine package for that.

# rag_orchestrator/main.py (Running on GCP Cloud Run)

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

# Initialize clients (assuming environment variables for configuration)
# For Vertex AI Search, you would typically authenticate via Workload Identity
# when running on Cloud Run. The SDK handles this automatically.

def get_vertex_ai_search_results(query: str, project_id: str, location: str, data_store_id: str) -> List[Document]:
    """Queries Vertex AI Search (Discovery Engine) for relevant documents."""
    print(f"Querying Vertex AI Search in {location} for: {query}")
    try:
        client = discoveryengine.SearchServiceClient()

        # Construct the serving config path for the data store
        serving_config = client.serving_config_path(
            project=project_id,
            location=location, # e.g., europe-west3
            data_store=data_store_id, # Your data store ID
            serving_config="default_config", # Default serving config
        )

        request = discoveryengine.SearchRequest(
            serving_config=serving_config,
            query=query,
            page_size=3, # Requesting top 3 results
            query_params=discoveryengine.SearchRequest.QueryParameters(
                # You can configure query expansion or grounding here if enabled in your data store
                # For 'Grounding with Google Search', ensure it's configured in your Vertex AI Search data store.
                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:
                # Assuming 'content' holds the main text. Adjust based on your data store schema.
                results.append(Document(page_content=result.document.content, metadata={"source": result.document.id}))
        return results
    except Exception as e:
        print(f"Error querying Vertex AI Search: {e}")
        # In a production system, implement robust error handling and fallback logic
        return [Document(page_content=f"Vertex AI Search (mock): Could not retrieve for '{query}' due to error: {e}")]

def get_bedrock_knowledge_base_results(query: str, kb_id: str, region: str) -> List[Document]:
    """Queries an Amazon Bedrock Knowledge Base for relevant documents."""
    print(f"Querying Bedrock Knowledge Base '{kb_id}' in {region} for: {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"Error querying Bedrock KB: {e}")
        return [Document(page_content=f"Bedrock KB (mock): Could not retrieve for '{query}' due to error: {e}")]

async def parallel_rag_query(query: str) -> List[Document]:
    """Executes RAG queries against both GCP and AWS in parallel."""
    gcp_project_id = os.environ.get("GCP_PROJECT_ID", "your-gcp-project") # Ensure this is set via env var
    gcp_region = os.environ.get("GCP_REGION", "europe-west3")
    gcp_datastore_id = os.environ.get("GCP_DATASTORE_ID", "your-datastore-id")
    aws_kb_id = os.environ.get("AWS_BEDROCK_KB_ID", "your-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),
    )

    # Combine and deduplicate results for comprehensive context
    all_results = gcp_results + aws_results
    return all_results

# Example usage (e.g., in a FastAPI or Flask endpoint)
async def handle_rag_request(query: str):
    """Simulates handling an incoming RAG request."""
    context_documents = await parallel_rag_query(query)
    # Further process with LangChain/LlamaIndex for prompt construction
    # and then send to LLM via OpenRouter proxy
    return context_documents

if __name__ == "__main__":
    # This part would typically be part of a web server or function invocation
    # For local testing, ensure dummy environment variables are set or values are passed.
    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 = "latest GDPR changes for AI data processing"
    print(f"\nRunning parallel RAG query for: {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]}...")

Explanation: This rag_orchestrator demonstrates parallel querying of two distinct RAG sources. asyncio.to_thread is key for efficiently offloading blocking I/O calls to boto3 and the Vertex AI Search (Discovery Engine) SDK client, allowing the main event loop to remain responsive. The results are then combined, ready for integration into a final prompt for an LLM.

To manage multiple LLM APIs efficiently and implement failover between Gemini and Claude, I deploy a small Python proxy service. This abstracts away the complexity of different API endpoints and allows for dynamic model selection based on cost, performance, or availability. It's a robust pattern for improving the reliability of LLM integrations and managing costs.

# openrouter_proxy/app.py (Running on GCP Cloud Run, alongside or as part of 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: For OpenRouter analytics
            "X-Title": "Multi-Cloud AI Service", # Optional: For OpenRouter analytics
            "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() # Raise an exception for HTTP errors
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error calling OpenRouter with model {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"Attempting chat completion with model: {chosen_model}")
            return self._make_request(chosen_model, messages, **kwargs)
        except Exception as e:
            print(f"Primary model {chosen_model} failed. Falling back to {self.fallback_model}. Error: {e}")
            if chosen_model != self.fallback_model: # Prevent infinite fallback if fallback also fails
                return self._make_request(self.fallback_model, messages, **kwargs)
            else:
                raise # Re-raise if fallback also fails

# Example usage
if __name__ == "__main__":
    openrouter_api_key = os.environ.get("OPENROUTER_API_KEY")
    if not openrouter_api_key:
        # For local testing, replace with a valid key or set environment variable
        print("OPENROUTER_API_KEY environment variable not set. Using a dummy key for 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": "Write a Python function to parse a JSON string into a dictionary, handling potential errors."}
    ]

    try:
        response = proxy.chat_completion(messages)
        print("\n--- Primary Model Response ---")
        print(response["choices"][0]["message"]["content"])
    except Exception as e:
        print(f"Failed to get response from any model: {e}")

    # Simulate primary model failure to test fallback
    print("\n--- Simulating Primary Model Failure and Testing Fallback ---")
    # In a real scenario, you'd integrate this with a circuit breaker or health check.
    # For this example, let's pretend to explicitly request a non-existent model to trigger fallback
    try:
        # Using a dummy API key for this example, which will likely cause failure
        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 Model Response ---")
        print(response_fallback["choices"][0]["message"]["content"])
    except Exception as e:
        print(f"Failed even with fallback, due to initial API key issue or actual model failure: {e}")

Before sending any user-generated prompts or RAG-extracted content to external LLM APIs (even via OpenRouter), I ensure sensitive data is removed or anonymized. This is a critical GDPR requirement for maintaining a Compliance & Privacy Fortress.

# 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 patterns for common PII. This is illustrative; a production system
        # would use a dedicated PII detection library (e.g., 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", # Simplified EU phone pattern
            "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]))", # Illustrative, IBANs are complex
            "address": r"\b(\d{1,4}[ ](?:[A-Za-z0-9'-]+[ ]?){1,}[A-Za-z]{2,})\b" # Basic address pattern
        }

    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():
            # Using re.sub with a replacer function for correct replacement in 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:
                # Create a copy to avoid modifying the original message dict in place if not intended
                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

# Example usage
if __name__ == "__main__":
    scrubber = PIIScrubber(replace_with_hash=True)
    sample_messages = [
        {"role": "user", "content": "My email is mark@thecloudarchitect.io and my phone is +352 176 12345678. The order was placed with card 4111222233334444."},
        {"role": "system", "content": "Hello, how can I help you?"}
    ]

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

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

    # Example with hashing disabled
    scrubber_redacted = PIIScrubber(replace_with_hash=False)
    sample_text = "My IBAN is LU89370400440532013000."
    scrubbed_text = scrubber_redacted.scrub_text(sample_text)
    print(f"\n--- Scrubbed Text (Redacted): {scrubbed_text} ---")

Explanation: This PIIScrubber class uses regular expressions to detect and redact or hash common PII types within text. While illustrative, a production-grade system would integrate with cloud-native DLP services (like Google Cloud DLP or AWS Macie) or specialized libraries for more robust and configurable PII detection across various data types. The key is to apply this scrubbing before data leaves your trusted EU zone. Regular expressions are a starting point; a comprehensive solution requires continuous refinement and testing against real-world data.

Tracking costs and latency across clouds is critical for FinOps and performance optimization. I integrate OpenTelemetry for tracing, and then push metrics to both CloudWatch (AWS) and Cloud Monitoring (GCP). This gives us a unified view while respecting native tooling.

# 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
# The OpenTelemetry Python client for Google Cloud Monitoring is part of 'opentelemetry-exporter-google-cloud'
# from opentelemetry.exporter.google_cloud import CloudMonitoringMetricsExporter
# For AWS, typically use boto3 for CloudWatch or OpenTelemetry exporter for X-Ray/CloudWatch metrics.
import boto3

import time
import random
from typing import Dict

# Configure OpenTelemetry Tracer
resource = Resource.create({
    "service.name": "multi-cloud-ai-service",
    "service.version": "1.0.0",
    "cloud.provider": "gcp", # Can be dynamically set based on deployment context
    "cloud.region": os.environ.get("GCP_REGION", "europe-west3"),
})

provider = TracerProvider(resource=resource)
span_processor = BatchSpanProcessor(ConsoleSpanExporter()) # For console output during development
provider.add_span_processor(span_processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# Assuming a conversion rate of $1 \approx \u20ac0.92 for illustrative pricing calculations.
USD_TO_EUR_RATE = 0.92

def record_llm_call_metrics(model_name: str, tokens_used: int, latency_ms: float, cost_usd: float):
    """Records LLM call metrics to both GCP Cloud Monitoring and AWS CloudWatch (conceptually)."""
    cost_eur = cost_usd * USD_TO_EUR_RATE

    # --- GCP Cloud Monitoring (via OpenTelemetry Exporter) ---
    # In a real setup, I would initialize CloudMonitoringMetricsExporter and use it to push custom metrics.
    # For this to work, ensure GCP credentials are set up with Cloud Monitoring write scope.
    print(f"[GCP Cloud Monitoring] Exporting metrics for {model_name}: tokens={tokens_used}, latency={latency_ms:.2f}ms, cost=\u20ac{cost_eur:.4f} (${cost_usd:.4f})")

    # --- AWS CloudWatch (via Boto3) ---
    # I use boto3 to push custom metrics to 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' # Use 'Count' for currency unless a specific 'Currency' unit is supported and desired
                }
            ]
        )
        print(f"[AWS CloudWatch] Exported metrics for {model_name}")
    except Exception as e:
        print(f"[AWS CloudWatch] Error exporting metrics: {e}")

def perform_llm_call(model: str, prompt: str) -> Dict:
    """Simulates an LLM call and records its performance metrics."""
    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"Performing LLM call with {model} for prompt: {prompt[:50]}...")
        start_time = time.time()
        time.sleep(random.uniform(0.5, 2.0)) # Simulate network latency and processing
        end_time = time.time()

        latency_ms = (end_time - start_time) * 1000
        tokens = random.randint(50, 500)
        # Approximate cost in USD: Claude 3.5 Sonnet might be ~$0.003/1K tokens input, Gemini 2.5 Flash ~$0.00035/1K tokens
        # (These are illustrative; verify against current vendor docs and OpenRouter's pricing).
        cost_per_token_usd = 0.000003 # e.g., $0.003 per 1k tokens, so $0.000003 per 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": "Simulated LLM output"}

if __name__ == "__main__":
    print("\n--- Starting Observability Example ---")
    # Set dummy environment variables if running locally and not configured elsewhere
    os.environ["GCP_REGION"] = os.environ.get("GCP_REGION", "europe-west3")
    os.environ["AWS_REGION"] = os.environ.get("AWS_REGION", "eu-central-1")

    # Simulate a series of LLM calls
    perform_llm_call("anthropic/claude-3.5-sonnet", "Summarize recent AI safety research")
    perform_llm_call("google/gemini-2.5-flash", "Generate a marketing slogan for cloud architecture blog")
    print("--- Observability Example Finished ---")

Explanation: This script illustrates how OpenTelemetry can be used to instrument LLM calls. It demonstrates how to record custom metrics like tokens_used, latency_ms, and cost_eur and push them to both Google Cloud Monitoring and AWS CloudWatch. This provides the necessary visibility for FinOps teams and engineers to optimize costs and performance across their multi-cloud AI stack. For actual cost metrics, I typically set up specific cost monitoring dashboards in each cloud, cross-referenced with OpenRouter's billing data. I've noted the conversion rate of $1 \approx \u20ac0.92 for illustrative pricing calculations.

After deployment, I use a series of commands to ensure everything is connected and functioning as expected.

# 1. Verify GCP Cloud Run service deployment and status
gcloud run services describe multi-cloud-ai-service --platform managed --region europe-west3 --format='value(status.url)'
# Expected output:
# https://multi-cloud-ai-service-xxxxxxxx-ew.a.run.app

# 2. Test the Cloud Run endpoint (replace with your actual URL)
# Ensure your application exposes a '/rag-query' endpoint for this example.
curl -X POST -H "Content-Type: application/json" \n     -d '{"query": "What is the latest advancement in transformer models?"}' \n     "$(gcloud run services describe multi-cloud-ai-service --platform managed --region europe-west3 --format='value(status.url)')/rag-query"
# Expected output (simplified, actual response will depend on your app logic):
# {"context_documents": [...], "llm_response": "..."}

# 3. Verify AWS Lambda function existence
aws lambda get-function --function-name bedrock-rag-proxy --region eu-central-1
# Expected output (partial):
# {
#     "Configuration": {
#         "FunctionName": "bedrock-rag-proxy",
#         "Runtime": "python3.12",
#         "Handler": "lambda_function.handler",
#         ...
#     }
# }

# 4. Check OpenRouter API connectivity (example using curl)
# Replace $OPENROUTER_API_KEY with your actual API key.
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": "Hello"}] }'
# Expected output (partial):
# {"choices": [{"message": {"content": "Hello! How can I help you today?"}, ...}]}

1. Error: Cross-cloud IAM permission denied

# Error message example (from GCP logs when calling AWS Bedrock)
google.auth.exceptions.RefreshError: ('Failed to retrieve access token: {"error":"invalid_grant", "error_description":"Invalid AWS credential for Workload Identity Federation"}', '...')

**Solution:** This typically means your Workload Identity Federation setup is incorrect. Double-check your GCP Service Account's IAM policy, the AWS IAM Role's trust policy (especially the `Federated` principal and `StringEquals` conditions for the `google.subject` and `google.aud`), and the policy attached to the AWS role to ensure it grants appropriate `bedrock` permissions. Also ensure the GCP Service Account is correctly linked to the Cloud Run service account, as specified in [Google Cloud's Workload Identity Federation documentation](https://cloud.google.com/iam/docs/manage-workload-identity-federation).

1. Error: PII Leakage Detected

# This error might not be a system error, but an alert from a DLP system (e.g. Cloud DLP, AWS Macie)
# or a security audit finding.

**Solution:** If PII is detected downstream of your scrubbing layer, review your `PIIScrubber` patterns and ensure they are comprehensive enough for your data. Consider integrating a more robust, cloud-native DLP service. Remember that regex-based scrubbing is never 100% foolproof; dedicated DLP services offer higher accuracy due to their context-aware detection capabilities. Validate your scrubbing process with regular data audits and penetration testing.

• Hybrid RAG is crucial for performance and compliance: Combine GCP Vertex AI Search with AWS Bedrock Knowledge Bases to leverage their respective strengths and meet data residency requirements.
• OpenRouter provides critical LLM abstraction: Use OpenRouter for failover between models like Claude Sonnet and Gemini 2.5 Flash/Pro, optimizing for cost and availability without application code changes.
• GDPR compliance requires PII scrubbing and EU residency: Implement robust PII anonymization before sending data to external APIs and provision all RAG data sources exclusively in EU-central cloud regions.
• Workload Identity Federation is essential for secure cross-cloud access: Securely allow GCP services to interact with AWS resources without managing long-lived access keys.
• Comprehensive observability is non-negotiable: Instrument your multi-cloud AI systems with OpenTelemetry to track costs and latency across AWS CloudWatch and GCP Cloud Monitoring.