Claude Batch API — Xử lý hàng loạt với chi phí giảm 50%

Khi bạn cần xử lý hàng trăm hoặc hàng nghìn request tới Claude API, việc gửi từng request một không chỉ chậm mà còn tốn kém. Anthropic cung cấp Message Batches API cho phép bạn gửi nhiều request cùng lúc và nhận kết quả trong vòng 24 giờ, với mức giá chỉ bằng 50% so với API thông thường. Bài viết này hướng dẫn chi tiết cách triển khai Batch API trong thực tế.

Batch API là gì?

Message Batches API là endpoint bất đồng bộ (asynchronous) của Claude API, cho phép bạn gửi một tập hợp nhiều request cùng một lúc thay vì gửi từng request riêng lẻ. Hệ thống sẽ xử lý toàn bộ batch và trả kết quả trong thời gian tối đa 24 giờ.

Các đặc điểm chính của Batch API:

• Giảm 50% chi phí: mỗi token input và output đều được tính với giá bằng một nửa so với API đồng bộ (synchronous)
• Thời gian xử lý tối đa 24 giờ: hệ thống cam kết hoàn thành trong khung thời gian này, thực tế thường nhanh hơn nhiều
• Tối đa 100.000 request mỗi batch: đủ cho hầu hết các tác vụ xử lý hàng loạt
• Hỗ trợ tất cả các model: Claude Opus, Sonnet, Haiku đều có thể dùng Batch API
• Kết quả trả về theo từng request: mỗi request trong batch có kết quả riêng, bao gồm cả trường hợp lỗi

Khi nào nên dùng Batch API?

Batch API phù hợp với các tác vụ không yêu cầu phản hồi ngay lập tức. Dưới đây là những use case phổ biến nhất:

1. Sinh nội dung hàng loạt

Bạn cần tạo mô tả sản phẩm cho 5.000 SKU, viết meta description cho 2.000 trang web, hoặc dịch 500 bài viết sang nhiều ngôn ngữ. Thay vì chờ từng request hoàn thành, bạn gửi toàn bộ vào một batch và nhận kết quả sau vài giờ.

2. Phân tích và phân loại dữ liệu

Phân loại 10.000 đánh giá khách hàng theo sentiment, trích xuất thông tin từ hàng nghìn hợp đồng, hoặc gắn nhãn dữ liệu training cho model machine learning. Đây là những tác vụ vốn không cần kết quả tức thì.

3. Đánh giá và kiểm duyệt nội dung

Kiểm tra hàng nghìn bình luận, đánh giá chất lượng bài viết, hoặc phát hiện nội dung vi phạm chính sách. Batch API giúp bạn xử lý khối lượng lớn với chi phí thấp.

4. Tạo dữ liệu tổng hợp (synthetic data)

Khi cần tạo dữ liệu huấn luyện, test case, hoặc dữ liệu mẫu cho phát triển ứng dụng, Batch API là lựa chọn kinh tế nhất.

Triển khai Batch API với Python

Dưới đây là quy trình hoàn chỉnh từ tạo batch đến xử lý kết quả.

Cài đặt và cấu hình

Trước tiên, cài đặt thư viện Anthropic Python SDK:

pip install anthropic

Bước 1: Chuẩn bị danh sách request

Mỗi request trong batch cần có một custom_id duy nhất để bạn có thể ghép kết quả với input sau khi batch hoàn thành:

import anthropic

client = anthropic.Anthropic()

# Danh sach san pham can tao mo ta
products = [
    {"id": "prod_001", "name": "Ao thun cotton", "category": "Thoi trang"},
    {"id": "prod_002", "name": "Tai nghe Bluetooth", "category": "Cong nghe"},
    {"id": "prod_003", "name": "Binh giu nhiet 500ml", "category": "Gia dung"},
    # ... hang nghin san pham khac
]

