Claude 10 Amazon Bedrock: Panduan Lengkap dari Setup hingga Production AI System

pip install boto3 botocore
pip install rank-bm25 numpy  # Untuk RAG

# Untuk development: AWS CLI
aws configure
# AWS Access Key ID: [your-key]
# AWS Secret Access Key: [your-secret]
# Default region: us-east-1

# Untuk production: gunakan IAM Role (tidak perlu credentials manual)
# Attach policy berikut ke EC2/Lambda/ECS role:

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream"],
    "Resource": "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-*"
  }]
}

from botocore.config import Config

config = Config(
    read_timeout=3600,
    connect_timeout=10,
    retries={'mode': 'adaptive', 'max_attempts': 3}
)

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

import boto3
import json
from botocore.config import Config

config = Config(read_timeout=3600, retries={'mode': 'adaptive', 'max_attempts': 3})
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1', config=config)
MODEL_ID = 'anthropic.claude-sonnet-4-6-20250514-v1:0'


def invoke_model(prompt: str, max_tokens: int = 1024, temperature: float = 0.7) -> str:
    """Kirim request ke Claude via InvokeModel — native Anthropic format."""
    payload = {
        "anthropic_version": "bedrock-2023-05-31",  # WAJIB! Jangan hapus ini
        "max_tokens": max_tokens,
        "temperature": temperature,
        "messages": [{"role": "user", "content": prompt}]
    }

    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps(payload),
        contentType='application/json',
        accept='application/json'
    )
    result = json.loads(response['body'].read())

    # Log token usage untuk monitoring cost
    usage = result['usage']
    print(f"Tokens: in={usage['input_tokens']}, out={usage['output_tokens']}")

    return result['content'][0]['text']

def converse(
    user_message: str,
    system: str = None,
    max_tokens: int = 1024,
    temperature: float = 0.7
) -> str:
    """Converse API — unified dan portabel lintas model."""
    kwargs = {
        'modelId': MODEL_ID,
        'messages': [{'role': 'user', 'content': [{'text': user_message}]}],
        'inferenceConfig': {'maxTokens': max_tokens, 'temperature': temperature}
    }

    if system:
        kwargs['system'] = [{'text': system}]

    response = bedrock.converse(**kwargs)

    usage = response['usage']
    stop = response['stopReason']
    print(f'Stop: {stop} | Tokens: in={usage["inputTokens"]} out={usage["outputTokens"]}')

    return response['output']['message']['content'][0]['text']


# Contoh dengan system prompt
result = converse(
    user_message='Apa best practice keamanan untuk API Bedrock di production?',
    system='Kamu adalah AWS Solutions Architect yang fokus pada security. Jawab ringkas.',
    temperature=0.3
)

def stream_converse(user_message: str, system: str = None, max_tokens: int = 2048) -> str:
    """Stream response token per token — ideal untuk user-facing interface."""
    kwargs = {
        'modelId': MODEL_ID,
        'messages': [{'role': 'user', 'content': [{'text': user_message}]}],
        'inferenceConfig': {'maxTokens': max_tokens}
    }
    if system:
        kwargs['system'] = [{'text': system}]

    response = bedrock.converse_stream(**kwargs)
    collected = []

    for event in response['stream']:
        if 'contentBlockDelta' in event:
            delta = event['contentBlockDelta']['delta']
            if 'text' in delta:
                print(delta['text'], end='', flush=True)
                collected.append(delta['text'])
        elif 'messageStop' in event:
            print()  # Newline setelah selesai

    return ''.join(collected)

from dataclasses import dataclass, field
from typing import List


