OpenAI Batch APIの使い方を解説。JSONLファイルの作り方からPython実装例、料金50%オフの条件、大量翻訳やデータ分類の実用ユースケースまでステップバイステップで紹介します。
使い方ガイド
公開: by ToolCraft Lab 約10分で読めます

ChatGPT API Batch完全ガイド — 料金50%オフで大量処理を実行する方法【2026年版】

OpenAI Batch APIの使い方を解説。JSONLファイルの作り方からPython実装例、料金50%オフの条件、大量翻訳やデータ分類の実用ユースケースまでステップバイステップで紹介します。

#OpenAI#Batch API#コスト削減

「同じAPIを使って料金が半額になる方法、知っていますか?」 OpenAIのBatch APIを使えば、大量のリクエストを50%オフの料金で処理できます。即時性が不要なタスクなら、使わない手はありません。

筆者が実際にBatch APIを使って1,000件の翻訳を処理したところ、通常APIと比べて約48%のコスト削減になりました。gpt-4o-miniを使った場合で、$1.2が$0.63に。「たったそれだけ?」と思うかもしれませんが、これが月に数万件のリクエストになると話は変わってきます。

本記事では、Batch APIの仕組みから、JSONLファイルの作り方、Pythonでの実装例、料金体系、そして実際に業務で使える大量翻訳・データ分類のユースケースまで、ステップバイステップで解説します。ChatGPT APIの基本的な使い方をまだ押さえていない方は、先に「ChatGPT APIの使い方 Pythonガイド」を読んでおくとスムーズです。

Batch APIとは

OpenAI Batch APIは、大量のAPIリクエストを一括で非同期処理するための仕組みです。通常のAPIとは異なり、リクエストをまとめてファイルとして送信し、24時間以内に結果が返ってきます。

通常APIとBatch APIの違い

項目通常APIBatch API
処理方式同期(リアルタイム)非同期(バッチ処理)
レスポンス時間数秒〜数十秒最大24時間以内
料金通常料金通常の50%オフ
レート制限通常のTPM/RPM制限別枠のレート制限
入力形式個別のAPIリクエストJSONLファイル
適したユースケースチャット、リアルタイム応答大量データ処理、バッチ翻訳

なぜ50%オフなのか

Batch APIが半額で提供される理由は、OpenAIがリクエストの実行タイミングを自由にスケジューリングできるためです。サーバーの負荷が低い時間帯に処理を回すことで、インフラコストを削減し、その分をユーザーに還元しています。

対応モデル

Batch APIは以下のモデルで利用可能です。

モデル通常料金(入力/出力)Batch料金(50%オフ)
gpt-4o$2.50 / $10.00 per 1M tokens$1.25 / $5.00 per 1M tokens
gpt-4o-mini$0.15 / $0.60 per 1M tokens$0.075 / $0.30 per 1M tokens
gpt-4.1$2.00 / $8.00 per 1M tokens$1.00 / $4.00 per 1M tokens
gpt-4.1-mini$0.40 / $1.60 per 1M tokens$0.20 / $0.80 per 1M tokens
gpt-4.1-nano$0.10 / $0.40 per 1M tokens$0.05 / $0.20 per 1M tokens

JSONLファイルの作り方

Batch APIへのリクエストは、JSONL(JSON Lines) 形式のファイルで送信します。1行につき1リクエストを記述します。

JSONLファイルの基本構造

{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "日本語で回答してください"}, {"role": "user", "content": "Pythonのリスト内包表記とは何ですか?"}], "max_tokens": 500}}
{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "日本語で回答してください"}, {"role": "user", "content": "Dockerのメリットを3つ教えてください"}], "max_tokens": 500}}
{"custom_id": "request-3", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "日本語で回答してください"}, {"role": "user", "content": "REST APIの設計原則を説明してください"}], "max_tokens": 500}}

各フィールドの説明

フィールド説明必須
custom_idリクエストを識別するための一意なID
methodHTTPメソッド(POST固定)
urlAPIエンドポイント
body通常のAPI リクエストボディと同じ

PythonでJSONLファイルを生成する

import json

