Claude 11 dengan Google Cloud Vertex AI: Panduan Lengkap untuk Developer Indonesia

# Step 1: Install dependencies
pip install 'anthropic[vertex]'  # SDK dengan Vertex AI support
pip install google-cloud-aiplatform

# Step 2: Enable API di GCP
gcloud services enable aiplatform.googleapis.com
gcloud config set project YOUR-PROJECT-ID

# Step 3: Enable Claude di Model Garden
# Buka: console.cloud.google.com/vertex-ai/model-garden
# Cari 'Claude' → Pilih model → Klik 'Enable'
# Perlu setujui Terms of Service Anthropic

# Step 4: Setup Authentication
# Development (local machine):
gcloud auth application-default login

# Production (Cloud Run/GCE/GKE):
# Credentials otomatis dari Service Account attached ke resource

gcloud iam service-accounts create claude-vertex-sa \
  --display-name='Claude Vertex AI Service Account'

gcloud projects add-iam-policy-binding YOUR-PROJECT \
  --member='serviceAccount:claude-vertex-sa@YOUR-PROJECT.iam.gserviceaccount.com' \
  --role='roles/aiplatform.user'

from anthropic import AnthropicVertex

client = AnthropicVertex(
    project_id='YOUR-PROJECT-ID',
    region='us-east5'  # Region utama Claude: us-east5, us-central1, europe-west1
)

msg = client.messages.create(
    model='claude-sonnet-4-6@20250514',
    max_tokens=50,
    messages=[{'role': 'user', 'content': 'Hello! Koneksi berhasil?'}]
)
print(msg.content[0].text)

import os
from anthropic import AnthropicVertex

client = AnthropicVertex(
    project_id=os.environ.get('GOOGLE_CLOUD_PROJECT', 'your-project-id'),
    region='us-east5'
)
MODEL = 'claude-sonnet-4-6@20250514'


def ask(prompt: str, max_tokens: int = 1024, temperature: float = 0.7) -> str:
    """Kirim single message ke Claude via Vertex AI."""
    msg = client.messages.create(
        model=MODEL,
        max_tokens=max_tokens,
        temperature=temperature,
        messages=[{'role': 'user', 'content': prompt}]
    )
    usage = msg.usage
    print(f'Tokens: in={usage.input_tokens}, out={usage.output_tokens}')
    return msg.content[0].text


def ask_with_system(user_message: str, system: str, temperature: float = 0.5) -> str:
    """Request dengan system prompt."""
    msg = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        temperature=temperature,
        system=system,
        messages=[{'role': 'user', 'content': user_message}]
    )
    return msg.content[0].text


# Contoh: GCP Architect assistant
result = ask_with_system(
    user_message='Bagaimana cara deploy Claude di Cloud Run dengan best practices?',
    system='''Kamu adalah Google Cloud Solutions Architect yang ahli AI/ML.
Berikan jawaban teknikal yang konkret dengan pertimbangan cost dan security.
Sertakan gcloud commands jika relevan.'''
)

def stream_response(prompt: str, system: str = None) -> str:
    """Stream response token per token — ideal untuk user-facing interfaces."""
    kwargs = {
        'model': MODEL,
        'max_tokens': 2048,
        'messages': [{'role': 'user', 'content': prompt}]
    }
    if system:
        kwargs['system'] = system

    collected = []
    with client.messages.stream(**kwargs) as stream:
        for text in stream.text_stream:
            print(text, end='', flush=True)
            collected.append(text)

    final = stream.get_final_message()
    print(f'\n\nTokens: in={final.usage.input_tokens}, out={final.usage.output_tokens}')
    return ''.join(collected)

from dataclasses import dataclass, field
from typing import List