@dataclass
class BedrockChatbot:
    """Production-ready chatbot dengan conversation history."""
    system_prompt: str
    model_id: str = MODEL_ID
    max_tokens: int = 2048
    temperature: float = 0.7
    max_history: int = 20  # Batas history untuk kontrol context window
    history: List[dict] = field(default_factory=list)
    total_input_tokens: int = 0
    total_output_tokens: int = 0

    def chat(self, user_message: str) -> str:
        self.history.append({
            'role': 'user',
            'content': [{'text': user_message}]
        })

        # Trim history jika terlalu panjang (FIFO)
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

        response = bedrock.converse(
            modelId=self.model_id,
            system=[{'text': self.system_prompt}],
            messages=self.history,
            inferenceConfig={'maxTokens': self.max_tokens, 'temperature': self.temperature}
        )

        assistant_msg = response['output']['message']
        assistant_text = assistant_msg['content'][0]['text']
        self.history.append(assistant_msg)

        usage = response['usage']
        self.total_input_tokens += usage['inputTokens']
        self.total_output_tokens += usage['outputTokens']

        return assistant_text

    @property
    def total_cost_estimate(self) -> float:
        # Estimasi berdasarkan Sonnet 4.6 pricing
        return (self.total_input_tokens * 3 + self.total_output_tokens * 15) / 1_000_000


# Demo
bot = BedrockChatbot(system_prompt='Kamu adalah AWS expert. Jawab teknikal tapi mudah dipahami.')
print(bot.chat('Apa itu Amazon Bedrock?'))
print(bot.chat('Bagaimana cara enable model access?'))  # Context terjaga
print(f'Estimasi biaya: ${bot.total_cost_estimate:.4f}')

def analyze_document(document: str, focus: str) -> str:
    prompt = f'''Analisis dokumen dan fokus pada aspek: {focus}

<document>
{document}
</document>

<instructions>
1. Identifikasi 3 poin utama tentang {focus}
2. Evaluasi kekuatan dan kelemahan
3. Berikan 2 rekomendasi actionable yang spesifik
</instructions>

<output_format>
Gunakan struktur:
## Poin Utama
## Evaluasi
## Rekomendasi
</output_format>'''
    return converse(prompt)

def classify_ticket(ticket_text: str) -> dict:
    prompt = f'''Klasifikasikan tiket customer support.

<examples>
Tiket: "Pesanan saya belum sampai sudah 3 minggu"
Output: {{"category": "pengiriman", "priority": "high", "sentiment": "negatif", "requires_escalation": true}}

Tiket: "Bagaimana cara mengubah alamat pengiriman default?"
Output: {{"category": "akun", "priority": "low", "sentiment": "netral", "requires_escalation": false}}
</examples>

Klasifikasikan tiket berikut:
<ticket>{ticket_text}</ticket>

Respons HANYA dengan JSON, tanpa penjelasan tambahan.'''

    result = converse(prompt, temperature=0.0)
    return json.loads(result)

from typing import Callable, List


def generate_test_dataset(task_description: str, n: int = 20) -> List[dict]:
    """Generate test cases menggunakan Claude untuk evaluasi prompt."""
    prompt = f'''Generate {n} diverse test cases untuk mengevaluasi prompt yang: {task_description}

Format output (JSON array valid):
[
  {{"input": "input contoh", "expected": "output yang benar", "difficulty": "easy|medium|hard"}},
  ...
]

Pastikan test cases mencakup: Normal cases (60%), Edge cases (20%), Ambiguous cases (20%).
Respons HANYA dengan JSON array yang valid.'''

    result = converse(prompt, temperature=0.3)
    return json.loads(result)


def evaluate_prompt(
    prompt_template: str,
    test_cases: List[dict],
    grader: Callable,
    verbose: bool = False
) -> dict:
    """Evaluasi prompt terhadap test dataset."""
    results = []

    for i, case in enumerate(test_cases):
        prompt = prompt_template.format(input=case['input'])
        actual = converse(prompt, temperature=0.0)
        passed = grader(actual, case['expected'])
        results.append({'case': case, 'actual': actual, 'passed': passed})

        if verbose:
            status = '✅' if passed else '❌'
            print(f'[{i+1}/{len(test_cases)}] {status} | Input: {case["input"][:50]}...')

    accuracy = sum(r['passed'] for r in results) / len(results)
    return {
        'accuracy': accuracy,
        'total': len(results),
        'passed': sum(r['passed'] for r in results),
        'failed_cases': [r for r in results if not r['passed']]
    }


