◈ API / SDK/2026-03-21上級

Gemini Batch Processing API — 大量リクエストを50%オフで効率処理する

Gemini APIのBatch Processing機能を徹底解説。数千～数万件のリクエストを非同期で一括処理し、コストを50%削減する実装パターン。Python/TypeScript実装例、エラーハンドリング、本番運用のベストプラクティスまで。

Gemini API¹⁷⁹ Batch Processing コスト最適化¹⁹ 非同期処理⁵ 大規模データ

✦ プレミアム記事

背景と前提

AIアプリケーションが本番環境でスケールすると、リアルタイム処理では対応しきれないユースケースが必ず発生します。数千件の顧客レビューの感情分析、数万件のドキュメントの要約生成、大量の画像に対するキャプション付与——こうした大規模な非同期処理には、Gemini APIの Batch Processing（バッチ処理） が最適解です。

Batch Processing APIを使えば、通常の同期APIと比較して 50%のコスト削減 が実現でき、レート制限を気にすることなく大量のリクエストを一括送信できます。処理結果は24時間以内に返却され、その間アプリケーションは他のタスクに集中できます。

このガイドでは、Batch Processing APIの仕組みから、本番環境で活用するための実装パターン、エラーハンドリング、コスト最適化戦略までを包括的に解説します。

Batch Processing APIの基本概念

なぜバッチ処理が必要なのか

同期APIでは、1リクエストごとにレスポンスを待つ必要があります。大量処理の場合、以下の問題が発生します。

レート制限: モデルやティアによって1分あたりのリクエスト数（RPM）が制限される
コスト増大: 同期APIのフル料金が適用される
タイムアウト: 長時間実行のリクエストがタイムアウトする可能性
エラー処理の複雑さ: 個別リクエストの失敗をリアルタイムで処理する必要がある

Batch Processing APIは、これらの問題をすべて解決します。

料金体系

Batch Processing APIの最大のメリットはコストです。

| 処理方式 | 入力コスト | 出力コスト | 備考 | |---------|----------|----------|------| | 同期API（リアルタイム） | 標準料金 | 標準料金 | 即座にレスポンス | | Batch Processing | 標準の 50% | 標準の 50% | 24時間以内にレスポンス |

ℹ️

**コスト試算例**: Gemini 2.5 Flash で10万件のテキスト分類（各500トークン入力・100トークン出力）を実行する場合、同期APIでは約$3.75のところ、バッチ処理なら約$1.88で処理できます。

処理フロー

Batch Processing APIの処理フローは以下の3ステップです。

バッチジョブの作成: リクエスト群をまとめてジョブを作成
非同期処理: Googleのインフラが自動的にリクエストを処理（最大24時間）
結果の取得: ジョブ完了後に結果を一括取得

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦Gemini Batch Processing APIを活用した大量処理の最適化

✦バッチ処理のコスト削減と処理時間の短縮

✦本番環境での信頼性と監視

Stripe による安全な決済 · いつでもキャンセル可能

Python でのバッチ処理実装

基本的なバッチジョブの作成

import google.generativeai as genai
import json
import time
 
genai.configure(api_key="YOUR_API_KEY")
 
def create_batch_job(requests: list[dict], model: str = "gemini-2.5-flash") -> str:
    """
    バッチジョブを作成し、ジョブIDを返す。
    各リクエストはカスタムIDとコンテンツを含む。
    """
    batch_requests = []
    for req in requests:
        batch_requests.append(
            genai.types.BatchRequest(
                custom_id=req["id"],
                request=genai.types.GenerateContentRequest(
                    model=f"models/{model}",
                    contents=[
                        genai.types.Content(
                            parts=[genai.types.Part(text=req["prompt"])]
                        )
                    ],
                    config=genai.types.GenerateContentConfig(
                        temperature=0.3,
                        max_output_tokens=1024,
                    ),
                ),
            )
        )
 
    batch_job = genai.batches.create(
        model=f"models/{model}",
        requests=batch_requests,
        config=genai.types.CreateBatchJobConfig(
            display_name="my-batch-job",
        ),
    )
 
    print(f"バッチジョブ作成: {batch_job.name}")
    print(f"状態: {batch_job.state}")
    return batch_job.name

