Batch Processing — Xử lý hàng loạt request với Claude API

Bạn đang xử lý hàng nghìn tài liệu, phân loại hàng loạt email, hay chạy evaluations trên dataset lớn? Thay vì gọi API từng request một và chờ đợi, Message Batches API của Anthropic cho phép bạn gửi tối đa 10.000 request trong một lần, xử lý bất đồng bộ, và tiết kiệm đến 50% chi phí.

Bài viết này hướng dẫn toàn diện về Batch Processing — từ cách tạo batch, theo dõi trạng thái, lấy kết quả, đến xử lý lỗi và các kỹ thuật nâng cao.

Tại sao cần Batch Processing?

Khi xử lý dữ liệu quy mô lớn, bạn thường gặp ba vấn đề:

• Tốc độ: Gọi API tuần tự (sequential) mất rất nhiều thời gian — 1.000 request × 2 giây/request = gần 34 phút
• Rate limits: Gọi song song (concurrent) thường bị throttle ở tốc độ cao
• Chi phí: Standard API pricing không có discount cho volume lớn

Message Batches API giải quyết cả ba: bạn submit tất cả cùng lúc, Anthropic xử lý trong nền, và bạn trả 50% giá so với standard API.

Giới hạn và điều kiện

• Tối đa 10.000 requests mỗi batch
• Kích thước batch tối đa: 32 MB
• Thời gian xử lý: thường hoàn thành trong vòng 24 giờ
• Kết quả được lưu trong 29 ngày sau khi tạo
• Batch không thể bị hủy sau khi 60% requests đã xử lý

Bước 1: Tạo một Batch

Mỗi request trong batch cần có một custom_id duy nhất để bạn map kết quả về sau:

import anthropic

client = anthropic.Anthropic()

# Tạo danh sách requests
requests = []
documents = [
    {"id": "doc_001", "text": "Sản phẩm này rất tốt, tôi rất hài lòng!"},
    {"id": "doc_002", "text": "Giao hàng chậm và đóng gói kém chất lượng."},
    {"id": "doc_003", "text": "Giá cả hợp lý, chất lượng tạm ổn."},
    # ... thêm hàng nghìn documents
]

for doc in documents:
    requests.append(
        anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
            custom_id=doc["id"],
            params={
                "model": "claude-haiku-4-5",
                "max_tokens": 100,
                "messages": [
                    {
                        "role": "user",
                        "content": f"Phân loại cảm xúc: POSITIVE, NEGATIVE, hay NEUTRAL?

Text: {doc['text']}

Chỉ trả lời một từ."
                    }
                ]
            }
        )
    )

# Tạo batch
batch = client.messages.batches.create(requests=requests)
print(f"Batch ID: {batch.id}")
print(f"Trạng thái: {batch.processing_status}")
print(f"Tổng requests: {batch.request_counts.processing}")

API trả về ngay lập tức với batch.id — bạn dùng ID này để theo dõi và lấy kết quả.

Bước 2: Theo dõi trạng thái

Batch có thể mất từ vài phút đến vài giờ tùy kích thước. Có hai cách theo dõi:

Polling thủ công

import time

def wait_for_batch(batch_id, poll_interval=60):
    """Chờ batch hoàn thành, poll mỗi poll_interval giây."""
    while True:
        batch = client.messages.batches.retrieve(batch_id)

        counts = batch.request_counts
        total = counts.processing + counts.succeeded + counts.errored + counts.canceled + counts.expired

        print(f"[{batch.processing_status}] "
              f"Đang xử lý: {counts.processing}/{total} | "
              f"Thành công: {counts.succeeded} | "
              f"Lỗi: {counts.errored}")

        if batch.processing_status == "ended":
            print("Batch hoàn thành!")
            return batch

        time.sleep(poll_interval)

batch = wait_for_batch(batch.id, poll_interval=30)

Kiểm tra status counts

Trường request_counts cho biết chi tiết:

Bước 3: Lấy kết quả

Sau khi batch ended, dùng stream_results() để iterate qua từng kết quả:

results = {}

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

    if result.result.type == "succeeded":
        message = result.result.message
        text = message.content[0].text
        results[custom_id] = {
            "status": "success",
            "output": text,
            "input_tokens": message.usage.input_tokens,
            "output_tokens": message.usage.output_tokens,
        }
    elif result.result.type == "errored":
        error = result.result.error
        results[custom_id] = {
            "status": "error",
            "error_type": error.type,
            "error_message": error.message,
        }