@dataclass
class VertexChatbot:
    """Production chatbot dengan persistent conversation history."""
    system_prompt: str
    model: str = MODEL
    max_tokens: int = 2048
    temperature: float = 0.7
    max_history_turns: int = 10
    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': user_message})

        # Trim history untuk kontrol context window
        if len(self.history) > self.max_history_turns * 2:
            self.history = self.history[-(self.max_history_turns * 2):]

        msg = client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            temperature=self.temperature,
            system=self.system_prompt,
            messages=self.history
        )

        response_text = msg.content[0].text
        self.history.append({'role': 'assistant', 'content': response_text})
        self.total_input_tokens += msg.usage.input_tokens
        self.total_output_tokens += msg.usage.output_tokens
        return response_text

    @property
    def estimated_cost_usd(self) -> float:
        return (self.total_input_tokens * 3 + self.total_output_tokens * 15) / 1_000_000


# Demo
bot = VertexChatbot(system_prompt='Kamu adalah GCP expert. Jawab teknikal, ringkas, dengan contoh konkret.')
print(bot.chat('Apa perbedaan Cloud Run vs Cloud Functions?'))
print(bot.chat('Kapan sebaiknya saya pilih Cloud Run?'))  # Context terjaga
print(f'Estimasi biaya: ${bot.estimated_cost_usd:.4f}')

def analyze_with_xml(document: str, criteria: str) -> str:
    """XML tags membantu model membedakan instruksi dari data."""
    prompt = f'''Analisis dokumen berdasarkan kriteria yang diberikan.

<document>
{document}
</document>

<criteria>{criteria}</criteria>

<output_format>
## Temuan Utama
## Analisis Detail
## Rekomendasi
</output_format>

Berikan analisis yang spesifik dan berbasis bukti dari dokumen.'''

    return ask(prompt)

def classify_support_ticket(ticket: str) -> dict:
    """Few-shot examples meningkatkan konsistensi output secara signifikan."""
    import json
    prompt = f'''Klasifikasikan tiket customer support.

<examples>
Input: 'Paket belum tiba setelah 2 minggu, no tracking sudah tidak update'
Output: {{"type": "pengiriman", "priority": "high", "dept": "logistics"}}

Input: 'Cara ubah metode pembayaran default di akun saya?'
Output: {{"type": "akun", "priority": "low", "dept": "support"}}

Input: 'Aplikasi crash saat buka halaman produk, sudah coba reinstall'
Output: {{"type": "teknis", "priority": "high", "dept": "engineering"}}
</examples>

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

Respons HANYA dengan JSON, tanpa penjelasan.'''

    return json.loads(ask(prompt, temperature=0.0))

from typing import Callable, List
import json


def generate_test_cases(task: str, n: int = 25) -> List[dict]:
    """Claude menghasilkan test cases yang beragam untuk dirinya sendiri."""
    prompt = f'''Buat {n} test cases untuk tugas: {task}

Distribusi: 60% kasus normal, 20% edge cases, 20% ambiguous cases.

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

Respond ONLY dengan JSON array.'''

    return json.loads(ask(prompt, temperature=0.4))


def evaluate(
    prompt_template: str,
    test_cases: List[dict],
    grader: Callable,
    verbose: bool = True
) -> dict:
    results = []
    for i, case in enumerate(test_cases):
        response = ask(prompt_template.format(input=case['input']), temperature=0.0)
        passed = grader(response, case['expected'])
        results.append({'case': case, 'response': response, 'passed': passed})
        if verbose:
            print(f"[{i+1}/{len(test_cases)}] {'✅' if passed else '❌'} {case['input'][:40]}...")

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


# Bandingkan dua versi prompt
cases = generate_test_cases('mengklasifikasikan tiket support', n=30)

v1 = 'Klasifikasikan tiket support ini: {input}. Output: JSON dengan type dan priority.'
v2 = '''Kamu adalah support classifier. Klasifikasikan tiket berikut.
Berikan: type (billing/teknis/pengiriman/akun), priority (low/medium/high), dept.
Tiket: {input}
Output JSON: {{"type":..., "priority":..., "dept":...}}'''

r1 = evaluate(v1, cases, json_grader)
r2 = evaluate(v2, cases, json_grader)
print(f'v1: {r1["accuracy"]:.1%} | v2: {r2["accuracy"]:.1%}')
print(f'Improvement: {(r2["accuracy"]-r1["accuracy"])*100:.1f} percentage points')

