Akira

Posted on Jun 25 • Originally published at apidog.com

OpenAI Batch API の使い方

このガイドでは、OpenAIのBatch APIを使って、数千件のモデルリクエストを1つの非同期ジョブとして実行する手順を実装ベースで説明します。JSONLファイルを作成し、バッチを送信し、完了までポーリングし、結果をダウンロードします。本番に組み込む前に、各API呼び出しはApidogで検証できます。対話的な処理が必要な場合は、同期APIを使い、ApidogでChatGPT APIをテストしてください。

今すぐApidogを試す

Batch APIとは何か、いつ使うべきか

Batch APIは、即時応答が不要な大量のモデル呼び出しを非同期に処理するためのAPIです。

通常の同期APIでは、プロンプトごとにHTTPリクエストを送ります。一方、Batch APIでは複数のリクエストをJSONLファイルにまとめ、1つのジョブとして送信します。その後、ジョブのステータスをポーリングし、完了後に結果ファイルを取得します。

主なメリットは次の2つです。

同期APIと比べて、入力トークンと出力トークンの両方が50%割引になる
バッチジョブは個別のレート制限プールを使うため、ライブトラフィックと競合しにくい

トレードオフはレイテンシーです。OpenAIは24時間以内の完了を保証していますが、即時処理には向きません。

Batch APIが適している処理は次のようなものです。

過去データの分類やタグ付け
コーパス全体の埋め込み生成
商品説明、要約、翻訳などの大量コンテンツ生成
評価スイートやモデル比較の一括実行

チャットUI、オートコンプリート、ライブエージェントなど、ユーザーが応答を待つ処理には同期エンドポイントを使ってください。多数のモデル設定やエージェント設定をまとめて生成する場合は、バッチ処理で100以上のエージェント設定を生成する方法も参考になります。

全体フロー

Batch APIの基本フローは、/v1/files と /v1/batches を使う4ステップです。

ステップ	エンドポイント	内容
1. アップロード	`POST /v1/files`	`purpose: "batch"` を付けて `.jsonl` ファイルを送信し、ファイルIDを取得する
2. 作成	`POST /v1/batches`	ファイルID、対象エンドポイント、完了ウィンドウを指定してバッチを作成する
3. ポーリング	`GET /v1/batches/{id}`	`status` が `completed` になるまで確認する
4. 取得	`GET /v1/files/{id}/content`	`output_file_id` を使って結果ファイルをダウンロードする

事前に用意するものは次の3つです。

OpenAI APIキー
リクエストを格納したJSONLファイル
API呼び出しを実行・検証するツール

以下では、APIキーを環境変数 OPENAI_API_KEY に設定している前提で進めます。

export OPENAI_API_KEY="your_api_key"

ステップ1: JSONLファイルを作成する

Batch APIの入力はJSONLファイルです。各行が1つの独立したリクエストになります。

各行には次のフィールドが必要です。

フィールド	内容
`custom_id`	入力と出力を照合するための一意なID
`method`	通常は `POST`
`url`	実行対象のエンドポイント
`body`	通常のAPIリクエスト本文

例:

{"custom_id": "req-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4.1-mini", "messages": [{"role": "user", "content": "Classify the sentiment of: 'shipping was slow but the product is great'"}]}}
{"custom_id": "req-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4.1-mini", "messages": [{"role": "user", "content": "Classify the sentiment of: 'returned it the same day'"}]}}

重要なポイント:

custom_id はファイル内で一意にする
結果の順序は保証されないため、必ず custom_id で照合する
1つのバッチは最大50,000リクエストまで
ファイルサイズは最大200MBまで

ステップ2: JSONLファイルをアップロードする

JSONLファイルをFiles APIにアップロードします。purpose には "batch" を指定します。

curl https://api.openai.com/v1/files \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -F purpose="batch" \
  -F file="@requests.jsonl"

レスポンス例:

{
  "id": "file-abc123",
  "object": "file",
  "purpose": "batch"
}

ここで返る id が、次のステップで使う input_file_id です。

ステップ3: バッチを作成する

アップロードしたファイルIDを使ってバッチジョブを作成します。

curl https://api.openai.com/v1/batches \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input_file_id": "file-abc123",
    "endpoint": "/v1/chat/completions",
    "completion_window": "24h",
    "metadata": {
      "job": "sentiment-backfill"
    }
  }'