def create_batch_file(requests: list[dict], output_path: str):
    """バッチリクエスト用のJSONLファイルを生成する"""
    with open(output_path, 'w', encoding='utf-8') as f:
        for i, req in enumerate(requests):
            line = {
                "custom_id": f"request-{i+1}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": req.get("model", "gpt-4o-mini"),
                    "messages": req["messages"],
                    "max_tokens": req.get("max_tokens", 500)
                }
            }
            f.write(json.dumps(line, ensure_ascii=False) + "\n")

# 使用例
requests = [
    {
        "messages": [
            {"role": "system", "content": "日本語で簡潔に回答してください"},
            {"role": "user", "content": "Pythonとは?"}
        ]
    },
    {
        "messages": [
            {"role": "system", "content": "日本語で簡潔に回答してください"},
            {"role": "user", "content": "JavaScriptとは?"}
        ]
    },
    {
        "messages": [
            {"role": "system", "content": "日本語で簡潔に回答してください"},
            {"role": "user", "content": "Rustとは?"}
        ]
    }
]

create_batch_file(requests, "batch_input.jsonl")
print("JSONLファイルを生成しました")

Python実装例 — バッチ処理の全手順

Batch APIを使った処理は、以下の4ステップで行います。

1. JSONLファイルをアップロード
2. バッチジョブを作成
3. ステータスを確認(完了を待つ)
4. 結果をダウンロード

完全な実装コード

import json
import time
from openai import OpenAI

client = OpenAI()

def run_batch_pipeline(input_file: str) -> list[dict]:
    """Batch APIの全パイプラインを実行する"""

    # ステップ1: ファイルのアップロード
    print("📤 ファイルをアップロード中...")
    with open(input_file, "rb") as f:
        uploaded_file = client.files.create(
            file=f,
            purpose="batch"
        )
    print(f"   ファイルID: {uploaded_file.id}")

    # ステップ2: バッチジョブの作成
    print("🚀 バッチジョブを作成中...")
    batch = client.batches.create(
        input_file_id=uploaded_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
        metadata={
            "description": "技術用語の説明バッチ"
        }
    )
    print(f"   バッチID: {batch.id}")
    print(f"   ステータス: {batch.status}")

    # ステップ3: 完了を待つ
    print("⏳ 処理完了を待機中...")
    while True:
        batch = client.batches.retrieve(batch.id)
        status = batch.status

        if status == "completed":
            print(f"   ✅ 完了! 処理件数: {batch.request_counts.completed}")
            break
        elif status == "failed":
            print(f"   ❌ 失敗: {batch.errors}")
            return []
        elif status == "expired":
            print("   ⚠️ 期限切れ")
            return []
        else:
            completed = batch.request_counts.completed
            total = batch.request_counts.total
            print(f"   ステータス: {status} ({completed}/{total})")
            time.sleep(30)  # 30秒ごとにチェック

    # ステップ4: 結果のダウンロード
    print("📥 結果をダウンロード中...")
    output_file_id = batch.output_file_id
    result_content = client.files.content(output_file_id)

    results = []
    for line in result_content.text.strip().split("\n"):
        result = json.loads(line)
        results.append({
            "custom_id": result["custom_id"],
            "status": result["response"]["status_code"],
            "content": result["response"]["body"]["choices"][0]["message"]["content"]
        })

    print(f"   {len(results)}件の結果を取得しました")
    return results


# 実行
results = run_batch_pipeline("batch_input.jsonl")

for r in results:
    print(f"\n--- {r['custom_id']} ---")
    print(r["content"])

実行結果の例

python batch_process.py

📤 ファイルをアップロード中...
   ファイルID: file-abc123
🚀 バッチジョブを作成中...
   バッチID: batch-xyz789
   ステータス: validating
 処理完了を待機中...
   ステータス: in_progress (0/3)
   ステータス: in_progress (1/3)
 完了! 処理件数: 3
📥 結果をダウンロード中...
   3件の結果を取得しました

--- request-1 ---
Pythonは、読みやすさを重視したプログラミング言語です...

--- request-2 ---
JavaScriptは、Web開発で広く使われるプログラミング言語です...

--- request-3 ---
Rustは、安全性とパフォーマンスを両立するシステムプログラミング言語です...

エラーハンドリング

バッチ処理では一部のリクエストが失敗する場合があります。エラーファイルの確認方法を解説します。