# In kết quả
for doc_id, result in results.items():
    if result["status"] == "success":
        print(f"{doc_id}: {result['output']}")
    else:
        print(f"{doc_id}: ERROR - {result['error_message']}")

Xử lý lỗi và Retry

Một số requests trong batch có thể thất bại vì nhiều lý do: timeout, invalid input, content policy. Chiến lược xử lý tốt nhất là tạo retry batch chỉ với những request lỗi:

def retry_failed_requests(original_requests, results):
    """Tạo batch mới với các requests bị lỗi."""
    failed_ids = {
        custom_id
        for custom_id, result in results.items()
        if result["status"] == "error"
    }

    if not failed_ids:
        print("Không có lỗi nào cần retry!")
        return None

    # Filter requests gốc
    retry_requests = [
        req for req in original_requests
        if req.custom_id in failed_ids
    ]

    print(f"Retry {len(retry_requests)} requests bị lỗi...")
    retry_batch = client.messages.batches.create(requests=retry_requests)
    return retry_batch.id

retry_batch_id = retry_failed_requests(requests, results)

Xử lý nhiều loại message khác nhau trong một batch

Sức mạnh thực sự của Batch API: mỗi request hoàn toàn độc lập — bạn có thể mix nhiều loại task, model, và cấu hình trong cùng một batch:

mixed_requests = [
    # Task 1: Phân loại ngắn gọn với Haiku
    anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
        custom_id="classify_001",
        params={
            "model": "claude-haiku-4-5",
            "max_tokens": 50,
            "messages": [{"role": "user", "content": "Classify: SPAM or HAM?
Email: Win a free iPhone now!"}]
        }
    ),
    # Task 2: Tóm tắt dài hơn với Sonnet
    anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
        custom_id="summarize_001",
        params={
            "model": "claude-sonnet-4-5",
            "max_tokens": 500,
            "system": "Bạn là chuyên gia tóm tắt văn bản. Tóm tắt ngắn gọn, súc tích.",
            "messages": [{"role": "user", "content": "Tóm tắt bài viết sau:

[Nội dung dài 5000 từ...]"}]
        }
    ),
    # Task 3: Translation với system prompt khác
    anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
        custom_id="translate_001",
        params={
            "model": "claude-haiku-4-5",
            "max_tokens": 200,
            "system": "Translate the following text to Vietnamese. Keep technical terms in English.",
            "messages": [{"role": "user", "content": "Machine learning models require large datasets."}]
        }
    ),
]

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

Tính chi phí và so sánh

Giả sử bạn cần phân loại 10.000 emails, mỗi email ~500 input tokens và response ~50 tokens:

Với 10.000 emails, bạn tiết kiệm gần $1.87. Với 1 triệu emails mỗi tháng, tiết kiệm hơn $187.

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

# Liệt kê tất cả batches gần đây
for batch in client.messages.batches.list(limit=10):
    print(f"ID: {batch.id}")
    print(f"  Status: {batch.processing_status}")
    print(f"  Created: {batch.created_at}")
    print(f"  Requests: {batch.request_counts}")
    print()

# Hủy batch đang chạy (nếu chưa quá 60%)
try:
    canceled = client.messages.batches.cancel(batch_id)
    print(f"Đã hủy batch: {canceled.processing_status}")
except Exception as e:
    print(f"Không thể hủy: {e}")

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

• Nên dùng: Phân tích dataset lớn, chạy evaluations, xử lý tài liệu offline, ETL pipelines
• Không nên dùng: Chatbots real-time, ứng dụng cần response ngay lập tức, tasks cần kết quả theo thứ tự

Batch Processing là công cụ không thể thiếu khi làm việc với dữ liệu quy mô lớn. Kết hợp với Prompt Caching để tối ưu hóa chi phí tối đa.

Bài viết liên quan

• Calculator Tool — Bài học đầu tiên về Tool Use với Claude
• Fine-tuning Claude trên AWS Bedrock — Hướng dẫn từng bước
• Content Moderation — Xây dựng bộ lọc nội dung với Claude
• Claude cho Engineering: System Design interviews và planning
• Building Effective Agents với Claude — Hướng dẫn kiến trúc