GCP_TOOLS = [
    {
        'name': 'query_bigquery',
        'description': '''Jalankan SQL query di BigQuery dan kembalikan hasil.
Gunakan untuk analisis data, business intelligence, dan reporting.
Selalu gunakan query yang efisien dengan LIMIT yang tepat.''',
        'input_schema': {
            'type': 'object',
            'properties': {
                'sql': {'type': 'string', 'description': 'SQL query yang valid untuk BigQuery'},
                'limit': {'type': 'integer', 'description': 'Maksimum baris. Default: 100', 'default': 100}
            },
            'required': ['sql']
        }
    },
    {
        'name': 'read_gcs_file',
        'description': 'Baca konten file dari Google Cloud Storage bucket.',
        'input_schema': {
            'type': 'object',
            'properties': {
                'bucket': {'type': 'string', 'description': 'Nama GCS bucket'},
                'path': {'type': 'string', 'description': 'Path file dalam bucket'}
            },
            'required': ['bucket', 'path']
        }
    }
]

from google.cloud import bigquery, storage
import json

bq_client = bigquery.Client()
gcs_client = storage.Client()


def query_bigquery(sql: str, limit: int = 100) -> str:
    """Jalankan BigQuery SQL dan kembalikan hasil sebagai JSON."""
    query = f'{sql} LIMIT {limit}' if 'LIMIT' not in sql.upper() else sql
    try:
        df = bq_client.query(query).to_dataframe()
        return df.to_json(orient='records', force_ascii=False)
    except Exception as e:
        return json.dumps({'error': str(e)})


def read_gcs_file(bucket: str, path: str) -> str:
    """Baca file dari GCS."""
    try:
        blob = gcs_client.bucket(bucket).blob(path)
        content = blob.download_as_text()
        return content[:10000]  # Limit 10K chars
    except Exception as e:
        return json.dumps({'error': str(e)})


TOOL_REGISTRY = {
    'query_bigquery': query_bigquery,
    'read_gcs_file': read_gcs_file
}

def gcp_analyst_agent(user_query: str, max_iter: int = 8) -> str:
    """Agent yang bisa query BigQuery dan baca GCS untuk menjawab pertanyaan data."""
    messages = [{'role': 'user', 'content': user_query}]
    system = '''Kamu adalah GCP Data Analyst.
Gunakan tools yang tersedia untuk mendapatkan data dari BigQuery dan GCS.
Analisis data secara mendalam dan berikan insight bisnis yang actionable.
Selalu verifikasi data sebelum membuat kesimpulan.'''

    for iteration in range(max_iter):
        msg = client.messages.create(
            model=MODEL,
            max_tokens=4096,
            system=system,
            messages=messages,
            tools=GCP_TOOLS
        )

        messages.append({'role': 'assistant', 'content': msg.content})

        # Model selesai
        if msg.stop_reason == 'end_turn':
            for block in msg.content:
                if hasattr(block, 'text'):
                    return block.text

        # Model mau pakai tools
        elif msg.stop_reason == 'tool_use':
            tool_results = []
            for block in msg.content:
                if block.type == 'tool_use':
                    print(f'[Iter {iteration+1}] Calling: {block.name}({block.input})')
                    try:
                        result = TOOL_REGISTRY[block.name](**block.input)
                    except Exception as e:
                        result = json.dumps({'error': str(e)})

                    tool_results.append({
                        'type': 'tool_result',
                        'tool_use_id': block.id,
                        'content': result
                    })

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

    return 'Max iterations reached'


# Contoh penggunaan
result = gcp_analyst_agent(
    'Analisis penjualan produk bulan lalu dari BigQuery. '
    'Tampilkan top 5 produk dan tren mingguannya.'
)
print(result)

