Anthropic Batch API: 50% diskon untuk async

Klasifikasi 50rb review produk Tokopedia per hari pakai Claude itu lumayan mahal kalau real-time. Padahal tidak ada user yang nungguin. Batch API kasih diskon 50% — sama pintarnya, sama akuratnya, cuma async. Snippet ini submit batch + poll status + parse hasil.

Kode

import json
import time
from dataclasses import dataclass
from typing import Optional

import anthropic

client = anthropic.Anthropic()


@dataclass
class BatchItem:
    custom_id: str  # ID kamu sendiri untuk match hasil ke input
    user_message: str
    system: Optional[str] = None


def submit_batch(items: list[BatchItem], model: str = "claude-opus-4-5") -> str:
    """Submit batch ke Anthropic. Return batch_id."""
    requests = []

    for item in items:
        params = {
            "model": model,
            "max_tokens": 256,
            "messages": [{"role": "user", "content": item.user_message}],
        }
        if item.system:
            params["system"] = item.system

        requests.append({
            "custom_id": item.custom_id,
            "params": params,
        })

    batch = client.messages.batches.create(requests=requests)
    return batch.id


def poll_batch(batch_id: str, poll_interval: int = 30) -> dict:
    """Poll status batch sampai selesai. Return final batch object."""
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        status = batch.processing_status

        counts = batch.request_counts
        print(
            f"[{status}] processing={counts.processing}, "
            f"succeeded={counts.succeeded}, errored={counts.errored}, "
            f"canceled={counts.canceled}, expired={counts.expired}"
        )

        if status == "ended":
            return batch

        time.sleep(poll_interval)


def fetch_results(batch) -> dict[str, str]:
    """Stream hasil JSONL, return dict custom_id -> response text."""
    hasil: dict[str, str] = {}

    for line in client.messages.batches.results(batch.id):
        custom_id = line.custom_id
        result = line.result

        if result.type == "succeeded":
            # Ambil text block pertama dari message
            text_blocks = [
                b.text for b in result.message.content if b.type == "text"
            ]
            hasil[custom_id] = "\n".join(text_blocks)
        elif result.type == "errored":
            hasil[custom_id] = f"[ERROR] {result.error.message}"
        else:
            hasil[custom_id] = f"[STATUS: {result.type}]"

    return hasil

Pemakaian

# Use case 1: klasifikasi sentiment 1000 review
review_list = [
    {"id": "rev-001", "text": "Barangnya bagus, pengiriman cepat banget"},
    {"id": "rev-002", "text": "Kemasan rusak parah, kecewa"},
    {"id": "rev-003", "text": "Produknya ok, biasa aja"},
    # ... 997 review lainnya
]

items = [
    BatchItem(
        custom_id=r["id"],
        user_message=r["text"],
        system="Klasifikasi review ke salah satu: POSITIF, NEGATIF, NETRAL. Jawab satu kata.",
    )
    for r in review_list
]

batch_id = submit_batch(items)
print(f"Batch submitted: {batch_id}")

batch = poll_batch(batch_id)
hasil = fetch_results(batch)

# Hasil siap dipakai
for rev_id, klasifikasi in hasil.items():
    print(f"{rev_id}: {klasifikasi}")

# Use case 2: summarization 500 artikel berita
artikel_list = load_artikel_from_db()  # dummy

items = [
    BatchItem(
        custom_id=f"art-{a['id']}",
        user_message=a["isi"][:8000],  # truncate kalau terlalu panjang
        system="Buat ringkasan 2 kalimat dalam Bahasa Indonesia.",
    )
    for a in artikel_list
]

batch_id = submit_batch(items, model="claude-opus-4-5")
batch = poll_batch(batch_id, poll_interval=60)
ringkasan = fetch_results(batch)

# Simpan ke DB
for art_id, summary in ringkasan.items():
    real_id = int(art_id.removeprefix("art-"))
    save_summary_to_db(real_id, summary)

Kapan dipakai

• Content moderation harian — review semua post baru dalam 24 jam.
• Summarization artikel bulk untuk newsletter.
• Klasifikasi / tagging product catalog.
• Translation bulk dokumen yang tidak urgent.
• Generate metadata SEO untuk seluruh inventaris produk.

Catatan

• custom_id wajib unique dalam satu batch. Pakai UUID atau composite ID untuk match hasil ke input record kamu.
• Tidak ada streaming di batch — response full sekaligus saat batch selesai.
• Cancellation — batches.cancel(batch_id) bisa cancel batch yang masih in_progress. Yang sudah processed tetap di-charge.
• Storage 30 hari — hasil disimpan 30 hari di Anthropic. Setelah itu hilang. Always fetch + simpan ke DB sendiri.
• Error per request — satu request fail tidak gagalkan batch. Cek result.type untuk handle errored items.
• Prompt caching tetap jalan — kalau pakai cache_control di system prompt, batch tetap dapat cache hit (diskon ganda).

Untuk job recurring (daily classification), schedule submit di cron job pagi-pagi, fetch hasil sore. Sekali setup, biaya AI bulanan bisa hemat 50%.