AWS Bedrock¶

# Install AWS SDK
pip install boto3

# Basic text generation with Claude 3.5 Sonnet
import boto3
import json

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

prompt = "Explain quantum computing in simple terms"

body = json.dumps({
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 1024,
    "messages": [
        {"role": "user", "content": prompt}
    ]
})  # (1)!

response = bedrock.invoke_model(
    modelId='anthropic.claude-3-5-sonnet-20240620-v1:0',  # (2)!
    body=body
)

response_body = json.loads(response['body'].read())
print(response_body['content'][0]['text'])

# Streaming responses (recommended for UX)
response = bedrock.invoke_model_with_response_stream(
    modelId='anthropic.claude-3-5-sonnet-20240620-v1:0',
    body=body
)

for event in response['body']:
    chunk = json.loads(event['chunk']['bytes'])
    if chunk['type'] == 'content_block_delta':
        if 'delta' in chunk and 'text' in chunk['delta']:
            print(chunk['delta']['text'], end='', flush=True)  # (1)!

# Multi-modal input (image + text)
import base64

with open('diagram.png', 'rb') as f:
    image_data = base64.b64encode(f.read()).decode('utf-8')

body = json.dumps({
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 2048,
    "messages": [{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": image_data
                }
            },
            {
                "type": "text",
                "text": "Explain this architecture diagram"
            }
        ]
    }]
})  # (1)!

# AWS CLI usage
aws bedrock-runtime invoke-model \
    --model-id anthropic.claude-3-5-sonnet-20240620-v1:0 \
    --body '{"anthropic_version":"bedrock-2023-05-31","max_tokens":1024,"messages":[{"role":"user","content":"Hello"}]}' \
    --cli-binary-format raw-in-base64-out \
    response.json  # (1)!

cat response.json | jq -r '.content[0].text'  # (2)!

# RAG with Bedrock Knowledge Bases
import boto3

bedrock_agent = boto3.client('bedrock-agent-runtime', region_name='us-east-1')

query = "What are the company policies on remote work?"

response = bedrock_agent.retrieve_and_generate(
    input={'text': query},
    retrieveAndGenerateConfiguration={
        'type': 'KNOWLEDGE_BASE',
        'knowledgeBaseConfiguration': {
            'knowledgeBaseId': 'KB123ABC',  # (1)!
            'modelArn': 'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0'
        }
    }
)

print(response['output']['text'])
print("\nSources:")
for citation in response['citations']:
    for reference in citation['retrievedReferences']:
        print(f"- {reference['location']['s3Location']['uri']}")  # (2)!

# Bedrock Agents (autonomous workflows)
agent_client = boto3.client('bedrock-agent-runtime')

response = agent_client.invoke_agent(
    agentId='AGENT123',
    agentAliasId='ALIAS456',
    sessionId='session-001',  # (1)!
    inputText='Book a flight to NYC next week'
)

for event in response['completion']:
    if 'chunk' in event:
        chunk = event['chunk']
        if 'bytes' in chunk:
            print(chunk['bytes'].decode('utf-8'), end='', flush=True)  # (2)!

# Guardrails (content filtering)
body = json.dumps({
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 1024,
    "messages": [{"role": "user", "content": prompt}]
})

response = bedrock.invoke_model(
    modelId='anthropic.claude-3-5-sonnet-20240620-v1:0',
    body=body,
    guardrailIdentifier='guardrail-123',  # (1)!
    guardrailVersion='1'
)

# Response blocked if violates guardrails
if response['ResponseMetadata']['HTTPStatusCode'] == 400:
    print("Content blocked by guardrails")  # (2)!

# Custom model import (fine-tuned models)
# 1. Upload model weights to S3
# 2. Create custom model in Bedrock console
# 3. Use like foundation model

response = bedrock.invoke_model(
    modelId='arn:aws:bedrock:us-east-1:123456789:custom-model/my-model-v1',  # (1)!
    body=body
)

# Error handling and retries
from botocore.exceptions import ClientError
import time

def invoke_with_retry(model_id, body, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = bedrock.invoke_model(
                modelId=model_id,
                body=body
            )
            return json.loads(response['body'].read())
        except ClientError as e:
            if e.response['Error']['Code'] == 'ThrottlingException':  # (1)!
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # Exponential backoff
                    continue
            raise