LLMの大量処理を同期APIでやらない｜バッチ推論・非同期処理の判断基準と失敗設計

過去3年分の問い合わせ12万件にタグを付ける、社内文書5万本を要約してインデックスを作り直す、既存データをモデル移行後の構成で再処理する。こうした仕事が来たとき、最初に書かれるコードはたいてい同じ形をしています。CSVを読み込んで、for文でAPIを呼んで、結果を配列に足していく。件数が100件なら問題ありません。ところがこれが5万件になると、実行に丸一日かかり、途中で429に当たって止まり、どこまで終わったのか分からないまま再実行して二重に課金される、という展開になります。

この記事は、その手前で持つべき判断の話です。どの処理を同期のままにし、どれをバッチAPIへ寄せ、どれを自前のジョブキューで回すか。そして非同期に寄せると決めたあとに必ず出てくる、冪等性・突合・部分失敗・品質検査をどう設計するか。所要時間や割引の目安は各社の公式ドキュメントで確認できた範囲だけを書き、価格そのものは改定が続くため記載しません。

実行モードを、まず言葉として分ける#

「非同期にする」という言葉は、実務では少なくとも4つの別のものを指します。混ざったまま議論すると噛み合いません。

低優先ティアの例として、OpenAIはservice_tierにflexを指定する方式を公開しています。公式ドキュメントによれば、トークンはBatch APIのレートで課金され、プロンプトキャッシュによる追加の割引も効きます。応答が遅くなることが前提で、リソースが確保できないときには429が返り、その場合は課金されないとされています。既定のタイムアウトは10分ですが、ドキュメントのコード例では15分へ引き上げる形が示されています。なお同ドキュメントはFlex processingがベータであり対応モデルも限られる旨を明記しているため、本番採用の可否は最新の記載と対応モデル一覧を確認したうえで判断してください。

この4つのうち、コードの書き換え量が最小なのは低優先ティアです。バッチAPIは投入と回収という別のフローになるため、いちばん書き換えが大きくなります。自前ジョブキューはその中間で、同期APIの呼び出しはそのまま流用できます。

判断軸は4つ:締切・件数・失敗許容度・コスト#

どこへ寄せるかは、次の4つで決まります。順番に重要度が高いわけではなく、締切だけが性質の違う軸です。

締切(いつまでに全件終わっていればよいか)#

これだけが足切りとして働きます。バッチAPIの完了ウィンドウは各社とも24時間の水準にあり、しかもそれは保証ではなく目安です。したがって「明日の朝までに終わっていればよい」なら候補に入りますが、「30分後の会議で使う」なら候補から外れます。ここで重要なのは、締切を人の都合ではなく処理の下流で決めることです。バッチの結果を受けて別のジョブが動く構成なら、そのジョブの開始時刻が実質の締切になります。

件数(1回のジョブで何件処理するか)#

数十件なら同期でループさせても数分で終わり、失敗しても最初からやり直せます。数千件を超えたあたりから、途中失敗のコストが「やり直せばいい」で済まなくなり、進捗の保存が必要になります。数万件を超えると、レート制限に当たらないように送出を絞ること自体が仕事になり、バッチAPIの守備範囲に入ってきます。

失敗許容度(一部が落ちたときに何が起きるか)#

全件そろわないと次に進めない処理と、9割そろえば十分な処理では設計が変わります。前者は再投入と全件突合の仕組みが必須で、後者は落ちた分を記録して次回に回す運用ができます。この見極めを先にやらないと、必要のない厳密さを作り込んだり、逆に静かに欠落したまま下流へ流したりします。

コスト(削減幅が設計変更の手間に見合うか)#

各社ともバッチ処理は同期比で半分の水準としています。月に数千円しか使っていない処理なら、半額になっても設計変更の手間に見合いません。逆に、数万件の再処理を毎月回すような処理では、この差がそのまま予算の話になります。

各社のバッチAPIに共通する形と、違いが出るところ#

公式ドキュメントで確認できる範囲を整理すると、共通点と相違点がはっきりします。