ジョブの監視と結果取得

def wait_for_batch_completion(job_name: str, poll_interval: int = 30) -> dict:
    """
    バッチジョブの完了を待ち、結果を取得する。
    """
    while True:
        job = genai.batches.get(name=job_name)
        state = job.state.name
 
        if state == "JOB_STATE_SUCCEEDED":
            print(f"ジョブ完了: 成功={job.succeeded_count}, 失敗={job.failed_count}")
            return collect_results(job)
 
        elif state == "JOB_STATE_FAILED":
            raise RuntimeError(f"バッチジョブ失敗: {job.error}")
 
        elif state == "JOB_STATE_CANCELLED":
            raise RuntimeError("バッチジョブがキャンセルされました")
 
        else:
            total = job.total_count or 0
            succeeded = job.succeeded_count or 0
            progress = (succeeded / total * 100) if total > 0 else 0
            print(f"処理中... {succeeded}/{total} ({progress:.1f}%)")
            time.sleep(poll_interval)
 
def collect_results(job) -> dict:
    """
    完了したバッチジョブから結果を収集する。
    """
    results = {}
    for response in job.responses:
        custom_id = response.custom_id
        if response.response:
            text = response.response.candidates[0].content.parts[0].text
            results[custom_id] = {
                "status": "success",
                "text": text,
                "usage": {
                    "input_tokens": response.response.usage_metadata.prompt_token_count,
                    "output_tokens": response.response.usage_metadata.candidates_token_count,
                },
            }
        else:
            results[custom_id] = {
                "status": "error",
                "error": str(response.error),
            }
    return results

実践例：大量テキストの感情分析

def batch_sentiment_analysis(texts: list[dict]) -> dict:
    """
    大量のテキストに対して感情分析をバッチ実行する。
    texts: [{"id": "review_001", "text": "この製品は素晴らしい..."}, ...]
    """
    requests = []
    for item in texts:
        prompt = f"""以下のテキストの感情を分析し、JSON形式で回答してください。
 
テキスト: {item["text"]}
 
出力形式:
{{"sentiment": "positive|negative|neutral", "confidence": 0.0-1.0, "summary": "一行要約"}}"""
 
        requests.append({"id": item["id"], "prompt": prompt})
 
    # 1バッチあたり最大100件に分割
    BATCH_SIZE = 100
    all_results = {}
 
    for i in range(0, len(requests), BATCH_SIZE):
        chunk = requests[i : i + BATCH_SIZE]
        job_name = create_batch_job(chunk)
        results = wait_for_batch_completion(job_name)
        all_results.update(results)
        print(f"バッチ {i // BATCH_SIZE + 1} 完了: {len(chunk)}件処理")
 
    return all_results

TypeScript でのバッチ処理実装

import { GoogleGenAI } from "@google/genai";
 
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY! });
 
interface BatchRequest {
  id: string;
  prompt: string;
}
 
interface BatchResult {
  status: "success" | "error";
  text?: string;
  error?: string;
  usage?: { inputTokens: number; outputTokens: number };
}
 
async function createAndRunBatch(
  requests: BatchRequest[],
  model = "gemini-2.5-flash"
): Promise<Map<string, BatchResult>> {
  // バッチジョブの作成
  const batchRequests = requests.map((req) => ({
    customId: req.id,
    request: {
      model: `models/${model}`,
      contents: [{ parts: [{ text: req.prompt }] }],
      generationConfig: {
        temperature: 0.3,
        maxOutputTokens: 1024,
      },
    },
  }));
 
  const batchJob = await ai.batches.create({
    model: `models/${model}`,
    requests: batchRequests,
    displayName: "ts-batch-job",
  });
 
  console.log(`バッチジョブ作成: ${batchJob.name}`);
 
  // ポーリングで完了を待つ
  const results = new Map<string, BatchResult>();
  let job = batchJob;
 
  while (true) {
    job = await ai.batches.get({ name: job.name! });
 
    if (job.state === "JOB_STATE_SUCCEEDED") {
      for (const resp of job.responses ?? []) {
        if (resp.response) {
          results.set(resp.customId!, {
            status: "success",
            text: resp.response.candidates?.[0]?.content?.parts?.[0]?.text,
            usage: {
              inputTokens: resp.response.usageMetadata?.promptTokenCount ?? 0,
              outputTokens:
                resp.response.usageMetadata?.candidatesTokenCount ?? 0,
            },
          });
        } else {
          results.set(resp.customId!, {
            status: "error",
            error: String(resp.error),
          });
        }
      }
      break;
    }
 
    if (job.state === "JOB_STATE_FAILED" || job.state === "JOB_STATE_CANCELLED") {
      throw new Error(`Batch job ${job.state}: ${job.error}`);
    }
 
    console.log(`処理中... ${job.succeededCount ?? 0}/${job.totalCount ?? 0}`);
    await new Promise((r) => setTimeout(r, 30_000));
  }
 
  return results;
}

