Batch API で Gemini を 50% 安く使う — 非同期バッチ処理とマルチモーダル対応

🎯 はじめに：即時応答が要らないなら、必ず使うべき機能

本記事はシリーズ 「Gemini AI プラットフォーム活用」の第 3 回 です。前回 第 2 回 で Gemini ファミリーのモデル選定を扱いました。今回は、そのすべてのモデルを 50% オフで使えるようにする Batch API を扱います。

⚡ 結論を先に: 即時応答が不要なタスクで Batch API を使わないのは、課金的に「半額クーポンを捨てている」のと同じです。夜間バッチ、商品説明の一括生成、評価データセット作成など、24 時間以内に終わればよいタスクは全部 Batch に寄せましょう。

📦 Batch API とは

通常 API との比較

非同期な代わりに、通常料金の半額でリクエストを処理します。Anthropic の Message Batch API と思想は同じで、Google も「即時性が要らないなら半額にする」というインセンティブ設計を採用しています。

💰 料金比較：Batch API は単純に半額

※ 料金は per 1M tokens（USD）

月間コストのインパクト試算

たとえば EC サイトで 毎日 10,000 件の商品説明を Flash で生成するケース。

入力 500 トークン × 出力 200 トークン × 10,000 件 × 30 日

通常:
入力: 500 × 10,000 × 30 / 1,000,000 = 150M tokens × $0.15 = $22.50
出力: 200 × 10,000 × 30 / 1,000,000 =  60M tokens × $0.60 = $36.00
合計: $58.50/月

Batch API:
入力: 150M × $0.075 = $11.25
出力:  60M × $0.30  = $18.00
合計: $29.25/月

→ 月 $29.25（約 4,400 円）削減

商品数が 10 倍 100 倍になればこの差はそのままスケールします。

🎬 マルチモーダルにも 50% オフが効く

意外と知られていませんが、画像・動画・音声・Embedding すべてのモダリティで Batch 50% オフが適用されます。

💡 「音声 100 時間の文字起こしを夜間に投げて朝に結果回収」のようなワークフローが、通常の半額で組めます。これが BPO 業務の自動化の現実解です。

✅ 向き / 不向きの判断

✅ Batch API が刺さるケース

• EC: 数万 SKU の商品説明・SEO テキスト一括生成
• コールセンター: 録音データの文字起こし + 要約
• 法務: 契約書ライブラリの一括レビュー / 抽出
• 動画: 監視カメラ / 学習動画の自動メタデータ付与
• RAG 構築: 数百万件の埋め込み生成
• 評価: LLM-as-a-Judge での品質スコアリング
• 夜間バッチ: 22 時投入 → 朝には結果

❌ 向かないケース

• リアルタイムチャット（ユーザーが待てない）
• インタラクティブな UI（即時応答が UX 要件）
• ヒトのレビューを挟むワークフロー

🛠 実装：2 つの投入方式

Batch API は規模に応じて 2 つの投入方式を使い分けます。

方式 A: インラインリクエスト（小規模、20MB 以下）

from google import genai

client = genai.Client()

batch_job = client.batches.create(
    model="gemini-2.5-flash",
    requests=[
        {"contents": [{"parts": [{"text": "商品A: 高品質なワイヤレスイヤホン"}]}]},
        {"contents": [{"parts": [{"text": "商品B: 軽量で持ち運びやすいノートPC"}]}]},
        {"contents": [{"parts": [{"text": "商品C: 防水機能付きスマートウォッチ"}]}]},
    ],
    config={"display_name": "product-descriptions-batch"}
)

print(f"バッチジョブ作成: {batch_job.name}")

数百〜数千件、合計 20MB 以下の小〜中規模バッチ向き。スクリプト 1 ファイルで完結します。

方式 B: JSONL ファイル投入（大規模）

事前に JSONL ファイル（1 行 1 リクエスト）を作って File API でアップロードします。

# batch_requests.jsonl の中身
# {"key": "req_1", "request": {"contents": [{"parts": [{"text": "質問1"}]}]}}
# {"key": "req_2", "request": {"contents": [{"parts": [{"text": "質問2"}]}]}}

from google import genai
import time

client = genai.Client()

# 1. JSONL ファイルをアップロード
uploaded_file = client.files.upload(file="batch_requests.jsonl")

# 2. バッチジョブを作成
batch_job = client.batches.create(
    model="gemini-2.5-flash",
    src=uploaded_file.name,
    config={"display_name": "large-batch-job"}
)

# 3. ポーリングで完了待ち
while batch_job.state.name not in ['JOB_STATE_SUCCEEDED', 'JOB_STATE_FAILED']:
    time.sleep(60)
    batch_job = client.batches.get(name=batch_job.name)
    print(f"状態: {batch_job.state.name}")