共通しているのは、24時間という時間軸、半額という価格帯、そして結果が投入順に返らないことです。Anthropicのドキュメントは「バッチ結果は任意の順序で返ることがあり、バッチ作成時のリクエスト順と一致しないことがある。結果を正しく対応付けるには常にcustom_idフィールドを使うこと」と明記しています。OpenAIも同様に、出力の行順が入力の行順と一致しないことがあるため順序に頼らずcustom_idを使うよう書いています。

期限切れと部分失敗:課金の考え方は共通、結果の取り方が違う#

バッチが24時間で終わらなかったときの課金は、各社とも同じ考え方です。モデルへ送られる前に期限切れになった個々のリクエストは課金されず、期限切れまでに完了した分は課金されます。設計に効いてくる差分は金額ではなく、「未処理分をどうやって知るか」という結果の受け取り方のほうにあります。

Anthropicのドキュメントは、個々のリクエストの結果をsucceeded・errored・canceled・expiredの4種類に分け、errored(無効なリクエストや内部サーバエラー)・canceled(ユーザーがモデルへ送られる前にキャンセル)・expired(24時間の期限に到達し、モデルへ送られなかった)については課金されないとしています。またキャンセルしたバッチはendedという状態で終わり、キャンセル前に処理された分の部分結果を含むことがあるとされています。

OpenAIのドキュメントは、バッチが期限切れになると未完了のリクエストはキャンセルされ、完了済みリクエストの応答は出力ファイルから取得でき、完了したリクエストで消費されたトークンについては課金される、としています。つまり期限切れは「全損」ではなく「途中まで取れる、その分は払う」という扱いです。

Geminiは、ジョブ状態としてJOB_STATE_PENDING・JOB_STATE_RUNNING・JOB_STATE_SUCCEEDED・JOB_STATE_FAILED・JOB_STATE_CANCELLED・JOB_STATE_EXPIREDを定義しており、実行中または保留のまま48時間を超えたジョブは期限切れになるとしています。

つまり、未処理分の知り方が三者三様です。Anthropicはexpiredが結果JSONLの1行としてcustom_idつきで返り、OpenAIはbatch_expiredのエラーとしてエラーファイルへ書き出され、Geminiはジョブ全体がJOB_STATE_EXPIREDになり結果を取得できません。台帳側の未処理判定を、どの経路から埋めるかがここで決まります。

実務上の結論はシンプルです。期限切れは異常系ではなく、起こりうる正常な終了状態のひとつとして扱ってください。そのうえで、「期限切れになったとき、何件が未処理で残り、それを次にどう投入するか」を先に決めておきます。Anthropicのドキュメントには、需要と自分のリクエスト量によっては処理が遅くなり、24時間後に期限切れになるリクエストが増えることがある、という趣旨の記載もあります。混雑は自分では制御できない変数です。

ジョブ設計:主役はプロンプトではなく入力ID#

非同期に寄せると決めた瞬間から、設計の中心はプロンプトの書き方ではなく、入力1件を一意に指す識別子になります。これが決まっていないと、突合も再投入も冪等性も成立しません。

入力IDの決め方#

業務側にすでに一意キーがあるなら、それを使うのがいちばん安全です(問い合わせID、文書ID、注文番号など)。無い場合は、入力内容のハッシュを使う手もありますが、同じ内容が別の意味を持つ場合は衝突するため注意が必要です。IDには文字種の制約が付くことがあります。Anthropicのcustom_idは1〜64文字で、英数字・ハイフン・アンダースコアのみ(^[a-zA-Z0-9_-]{1,64}$)と定められています。日本語の文書名やスラッシュを含むパスをそのまま入れることはできません。

さらに実務では、IDに処理のバージョンを含めておくと後で助かります。doc-10482__v3のように、プロンプトのバージョンやモデル世代を含めておけば、同じ文書を条件を変えて再処理した結果を並べて保持できます。

台帳を先に作る#

バッチジョブの中心は、投入したい入力の一覧と、その状態を持つテーブルです。これを先に作ってからAPIを呼びます。最低限、次の列があれば回ります。