def batch_classify(items: list, categories: list) -> list:
    """Klasifikasikan beberapa item dalam satu request — lebih hemat dari N calls."""
    classify_tool = {
        'name': 'classify_item',
        'description': f'Klasifikasikan item ke salah satu: {categories}',
        'input_schema': {
            'type': 'object',
            'properties': {
                'item_id': {'type': 'integer'},
                'classification': {'type': 'string', 'enum': categories},
                'confidence': {'type': 'number', 'minimum': 0, 'maximum': 1}
            },
            'required': ['item_id', 'classification', 'confidence']
        }
    }

    items_text = '\n'.join(f'{i}. {item}' for i, item in enumerate(items, 1))
    prompt = f'Klasifikasikan setiap item ke {categories}:\n{items_text}'

    msg = client.messages.create(
        model=MODEL,
        max_tokens=2048,
        temperature=0.0,
        tools=[classify_tool],
        messages=[{'role': 'user', 'content': prompt}]
    )

    return [b.input for b in msg.content if b.type == 'tool_use']

import numpy as np
import vertexai
from vertexai.language_models import TextEmbeddingModel
from typing import List

vertexai.init(project='YOUR-PROJECT-ID', location='us-central1')
embed_model = TextEmbeddingModel.from_pretrained('text-embedding-005')


def get_embedding(text: str) -> list:
    """Generate embedding menggunakan Vertex AI text-embedding-005."""
    embeddings = embed_model.get_embeddings([text])
    return embeddings[0].values


def get_batch_embeddings(texts: list, batch_size: int = 250) -> list:
    """Batch embedding — Vertex AI support hingga 250 items per call."""
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        embs = embed_model.get_embeddings(batch)
        all_embeddings.extend([e.values for e in embs])
    return all_embeddings


def cosine_similarity(v1: list, v2: list) -> float:
    a, b = np.array(v1), np.array(v2)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-10))

import re
from rank_bm25 import BM25Okapi


class HybridSearch:
    def __init__(self, chunks: List[str], alpha: float = 0.6):
        """alpha=0.6: lebih berat ke semantic; alpha=0.0: pure BM25"""
        self.chunks, self.alpha = chunks, alpha
        self.bm25 = BM25Okapi([re.findall(r'\w+', c.lower()) for c in chunks])
        print(f'Building {len(chunks)} embeddings...')
        self.embeddings = get_batch_embeddings(chunks)

    def search(self, query: str, k: int = 5) -> List[dict]:
        tokens = re.findall(r'\w+', query.lower())
        bm25_s = self.bm25.get_scores(tokens)
        q_emb = get_embedding(query)
        sem_s = [cosine_similarity(q_emb, e) for e in self.embeddings]

        def norm(arr):
            mn, mx = min(arr), max(arr)
            return [(x - mn) / (mx - mn + 1e-8) for x in arr]

        combined = [(1 - self.alpha) * b + self.alpha * s
                    for b, s in zip(norm(bm25_s), norm(sem_s))]
        top = sorted(range(len(combined)), key=lambda i: combined[i], reverse=True)[:k]
        return [{'text': self.chunks[i], 'score': combined[i]} for i in top]


def rag_answer(question: str, search: HybridSearch, k: int = 4) -> str:
    chunks = search.search(question, k=k)
    context = '\n\n---\n\n'.join(c['text'] for c in chunks)
    prompt = f'''Jawab pertanyaan HANYA dari konteks. Jika tidak ada, katakan demikian.

<context>
{context}
</context>

Pertanyaan: {question}'''

    return ask_with_system(prompt, 'Kamu adalah asisten berbasis dokumen yang akurat.')

def contextualize_chunk(chunk: str, doc_summary: str) -> str:
    """Tambahkan konteks dokumen ke chunk."""
    prompt = f'''Dokumen ini berisi: {doc_summary}

Berikan konteks singkat (2-3 kalimat) yang menjelaskan bagaimana chunk ini
berhubungan dengan dokumen secara keseluruhan. Fokus pada informasi yang
membantu retrieval, bukan ringkasan ulang.

<chunk>{chunk}</chunk>

Respond hanya dengan teks konteks, tanpa penjelasan tambahan.'''

    return ask(prompt, temperature=0.0) + '\n\n' + chunk


