【AI】Claude APIのBatch APIで大量処理をする方法

import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()

reviews = [
    "配送が早くて満足しています",
    "梱包が雑で商品が傷んでいた",
    "可もなく不可もなくといった印象",
]

batch = client.messages.batches.create(
    requests=[
        Request(
            custom_id=f"review-{i}",
            params=MessageCreateParamsNonStreaming(
                model="claude-haiku-4-5",
                max_tokens=50,
                messages=[
                    {
                        "role": "user",
                        "content": f"次のレビューをポジティブ・ネガティブ・中立のいずれかで分類して: {text}",
                    }
                ],
            ),
        )
        for i, text in enumerate(reviews)
    ]
)

print(batch.id)
print(batch.processing_status)

import time

while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    print(batch.request_counts)
    time.sleep(60)

for result in client.messages.batches.results(batch.id):
    match result.result.type:
        case "succeeded":
            message = result.result.message
            text = next((b.text for b in message.content if b.type == "text"), "")
            print(result.custom_id, text)
        case "errored":
            print(result.custom_id, "エラー:", result.result.error)
        case "expired":
            print(result.custom_id, "期限切れ。再投入が必要")

client.messages.batches.cancel(batch.id)

for b in client.messages.batches.list(limit=20):
    print(b.id, b.processing_status)