# Bandingkan dua versi prompt
test_cases = generate_test_dataset('mengklasifikasikan tiket customer support', n=30)

prompt_v1 = 'Klasifikasikan tiket: {input}. Kategori: pengiriman/akun/produk/umum.'
prompt_v2 = '''Kamu adalah customer support classifier. Baca tiket dan klasifikasikan ke:
kategori (pengiriman/akun/produk/umum), prioritas (low/medium/high), dan sentimen.
Tiket: {input}
Output JSON: {{"category": ..., "priority": ..., "sentiment": ...}}'''

eval_v1 = evaluate_prompt(prompt_v1, test_cases, json_grader, verbose=True)
eval_v2 = evaluate_prompt(prompt_v2, test_cases, json_grader, verbose=True)

print(f'Prompt v1: {eval_v1["accuracy"]:.1%}')
print(f'Prompt v2: {eval_v2["accuracy"]:.1%}')
print(f'Improvement: {(eval_v2["accuracy"]-eval_v1["accuracy"])*100:.1f} percentage points')

TOOLS = [
    {
        'toolSpec': {
            'name': 'get_product_info',
            'description': '''Dapatkan informasi lengkap tentang produk berdasarkan ID.
Gunakan tool ini saat user bertanya tentang produk, harga, stok, atau spesifikasi.
Jangan gunakan untuk hal lain di luar informasi produk.''',
            'inputSchema': {
                'json': {
                    'type': 'object',
                    'properties': {
                        'product_id': {
                            'type': 'string',
                            'description': 'ID produk (format: PROD-XXXX)'
                        },
                        'include_reviews': {
                            'type': 'boolean',
                            'description': 'Sertakan rating dan review? Default: false',
                            'default': False
                        }
                    },
                    'required': ['product_id']
                }
            }
        }
    }
]

def run_agent(user_query: str, max_iterations: int = 10) -> str:
    """Agentic loop dengan tool use."""
    messages = [{'role': 'user', 'content': [{'text': user_query}]}]

    for iteration in range(max_iterations):
        response = bedrock.converse(
            modelId=MODEL_ID,
            messages=messages,
            toolConfig={'tools': TOOLS},
            inferenceConfig={'maxTokens': 2048}
        )

        stop_reason = response['stopReason']
        assistant_msg = response['output']['message']
        messages.append(assistant_msg)

        # Model selesai — tidak butuh tool lagi
        if stop_reason == 'end_turn':
            for block in assistant_msg['content']:
                if 'text' in block:
                    return block['text']

        # Model ingin menggunakan tool
        elif stop_reason == 'tool_use':
            tool_results = []

            for block in assistant_msg['content']:
                if 'toolUse' in block:
                    tool_call = block['toolUse']
                    name = tool_call['name']
                    inp = tool_call['input']
                    tid = tool_call['toolUseId']

                    print(f'[Iter {iteration+1}] Tool: {name}({inp})')

                    try:
                        result = TOOL_REGISTRY[name](**inp)
                    except Exception as e:
                        result = {'error': str(e)}

                    tool_results.append({
                        'toolResult': {
                            'toolUseId': tid,
                            'content': [{'text': json.dumps(result, ensure_ascii=False)}]
                        }
                    })

            messages.append({'role': 'user', 'content': tool_results})

    return 'Max iterations reached without final response'

def extract_invoice(invoice_text: str) -> dict:
    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=[{'role': 'user', 'content': [{'text': f'Ekstrak data dari invoice: {invoice_text}'}]}],
        toolConfig={
            'tools': EXTRACTION_TOOLS,
            'toolChoice': {'tool': {'name': 'extract_invoice_data'}}  # Force pakai tool ini
        },
        inferenceConfig={'maxTokens': 1024, 'temperature': 0.0}
    )

    for block in response['output']['message']['content']:
        if 'toolUse' in block:
            return block['toolUse']['input']
    return {}

import re
from typing import List