def build_contextual_index(documents: dict) -> HybridSearch:
    """Build RAG index dengan contextual enrichment per chunk."""
    all_chunks = []
    for name, text in documents.items():
        summary = ask(f'Ringkas dokumen {name} dalam 2 kalimat: {text[:2000]}')
        for chunk in paragraph_chunk(text):
            enriched = contextualize_chunk(chunk, f'{name}: {summary}')
            all_chunks.append(enriched)
    return HybridSearch(all_chunks)

def bigquery_rag(question: str, table: str) -> str:
    """Gabungkan BigQuery data dengan Claude untuk data Q&A."""
    # 1. Generate SQL dari pertanyaan natural language
    sql = ask(f'Buat SQL BigQuery untuk: {question}. Tabel: {table}. Output SQL saja.')

    # 2. Eksekusi query
    data = query_bigquery(sql)

    # 3. Claude analisis hasilnya
    analysis_prompt = f'''Analisis data berikut dan jawab: {question}
Data (JSON): {data}
Berikan insight yang konkret dan actionable.'''

    return ask(analysis_prompt)

def cached_analysis_session(document: str, questions: list) -> list:
    """Analisis dokumen yang sama dengan beberapa pertanyaan — hemat biaya."""
    answers = []
    for question in questions:
        msg = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=[{
                'type': 'text',
                'text': f'''Kamu adalah analis dokumen yang teliti.
Dokumen referensi:

{document}

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

        # Log untuk verifikasi cache hit
        usage = msg.usage
        cache_read = getattr(usage, 'cache_read_input_tokens', 0)
        cache_write = getattr(usage, 'cache_creation_input_tokens', 0)
        print(f'Q: {question[:50]}... | Cache read: {cache_read}, write: {cache_write}')

        answers.append(msg.content[0].text)

    return answers


# Request pertama: cache MISS (write cache)
# Request berikutnya dengan dokumen yang sama: cache HIT (bayar 10%)
long_doc = "... dokumen panjang ..."
questions = [
    'Apa kesimpulan utama dokumen ini?',
    'Apa rekomendasi konkretnya?',
    'Siapa stakeholder yang terdampak?'
]
answers = cached_analysis_session(long_doc, questions)

def deep_analysis(problem: str, budget_tokens: int = 10000) -> dict:
    """Analisis kompleks dengan extended thinking."""
    msg = client.messages.create(
        model=MODEL,
        max_tokens=budget_tokens + 4000,
        thinking={
            'type': 'enabled',
            'budget_tokens': budget_tokens
        },
        messages=[{'role': 'user', 'content': problem}]
    )

    thinking_text = ''
    answer_text = ''

    for block in msg.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': {'input': msg.usage.input_tokens, 'output': msg.usage.output_tokens}
    }


# Cocok untuk:
# - Analisis arsitektur sistem yang kompleks
# - Keputusan strategis bisnis
# - Debugging masalah teknikal yang sulit
# - Review keamanan yang mendalam
result = deep_analysis(
    'Evaluasi arsitektur microservices kami dan identifikasi risiko scalability...',
    budget_tokens=15000
)
print(result['answer'])

# Gunakan IAM Conditions untuk membatasi akses ke model tertentu
# Policy IAM yang disarankan:
policy_document = {
    "bindings": [
        {
            "role": "roles/aiplatform.user",
            "members": ["serviceAccount:claude-vertex-sa@PROJECT.iam.gserviceaccount.com"],
            "condition": {
                "title": "Only Claude models",
                "expression": "resource.name.startsWith('projects/PROJECT/locations/us-east5/publishers/anthropic')"
            }
        }
    ]
}

# Di Cloud Run: gunakan --service-account flag
# gcloud run deploy my-service --service-account=claude-vertex-sa@PROJECT.iam.gserviceaccount.com

import logging
from google.cloud import logging as cloud_logging

cloud_logger = cloud_logging.Client()

def log_claude_usage(model: str, input_tokens: int, output_tokens: int, latency_ms: float):
    logger = cloud_logger.logger('claude-vertex-usage')
    logger.log_struct({
        'model': model,
        'input_tokens': input_tokens,
        'output_tokens': output_tokens,
        'latency_ms': latency_ms,
        'estimated_cost_usd': (input_tokens * 3 + output_tokens * 15) / 1_000_000
    })