AI Services Across Clouds

import boto3

client = boto3.client("bedrock-runtime", region_name="us-east-1")

response = client.converse(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[
        {
            "role": "user",
            "content": [{"text": "What is the capital of France?"}],
        }
    ],
    system=[{"text": "Answer in one sentence."}],
    inferenceConfig={"maxTokens": 256, "temperature": 0.3},
)

print(response["output"]["message"]["content"][0]["text"])
print(f"Tokens: {response['usage']['inputTokens']} in, {response['usage']['outputTokens']} out")

from openai import AzureOpenAI

client = AzureOpenAI(
    api_key="your-key",
    api_version="2024-10-21",
    azure_endpoint="https://my-resource.openai.azure.com",
)

response = client.chat.completions.create(
    model="gpt-4o",  # deployment name
    messages=[
        {"role": "system", "content": "Answer in one sentence."},
        {"role": "user", "content": "What is the capital of France?"},
    ],
    max_tokens=256,
    temperature=0.3,
)

print(response.choices[0].message.content)
print(f"Tokens: {response.usage.prompt_tokens} in, {response.usage.completion_tokens} out")

import vertexai
from vertexai.generative_models import GenerativeModel

vertexai.init(project="my-project", location="us-central1")

model = GenerativeModel(
    "gemini-2.0-flash",
    system_instruction="Answer in one sentence.",
)

response = model.generate_content(
    "What is the capital of France?",
    generation_config={"max_output_tokens": 256, "temperature": 0.3},
)

print(response.text)
print(f"Tokens: {response.usage_metadata.prompt_token_count} in, "
      f"{response.usage_metadata.candidates_token_count} out")

# ── AWS Bedrock ──
import boto3, json

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
response = bedrock.invoke_model(
    modelId="amazon.titan-embed-text-v2:0",
    body=json.dumps({"inputText": "Cloud computing fundamentals", "dimensions": 1024}),
)
embedding_aws = json.loads(response["body"].read())["embedding"]
print(f"Bedrock Titan: {len(embedding_aws)} dimensions")

# ── Azure OpenAI ──
from openai import AzureOpenAI

azure_client = AzureOpenAI(api_key="key", api_version="2024-10-21",
                            azure_endpoint="https://my.openai.azure.com")
response = azure_client.embeddings.create(
    model="text-embedding-3-large", input=["Cloud computing fundamentals"], dimensions=1024,
)
embedding_azure = response.data[0].embedding
print(f"Azure OpenAI: {len(embedding_azure)} dimensions")

# ── GCP Vertex AI ──
from vertexai.language_models import TextEmbeddingModel

model = TextEmbeddingModel.from_pretrained("text-embedding-005")
embeddings = model.get_embeddings(["Cloud computing fundamentals"], output_dimensionality=1024)
embedding_gcp = embeddings[0].values
print(f"Vertex AI: {len(embedding_gcp)} dimensions")

# Abstract AI provider behind a common interface
from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class AIResponse:
    content: str
    input_tokens: int
    output_tokens: int
    provider: str
    model: str

class AIProvider(ABC):
    @abstractmethod
    def complete(self, messages: list, max_tokens: int = 1024) -> AIResponse:
        pass

class BedrockProvider(AIProvider):
    def __init__(self):
        import boto3
        self.client = boto3.client("bedrock-runtime", region_name="us-east-1")

    def complete(self, messages, max_tokens=1024):
        response = self.client.converse(
            modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
            messages=[{"role": m["role"], "content": [{"text": m["content"]}]} for m in messages],
            inferenceConfig={"maxTokens": max_tokens},
        )
        return AIResponse(
            content=response["output"]["message"]["content"][0]["text"],
            input_tokens=response["usage"]["inputTokens"],
            output_tokens=response["usage"]["outputTokens"],
            provider="bedrock", model="claude-3.5-sonnet",
        )

class AzureOpenAIProvider(AIProvider):
    def __init__(self):
        from openai import AzureOpenAI
        self.client = AzureOpenAI(api_key="key", api_version="2024-10-21",
                                   azure_endpoint="https://my.openai.azure.com")

    def complete(self, messages, max_tokens=1024):
        response = self.client.chat.completions.create(
            model="gpt-4o", messages=messages, max_tokens=max_tokens,
        )
        return AIResponse(
            content=response.choices[0].message.content,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            provider="azure", model="gpt-4o",
        )

class VertexProvider(AIProvider):
    def __init__(self):
        import vertexai
        from vertexai.generative_models import GenerativeModel
        vertexai.init(project="my-project", location="us-central1")
        self.model = GenerativeModel("gemini-2.0-flash")

    def complete(self, messages, max_tokens=1024):
        prompt = "\n".join(f"{m['role']}: {m['content']}" for m in messages)
        response = self.model.generate_content(
            prompt, generation_config={"max_output_tokens": max_tokens},
        )
        return AIResponse(
            content=response.text,
            input_tokens=response.usage_metadata.prompt_token_count,
            output_tokens=response.usage_metadata.candidates_token_count,
            provider="vertex", model="gemini-2.0-flash",
        )

# Failover chain
PROVIDERS = [BedrockProvider(), AzureOpenAIProvider(), VertexProvider()]

def complete_with_failover(messages, max_tokens=1024):
    for provider in PROVIDERS:
        try:
            return provider.complete(messages, max_tokens)
        except Exception as e:
            print(f"Provider {provider.__class__.__name__} failed: {e}")
    raise Exception("All providers failed")