# Tao danh sach request cho batch
requests = []
for product in products:
    requests.append({
        "custom_id": product["id"],
        "params": {
            "model": "claude-sonnet-4-20250514",
            "max_tokens": 1024,
            "messages": [
                {
                    "role": "user",
                    "content": f"Viet mo ta san pham ngan gon (100-150 tu) "
                               f"cho san pham: {product['name']}, "
                               f"danh muc: {product['category']}. "
                               f"Viet bang tieng Viet, gong dieu than thien, "
                               f"tap trung vao loi ich cho nguoi dung."
                }
            ]
        }
    })

Bước 2: Tạo batch

Gửi danh sách request tới Batch API:

# Tao batch
batch = client.messages.batches.create(requests=requests)

print(f"Batch ID: {batch.id}")
print(f"Trang thai: {batch.processing_status}")
print(f"So request: {batch.request_counts.total}")
# Output:
# Batch ID: msgbatch_01HG5K...
# Trang thai: in_progress
# So request: 3

Bước 3: Theo dõi tiến trình

Batch API cung cấp thông tin chi tiết về tiến trình xử lý. Bạn có thể polling hoặc dùng webhook:

import time

def wait_for_batch(client, batch_id, poll_interval=30):
    """Cho doi batch hoan thanh voi polling."""
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        counts = batch.request_counts

        print(f"Tien trinh: {counts.succeeded + counts.errored + counts.expired}"
              f"/{counts.total} "
              f"(thanh cong: {counts.succeeded}, "
              f"loi: {counts.errored}, "
              f"het han: {counts.expired})")

        if batch.processing_status == "ended":
            print("Batch da hoan thanh!")
            return batch

        time.sleep(poll_interval)

# Su dung
completed_batch = wait_for_batch(client, batch.id)

Bước 4: Xử lý kết quả

Sau khi batch hoàn thành, lấy kết quả và xử lý từng response:

def process_batch_results(client, batch_id):
    """Xu ly ket qua batch va phan loai theo trang thai."""
    results = {
        "succeeded": [],
        "errored": [],
        "expired": []
    }

    for result in client.messages.batches.results(batch_id):
        custom_id = result.custom_id
        result_type = result.result.type

        if result_type == "succeeded":
            message = result.result.message
            content = message.content[0].text
            results["succeeded"].append({
                "id": custom_id,
                "content": content,
                "input_tokens": message.usage.input_tokens,
                "output_tokens": message.usage.output_tokens
            })
        elif result_type == "errored":
            error = result.result.error
            results["errored"].append({
                "id": custom_id,
                "error_type": error.type,
                "error_message": error.message
            })
        elif result_type == "expired":
            results["expired"].append({"id": custom_id})

    return results

# Su dung
results = process_batch_results(client, batch.id)
print(f"Thanh cong: {len(results['succeeded'])}")
print(f"Loi: {len(results['errored'])}")
print(f"Het han: {len(results['expired'])}")

Xử lý lỗi và retry cho batch

Trong thực tế, một số request trong batch có thể thất bại. Bạn cần có chiến lược retry cho những request lỗi:

def retry_failed_requests(client, results, original_requests, max_retries=3):
    """Gui lai cac request that bai trong batch moi."""
    failed_ids = set()
    for item in results["errored"]:
        failed_ids.add(item["id"])
    for item in results["expired"]:
        failed_ids.add(item["id"])

    if not failed_ids:
        print("Khong co request nao can retry.")
        return results

    # Loc ra cac request can gui lai
    retry_requests = [
        req for req in original_requests
        if req["custom_id"] in failed_ids
    ]

    print(f"Retry {len(retry_requests)} request...")

    for attempt in range(max_retries):
        retry_batch = client.messages.batches.create(requests=retry_requests)
        completed = wait_for_batch(client, retry_batch.id)
        retry_results = process_batch_results(client, retry_batch.id)

        # Cap nhat ket qua
        results["succeeded"].extend(retry_results["succeeded"])

        # Kiem tra con request loi khong
        remaining_failed = retry_results["errored"] + retry_results["expired"]
        if not remaining_failed:
            print(f"Tat ca request da thanh cong sau {attempt + 1} lan retry.")
            break

        failed_ids = set(item["id"] for item in remaining_failed)
        retry_requests = [
            req for req in retry_requests
            if req["custom_id"] in failed_ids
        ]

    return results

