Claude 3.5 Sonnet on Vertex AI: Python quickstart

gcloud components update
gcloud components install beta

python -m pip install -U 'anthropic[vertex]'
python -m pip install -U Weave 

import base64
from anthropic import AnthropicVertex
import weave


# Initialize Weave for logging
weave.init('your_project_name')


# Set up the Anthropic client
LOCATION = "us-east5"
PROJECT_ID = "your-gcp-project-id"
client = AnthropicVertex(region=LOCATION, project_id=PROJECT_ID)


@weave.op()
def get_response_from_anthropic(query):
    response = client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": query
            }
        ],
        model="claude-3-5-sonnet@20240620",
    )
    return response.model_dump_json(indent=2)


# Define the query
query = "How can I improve my eating habits?"

# Call the function and log the response
response = get_response_from_anthropic(query)
print(response)


import base64
import requests
from pathlib import Path
from anthropic import AnthropicVertex
import weave
import httpx

# Initialize Weave for logging
weave.init('claude_sonnet')

# Set up the Anthropic client
LOCATION = "us-east5"
PROJECT_ID = "your-gcp-project-id"
client = AnthropicVertex(region=LOCATION, project_id=PROJECT_ID)

def image_to_data_url(file_path_or_url):
    EXT_TO_MIMETYPE = {
        '.jpg': 'image/jpeg',
        '.png': 'image/png',
        '.svg': 'image/svg+xml',
    }

    if file_path_or_url.startswith('http://') or file_path_or_url.startswith('https://'):
        response = requests.get(file_path_or_url)
        if response.status_code != 200:
            raise ValueError(f"Unable to fetch image from URL: {file_path_or_url}")
        content_type = response.headers['Content-Type']
        mimetype = content_type if content_type in EXT_TO_MIMETYPE.values() else None
        image_data = response.content
    else:
        ext = Path(file_path_or_url).suffix
        mimetype = EXT_TO_MIMETYPE.get(ext)
        if not mimetype:
            raise ValueError(f"Unsupported file extension: {ext}")
        with open(file_path_or_url, 'rb') as image_file:
            image_data = image_file.read()

    if not mimetype:
        raise ValueError(f"Unsupported mime type: {mimetype}")

    encoded_string = base64.b64encode(image_data).decode('utf-8')
    data_url = f"data:{mimetype};base64,{encoded_string}"
    return data_url

@weave.op()
def analyze_image(image_data, prompt, wv_img): # wv_image added for tracking 
    response = client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "data": image_data,
                            "media_type": "image/jpeg"  # Add the media type here
                        },

                    },
                    {
                    "type": "text",
                    "text": prompt, 
                    },
                ]
            }
        ],
        model="claude-3-5-sonnet@20240620",
    )
    return response.model_dump_json(indent=2)



image_url = "https://upload.wikimedia.org/wikipedia/commons/a/a7/Camponotus_flavomarginatus_ant.jpg"
image_b64 = base64.b64encode(httpx.get(image_url).content).decode("utf-8")
weave_img = image_to_data_url(image_url)


# Call the function and log the response
image_analysis_response = analyze_image(image_b64, "describe this image", weave_img)
print(image_analysis_response)