# 4. 結果ダウンロード
if batch_job.state.name == 'JOB_STATE_SUCCEEDED':
    result_file = client.files.download(file=batch_job.dest.file_name)
    for line in result_file.decode('utf-8').splitlines():
        print(json.loads(line))

数万〜数百万件、または 20MB 超のバッチに使います。key を付けることで結果と元リクエストを照合できます。

マルチモーダル × Batch（画像分析）

from google import genai
import json, time

client = genai.Client()

# 1. 画像を File API でアップロード
images = [client.files.upload(file=p) for p in ["p1.jpg", "p2.jpg", "p3.jpg"]]

# 2. PROCESSING → ACTIVE を待つ
for f in images:
    while f.state.name == "PROCESSING":
        time.sleep(2)
        f = client.files.get(name=f.name)

# 3. JSONL でバッチリクエストを作成
with open("image_batch.jsonl", "w") as out:
    for i, img in enumerate(images):
        req = {
            "key": f"img_{i}",
            "request": {
                "contents": [{
                    "parts": [
                        {"file_data": {"file_uri": img.uri, "mime_type": "image/jpeg"}},
                        {"text": "この商品画像を分析して、商品説明を生成してください。"}
                    ]
                }]
            }
        }
        out.write(json.dumps(req) + "\n")

# 4. バッチ投入（Flash + 画像でも 50% オフ）
batch_file = client.files.upload(file="image_batch.jsonl")
batch_job = client.batches.create(
    model="gemini-2.5-flash",
    src=batch_file.name,
    config={"display_name": "image-analysis-batch"}
)

Embeddings × Batch

RAG の初期構築で大量ベクトル化が必要なケースに最適です。

embedding_batch = client.batches.create_embeddings(
    model="gemini-embedding-001",
    requests=[
        {"content": "商品Aの説明文..."},
        {"content": "商品Bの説明文..."},
        {"content": "商品Cの説明文..."},
    ],
    config={"display_name": "embeddings-batch"}
)

🧰 他のコスト最適化との組み合わせ

「Flash-Lite + Batch」が事実上の最強

Flash-Lite 通常:    $0.075 入力 / $0.30 出力
Flash-Lite + Batch: $0.0375 入力 / $0.15 出力
                    ↑ 通常 Flash の 1/4 のコスト！

単純分類・抽出・タグ付けのような Flash で十分なタスクは、Flash-Lite + Batch に倒すと劇的にコストが下がります。

Context Caching との使い分け

Context Caching は「同じ長いプロンプトを何度も使う」用途に特化（90% オフ）。逆に Batch API は「個別の異なるリクエストを大量にまとめる」用途に特化（50% オフ）。両者は併用できません（Cache が優先）が、性質も使い所も異なります。

🧪 ベストプラクティス

1. 小ジョブを統合する

NG: 200 リクエスト × 1,000 ジョブ
OK: 200,000 リクエスト × 1 ジョブ

ジョブ単位で管理オーバーヘッドが発生するため、まとめて投げるほうが効率的です。

2. Flash-Lite と組み合わせる

単純なタスク（分類、抽出、タグ付け）は迷わず Flash-Lite + Batch。コスト 1/4 で十分な品質が出ます。

3. 夜間バッチに最適化

24 時間 SLO なので、22 時投入 → 朝 9 時に結果回収のワークフローが自然に組めます。Cloud Scheduler や GitHub Actions の schedule で運用化しましょう。

4. エラーハンドリングは結果ファイル単位で

一部のリクエストが失敗しても全体は継続します。完了後に結果 JSONL を走査し、error フィールドがある行を抽出して再投入する方式が定石です。

for line in result_file.decode('utf-8').splitlines():
    obj = json.loads(line)
    if 'error' in obj:
        # 再投入用に退避
        retry_queue.append(obj['key'])
    else:
        save_to_db(obj['key'], obj['response'])

5. key フィールドで突合性を担保

JSONL 投入では各リクエストに "key": "your_id" を付与すると、結果と元データの紐付けが楽になります（DB の主キー、ファイル名などを使う）。

✅ 第 3 回まとめ

• Batch API は 通常料金の 50% オフ。即時性不要なタスクは全部寄せるべき
• マルチモーダル全部対応（画像・動画・音声・Embedding）
• 規模で 2 方式を使い分け: 小規模ならインライン、大規模なら JSONL + File API
• Flash-Lite + Batch が最強コスパ（通常 Flash の 1/4）
• Context Caching とは併用不可（Cache が優先）。性質が違うので使い分け
• 夜間バッチで運用化、結果ファイルで一括エラーハンドリング、key で突合性確保

次回 Python で Gemini を扱う実装ガイド では、ここまで紹介した API（テキスト生成、Embedding、マルチモーダル、Batch）を Python から使うコードを総まとめします。

📚 シリーズ記事

🔗 関連リソース

• Gemini Batch API 公式ドキュメント
• Batch API（Gemini） — 用語集