はじめに / 対象と前提

大量のテキストを Claude で処理するとき、1 件ずつ Messages API を叩くとコストとレート制限の両方がつらい。Claude API には Message Batches API という非同期バッチ処理の仕組みがあり、入力・出力トークンともに通常の半額で処理できる。自分は数千件のログ要約をこれに載せ替えて、請求額をほぼ半分にした。

想定読者:Claude API(Messages API)を Python から叩いたことがある人
環境:Python 3.13 / anthropic(公式 Python SDK)0.69 系
即時レスポンスが不要な処理(夜間の一括要約・分類・埋め込み前処理など)が対象。チャット UI のような同期用途には使えない

TL;DR

client.messages.batches.create() に最大 10 万件(または 256MB)のリクエストを渡すと、24 時間以内に処理されてトークン単価が 50% オフになる
結果は JSONL で返るが 投入順は保証されないので、custom_id で突き合わせる
バッチ全体が ended でも個別リクエストは errored / expired があり得る。結果は必ず 1 件ずつ result.type を見る

手順 / 動かし方

1. バッチを作る

各リクエストは通常の Messages API のパラメータを params に包み、一意な custom_id を付ける。

import anthropic

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY を環境変数から読む

docs = load_documents()  # 要約したいテキストのリスト

batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"doc-{i}",  # バッチ内で一意にする
            "params": {
                "model": "claude-sonnet-5",
                "max_tokens": 1024,
                "messages": [
                    {"role": "user", "content": f"次の文書を3行で要約して:\n{doc}"}
                ],
            },
        }
        for i, doc in enumerate(docs)
    ]
)
print(batch.id)                 # msgbatch_01...
print(batch.processing_status)  # in_progress

2. 完了をポーリングする

Webhook はないので retrieve() で processing_status を見る。体感では数百件なら数分〜1 時間以内に終わることが多い。

import time

while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    print(batch.request_counts)  # processing/succeeded/errored/canceled/expired の内訳
    time.sleep(60)

3. 結果を custom_id で突き合わせて回収する

summaries = {}
for entry in client.messages.batches.results(batch.id):
    if entry.result.type == "succeeded":
        summaries[entry.custom_id] = entry.result.message.content[0].text
    else:
        # errored / canceled / expired
        print(f"{entry.custom_id}: {entry.result.type}")

実行結果(抜粋):

doc-0: succeeded
doc-1: succeeded
doc-7: errored

succeeded の中身は通常の Messages API のレスポンスと同じ構造なので、既存のパース処理がそのまま使える。

ハマりどころ

1. 結果の順序が投入順と一致しない

最初、結果を投入時のリストと zip() で突き合わせるコードを書いて、要約と元文書の対応がズレるというバグを踏んだ。処理は並列に走るため、結果 JSONL の並び順は保証されない。対応付けは必ず custom_id で行う。custom_id を連番でなく元データの主キー(ファイル名や DB の ID)にしておくと、復元処理が一気に楽になる。

2. バッチが ended でも「全件成功」ではない

processing_status == "ended" は「全リクエストの処理が終わった」であって「全件成功した」ではない。個別の結果には succeeded / errored / canceled / expired の 4 種類がある。自分は ended を見て安心して succeeded 前提で回したら、errored(リクエスト中の 1 件が max_tokens 超過の入力バリデーションエラー)で AttributeError になった。result.type の分岐を最初から書いておくこと。失敗分の custom_id を集めておけば、その分だけ再投入できる。

3. 24 時間の処理期限と 29 日の結果保持期限

24 時間以内に処理されなかったリクエストは expired になる(課金もされない)。また結果データの保持はバッチ作成から 29 日間で、それを過ぎると取得できなくなる。「後でまとめて回収しよう」と放置して消えると再実行しかない。結果はポーリング完了直後にローカルへ保存しておくのが安全。ちなみにレート制限は通常の Messages API とは別枠なので、バッチを投げても対話用途の枠は減らない。

背景・補足

半額になる理屈は単純で、Anthropic 側が空いている計算リソースで非同期に処理するため。だから「いつ終わるか」は保証されず、「24 時間以内」だけが約束される。プロンプトキャッシュもバッチ内で併用できるが、並列処理の都合でキャッシュヒットは保証されない。確実に効かせたいならキャッシュ形成用のリクエストを先に単発で送っておく手がある。

まとめ

即時性が要らない大量処理は Message Batches API に載せるだけでトークン単価が半額になる
結果は custom_id で突き合わせる。zip() での突き合わせは事故のもと
ended ≠ 全件成功。result.type を必ず分岐し、失敗分は custom_id で再投入
期限は「処理 24 時間・結果保持 29 日」。回収した結果は即ローカル保存