この台帳があると、「まだ処理していないものだけを投入する」という操作が単なるSQLになります。冪等性の土台はこの一文ですが、これだけでは足りません。APIを呼んだ直後、状態を書き戻す前にプロセスが落ちると、行は未処理のまま残り、次回の投入で同じ入力をもう一度投げてしまうからです。そこで投入は二段構えにします。

# 投入と回収を分けた最小構成の擬似コード(エラー処理は簡略化)
# 前提: 台帳テーブル jobs(input_id, status, batch_id, attempts, ...)

def submit_pending(limit=50_000):
    # 1. まだ終わっていないものだけを取り出す(冪等性の本体はこのクエリ)
    rows = db.query(
        "SELECT input_id, payload FROM jobs "
        "WHERE status IN ('pending', 'retryable') AND attempts < 3 "
        "LIMIT ?", limit)
    if not rows:
        return None

    requests = [
        {"custom_id": r.input_id, "params": build_params(r.payload)}
        for r in rows
    ]

    # 2. 投入する前に「投入しようとしている」ことを書く(二段書きの前半)
    #    これがないと、create() の直後に落ちたとき行が pending のまま残り、
    #    次回の submit_pending が同じ入力を再投入して二重課金になる
    token = new_uuid()   # この試行を識別する値(回収時のグループキー)
    db.execute(
        "UPDATE jobs SET status='submitting', submit_token=?, "
        "attempts=attempts+1, submitted_at=now() WHERE input_id IN (?)",
        token, [r.input_id for r in rows])

    batch = client.messages.batches.create(requests=requests)

    # 3. 成功したらバッチIDを埋めて submitted に進める(二段書きの後半)
    db.execute(
        "UPDATE jobs SET status='submitted', batch_id=? "
        "WHERE submit_token=?", batch.id, token)
    return batch.id


def recover_submitting(grace="30min"):
    # 4. submitting のまま宙に浮いた行を回収する。バッチ一覧APIで
    #    submitted_at 以降に作られたバッチを列挙し、その custom_id に
    #    自分の入力IDが含まれていれば submitted へ、
    #    どこにも無ければ create() 前に落ちたので pending へ戻す
    for token, since, ids in db.stale_submit_batches(older_than=grace):
        batch = find_batch_containing(client, since, ids)
        if batch:
            db.execute("UPDATE jobs SET status='submitted', batch_id=? "
                       "WHERE submit_token=?", batch.id, token)
        else:
            db.execute("UPDATE jobs SET status='pending', "
                       "attempts=attempts-1 WHERE submit_token=?", token)


def collect(batch_id):
    # 5. 結果は投入順に返らない。必ず custom_id で突き合わせる
    for result in client.messages.batches.results(batch_id):
        input_id = result.custom_id
        kind = result.result.type  # succeeded / errored / expired / canceled

        if kind == "succeeded":
            body = extract_text(result)
            if not passes_quality_check(body):        # 6. 検査もここで
                mark(input_id, "failed", reason="quality")
                continue
            save_output(input_id, body)
            mark(input_id, "succeeded")
        elif kind == "errored" and is_validation_error(result):
            mark(input_id, "failed", reason="validation")   # 再投入しない
        else:
            mark(input_id, "retryable", reason=kind)        # 再投入する

ポイントは3つです。投入する対象を台帳から決めること、API呼び出しの前後で状態を二段に書くこと、そして回収時にcustom_idだけで紐付けることです。二段書きにしておけば、どのタイミングで落ちても行はpendingかsubmittingかsubmittedのいずれかに残ります。再実行の前にrecover_submittingを通し、宙に浮いたsubmittingの行をバッチ一覧APIとcustom_idで突き合わせてから投入する、という順番にしてください(バッチ一覧の取得と、投入済みバッチの結果に含まれるcustom_idの確認は各社とも用意されています)。この復旧処理を挟まずに同じコマンドをもう一度叩くと、二重投入の穴がそのまま残ります。

部分失敗は例外ではなく通常状態#

数万件を投げれば、必ず何件かは失敗します。ここで「失敗したので全部やり直す」を選ぶと、成功した分の課金が丸ごと無駄になります。再投入の可否は、エラーの性質で分けてください。