マルチモーダルバッチ処理

Batch Processing APIはテキストだけでなく、画像や動画などのマルチモーダル入力にも対応しています。

import base64
from pathlib import Path
 
def create_multimodal_batch(image_tasks: list[dict]) -> str:
    """
    画像を含むマルチモーダルバッチジョブを作成する。
    image_tasks: [{"id": "img_001", "image_uri": "gs://bucket/img.jpg", "prompt": "画像を説明して"}]
    """
    batch_requests = []
 
    for task in image_tasks:
        parts = []
 
        # Cloud Storage のURIを使用
        if task["image_uri"].startswith("gs://"):
            parts.append(
                genai.types.Part(
                    file_data=genai.types.FileData(
                        file_uri=task["image_uri"],
                        mime_type="image/jpeg",
                    )
                )
            )
        # ローカルファイルをBase64エンコード
        else:
            image_bytes = Path(task["image_uri"]).read_bytes()
            parts.append(
                genai.types.Part(
                    inline_data=genai.types.Blob(
                        data=base64.b64encode(image_bytes).decode(),
                        mime_type="image/jpeg",
                    )
                )
            )
 
        parts.append(genai.types.Part(text=task["prompt"]))
 
        batch_requests.append(
            genai.types.BatchRequest(
                custom_id=task["id"],
                request=genai.types.GenerateContentRequest(
                    model="models/gemini-2.5-flash",
                    contents=[genai.types.Content(parts=parts)],
                ),
            )
        )
 
    batch_job = genai.batches.create(
        model="models/gemini-2.5-flash",
        requests=batch_requests,
    )
 
    return batch_job.name

⚠️

**注意**: マルチモーダルバッチ処理では、Files APIでアップロードしたファイルのURIまたはCloud StorageのURIを使用してください。Base64エンコードはリクエストサイズが増大するため、大量処理には向きません。

エラーハンドリングとリトライ戦略

本番環境では、個別リクエストの失敗に対する堅牢なエラーハンドリングが不可欠です。

from dataclasses import dataclass, field
from datetime import datetime
 
@dataclass
class BatchJobTracker:
    """バッチジョブの実行を追跡し、失敗したリクエストのリトライを管理する"""
 
    max_retries: int = 3
    failed_requests: list = field(default_factory=list)
    retry_count: dict = field(default_factory=dict)
 
    def process_results(self, results: dict, original_requests: list[dict]):
        """結果を処理し、失敗したリクエストを記録する"""
        for req in original_requests:
            rid = req["id"]
            result = results.get(rid, {})
 
            if result.get("status") == "error":
                current_retries = self.retry_count.get(rid, 0)
                if current_retries < self.max_retries:
                    self.retry_count[rid] = current_retries + 1
                    self.failed_requests.append(req)
                    print(f"リトライ対象: {rid} (試行 {current_retries + 1}/{self.max_retries})")
                else:
                    print(f"リトライ上限到達: {rid}")
 
    def get_retry_requests(self) -> list[dict]:
        """リトライ対象のリクエストを返す"""
        requests = self.failed_requests.copy()
        self.failed_requests.clear()
        return requests
 
