AI Batch API完全ガイド：OpenAI・Claude・Geminiの大量処理術

OpenAI Batch APIで大量リクエストを一括処理する：JONLファイル作成からステータス確認まで

Batch APIを使い、数百〜数千件のリクエストを一括で送信してコストを50%削減する方法を実装します。JONLファイルの作成、バッチジョブの投入、完了確認、結果取得までの全工程を確認しましょう。

from openai import OpenAI
import json
import time

client = OpenAI()

# ステップ1: バッチリクエスト用のJSONLファイルを作成
requests_data = [
    {
        "custom_id": f"req-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "技術用語を小学生にもわかるように1文で説明してください。"},
                {"role": "user", "content": term}
            ],
            "max_tokens": 200
        }
    }
    for i, term in enumerate([
        "ニューラルネットワーク", "勾配降下法", "過学習",
        "正則化", "バッチ正規化", "ドロップアウト",
        "転移学習", "ファインチューニング", "蒸留",
        "アテンション機構"
    ])
]

jsonl_path = "batch_requests.jsonl"
with open(jsonl_path, "w", encoding="utf-8") as f:
    for req in requests_data:
        f.write(json.dumps(req, ensure_ascii=False) + "\n")
print(f"JONLファイル作成完了: {len(requests_data)}件")

# ステップ2: ファイルをアップロード
batch_file = client.files.create(
    file=open(jsonl_path, "rb"),
    purpose="batch"
)
print(f"ファイルアップロード完了: {batch_file.id}")

# ステップ3: バッチジョブを作成
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    metadata={"description": "AI用語の簡単説明を一括生成"}
)
print(f"バッチジョブ作成: {batch_job.id} (status={batch_job.status})")

# ステップ4: 完了をポーリングで待機
while batch_job.status in ["validating", "in_progress", "finalizing"]:
    time.sleep(30)
    batch_job = client.batches.retrieve(batch_job.id)
    completed = batch_job.request_counts.completed
    total = batch_job.request_counts.total
    print(f"  status={batch_job.status}  ({completed}/{total})")

# ステップ5: 結果を取得
if batch_job.status == "completed":
    result_file = client.files.content(batch_job.output_file_id)
    results = result_file.text.strip().split("\n")
    for line in results:
        res = json.loads(line)
        custom_id = res["custom_id"]
        content = res["response"]["body"]["choices"][0]["message"]["content"]
        print(f"  {custom_id}: {content}")
    print(f"\n合計 {len(results)} 件の結果を取得")
elif batch_job.status == "failed":
    print(f"バッチ失敗: {batch_job.errors}")
    # エラーファイルがある場合
    if batch_job.error_file_id:
        errors = client.files.content(batch_job.error_file_id)
        print(errors.text)

Batch APIの最大のメリットはコスト削減です。通常のAPI呼び出しと比べて50%安い料金で同じモデル・同じ品質の結果が得られます。ただし、完了まで最大24時間かかるため、リアルタイム性が不要な用途に限定されます。典型的なユースケースは、大量テキストの分類・翻訳、データセットへのラベル付け、コンテンツの一括生成です。JONLファイルの各行が独立したリクエストのため、1件のエラーが他に影響しない点も堅牢性の面で優れています。