• 再投入してはいけないもの: リクエスト自体が無効なケースです。入力が長すぎてコンテキスト上限を超えている、必須パラメータが欠けている、といったバリデーションエラーは、何度投げても同じ結果になります。Anthropicの公式サンプルコードも、erroredのうちinvalid_requestは「再送する前にリクエスト本文を修正する必要がある」、それ以外のサーバエラーは「そのまま再試行できる」とコメントで書き分けています。
• 再投入してよいもの: 一時的なサーバエラーと、期限切れです。期限切れになった個々のリクエストは各社とも課金されず、期限切れまでに完了した分は課金される、というのが共通の形です。いずれにせよ未処理として次のバッチに載せ直すのが素直です。
• 投げ方を変えるべきもの: 同じ入力で毎回タイムアウトや上限超過を起こす少数の巨大な入力です。これは全体のバッチから外し、分割してから別枠で処理します。1件の外れ値のために全体を細切れにするのは無駄が大きくなります。

再投入には上限を持たせてください。試行回数を台帳に持ち、3回失敗したものはfailedに固定して人の目に回します。無限に再投入するループは、失敗の原因が入力側にある場合に永久に回り続け、しかも毎回課金されます。

レート制限との付き合い方#

バッチAPIの枠が同期の枠とは別に管理される点は、リトライ設計の記事で触れたとおりです(OpenAIのドキュメントも、Batch APIのレート制限は標準のモデル別レート制限とは別だと明記しています)。ここで押さえたいのは、その先にあるバッチ固有の制限です。AnthropicのMessage Batches APIには、全モデル共通のRPMに加えて「処理キューに同時に入れられるバッチリクエスト数」という軸があり、公式のレート制限ページではStart/Build/Scaleの各tierに対して200,000/300,000/500,000という値が、1バッチあたりの上限100,000件とあわせて示されています。ここでいうバッチリクエストとは、Message Batchの中の1件を指します。つまり「1バッチに何件入れられるか」と「同時に何件を待ち行列に置けるか」は別の制限です。

したがって、100万件を処理したいときの分割は、1バッチの上限だけでなく、キューに置ける総数も見て決めることになります。実務的には、1バッチを上限いっぱいまで詰めるより、回収と再投入の粒度として扱いやすいサイズ(数千から数万件)に区切り、前のバッチが終わってから次を投げる形のほうが運用が楽です。1つのバッチが期限切れになったときの被害も、そのぶん小さくなります。

コスト見積もりは、単価ではなくパイロットから立てる#

見積もりを「単価×件数」で立てると、たいてい外れます。出力トークン数が入力によって大きくばらつくためです。順番としては次のようにします。

そのうえで、実行中は台帳に記録したトークン数から実績を積み上げ、見積もりとの乖離を監視します。乖離が大きいときは、想定より長い出力が出ているか、リトライが想定より多いかのどちらかで、原因が違えば打ち手も違います。

なお、Anthropicのドキュメントには、高いスループットと並行処理のため、バッチがワークスペースに設定された支出上限をわずかに超えることがある、という記載があります。上限を安全弁として使っている場合は、この挙動を前提に少し余裕を持たせてください。

品質検査をバッチの中に組み込む#

同期処理なら、おかしな出力はその場で人の目に触れます。バッチではそれが起きません。5万件の結果が静かにデータベースへ書き込まれ、3週間後に「そういえばあのタグ、全部おかしくないですか」と言われるのが典型的な失敗です。検査は後工程ではなく、回収処理の中に入れてください。

まず、機械的に判定できるものは全件チェックします。JSONとしてパースできるか、想定したスキーマに合っているか、分類ラベルが定義した集合に入っているか、出力が空でないか、停止理由が上限到達になっていないか。前掲の擬似コードでpasses_quality_checkと書いた部分です。ここで落ちたものはfailedにして再投入対象に回します。

次に、内容の妥当性は全件では見られないので、無作為抽出で人が見ます。件数は母集団の規模と失敗許容度で決めますが、重要なのは「抽出をランダムにすること」と「見た結果を記録して次回と比較できるようにすること」です。バッチは繰り返し実行されるので、初回に基準を作っておけば2回目以降は差分だけを見ればよくなります。