So sánh chi phí: Batch API vs Real-time API

Dưới đây là bảng so sánh chi phí cho một tác vụ điển hình: xử lý 10.000 request, mỗi request trung bình 500 token input và 300 token output.

Với model Claude Opus, mức chênh lệch còn lớn hơn vì đơn giá cao hơn, nhưng tỷ lệ giảm vẫn là 50%.

Ví dụ thực tế: Phân loại đánh giá khách hàng

Giả sử bạn có 5.000 đánh giá sản phẩm cần phân loại theo sentiment (tích cực, tiêu cực, trung tính) và trích xuất các chủ đề chính:

import json

def create_review_classification_batch(client, reviews):
    """Tao batch phan loai danh gia khach hang."""
    system_prompt = (
        "Ban la chuyen gia phan tich feedback khach hang. "
        "Voi moi danh gia, hay tra ve JSON voi format:
"
        '{"sentiment": "positive|negative|neutral", '
        '"topics": ["topic1", "topic2"], '
        '"summary": "tom tat 1 cau", '
        '"action_required": true|false}'
    )

    requests = []
    for review in reviews:
        requests.append({
            "custom_id": f"review_{review['id']}",
            "params": {
                "model": "claude-haiku-3-5-20241022",
                "max_tokens": 256,
                "messages": [
                    {"role": "user", "content": review["text"]}
                ],
                "system": system_prompt
            }
        })

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

# Sau khi batch hoan thanh, parse ket qua
def parse_classification_results(client, batch_id):
    """Parse ket qua phan loai thanh structured data."""
    classifications = []
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            text = result.result.message.content[0].text
            try:
                data = json.loads(text)
                data["review_id"] = result.custom_id
                classifications.append(data)
            except json.JSONDecodeError:
                print(f"Loi parse JSON cho {result.custom_id}")

    return classifications

Ví dụ thực tế: Dịch nội dung đa ngôn ngữ

Khi cần dịch toàn bộ nội dung website sang nhiều ngôn ngữ, Batch API giúp tiết kiệm đáng kể:

def create_translation_batch(client, contents, target_languages):
    """Tao batch dich noi dung sang nhieu ngon ngu."""
    requests = []
    for content in contents:
        for lang in target_languages:
            requests.append({
                "custom_id": f"{content['id']}_{lang}",
                "params": {
                    "model": "claude-sonnet-4-20250514",
                    "max_tokens": 4096,
                    "messages": [
                        {
                            "role": "user",
                            "content": (
                                f"Dich doan van sau sang {lang}. "
                                f"Giu nguyen format HTML neu co. "
                                f"Dam bao ban dich tu nhien, "
                                f"khong dich mot-mot:

"
                                f"{content['text']}"
                            )
                        }
                    ]
                }
            })

    # Chia thanh nhieu batch neu vuot qua 100K request
    batch_size = 100000
    batches = []
    for i in range(0, len(requests), batch_size):
        chunk = requests[i:i + batch_size]
        batch = client.messages.batches.create(requests=chunk)
        batches.append(batch)
        print(f"Da tao batch {batch.id} voi {len(chunk)} request")

    return batches

Khi nào chọn Batch API, khi nào chọn Real-time API?

Quyết định giữa hai phương thức phụ thuộc vào bốn yếu tố chính:

Chọn Real-time API khi:

• Người dùng đang chờ phản hồi (chat, search, autocomplete)
• Kết quả cần trong vòng vài giây
• Số lượng request nhỏ (dưới 100 request mỗi lần)
• Cần streaming response (hiển thị từng chữ)
• Logic ứng dụng phụ thuộc vào kết quả request trước để tạo request sau

Chọn Batch API khi:

• Không cần kết quả ngay lập tức (có thể chờ vài giờ)
• Số lượng request lớn (hàng trăm đến hàng trăm nghìn)
• Tối ưu chi phí là ưu tiên hàng đầu
• Các request độc lập với nhau (kết quả request A không ảnh hưởng request B)
• Chạy trong pipeline ETL, cron job, hoặc quy trình backend

Kết hợp cả hai

