Batch APIを活用してOpenAIの利用料金を50%削減した方法

支払う必要のなかった月額5,000ドルの請求書

前四半期、私のチームは壁に突き当たりました。大手小売クライアント向けに240万件の顧客フィードバックを分類するタスクを任されたのですが、標準のOpenAI chat.completions エンドポイントを使用した場合、推定コストは5,100ドルという痛い出費になることが判明しました。さらに悪いことに、数分おきにレート制限（TPM）に達してしまい、データ取り込みパイプラインが頻繁に停止してしまいました。

このプロジェクトは即時性を必要とするものではありませんでした。500ミリ秒でレスポンスが返ってくる必要はなく、24時間以内にデータが処理されていればよかったのです。ワークロードをOpenAI’s Batch APIに移行したところ、請求額はきっかり2,550ドルまで下がりました。レート制限の悩みも一瞬で解消されました。もし予算を抑えつつAI機能をスケールさせたいのであれば、これは検討すべき最も重要なアーキテクチャの転換です。

「即時性プレミアム」が利益を圧迫している

多くの開発者はデフォルトで同期API呼び出しを使用します。リクエストを送り、レスポンスを待ち、次のタスクに進む。これはチャットボットには正しい選択です。しかし、データのエンリッチメント、大規模なSEOコンテンツ生成、過去データの分析といったバックグラウンドタスクにとっては、大きな足かせとなります。

OpenAIは即時レスポンスに対してプレミアム料金を課しています。なぜなら、彼らは計算リソースを即座に確保しなければならないからです。また、同期呼び出しは厳しいTier制限に縛られます。標準的なパイプで100万件のプロンプトを処理しようとすれば、価値を生み出す時間よりも 429: Too Many Requests エラーのデバッグに時間を費やすことになるでしょう。バッチ処理は、OpenAIが「閑散期」にジョブを実行できるようにすることで、この問題を解決します。

オプションの比較：同期 vs バッチ

パイプラインのリファクタリングを行う前に、240万行のデータを処理するための3つの戦略を比較検討しました。

• 標準の同期API: 通常料金（GPT-4oで100万トークンあたり5.00ドル）。レート制限による摩擦が大きい。結果は即時。
• マルチスレッド/AsyncIO: スループットは向上するが、料金は通常通り。レート制限のバーストを処理するために、複雑なリトライロジックやエクスポネンシャルバックオフが必要。
• OpenAI Batch API: 50%割引（GPT-4oで100万トークンあたり2.50ドル）。大規模で独立したレート制限枠。結果は24時間以内に返却。

実装：バッチワークフロー

Batch APIは、リクエスト自体に標準的なJSON POSTボディを使用しません。代わりに、各行が個別のリクエストとなっている .jsonl ファイルをアップロードします。OpenAIは、余剰リソースがあるときにこれらをバックグラウンドで処理します。

1. リクエストファイルの作成 (.jsonl)

すべてのリクエストには custom_id が必要です。出力ファイルの順序は入力時と同じとは限らないため、これは非常に重要です。このIDを使用して、結果をデータベースにマッピングし直します。

{"custom_id": "user-rev-101", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "感情を分類してください。"}, {"role": "user", "content": "素晴らしい製品です！"}]}}
{"custom_id": "user-rev-102", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "感情を分類してください。"}, {"role": "user", "content": "壊れて届きました。"}]}}

2. ジョブのアップロードと実行

ファイルの準備ができたら、Python SDKを使用してOpenAIに送信します。以下は、私が本番環境で使用しているボイラープレートです。

import openai

client = openai.OpenAI(api_key="your_api_key")

# JSONLファイルをOpenAIのサーバーにアップロード
batch_file = client.files.create(
  file=open("tasks.jsonl", "rb"),
  purpose="batch"
)

# バッチ処理ジョブを開始
batch_job = client.batches.create(
  input_file_id=batch_file.id,
  endpoint="/v1/chat/completions",
  completion_window="24h"
)

print(f"ジョブが開始されました: {batch_job.id}")

3. データの取得

1秒ごとにAPIをポーリングする必要はありません。私は通常、15分ごとにステータスを確認するシンプルなスクリプトを設定しています。ステータスが completed になれば、結果ファイルをダウンロードできます。

# ジョブが完了したか確認
job_status = client.batches.retrieve(batch_job.id)

if job_status.status == "completed":
    # 実際のレスポンスデータを取得
    results = client.files.content(job_status.output_file_id)
    with open("final_results.jsonl", "wb") as f:
        f.write(results.read())
    print("処理が完了しました。結果を保存しました。")

本番環境で得た教訓

数千万トークンを処理して気づいた、ドキュメントには書かれていないパターンがあります。まず、24時間という枠はかなり控えめな見積もりだということです。5万件のリクエストを含むバッチのほとんどは、45分以内に完了します。ただし、ステークホルダーに対して「バッチジョブは1時間以内に終わる」と約束してはいけません。

次に、バリデーション（検証）が非常に重要です。500MBのJSONLファイルのたった1行にカンマが抜けているだけで、バッチジョブ全体が開始前に失敗する可能性があります。私は現在、アップロードボタンを押す前に、Pydanticスクリプトを使ってローカルですべての行を検証しています。これで何時間ものフラストレーションを回避できます。

部分的な成功の処理

バッチ内のすべてのリクエストが成功するとは限りません。特定のプロンプトがコンテンツフィルタに抵触したり、内部サーバーエラーが発生したりしても、OpenAIは残りの9,999件のリクエストについては結果を返します。後処理のコードでは、出力ファイル内の各 custom_id ごとに status_code を確認する必要があります。

コスト・メリットの分析

最後に：Hot vs Cold アーキテクチャ

Batch APIを採用したことで、AIシステムの設計方法が変わりました。現在、私はタスクを2つのバケツに分けています。チャットボックスでユーザーが質問するような「Hot（即時）」なタスクは同期APIのままにします。週次レポートの生成やデータベースへのタグ付けのような「Cold（非即時）」なタスクはBatch APIに移行します。このシンプルな切り分けにより、月間予算を増やすことなく、処理能力を10倍にスケールさせることができました。もし大量のデータ処理にまだ同期呼び出しを使っているなら、予算の半分を捨てているようなものです。