エラーファイルの取得

def check_batch_errors(batch_id: str):
    """バッチジョブのエラーを確認する"""
    batch = client.batches.retrieve(batch_id)

    if batch.error_file_id:
        error_content = client.files.content(batch.error_file_id)
        errors = []
        for line in error_content.text.strip().split("\n"):
            error = json.loads(line)
            errors.append({
                "custom_id": error["custom_id"],
                "error_code": error["response"]["body"]["error"]["code"],
                "error_message": error["response"]["body"]["error"]["message"]
            })

        print(f"エラー件数: {len(errors)}")
        for e in errors:
            print(f"  {e['custom_id']}: [{e['error_code']}] {e['error_message']}")

        return errors

    print("エラーなし")
    return []

よくあるエラーと対処法

エラー原因対処法
invalid_request_errorJSONLの形式が不正ファイルのバリデーションを実行
rate_limit_exceededバッチのレート制限に到達時間をおいて再送信
context_length_exceeded入力トークンが上限超過テキストを短縮するか分割
server_errorOpenAI側の一時的なエラー該当リクエストのみ再送信

リトライの実装

def retry_failed_requests(
    original_file: str,
    error_ids: list[str],
    output_file: str
):
    """失敗したリクエストだけを抽出してリトライ用ファイルを生成"""
    with open(original_file, 'r') as f:
        lines = f.readlines()

    retry_lines = []
    for line in lines:
        data = json.loads(line)
        if data["custom_id"] in error_ids:
            retry_lines.append(line)

    with open(output_file, 'w') as f:
        f.writelines(retry_lines)

    print(f"{len(retry_lines)}件のリトライファイルを生成しました")

料金50%オフの条件と注意点

割引が適用される条件

Batch APIの50%割引は、以下の条件で自動的に適用されます。

  1. Batch APIのエンドポイントを使用すること — 通常のChat Completions APIではなく、Batches APIを経由する
  2. completion_windowを24hに設定すること — 24時間以内の完了が保証される
  3. JSONLファイルでリクエストを送信すること — 個別リクエストではなくバッチ形式

特別な申請やプランの契約は不要です。

コスト試算の例

1万件のテキスト分類タスクを、gpt-4o-miniで処理する場合のコスト比較です。

前提条件:
- リクエスト数: 10,000件
- 平均入力トークン: 200 tokens/件
- 平均出力トークン: 50 tokens/件

通常API:
  入力: 200 × 10,000 = 2,000,000 tokens × $0.15/1M = $0.30
  出力: 50 × 10,000 = 500,000 tokens × $0.60/1M = $0.30
  合計: $0.60

Batch API(50%オフ):
  入力: 2,000,000 tokens × $0.075/1M = $0.15
  出力: 500,000 tokens × $0.30/1M = $0.15
  合計: $0.30

→ 節約額: $0.30(50%削減)

gpt-4oのような高額モデルでは、節約効果がさらに大きくなります。各モデルの料金を詳しく比較したい方は「ChatGPT vs Claude API料金比較」も参考にしてください。

gpt-4oで同じ処理をする場合:
  通常API合計: $30.00
  Batch API合計: $15.00
  → 節約額: $15.00

注意点

  • 即時レスポンスが必要なケースには使えない: 最大24時間かかる可能性がある
  • ストリーミングは非対応: 結果は一括で返却される
  • ファイルサイズの上限: 1ファイルあたり最大200MBまたは50,000リクエスト
  • 同時実行制限: アカウントごとに同時に実行できるバッチ数に制限がある

ユースケース1: 大量テキストの翻訳

ドキュメントやUIテキストの大量翻訳にBatch APIは最適です。

実装例

import json
import csv

def create_translation_batch(
    csv_path: str,
    source_lang: str,
    target_lang: str,
    output_path: str
):
    """CSVファイルのテキストを翻訳するバッチファイルを生成"""
    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        rows = list(reader)

    with open(output_path, 'w', encoding='utf-8') as f:
        for i, row in enumerate(rows):
            line = {
                "custom_id": f"translate-{i+1}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o-mini",
                    "messages": [
                        {
                            "role": "system",
                            "content": f"""{source_lang}から{target_lang}に翻訳してください。
翻訳のルール:
- 自然で流暢な{target_lang}にする
- 技術用語はそのまま残す
- 原文のトーンを維持する
- 翻訳結果のみを出力する(説明不要)"""
                        },
                        {
                            "role": "user",
                            "content": row["text"]
                        }
                    ],
                    "max_tokens": 1000,
                    "temperature": 0.3
                }
            }
            f.write(json.dumps(line, ensure_ascii=False) + "\n")

    print(f"{len(rows)}件の翻訳バッチファイルを生成しました")

