壁紙アプリの素材を整理していた夜のことでした。分類待ちの画像が数千枚たまっていて、一枚ずつ Claude に投げて説明文とカテゴリを付ける手もありますが、リアルタイムの応答は要りません。寝ている間に走らせて、朝に揃っていれば十分です。
この「急がないけれど量がある」処理に合うのが、Claude API の Message Batches です。リクエストをまとめて投げ、非同期で処理してもらう仕組みになっています。
ただ、最初に回したときの私は、割引率も、期限の意味も、custom_id に入れられる文字も、正しく把握していませんでした。以下は、その修正の記録です。
割引は50%。9割落ちるのは、重ねたとき
バッチに載せるだけで費用が9割落ちる、と私は思い込んでいました。実際の Message Batches の割引は、公式ドキュメントにある通り一律50%です。入力トークン・出力トークン・特別なトークンのいずれにも同じ率で効きます。
では9割という数字はどこから来るのか。手元の分類ジョブで実際に計算してみました。条件は、リクエスト3,000件、共有プロンプト900トークン＋画像ごとの説明文60トークン、出力は一語のカテゴリなので8トークン、モデルは Claude Haiku 4.5（同期の単価は 100万トークンあたり入力 $1・出力 $5、バッチはその半額）です。
| 構成 | 3,000件あたりの費用 | 同期API比 |
|---|---|---|
| 同期 API（キャッシュなし） | $3.0000 | — |
| Message Batches | $1.5000 | 50% 減 |
| Message Batches ＋ 1時間キャッシュ | $0.2859 | 90.5% 減 |
三段目の内訳は、キャッシュ書き込み $0.0009・キャッシュ読み出し $0.1350・固有部分の入力 $0.0900・出力 $0.0600 です。共有プロンプト900トークンを一度だけ書き込み、残り2,999件がそれを読み出す前提で計算しています。
つまり9割に届いたのは、バッチ・モデル選択・プロンプトキャッシュの三つを重ねた結果であって、バッチ単体の効果ではありません。キャッシュとバッチの割引は併用できると明記されているので、この重ね方自体は正攻法です。
ただしバッチのキャッシュヒットはベストエフォートで、公式には30%〜98%という幅が示されています。5分の TTL ではバッチの処理中に消えてしまうため、1時間の TTL を選ぶ、という指針も併記されています。私の場合も、5分キャッシュのままでは読み出しがほとんど発生しませんでした。キャッシュ側の設計はプロンプトキャッシュの実装メモに別途まとめています。
最小構成で往復を確かめる
いきなり数千件を投げず、2〜3件で形を確かめます。custom_id は後で結果を元データへ突き合わせる鍵になるので、必ず意味のある値を入れます。
import anthropic
client = anthropic.Anthropic()
batch = client.messages.batches.create(
requests=[
{
"custom_id": "wallpaper_0001",
"params": {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 512,
"messages": [
{"role": "user", "content": "この壁紙画像の雰囲気を一語のカテゴリで分類してください: 夜空に淡い光の輪"}
],
},
},
{
"custom_id": "wallpaper_0002",
"params": {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 512,
"messages": [
{"role": "user", "content": "この壁紙画像の雰囲気を一語のカテゴリで分類してください: 朝もやの中の山並み"}
],
},
},
]
)
print(f"Batch ID: {batch.id}")分類のような定型タスクで上位モデルに手を伸ばす理由はありませんでした。Haiku で十分で、バッチ割引と合わせるとここでもう一段下がります。
ファイル名をそのまま custom_id に入れて弾かれた
最初に書いたコードで、私は画像のファイル名をそのまま custom_id に入れていました。wallpaper_0001.jpg のような値です。元データとの対応が自明になるので、良い設計のつもりでした。
ところが custom_id に使える文字は ^[a-zA-Z0-9_-]{1,64}$ と決まっています。英数字とハイフンとアンダースコアのみで、拡張子のピリオドを含んだ時点で条件から外れます。長さの上限も64文字です。
さらに厄介なのは、params の検証が非同期で行われる点です。公式の記述では、検証エラーはバッチ全体の処理が終わってから返ります。形式の誤りを、翌朝になって知る場合があるということです。
そこで、custom_id は安全な文字だけに正規化し、元のファイル名との対応表を手元に持つ形へ変えました。
import re
def to_custom_id(filename: str) -> str:
"""英数字・ハイフン・アンダースコアのみに正規化し、64文字に収める"""
stem = filename.rsplit(".", 1)[0]
safe = re.sub(r"[^A-Za-z0-9_-]", "_", stem)
return safe[:64]
# 逆引き表を必ず残す（結果は custom_id でしか戻せないため）
index = {}
for name in image_files:
cid = to_custom_id(name)
if cid in index:
raise ValueError(f"custom_id が衝突しました: {cid} <- {name}")
index[cid] = name衝突検出を入れているのは、正規化で別々のファイル名が同じ値に潰れる場合があるからです。img.001.jpg と img-001.jpg は、どちらも img_001 になります。数千件の中に一組でも紛れると、結果の突き合わせが静かに壊れます。
ended は「終わった」であって「成功した」ではない
作成直後のバッチは処理中です。多くは1時間以内に終わりますが、混雑時は時間単位になります。数秒おきに叩きにいく意味はないので、私は最初の数分を空けてから、60秒間隔で確認しています。
import time
while True:
batch = client.messages.batches.retrieve(batch.id)
if batch.processing_status == "ended":
break
counts = batch.request_counts
print(f"処理中… 成功 {counts.succeeded} / 失敗 {counts.errored} / 処理中 {counts.processing}")
time.sleep(60)
print("バッチ完了")確認しているのは「全部成功したか」ではなく「処理が ended に達したか」です。request_counts には processing succeeded errored canceled expired の五つが並びます。ended になっても中身は成功と失敗の混在で、ここを取り違えると失敗分をそのまま取りこぼします。
結果の順番は保証されない
次につまずいたのがこれでした。結果は投げた順に返るとは限りません。公式にも「バッチの結果は入力順と一致しない場合がある」と明記されています。ストリームで一件ずつ受け取り、custom_id をキーに元データへ突き合わせる前提で書くのが安全です。
results = {}
for item in client.messages.batches.results(batch.id):
cid = item.custom_id
if item.result.type == "succeeded":
results[cid] = item.result.message.content[0].text
elif item.result.type == "errored":
# ここで握りつぶさず、再投入リストに積む
print(f"失敗: {cid} -> {item.result.error}")
results[cid] = None
elif item.result.type == "expired":
print(f"期限切れ: {cid}")
results[cid] = None
ok = sum(1 for v in results.values() if v is not None)
print(f"取得 {ok} / {len(results)} 件")result.type の意味は次の通りです。
| type | 意味 | 課金 |
|---|---|---|
| succeeded | 成功。メッセージ本体が含まれる | あり |
| errored | 不正なリクエスト、またはサーバ側のエラー | なし |
| canceled | モデルへ送られる前にキャンセルされた | なし |
| expired | 24時間の期限までに送られなかった | なし |
失敗分に課金されないのは、再投入をためらわなくてよいという意味です。私は errored と expired の custom_id だけを集めて、小さなバッチに分け直して投げ直しています。ここを自動化しておくと、翌朝の作業が「確認」だけで終わります。
24時間の期限と、29日の保持期間
期限まわりの仕様は、設計に直接効きます。処理が24時間で終わらないバッチは期限切れになります。結果を取れるのは、全件完了した時点か24時間経過時点の、早いほうです。
そして結果のダウンロードには29日という保持期間があります。起点は作成時刻であって、処理が終わった時刻ではありません。この差を見落とすと、長く走ったバッチほど手元に置ける期間が短くなります。
私の運用では、結果を受け取ったその場で自前のストレージへ書き出し、Anthropic 側の保持期間には依存しない形にしました。処理済みのバッチは削除もできます。処理中のものを消したい場合は、先にキャンセルしてからになります。
もう一点、公式に注意書きがあるのが利用上限との関係です。高い並列度で処理されるため、ワークスペースに設定した支出上限をわずかに超える場合があります。上限ぎりぎりで運用している場合は、余白を見ておいたほうが安全です。
分割の単位は、巻き戻したい単位で決める
一つのバッチには10万リクエスト、または256MBまで入ります。先に到達したほうが上限です。256MBを超えると request_too_large が返ります。
とはいえ私は上限まで詰めず、2,000件ずつに区切って投げています。理由は単純で、途中で投げ直したくなったときに、巻き戻す単位が小さいほど扱いやすいからです。連番でチャンクを切り、チャンクごとにバッチ ID をログへ残しておくと、後から「どこまで終わったか」が一目で分かります。
def chunk(items, size=2000):
for i in range(0, len(items), size):
yield items[i:i + size]
for n, group in enumerate(chunk(all_requests)):
b = client.messages.batches.create(requests=group)
print(f"chunk {n}: {b.id} ({len(group)}件)")レート制限は、同期の Messages API とは別枠です。バッチ側の消費が同期側の上限を削ることはありません。アプリの通常動線を止めずに大量処理を流せるのは、この分離のおかげです。より踏み込んだ本番向けの設計は、Messages Batches API の非同期処理設計にまとめてあります。
バッチに寄せない処理を決めておく
バッチには載せられないものもあります。stream: true は結果が単一のファイルで返る性質上、指定できません。max_tokens: 0 によるキャッシュの事前ウォームアップも、バッチ処理中に書き込んだ一時的なキャッシュが先に消えてしまうため対象外です。
仕様として弾かれるもの以上に効いてくるのは、寄せない判断のほうでした。ユーザーの操作に同期して即座に返したい処理までバッチに回すと、体験が確実に損なわれます。私の線引きは「人が待っていない処理だけを寄せる」です。App Store のレビュー返信文の下書きや、素材の分類のように、朝に揃っていれば足りる仕事がそれに当たります。
個人開発では、実行中の待ち時間そのものが、他の作業に使える時間になります。夜のうちに走らせて朝に受け取る形へ移してから、私自身の一日の組み立てが少し楽になりました。
次の一手としては、手元の「急がない処理」を一つだけ選び、2〜3件の最小バッチで往復を確かめてみてください。custom_id を正規化して逆引き表を持つところまで作れば、あとは件数を増やしていくだけです。
お読みいただきありがとうございました。