Nhiều hệ thống sử dụng hybrid approach: Real-time API cho trải nghiệm người dùng trực tiếp, Batch API cho các tác vụ nền chạy hàng đêm. Ví dụ, một sàn thương mại điện tử có thể dùng Real-time API cho chatbot hỗ trợ khách hàng ban ngày, và Batch API để phân tích toàn bộ đánh giá sản phẩm mới vào ban đêm.

Best practices khi dùng Batch API

1. Thiết kế custom_id có ý nghĩa

Custom ID giúp bạn ghép kết quả với dữ liệu gốc. Nên dùng format rõ ràng và có thể parse được:

# Tot: co the parse de ghep voi database
custom_id = f"product_{product_id}_description_{language}"

# Khong tot: khong biet thuoc du lieu nao
custom_id = f"req_{uuid4()}"

2. Chia batch lớn thành nhiều batch nhỏ

Mặc dù giới hạn là 100.000 request mỗi batch, nên chia thành các batch 10.000-20.000 request để dễ theo dõi và retry khi có lỗi.

3. Lưu batch_id và trạng thái vào database

Trong môi trường production, luôn lưu batch_id vào database để có thể truy vấn lại kết quả nếu ứng dụng bị crash hoặc restart.

4. Xử lý idempotency

Nếu batch bị timeout hoặc lỗi, bạn cần đảm bảo việc retry không tạo ra dữ liệu trùng lặp. Sử dụng custom_id để kiểm tra request nào đã có kết quả trước khi gửi lại.

5. Monitor chi phí

Theo dõi usage từ kết quả batch để tính tổng chi phí thực tế:

def calculate_batch_cost(results, model="claude-sonnet-4-20250514"):
    """Tinh chi phi thuc te cua batch."""
    # Gia Batch API (50% gia thuong)
    pricing = {
        "claude-sonnet-4-20250514": {
            "input": 1.5 / 1_000_000,
            "output": 7.5 / 1_000_000
        },
        "claude-haiku-3-5-20241022": {
            "input": 0.4 / 1_000_000,
            "output": 2.0 / 1_000_000
        }
    }

    total_input = sum(r["input_tokens"] for r in results["succeeded"])
    total_output = sum(r["output_tokens"] for r in results["succeeded"])

    cost_input = total_input * pricing[model]["input"]
    cost_output = total_output * pricing[model]["output"]

    print(f"Input tokens: {total_input:,}")
    print(f"Output tokens: {total_output:,}")
    print(f"Chi phi input: ${cost_input:.4f}")
    print(f"Chi phi output: ${cost_output:.4f}")
    print(f"Tong chi phi: ${cost_input + cost_output:.4f}")

    return cost_input + cost_output

Hủy batch đang xử lý

Nếu bạn phát hiện lỗi trong dữ liệu đầu vào hoặc cần dừng batch vì lý do nào đó, bạn có thể hủy batch:

# Huy batch
client.messages.batches.cancel(batch_id)

# Luu y: cac request da xu ly xong se khong bi huy
# Chi nhung request chua bat dau moi bi huy

Sau khi hủy, bạn vẫn có thể lấy kết quả của các request đã hoàn thành trước thời điểm hủy.

Liệt kê và quản lý batch

Bạn có thể liệt kê tất cả các batch đã tạo để theo dõi và quản lý:

# Liet ke cac batch gan day
batches = client.messages.batches.list(limit=20)
for b in batches:
    print(f"ID: {b.id} | Status: {b.processing_status} | "
          f"Total: {b.request_counts.total} | "
          f"Succeeded: {b.request_counts.succeeded}")

Bước tiếp theo

Batch API là công cụ mạnh mẽ để tối ưu chi phí khi làm việc với Claude API ở quy mô lớn. Kết hợp với Prompt Caching để giảm thêm chi phí input token, hoặc sử dụng model cascade (bắt đầu với Haiku cho tác vụ đơn giản, nâng lên Sonnet cho tác vụ phức tạp) để tối ưu hóa ngân sách AI của bạn. Khám phá thêm tại Thư viện Ứng dụng Claude.