# 使用例
create_translation_batch(
    csv_path="texts_to_translate.csv",
    source_lang="英語",
    target_lang="日本語",
    output_path="translation_batch.jsonl"
)

結果のCSV出力

def save_translation_results(results: list[dict], output_csv: str):
    """翻訳結果をCSVに保存"""
    # custom_idでソート
    results.sort(key=lambda x: int(x["custom_id"].split("-")[1]))

    with open(output_csv, 'w', encoding='utf-8', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["id", "translated_text"])
        for r in results:
            writer.writerow([r["custom_id"], r["content"]])

    print(f"翻訳結果を {output_csv} に保存しました")

ユースケース2: データ分類・ラベリング

大量のテキストデータにカテゴリラベルを付与する作業も、Batch APIの得意分野です。

実装例

def create_classification_batch(
    texts: list[str],
    categories: list[str],
    output_path: str
):
    """テキスト分類のバッチファイルを生成"""
    categories_str = ", ".join(categories)

    with open(output_path, 'w', encoding='utf-8') as f:
        for i, text in enumerate(texts):
            line = {
                "custom_id": f"classify-{i+1}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o-mini",
                    "messages": [
                        {
                            "role": "system",
                            "content": f"""以下のテキストを分類してください。
カテゴリ: {categories_str}

必ず以下のJSON形式で出力してください:
{{"category": "カテゴリ名", "confidence": 0.0-1.0}}"""
                        },
                        {
                            "role": "user",
                            "content": text
                        }
                    ],
                    "max_tokens": 50,
                    "temperature": 0,
                    "response_format": {"type": "json_object"}
                }
            }
            f.write(json.dumps(line, ensure_ascii=False) + "\n")

    print(f"{len(texts)}件の分類バッチファイルを生成しました")

# 使用例
texts = [
    "商品が届かないのですが、配送状況を確認したいです",
    "この製品の保証期間はどのくらいですか?",
    "返品したいのですが手続きを教えてください",
    "素晴らしい商品でした!また購入したいです",
    "アプリが起動しません。エラーが表示されます",
]

categories = ["配送", "商品情報", "返品・交換", "フィードバック", "技術サポート"]

create_classification_batch(texts, categories, "classification_batch.jsonl")

ユースケース3: コンテンツ生成

大量のメタディスクリプションや商品説明文の生成にも活用できます。

SEOメタディスクリプション生成

def create_meta_description_batch(
    pages: list[dict],
    output_path: str
):
    """Webページのメタディスクリプションを一括生成"""
    with open(output_path, 'w', encoding='utf-8') as f:
        for i, page in enumerate(pages):
            line = {
                "custom_id": f"meta-{i+1}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o-mini",
                    "messages": [
                        {
                            "role": "system",
                            "content": """SEOに最適化されたメタディスクリプションを生成してください。
ルール:
- 120〜160文字以内
- ターゲットキーワードを自然に含める
- ユーザーがクリックしたくなる内容にする
- 日本語で出力"""
                        },
                        {
                            "role": "user",
                            "content": f"タイトル: {page['title']}\nキーワード: {page['keyword']}\n概要: {page['summary']}"
                        }
                    ],
                    "max_tokens": 200,
                    "temperature": 0.7
                }
            }
            f.write(json.dumps(line, ensure_ascii=False) + "\n")

# 使用例
pages = [
    {"title": "Pythonで始めるWeb開発", "keyword": "Python Web開発", "summary": "Flask/Djangoを使ったWeb開発入門"},
    {"title": "Docker入門ガイド", "keyword": "Docker 使い方", "summary": "コンテナ技術の基礎から実践まで"},
]

create_meta_description_batch(pages, "meta_batch.jsonl")

