Batch APIで大量データ処理を50%安くする方法

数千件規模のデータを同期APIで処理していると、料金とレート制限の両方が重くのしかかります。Batch APIに切り替えれば、実装を大きく変えずに料金を半分に抑えられます。

結論から：Batch APIは常に50%引き、1バッチ最大10万件

全ての利用料金は標準APIの50%で課金されます。この割引は利用量に応じた段階制ではなく、入力・出力トークンの両方に対して一律50%引きが適用される仕組みです。

1回のバッチで送信できる件数にも上限があり、1つのMessage Batchは10万件のリクエスト、またはサイズ256MBのいずれか早い方に制限されます。数千件程度のデータ処理であれば、1回のバッチ送信で十分にまかなえる規模です。

処理速度についても、システムは可能な限り高速にバッチを処理し、多くのバッチは1時間以内に完了します。ただし処理が24時間以内に完了しない場合、バッチは期限切れになります。つまり「即レスポンスは不要だが、当日中には結果が欲しい」という業務に向いた仕組みです。

なぜ半額になるのか

Batch APIは同期APIとは別の処理経路として設計されており、リアルタイム性を保証しない代わりにコストを下げています。Message Batches APIは大量のMessagesリクエストを非同期に処理し、コストを50%削減しながらスループットを高める仕組みです。

具体的な金額感をつかむために、サポートチケット1万件をHaiku 4.5で分類する場合を試算します。Haiku 4.5のPlatform料金は入力100万トークンあたり1ドル、出力100万トークンあたり5ドルから始まり、Batch処理では50%のコスト削減が適用されます。

1件あたり入力500トークン・出力200トークンと仮定すると、次のようになります。

• 入力合計：10,000件 × 500トークン ＝ 5,000,000トークン（500万トークン）
• 出力合計：10,000件 × 200トークン ＝ 2,000,000トークン（200万トークン）
• 同期API料金：入力 5,000,000 ÷ 1,000,000 × $1 ＝ $5.00、出力 2,000,000 ÷ 1,000,000 × $5 ＝ $10.00、合計 $15.00
• Batch API料金：上記の50%なので $7.50

1万件の分類処理だけで7.5ドル分の差が生まれます。件数が増えるほど、この差はそのまま比例して拡大します。

具体的な手順：custom_idで追跡可能なバッチを組む

業務自動化でBatch APIを使う際にまず設計すべきなのが、リクエストとレスポンスを紐づけるcustom_idです。custom_idはMessagesリクエストを識別するための一意な値で、1〜64文字の英数字・ハイフン・アンダースコアのみで構成する必要があります。

単なる連番ではなく、後から追跡・再送しやすい形式にしておくと運用が楽になります。

import anthropic
import json
import time

client = anthropic.Anthropic()

def build_batch_requests(records: list[dict]) -> list[dict]:
    """
    records: [{"source_system": "zendesk", "ticket_id": "8823", "body": "..."}]
    custom_id例: zendesk-2026-09-11-8823
    ソースシステム名と日付を含めることで、
    結果ファイルだけを見ても再送元を特定できるようにする
    """
    requests = []
    for r in records:
        custom_id = f"{r['source_system']}-2026-09-11-{r['ticket_id']}"
        requests.append({
            "custom_id": custom_id,
            "params": {
                "model": "claude-haiku-4-5",
                "max_tokens": 300,
                "messages": [
                    {
                        "role": "user",
                        "content": f"次の問い合わせ内容を分類してください: {r['body']}"
                    }
                ]
            }
        })
    return requests


def submit_batch(records: list[dict]) -> str:
    requests = build_batch_requests(records)
    batch = client.messages.batches.create(requests=requests)
    # batch.idは必ず永続化する（アプリ再起動時の追跡用）
    with open("batch_ids.log", "a") as f:
        f.write(f"{batch.id}\n")
    return batch.id


def wait_for_batch(batch_id: str, interval_sec: int = 60) -> None:
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            return
        time.sleep(interval_sec)


def process_results(batch_id: str) -> tuple[list, list]:
    succeeded = []
    failed_custom_ids = []
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            succeeded.append({
                "custom_id": result.custom_id,
                "text": result.result.message.content[0].text
            })
        else:
            # errored / canceled / expired のいずれか
            failed_custom_ids.append(result.custom_id)
    return succeeded, failed_custom_ids

結果の取得時には、成功したものだけでなく失敗したものをcustom_id単位で切り分けることが重要です。結果には4種類のタイプがあり、request_countsを見ればそれぞれの状態に到達した件数が分かります。API仕様上も、succeeded・errored・canceled・expiredという状態でリクエスト件数が集計されます。

失敗したcustom_idだけを抽出して元データと突き合わせれば、全件を再送する必要はなく、失敗分だけをピンポイントで再投入できます。

注意点：リアルタイム性を捨てる代わりに気をつけるべきこと

Batch APIは万能ではありません。導入前に把握しておくべき制約が複数あります。

最短時間の保証はない 全てのメッセージが完了するか24時間が経過した時点のいずれか早い方で結果にアクセスできます。多くは1時間以内に終わりますが、これは保証ではなく傾向です。当日中の即時性が求められる業務には向きません。

ストリーミングや一部パラメータが使えない Message Batches APIはほとんどのMessages API機能とベータ機能をサポートしますが、stream・speed・max_tokens: 0といった一部のパラメータは非対応です。チャットのように結果を逐次表示する用途には使えません。

バッチ専用のレート制限がある Message Batches APIには全モデルで共有される専用のレート制限があり、APIエンドポイントへのリクエスト数（RPM）と、処理待ちキューに入っているバッチリクエスト数の上限が設定されています。1つのMessage Batchには数千件のリクエストを含めることができ、その全てがこの上限にカウントされます。大量のバッチを短時間に連投すると、キュー上限に引っかかる可能性があります。

結果は29日で失効する バッチの結果は作成から29日間ダウンロード可能です。それ以降はバッチ自体の閲覧はできても、結果のダウンロードはできなくなるため、下流のストレージに移す処理を組み込んでおく必要があります。

ワークスペース単位で管理される バッチはワークスペースにスコープされ、そのリクエストが実行されたワークスペース内で作成された全てのバッチとその結果を閲覧できます。複数チームでAPIキーを共有している場合、バッチの棚卸しをどのワークスペースで行うか事前に決めておくとよいでしょう。

バッチジョブを定期実行する仕組みそのものを自動化する際には、CLAUDE.mdの指示が守られない時に見直すべき仕組みで紹介されているhooksの考え方も参考になります。指示を都度渡すのではなく、決まった処理をフックとして仕込む発想は、バッチ送信・結果取得・再送のパイプラインを安定させる上でも役立ちます。

まとめ

• Batch APIは全ての利用料金が標準APIの50%になり、1バッチ最大10万件・256MBまで送信できます
• custom_idにソースシステム名や日付を含めておくと、失敗分だけを特定して再送する運用が組みやすくなります
• 24時間以内の完了しか保証されないため、当日中の即時性が必須の業務には向きません