わさびです。
Claude APIを使っていて「大量のリクエストを安く処理したい」と思ったことはないだろうか。
Message Batches APIを使えば、通常APIの50%オフで処理できる。条件は「リアルタイム応答が不要」であること。24時間以内に結果を返してくれる。
英語のドキュメントしかないこの機能を、日本語で実装例つきで解説する。
Batch APIとは
通常のMessages APIは、リクエストを送ったらすぐにレスポンスが返ってくる（同期処理）。
Batch APIは違う。大量のリクエストをまとめて送って、Anthropic側が空いているリソースで順次処理する（非同期処理）。リアルタイム性を犠牲にする代わりに、料金が50%オフになる。
| 項目 | Messages API | Batch API |
|---|---|---|
| レスポンス | 即時 | 最大24時間 |
| 料金 | 通常 | 50%オフ |
| 最大リクエスト数 | 1件ずつ | 1バッチ最大100,000件 |
| ストリーミング | 対応 | 非対応 |
どんなときに使うか
Batch APIが合うユースケース:
- 1,000件の商品説明文を一括生成
- 過去の記事を全件翻訳
- 大量のカスタマーレビューを分類
- データセットに対するラベリング
- 夜間に回すバッチ処理全般
共通点は「結果がすぐに必要ない」こと。チャットボットのような即時応答には使えない。
バッチの作成（Python SDK）
インストール
pipinstallanthropic
基本的なバッチ作成
importanthropic
client = anthropic.Anthropic()
# バッチを作成
batch = client.messages.batches.create(
requests=[
{
"custom_id": "request-001",
"params": {
"model": "claude-sonnet-4-5",
"max_tokens": 1024,
"messages": [
{"role": "user", "content": "Pythonのリスト内包表記を説明して"}
]
}
},
{
"custom_id": "request-002",
"params": {
"model": "claude-sonnet-4-5",
"max_tokens": 1024,
"messages": [
{"role": "user", "content": "Pythonのデコレータを説明して"}
]
}
}
]
)
print(f"バッチID:{batch.id}")
print(f"ステータス:{batch.processing_status}")
custom_idは自分でつけるID。結果を受け取ったときに、どのリクエストの結果かを識別するために使う。
大量リクエストの場合
100件以上のリクエストを送る場合、リストを動的に組み立てる:
importanthropic
client = anthropic.Anthropic()
# 処理したいデータ
products = [
{"id": "prod-001", "name": "ワイヤレスイヤホン", "specs": "Bluetooth 5.3, ANC対応"},
{"id": "prod-002", "name": "モバイルバッテリー", "specs": "20000mAh, PD対応"},
# ... 数百件
]
# バッチリクエストを組み立て
requests = []
for product in products:
requests.append({
"custom_id": product["id"],
"params": {
"model": "claude-haiku-4-5",
"max_tokens": 512,
"messages": [
{
"role": "user",
"content": f"以下の商品の説明文を100字で書いて。\n商品名:{product['name']}\nスペック:{product['specs']}"
}
]
}
})
batch = client.messages.batches.create(requests=requests)
print(f"バッチID:{batch.id}")
print(f"リクエスト数:{batch.request_counts.total}")
ステータス確認
バッチの処理状況を確認する:
importanthropic
client = anthropic.Anthropic()
batch = client.messages.batches.retrieve("batch_abc123")
print(f"ステータス:{batch.processing_status}")
print(f"処理済み:{batch.request_counts.succeeded}")
print(f"エラー:{batch.request_counts.errored}")
print(f"未処理:{batch.request_counts.processing}")
processing_statusの値:
| ステータス | 意味 |
|---|---|
| in_progress | 処理中 |
| ended | 完了（成功・エラー混在の可能性あり） |
| canceling | キャンセル処理中 |
| canceled | キャンセル済み |
結果の取得
バッチが完了したら結果を取得する:
importanthropic
client = anthropic.Anthropic()
# 結果をイテレーション
for result in client.messages.batches.results("batch_abc123"):
if result.result.type == "succeeded":
message = result.result.message
print(f"ID:{result.custom_id}")
print(f"応答:{message.content[0].text}")
print(f"入力トークン:{message.usage.input_tokens}")
print(f"出力トークン:{message.usage.output_tokens}")
print("---")
elif result.result.type == "errored":
print(f"ID:{result.custom_id} — エラー:{result.result.error}")
結果はストリーミング形式で返ってくるので、メモリに全件載せる必要がない。10万件の結果でも1件ずつ処理できる。
ポーリングで完了を待つ
実用的には、バッチ作成後に完了を待ってから結果を取得するパターンが多い:
importanthropic
importtime
client = anthropic.Anthropic()
# バッチ作成
batch = client.messages.batches.create(requests=requests)
batch_id = batch.id
# 完了を待つ
while True:
batch = client.messages.batches.retrieve(batch_id)
if batch.processing_status == "ended":
break
print(f"処理中...{batch.request_counts.succeeded}/{batch.request_counts.total}")
time.sleep(30)
# 結果を取得
results = {}
for result in client.messages.batches.results(batch_id):
if result.result.type == "succeeded":
results[result.custom_id] = result.result.message.content[0].text
print(f"完了:{len(results)}件")
30秒間隔でポーリングしている。あまり頻繁にチェックするとAPI呼び出しが無駄に増えるので、30秒〜1分間隔が目安。
プロンプトキャッシュとの併用
Batch APIはプロンプトキャッシュと併用できる。同じシステムプロンプトを使うリクエストが大量にある場合、さらにコストを削減できる。
requests = []
for item in data:
requests.append({
"custom_id": item["id"],
"params": {
"model": "claude-sonnet-4-5",
"max_tokens": 1024,
"system": [
{
"type": "text",
"text": "あなたは商品レビューの分類器です。...(長いシステムプロンプト)",
"cache_control": {"type": "ephemeral"}
}
],
"messages": [
{"role": "user", "content": item["review"]}
]
}
})
cache_controlを設定しておくと、バッチ内で同じシステムプロンプトがキャッシュされる。
Batch APIの50%割引 + キャッシュの90%割引。組み合わせると、通常の5%程度のコストで処理できるケースもある。
制限事項
知っておくべき制限:
| 項目 | 制限 |
|---|---|
| 1バッチの最大リクエスト数 | 100,000件 |
| 処理期限 | 24時間以内 |
| ストリーミング | 非対応 |
| 結果の保持期間 | 29日間 |
| 同時実行バッチ数 | 制限あり（Tierによる） |
24時間以内に処理が完了しなかったリクエストは、期限切れ（expired）として扱われる。大規模バッチでも大半は数時間以内に完了するが、保証はない。
コスト試算の例
1,000件の記事翻訳をBatch APIで処理する場合:
| 項目 | 通常API | Batch API |
|---|---|---|
| 入力: 2,000トークン x 1,000件 | $6.00 | $3.00 |
| 出力: 3,000トークン x 1,000件 | $45.00 | $22.50 |
| 合計（Sonnet 4.5） | $51.00 | $25.50 |
半額。1,000件で$25の差。件数が増えるほど差が広がる。
まとめ
Batch APIは「大量 + リアルタイム不要」のタスクで威力を発揮する。
- 50%オフ（プロンプトキャッシュ併用でさらに削減）
- 1バッチ最大10万件
- Python SDKで数行のコードで使える
夜間バッチ処理、大量データのラベリング、コンテンツ一括生成。こういったタスクがあるなら、通常APIの前にBatch APIを検討する価値がある。
水槽の中で24時間待つのは得意なので、バッチ処理との相性は抜群だと思っている。
あわせて読みたい
- Claude API料金ガイド — トークン単価とコスト計算の基本
- Claudeプロンプトキャッシュで料金90%削減 — Batch APIと併用で最大効果
- Claude Streaming API実装ガイド — リアルタイム応答が必要ならこちら
- Claude本番運用ガイド — レート制限やエラー処理の実践
- Claude APIツール使用入門 — Batch内でTool Useも使える
見てもらえるだけで応援になります
このブログはアフィリエイトリンクで運営されています。以下のリンクから気になるサービスをチェックしてもらえると、僕たちの活動の支えになります。
この記事を書いたのは わさび（ニホンイシガメ / 3歳 / VTuberあかはら。の家族）です。
あかはらVラボ — Claude特化の情報を発信中。
この記事が参考になったら｜以下のリンクから見てもらえるだけで、ブログ運営の応援になります。
AI開発環境やブログ運営に。初期費用無料、月額296円から。- NordVPN
AI活用時のデータ保護に。VPNで通信を暗号化。
コメント