バッチジョブの管理

ジョブ一覧の取得

def list_batches(limit: int = 10):
    """バッチジョブの一覧を表示"""
    batches = client.batches.list(limit=limit)

    for batch in batches.data:
        print(f"ID: {batch.id}")
        print(f"  ステータス: {batch.status}")
        print(f"  作成日時: {batch.created_at}")
        print(f"  完了: {batch.request_counts.completed}/{batch.request_counts.total}")
        print()

ジョブのキャンセル

def cancel_batch(batch_id: str):
    """実行中のバッチジョブをキャンセル"""
    client.batches.cancel(batch_id)
    print(f"バッチ {batch_id} のキャンセルをリクエストしました")

バッチ処理のベストプラクティス

  1. 小さいバッチでテスト: まず10〜100件で動作確認する
  2. custom_idを活用: 入力データと結果を紐付けやすいIDを設定する
  3. エラーハンドリング: 部分的な失敗に備えてリトライロジックを用意する
  4. コスト見積もり: 大量処理の前にtiktokenでトークン数を見積もる
  5. 結果の検証: ランダムサンプリングで品質を確認する
import tiktoken

def estimate_batch_cost(
    input_file: str,
    model: str = "gpt-4o-mini"
):
    """バッチ処理のコストを見積もる"""
    enc = tiktoken.encoding_for_model(model)

    # モデルごとのBatch API料金
    pricing = {
        "gpt-4o-mini": {"input": 0.075, "output": 0.30},
        "gpt-4o": {"input": 1.25, "output": 5.00},
    }

    total_input_tokens = 0
    request_count = 0

    with open(input_file, 'r') as f:
        for line in f:
            data = json.loads(line)
            messages = data["body"]["messages"]
            for msg in messages:
                total_input_tokens += len(enc.encode(msg["content"]))
            request_count += 1

    # 出力トークンは入力の25%と仮定
    estimated_output_tokens = total_input_tokens * 0.25

    price = pricing.get(model, pricing["gpt-4o-mini"])
    input_cost = (total_input_tokens / 1_000_000) * price["input"]
    output_cost = (estimated_output_tokens / 1_000_000) * price["output"]

    print(f"リクエスト数: {request_count}")
    print(f"推定入力トークン: {total_input_tokens:,}")
    print(f"推定出力トークン: {int(estimated_output_tokens):,}")
    print(f"推定コスト: ${input_cost + output_cost:.4f}")

# 使用例
estimate_batch_cost("batch_input.jsonl", "gpt-4o-mini")

Responses APIでのBatch処理

Responses APIでもBatch処理が利用可能です。URLを変更するだけで対応できます。

{"custom_id": "resp-1", "method": "POST", "url": "/v1/responses", "body": {"model": "gpt-4o-mini", "input": "Pythonのデコレータとは何ですか?", "max_output_tokens": 500}}
{"custom_id": "resp-2", "method": "POST", "url": "/v1/responses", "body": {"model": "gpt-4o-mini", "input": "TypeScriptの型推論とは何ですか?", "max_output_tokens": 500}}

Chat Completions APIとResponses APIのどちらでも50%割引が適用されます。Responses APIの詳細は「OpenAI Responses APIガイド」で解説しています。

まとめ — Batch APIでコストを最適化しよう

OpenAI Batch APIは、以下の条件に当てはまるタスクでは必ず検討すべき選択肢です。

  • 即時レスポンスが不要: 数時間〜24時間待てるタスク
  • 大量のリクエスト: 数百〜数万件の処理
  • コスト削減が優先: 通常料金の50%オフ

特に効果的なユースケース

ユースケース推定コスト削減
大量翻訳(1万件)50%削減
データ分類・ラベリング50%削減
コンテンツ生成(メタデータ等)50%削減
テストデータの生成50%削減
文書の要約処理50%削減

まずは10件程度の小さなJSONLファイルで流れを掴んでみてください。筆者の経験では、最初のバッチ実行は完了まで15分ほどでした(数千件だと数時間かかることもあります)。一度パイプラインを組んでしまえば、翻訳でもデータ分類でも使い回せます。月のAPI費用が$50を超えるようなプロジェクトなら、Batch APIへの移行だけで年間$300以上の節約になる計算です。