3つ目の手として、LLM自身に検査させる方法があります。生成用のバッチとは別に、検査用のバッチをもう1本流す構成です。バッチは安いので、この二段構えのコストは同期で1回やるより安く収まることがあります。ただし検査側の判定が信用できるかどうかは別途評価が必要で、検査そのものが当てにならなければ意味がありません。

バッチや非同期に向かない場面#

寄せてはいけないものを先に決めておくと、判断が速くなります。

対話UIは論外です。ユーザーが画面の前で待っている処理を24時間ウィンドウの仕組みに載せることはできません。ここは同期とストリーミングの領域で、指標もコストではなくTTFTになります。

締切が分単位の処理も向きません。「入稿から10分以内にチェック結果を返す」といった約束がある処理は、目安として1時間未満で終わることが多いとされていても、混雑時に24時間側へ振れる可能性を許容できません。どうしても安く済ませたいなら、バッチに投げつつ一定時間で結果が揃わなければ同期へフォールバックする二重構成が考えられますが、実装と課金が複雑になるため、締切が本当に固いのかを先に確認するほうが建設的です。

入力が到着順や前の結果に依存する処理も向きません。バッチAPIは各リクエストを独立に処理する前提で、Anthropicのドキュメントも「各リクエストは独立して処理される」ことを明記しています。前の出力を次の入力に使う多段の処理は、1つのバッチには収まりません。この場合は、段ごとにバッチを分けて順に流すか、自前のジョブキューでワークフローとして組むことになります。

対話の途中経過を保持する処理も同様です。マルチターンの会話そのものはバッチに入れられますが(会話履歴を含めた1リクエストとして投げる形)、ユーザーの次の発話を待つタイプの処理は非同期化できません。

自前ジョブキューを選ぶべきとき#

バッチAPIがあるのに自前で組む理由は、主に次の3つです。

1つ目は、進捗を見せる必要がある場合です。管理画面に「1,240/5,000件完了」と出したいなら、キューとワーカーの構成のほうが素直です。バッチAPIでも件数の内訳は取れますが、粒度と更新頻度は自分では決められません。

2つ目は、複数ステップのワークフローになっている場合です。抽出してから分類し、分類結果によって要約の仕方を変える、といった処理は、LLM呼び出し以外の処理が間に挟まります。ジョブキューなら1つのジョブの中で完結させられます。

3つ目は、バッチAPIが対応していない機能を使う場合です。たとえばAnthropicのバッチではstream: trueやspeedパラメータ、max_tokens: 0が使えず、指定するとバリデーションエラーになるとされています。同期でしか使えない機能に依存しているなら、キュー+同期呼び出しの構成になります。

自前で組む場合、レート制限との付き合いはすべて自分の責任になります。ワーカー数で送出レートを握り、429のときは指数バックオフとジッタで待つ。この部分は別記事で詳しく扱っているので、そちらを参照してください。

よくある質問#

まとめ#

大量処理を同期APIでやらない、という判断そのものは難しくありません。難しいのは、非同期に寄せると決めたあとに何を作るかです。

判断は4軸で足ります。締切が時間単位まで許容できるか、件数が数千を超えるか、一部失敗を許容できるか、コスト差が手間に見合うか。締切だけが足切りで、残りは程度の問題です。そして寄せると決めたら、プロンプトより先に入力IDと台帳を設計します。結果は順不同で返り、一部は必ず失敗し、バッチは期限切れになりうる。この3つを前提にすれば、実装は「未処理を選ぶ・投入する・突き合わせて記録する」という単純な繰り返しに落ち着きます。

各社の仕様は共通点が多く、期限切れ時の課金の考え方も共通です。差が出るのは、未処理分をどの経路で知るかという部分結果の取り方と状態の表し方で、ここが台帳の設計に効いてきます。数値や挙動は改定されるため、採用時には必ず公式ドキュメントで現況を確認してください。この記事に書いた各社の記述も、確認できた範囲にとどめています。