def fixed_chunk(text: str, size: int = 512, overlap: int = 64) -> List[str]:
    """Fixed-size chunking dengan overlap — untuk log files, kode, data tabular."""
    words = text.split()
    chunks, step = [], size - overlap
    for i in range(0, len(words), step):
        chunks.append(' '.join(words[i:i+size]))
        if i + size >= len(words):
            break
    return chunks


def paragraph_chunk(text: str, max_words: int = 300, min_words: int = 50) -> List[str]:
    """Paragraph-based chunking — untuk artikel, dokumentasi, FAQ."""
    paragraphs = [p.strip() for p in re.split(r'\n{2,}', text) if p.strip()]
    chunks, current, current_size = [], [], 0

    for para in paragraphs:
        size = len(para.split())
        if current_size + size > max_words and current_size >= min_words:
            chunks.append(' '.join(current))
            current, current_size = [], 0
        current.append(para)
        current_size += size

    if current:
        chunks.append(' '.join(current))
    return chunks

import numpy as np


def get_embedding(text: str, model_id: str = 'amazon.titan-embed-text-v2:0') -> List[float]:
    """Generate text embedding menggunakan Amazon Titan Embeddings V2."""
    payload = {
        'inputText': text[:8000],  # Titan V2 max 8K tokens
        'dimensions': 1024,
        'normalize': True  # Normalize untuk cosine similarity
    }
    response = bedrock.invoke_model(
        modelId=model_id,
        body=json.dumps(payload),
        contentType='application/json',
        accept='application/json'
    )
    return json.loads(response['body'].read())['embedding']


class VectorStore:
    """In-memory vector store. Di production, gunakan OpenSearch atau FAISS."""
    def __init__(self):
        self.docs, self.embeddings, self.metadata = [], [], []

    def add(self, text: str, meta: dict = None):
        self.docs.append(text)
        self.embeddings.append(get_embedding(text))
        self.metadata.append(meta or {})

    def search(self, query: str, top_k: int = 5) -> List[dict]:
        q_emb = get_embedding(query)
        scored = sorted(
            [(self._cosine_sim(q_emb, e), i) for i, e in enumerate(self.embeddings)],
            reverse=True
        )
        return [{'text': self.docs[i], 'score': s, 'metadata': self.metadata[i]}
                for s, i in scored[:top_k]]

    def _cosine_sim(self, v1, v2) -> float:
        a, b = np.array(v1), np.array(v2)
        norm = np.linalg.norm(a) * np.linalg.norm(b)
        return float(np.dot(a, b) / norm) if norm > 0 else 0.0

from rank_bm25 import BM25Okapi


class HybridSearchEngine:
    """Gabungkan BM25 dan semantic search untuk retrieval terbaik."""

    def __init__(self):
        self.docs = []
        self.embeddings = []
        self.bm25 = None

    def add_documents(self, documents: List[str]):
        self.docs = documents
        self.embeddings = [get_embedding(doc) for doc in documents]
        tokenized = [doc.lower().split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized)

    def search(self, query: str, top_k: int = 5, alpha: float = 0.5) -> List[dict]:
        """
        alpha=0.5: seimbang antara BM25 dan semantic
        alpha=0.0: pure BM25 (keyword)
        alpha=1.0: pure semantic
        """
        # BM25 scores
        bm25_scores = self.bm25.get_scores(query.lower().split())
        bm25_norm = self._normalize(bm25_scores)

        # Semantic scores
        q_emb = get_embedding(query)
        sem_scores = [self._cosine_sim(q_emb, e) for e in self.embeddings]
        sem_norm = self._normalize(sem_scores)

        # Hybrid fusion
        combined = [(1-alpha) * bm + alpha * sem
                    for bm, sem in zip(bm25_norm, sem_norm)]

        top_indices = sorted(range(len(combined)), key=lambda i: combined[i], reverse=True)[:top_k]
        return [{'text': self.docs[i], 'score': combined[i]} for i in top_indices]

    def _normalize(self, scores):
        min_s, max_s = min(scores), max(scores)
        if max_s == min_s:
            return [0.0] * len(scores)
        return [(s - min_s) / (max_s - min_s) for s in scores]

    def _cosine_sim(self, v1, v2) -> float:
        a, b = np.array(v1), np.array(v2)
        norm = np.linalg.norm(a) * np.linalg.norm(b)
        return float(np.dot(a, b) / norm) if norm > 0 else 0.0

