Message Batches APIでリクエストを一括処理する

大量の分類・要約・抽出をまとめて処理したいとき、1件ずつ同期的にAPIを呼ぶのは非効率です。Message Batches APIを使うと、最大10万件のリクエストを非同期でまとめて投げ、標準価格の50%で処理できます。この記事では、バッチの作成・完了ポーリング・結果取得の一連の流れを解説します。

Message Batches APIとは

Message Batches API（POST /v1/messages/batches）は、Messages APIのリクエストを非同期でまとめて処理するための仕組みです。おもな特徴は次のとおりです。

1バッチあたり最大10万件、または256MBまで
ほとんどのバッチは1時間以内に完了（最大24時間）
結果は作成から29日間取得可能
全トークンについて50%のコスト削減
vision・tool use・プロンプトキャッシュなどMessages APIの機能をそのまま利用可能

レイテンシがそれほど重要でない一括処理（大量ドキュメントの要約・分類、評価データセットの生成など）に向いています。

バッチを作成する

各リクエストに custom_id と params（通常の messages.create と同じ内容）を渡します。custom_id は結果を突き合わせるための識別子で、後述のとおり結果は順不同で返るため必須です。

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

client = anthropic.Anthropic()

message_batch = client.messages.batches.create(
    requests=[
        Request(
            custom_id="request-1",
            params=MessageCreateParamsNonStreaming(
                model="claude-opus-4-8",
                max_tokens=1024,
                messages=[{"role": "user", "content": "気候変動の影響を要約して"}],
            ),
        ),
        Request(
            custom_id="request-2",
            params=MessageCreateParamsNonStreaming(
                model="claude-opus-4-8",
                max_tokens=1024,
                messages=[{"role": "user", "content": "量子コンピュータの基礎を説明して"}],
            ),
        ),
    ]
)
print(message_batch.id, message_batch.processing_status)

完了をポーリングする

processing_status が "ended" になるまで retrieve でポーリングします。request_counts で成功・失敗・処理中の件数を確認できます。

import time

while True:
    batch = client.messages.batches.retrieve(message_batch.id)
    if batch.processing_status == "ended":
        break
    print(f"status: {batch.processing_status}, 処理中: {batch.request_counts.processing}")
    time.sleep(60)

print("完了:", batch.request_counts.succeeded, "件成功")

結果を取得する

results() で結果をストリームします。結果は投入順ではなく順不同で返るため、必ず custom_id をキーにして突き合わせてください（位置で対応づけてはいけません）。各結果の result.type は succeeded / errored / canceled / expired のいずれかです。

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

使いどころと注意点

バッチは「即時性が不要で件数が多い」処理に最適です。分類・要約・抽出・評価データの一括生成などが典型例です。共有する大きなコンテキストがある場合は、各リクエストの system にプロンプトキャッシュを効かせるとさらにコストを下げられます。

一方で、リアルタイム応答が必要なチャットには向きません。結果が順不同で返る点、期限切れ（expired）はステータスで判定して再投入する点に注意してください。本記事は非公式翻訳です。最新の仕様は公式ドキュメントをご確認ください。