def robust_batch_processing(requests: list[dict], model: str = "gemini-2.5-flash") -> dict:
    """
    リトライ機能付きのバッチ処理。
    失敗したリクエストは最大3回までリトライする。
    """
    tracker = BatchJobTracker(max_retries=3)
    all_results = {}
    pending = requests.copy()
 
    round_num = 0
    while pending:
        round_num += 1
        print(f"\n--- ラウンド {round_num}: {len(pending)}件処理 ---")
 
        job_name = create_batch_job(pending, model=model)
        results = wait_for_batch_completion(job_name)
 
        # 成功した結果を保存
        for rid, result in results.items():
            if result["status"] == "success":
                all_results[rid] = result
 
        # 失敗したリクエストのリトライ判定
        tracker.process_results(results, pending)
        pending = tracker.get_retry_requests()
 
        if pending:
            print(f"リトライ対象: {len(pending)}件")
 
    return all_results

コスト最適化のベストプラクティス

1. モデルの適切な選択

タスクの複雑さに応じてモデルを使い分けることで、さらなるコスト削減が可能です。

| タスク種別 | 推奨モデル | 理由 | |----------|----------|------| | テキスト分類・感情分析 | Gemini 2.5 Flash | 低コスト・高速・十分な精度 | | 文書要約・翻訳 | Gemini 2.5 Flash | コストパフォーマンスが最良 | | 複雑な推論・コード生成 | Gemini 2.5 Pro | 精度が重要なタスク | | 画像分析・キャプション | Gemini 2.5 Flash | マルチモーダル対応・低コスト |

2. プロンプトの最適化

バッチ処理ではリクエスト数が多いため、プロンプトの最適化がコストに直結します。

# ❌ 冗長なプロンプト（不要なトークン消費）
bad_prompt = """
あなたはプロの感情分析の専門家です。
以下のテキストについて、非常に詳細な感情分析を行ってください。
まず、テキスト全体のトーンを評価し、次に各文の感情を分析し、
最後に総合的な感情スコアを算出してください。
出力は必ずJSON形式にしてください。
テキスト: {text}
"""
 
# ✅ 簡潔なプロンプト（同等の精度でトークン節約）
good_prompt = """テキストの感情をJSON出力: {{"sentiment":"positive|negative|neutral","confidence":0.0-1.0}}
 
テキスト: {text}"""

3. バッチサイズの最適化

大規模処理では、適切なバッチサイズの選択が処理効率に影響します。

def optimize_batch_size(total_requests: int) -> int:
    """
    リクエスト数に基づいて最適なバッチサイズを返す。
    """
    if total_requests <= 100:
        return total_requests  # 小規模：1バッチで処理
    elif total_requests <= 1000:
        return 100  # 中規模：100件ずつ
    else:
        return 500  # 大規模：500件ずつ（並列ジョブで処理）

本番環境での運用パターン

Webhook通知による非同期ワークフロー

ポーリングの代わりにWebhookを使用すれば、効率的な非同期ワークフローを構築できます。

from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
@app.route("/webhook/batch-complete", methods=["POST"])
def batch_complete():
    """バッチジョブ完了時のWebhookエンドポイント"""
    payload = request.json
    job_name = payload["name"]
    state = payload["state"]
 
    if state == "JOB_STATE_SUCCEEDED":
        # 結果を取得して後続処理を実行
        job = genai.batches.get(name=job_name)
        results = collect_results(job)
        process_completed_results(results)
 
    return jsonify({"status": "received"})

Cloud Schedulerとの連携

定期的なバッチ処理をCloud Schedulerと組み合わせることで、完全自動化が可能です。

from google.cloud import scheduler_v1
 
def schedule_daily_batch():
    """毎日深夜にバッチ処理を実行するスケジュールを作成"""
    client = scheduler_v1.CloudSchedulerClient()
 
    job = scheduler_v1.Job(
        name="projects/my-project/locations/us-central1/jobs/daily-batch",
        http_target=scheduler_v1.HttpTarget(
            uri="https://my-api.run.app/trigger-batch",
            http_method=scheduler_v1.HttpMethod.POST,
        ),
        schedule="0 2 * * *",  # 毎日午前2時
        time_zone="Asia/Tokyo",
    )
 
    client.create_job(parent="projects/my-project/locations/us-central1", job=job)