Amazon Bedrock Fundamentals¶

import boto3
import json

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

# Invoke Claude model
def invoke_claude(prompt, max_tokens=1000):
    body = json.dumps({
        "prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
        "max_tokens_to_sample": max_tokens,
        "temperature": 0.7,
        "top_p": 0.9
    })

    response = bedrock.invoke_model(
        modelId='anthropic.claude-v2',
        body=body,
        contentType='application/json'
    )

    response_body = json.loads(response['body'].read())
    return response_body['completion']

# Example usage
result = invoke_claude("Explain quantum computing in simple terms")
print(result)

def invoke_claude_streaming(prompt):
    body = json.dumps({
        "prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
        "max_tokens_to_sample": 1000,
        "temperature": 0.7
    })

    response = bedrock.invoke_model_with_response_stream(
        modelId='anthropic.claude-v2',
        body=body,
        contentType='application/json'
    )

    # Process streaming response
    for event in response['body']:
        chunk = json.loads(event['chunk']['bytes'])
        if 'completion' in chunk:
            print(chunk['completion'], end='', flush=True)

def generate_image(prompt, style="photographic"):
    body = json.dumps({
        "text_prompts": [{"text": prompt}],
        "cfg_scale": 10,
        "seed": 0,
        "steps": 50,
        "style_preset": style
    })

    response = bedrock.invoke_model(
        modelId='stability.stable-diffusion-xl-v1',
        body=body,
        contentType='application/json'
    )

    response_body = json.loads(response['body'].read())
    return response_body['artifacts'][0]['base64']

# Create fine-tuning job
response = bedrock.create_model_customization_job(
    jobName='my-custom-model',
    customModelName='my-model-v1',
    baseModelIdentifier='anthropic.claude-v2',
    trainingDataConfig={
        's3Uri': 's3://my-bucket/training-data.jsonl'
    },
    validationDataConfig={
        's3Uri': 's3://my-bucket/validation-data.jsonl'
    },
    hyperParameters={
        'epochCount': '3',
        'batchSize': '1',
        'learningRate': '0.00001'
    }
)

# Create provisioned model throughput
response = bedrock.create_provisioned_model_throughput(
    modelUnits=1,
    modelId='anthropic.claude-v2',
    provisionedModelName='my-provisioned-model',
    commitmentDuration='OneMonth'
)

# Create knowledge base
response = bedrock_agent.create_knowledge_base(
    name='my-knowledge-base',
    description='Company documentation knowledge base',
    knowledgeBaseConfiguration={
        'type': 'VECTOR',
        'vectorKnowledgeBaseConfiguration': {
            'embeddingModelArn': 'arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v1'
        }
    },
    storageConfiguration={
        'type': 'OPENSEARCH_SERVERLESS',
        'opensearchServerlessConfiguration': {
            'collectionArn': 'arn:aws:aoss:us-east-1:123456789012:collection/kb-collection',
            'vectorIndexName': 'knowledge-base-index',
            'fieldMapping': {
                'vectorField': 'vector',
                'textField': 'text',
                'metadataField': 'metadata'
            }
        }
    }
)

import numpy as np
from sentence_transformers import SentenceTransformer

class RAGSystem:
    def __init__(self):
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.knowledge_base = []
        self.embeddings = []

    def add_document(self, text, metadata=None):
        """Add document to knowledge base"""
        embedding = self.embedding_model.encode(text)
        self.knowledge_base.append({
            'text': text,
            'metadata': metadata or {},
            'embedding': embedding
        })
        self.embeddings.append(embedding)

    def search(self, query, top_k=3):
        """Search for relevant documents"""
        query_embedding = self.embedding_model.encode(query)

        # Calculate similarities
        similarities = []
        for doc_embedding in self.embeddings:
            similarity = np.dot(query_embedding, doc_embedding) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(doc_embedding)
            )
            similarities.append(similarity)

        # Get top-k most similar documents
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.knowledge_base[i] for i in top_indices]

    def generate_response(self, query):
        """Generate response using RAG"""
        # Retrieve relevant context
        relevant_docs = self.search(query)
        context = "\n".join([doc['text'] for doc in relevant_docs])

        # Create enhanced prompt
        prompt = f"""
        Context information:
        {context}

        Question: {query}

        Please answer the question based on the provided context.
        """

        # Generate response with foundation model
        return invoke_claude(prompt)

# Classification prompt
classification_prompt = """
Classify the following customer feedback into categories: positive, negative, or neutral.

Feedback: "{feedback}"

Classification: """

# Summarization prompt
summarization_prompt = """
Please provide a concise summary of the following text in 2-3 sentences:

Text: {text}

Summary: """

# Code generation prompt
code_prompt = """
Write a Python function that {description}.

Requirements:
- Include proper error handling
- Add docstrings
- Use type hints

Code: """

few_shot_prompt = """
Classify movie reviews as positive or negative:

Review: "This movie was absolutely fantastic!"
Sentiment: Positive

Review: "I fell asleep halfway through. Boring!"
Sentiment: Negative

Review: "The acting was great but the plot was confusing."
Sentiment: Negative

Review: "{user_review}"
Sentiment: """

cot_prompt = """
Solve this step by step:

Question: A restaurant has 24 tables. Each table can seat 4 people. If the restaurant is 75% full, how many people are dining?

Let me think through this step by step:
1) First, I need to find the total capacity
2) Then calculate 75% of that capacity
3) 

Step 1: Total capacity = 24 tables × 4 people per table = 96 people
Step 2: 75% of 96 = 0.75 × 96 = 72 people

Therefore, 72 people are dining.

Now solve this problem:
{problem}
"""

expert_prompt = """
You are a cybersecurity expert with 15 years of experience. 
A client asks: "{question}"

Provide a detailed, professional response that includes:
1. Technical explanation
2. Potential risks
3. Recommended solutions
4. Best practices

Response: """

# Calculate token costs
def calculate_cost(input_tokens, output_tokens, model_id):
    # Pricing varies by model - example for Claude
    pricing = {
        'anthropic.claude-v2': {
            'input': 0.00001102,   # per 1K input tokens
            'output': 0.00003268   # per 1K output tokens
        }
    }

    if model_id in pricing:
        input_cost = (input_tokens / 1000) * pricing[model_id]['input']
        output_cost = (output_tokens / 1000) * pricing[model_id]['output']
        return input_cost + output_cost

    return 0

import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def invoke_with_logging(prompt, model_id):
    logger.info(f"Invoking model {model_id}")
    logger.debug(f"Prompt: {prompt[:100]}...")

    try:
        response = invoke_model(prompt, model_id)
        logger.info(f"Successful invocation, response length: {len(response)}")
        return response
    except Exception as e:
        logger.error(f"Model invocation failed: {str(e)}")
        raise

# Enable content filtering
body = json.dumps({
    "prompt": prompt,
    "max_tokens_to_sample": 1000,
    "temperature": 0.7,
    "anthropic_version": "bedrock-2023-05-31"
})

response = bedrock.invoke_model(
    modelId='anthropic.claude-v2',
    body=body,
    contentType='application/json',
    guardrailIdentifier='my-guardrail-id',
    guardrailVersion='1'
)