def cached_analysis(document: str, question: str) -> str:
    """Analisis dokumen dengan prompt caching untuk efisiensi biaya."""
    response = bedrock.converse(
        modelId='anthropic.claude-sonnet-4-6-20250514-v1:0',
        system=[{
            'text': f'''Kamu adalah analis dokumen yang teliti.
Dokumen referensi yang akan kamu analisis:

{document}

Jawab semua pertanyaan berdasarkan dokumen di atas. Kutip bagian relevan.''',
            'cachePoint': {'type': 'default'}  # ← Cache system prompt + dokumen
        }],
        messages=[{'role': 'user', 'content': [{'text': question}]}],
        inferenceConfig={'maxTokens': 1024}
    )

    # Lihat token usage untuk memverifikasi cache hit
    usage = response['usage']
    cache_read = usage.get('cacheReadInputTokens', 0)
    cache_write = usage.get('cacheWriteInputTokens', 0)
    print(f'Cache read: {cache_read}, Cache write: {cache_write}')

    return response['output']['message']['content'][0]['text']


# Request pertama: cache MISS (write cache)
answer1 = cached_analysis(long_document, 'Apa kesimpulan utama dokumen ini?')

# Request berikutnya dengan dokumen yang sama: cache HIT (bayar 10%)
answer2 = cached_analysis(long_document, 'Apa rekomendasi konkretnya?')

def deep_analysis_with_thinking(problem: str, budget_tokens: int = 10000) -> dict:
    """Analisis kompleks dengan extended thinking."""
    payload = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": budget_tokens + 4000,
        "thinking": {
            "type": "enabled",
            "budget_tokens": budget_tokens  # Token yang dialokasikan untuk thinking
        },
        "messages": [{"role": "user", "content": problem}]
    }

    response = bedrock.invoke_model(
        modelId='anthropic.claude-sonnet-4-6-20250514-v1:0',
        body=json.dumps(payload),
        contentType='application/json',
        accept='application/json'
    )

    result = json.loads(response['body'].read())

    thinking_text = ''
    answer_text = ''

    for block in result['content']:
        if block['type'] == 'thinking':
            thinking_text = block['thinking']
        elif block['type'] == 'text':
            answer_text = block['text']

    return {
        'thinking': thinking_text,
        'answer': answer_text,
        'tokens_used': result['usage']
    }

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "BedrockInvokeSpecificModels",
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream"
      ],
      "Resource": [
        "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-sonnet-4-6-*",
        "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-haiku-4-5-*"
      ]
    }
  ]
}

import boto3
import json


def get_secret(secret_name: str, region: str = 'us-east-1') -> dict:
    """Ambil secrets dari AWS Secrets Manager."""
    client = boto3.client('secretsmanager', region_name=region)
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response['SecretString'])


# Penggunaan
secrets = get_secret('myapp/bedrock/config')

import boto3
from datetime import datetime


def log_bedrock_request(model_id: str, input_tokens: int,
                        output_tokens: int, latency_ms: float, success: bool):
    """Log metrics ke CloudWatch untuk monitoring dan cost tracking."""
    cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')

    cloudwatch.put_metric_data(
        Namespace='BedrockApp/Claude',
        MetricData=[
            {'MetricName': 'InputTokens', 'Value': input_tokens, 'Unit': 'Count',
             'Dimensions': [{'Name': 'ModelId', 'Value': model_id}]},
            {'MetricName': 'OutputTokens', 'Value': output_tokens, 'Unit': 'Count',
             'Dimensions': [{'Name': 'ModelId', 'Value': model_id}]},
            {'MetricName': 'Latency', 'Value': latency_ms, 'Unit': 'Milliseconds'},
            {'MetricName': 'SuccessCount', 'Value': 1 if success else 0, 'Unit': 'Count'}
        ]
    )