指定する主な項目は次の通りです。

フィールド	内容
`input_file_id`	アップロード済みJSONLファイルのID
`endpoint`	JSONL内の `url` と一致するエンドポイント
`completion_window`	現在は `"24h"`
`metadata`	任意のタグ情報

endpoint はJSONL内の url と一致している必要があります。

対応するターゲットには、次のようなエンドポイントがあります。

/v1/chat/completions
/v1/responses
/v1/embeddings
/v1/completions
/v1/moderations

レスポンス例:

{
  "id": "batch_abc123",
  "object": "batch",
  "endpoint": "/v1/chat/completions",
  "input_file_id": "file-abc123",
  "completion_window": "24h",
  "status": "validating",
  "output_file_id": null,
  "error_file_id": null,
  "request_counts": {
    "total": 0,
    "completed": 0,
    "failed": 0
  },
  "created_at": 1733452800,
  "metadata": {
    "job": "sentiment-backfill"
  }
}

この時点では、通常 status は validating です。

ステップ4: バッチステータスをポーリングする

バッチの状態は GET /v1/batches/{batch_id} で確認します。

curl https://api.openai.com/v1/batches/batch_abc123 \
  -H "Authorization: Bearer $OPENAI_API_KEY"

主なステータスは次の通りです。

ステータス	意味
`validating`	入力ファイルを検証中
`in_progress`	リクエストを処理中
`finalizing`	出力ファイルを準備中
`completed`	完了。結果をダウンロード可能
`failed`	検証に失敗。処理は実行されない
`expired`	24時間以内に全リクエストが完了しなかった
`cancelling` / `cancelled`	キャンセル処理中またはキャンセル済み

request_counts を見ると進捗を確認できます。

{
  "request_counts": {
    "total": 50000,
    "completed": 32000,
    "failed": 10
  }
}

毎秒ポーリングする必要はありません。数分ごとなど、余裕を持った間隔で確認するのが実用的です。

誤って送信したバッチはキャンセルできます。

curl -X POST https://api.openai.com/v1/batches/batch_abc123/cancel \
  -H "Authorization: Bearer $OPENAI_API_KEY"

ステップ5: 結果ファイルを取得する

status が completed になると、バッチオブジェクトに output_file_id が入ります。

{
  "status": "completed",
  "output_file_id": "file-output456",
  "error_file_id": "file-error789"
}

結果ファイルをダウンロードします。

curl https://api.openai.com/v1/files/file-output456/content \
  -H "Authorization: Bearer $OPENAI_API_KEY" > results.jsonl

出力もJSONL形式です。各行には、元の custom_id とレスポンス情報が含まれます。

例:

{"custom_id":"req-1","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"positive"}}]}}}
{"custom_id":"req-2","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"negative"}}]}}}

注意点:

出力順序は入力順と一致しない
必ず custom_id で入力データと結合する
失敗したリクエストは error_file_id のファイルも確認する

Pythonで結果を入力データに結合する例

結果の順序は保証されないため、custom_id をキーにして処理します。

import json

results = {}

with open("results.jsonl", "r", encoding="utf-8") as f:
    for line in f:
        row = json.loads(line)
        custom_id = row["custom_id"]
        results[custom_id] = row

print(results["req-1"])

入力側のデータと結合する場合は、同じ custom_id を持たせておくと扱いやすくなります。

inputs = [
    {"custom_id": "req-1", "text": "shipping was slow but the product is great"},
    {"custom_id": "req-2", "text": "returned it the same day"},
]

for item in inputs:
    result = results.get(item["custom_id"])
    if not result:
        continue

    item["batch_result"] = result["response"]["body"]

print(inputs)

コストと24時間ウィンドウの考え方

Batch APIでは、同期APIと比べて入力トークンと出力トークンの両方が50%割引になります。

一方で、最大24時間の待ち時間を受け入れる必要があります。夜間バッチ、バックフィル、評価処理などには向いていますが、ユーザー操作のクリティカルパスには向きません。

実装時は次の点を考慮してください。

24時間は上限であり、即時完了は保証されない
expired になった場合、完了済みリクエストのみが返され課金される
バッチはライブトラフィックとは別のエンキューされたトークン制限を使う
半額でも大量に処理すれば総額は増える
metadata を使ってジョブ単位で費用を追跡しやすくする

同期API側の制限を確認したい場合は、GPT APIのレート制限とそれらをテストする方法を参照してください。費用を機能やジョブごとに割り当てたい場合は、OpenAI費用に対するコストアトリビューションのプレイブックが参考になります。

ApidogでBatch APIをテストする

Batch APIは、通常の1回のチャットAPI呼び出しよりも失敗箇所が多くなります。

主な失敗ポイントは次の通りです。

JSONLの形式ミス
必須フィールドの欠落
endpoint と url の不一致
ファイルアップロードの失敗
バッチ検証エラー
ポーリング処理の実装ミス
出力ファイルとエラーファイルの処理漏れ

Apidogでは、各API呼び出しをリクエストとして作成し、レスポンスを確認しながらライフサイクル全体を検証できます。OpenAI SDKではなく、APIエンドポイントをテスト・モックするためのプラットフォームです。

実際の検証手順は次のようになります。

JSONLを事前検証する

各行に custom_id、method、url、body が含まれているか確認します。body.model や messages の欠落も事前に検出します。
ファイルアップロードを実行する

POST /v1/files をマルチパートで送信し、purpose=batch とJSONLファイルを指定します。返された id を環境変数として保存します。
バッチ作成を実行する

POST /v1/batches を送信し、status が validating であること、endpoint が期待値と一致することを確認します。
ステータスをポーリングする

GET /v1/batches/{id} を実行し、status と request_counts を確認します。
結果ファイルを取得する

completed 後に output_file_id を取得し、GET /v1/files/{id}/content で結果をダウンロードします。
エラー処理を確認する

意図的に壊したJSONLを送信し、failed や error_file_id の扱いを確認します。キャンセル処理も POST /v1/batches/{id}/cancel でテストします。

出力ファイルの取得処理を先に実装したい場合は、モックAPIを使って、完了済みバッチオブジェクトやサンプル結果ファイルを返すAPIを作れます。これにより、実際の24時間ジョブを待たずにパース処理や結合処理をテストできます。

仕様先行で開発する場合は、OpenAPI仕様から直接テストコレクションを生成し、バッチ関連エンドポイントをCIの回帰テストに含めることもできます。

よくある質問

バッチ処理にはどれくらい時間がかかりますか？

OpenAIは、バッチを24時間以内に完了することを保証しています。多くのジョブはそれより早く終わる場合がありますが、実装上は24時間かかる前提で設計してください。

24時間以内にすべてのリクエストが完了しなかった場合、バッチは expired になります。その場合、完了済みリクエストのみが返され、課金されます。

割引率はどれくらいですか？

Batch APIでは、同期エンドポイントと比べて入力トークンと出力トークンの両方が50%割引になります。

費用をジョブや機能ごとに追跡したい場合は、metadata を使って識別子を付けておくと後で集計しやすくなります。コストアトリビューションプレイブックも参考にしてください。

バッチで実行できるエンドポイントはどれですか？

JSONL内の url と、バッチ作成時の endpoint に同じエンドポイントを指定します。

サポートされているターゲットには、次のようなものがあります。

/v1/chat/completions
/v1/responses
/v1/embeddings
/v1/completions
/v1/moderations

画像や動画のエンドポイントも対象に含まれます。対応エンドポイントは追加される可能性があるため、最新のOpenAIドキュメントを確認してください。

結果の順序が入力と違うのはなぜですか？

Batch APIでは、出力JSONLの行順は保証されません。

そのため、すべてのリクエストに一意な custom_id を設定し、結果を custom_id で入力に結合する必要があります。同じ custom_id を複数行で使うと、レスポンスを安全に対応付けできません。

まとめ

Batch APIを使うと、JSONLにまとめた大量のモデルリクエストを非同期に実行し、24時間以内に結果を取得できます。同期APIよりレイテンシーは大きくなりますが、入力・出力トークンの両方で50%割引を受けられます。

実装の流れは次の通りです。

JSONLファイルを作成する
POST /v1/files に purpose=batch でアップロードする
POST /v1/batches でバッチを作成する
GET /v1/batches/{id} でステータスをポーリングする
output_file_id を使って結果をダウンロードする
custom_id で入力と出力を結合する

自動化する前に、Apidogをダウンロードして、アップロード、作成、ポーリング、キャンセル、結果取得の各エンドポイントを手動で検証してください。JSONLの不正な1行を事前に検出できれば、24時間の無駄な待ち時間を避けられます。

DEV Community