---
title: "Gemini Batch APIの使い方：JSONLで半額実行・完了待ち・失敗行の再投入をPythonで実装【2026年11月】"
url: "https://www.issoh.co.jp/tech/details/19447/"
published: 2026-11-07
updated: 2026-11-07
categories: ["AI"]
publisher: "株式会社一創"
---

# Gemini Batch APIの使い方：JSONLで半額実行・完了待ち・失敗行の再投入をPythonで実装【2026年11月】

Gemini Batch APIは、大量のリクエストをまとめて非同期で処理させ、通常料金の50%で結果を受け取る仕組みです。完了の目標は24時間で、即答が要らない問い合わせの分類や文書の要約を夜間に流す用途に向きます。本記事では、2026年11月時点の公式ドキュメントをもとに、半額になる条件とジョブの状態、JSONLの書き方、Pythonでの投入、完了待ちのポーリングとWebhook、結果の突き合わせ、失敗した行だけの再投入までを、そのまま動かせるコードで説明します。

## まとめ：Gemini Batch APIは24時間待てる大量処理を半額にする仕組み

件数が多く、結果を翌朝までに受け取れればよい処理は、Batch APIに移すと費用が半分になります。gemini-3.8-flashなら入力0.375ドル・出力1.875ドル（100万トークンあたり）です。ただし無料枠では使えず、課金を設定したプロジェクトが前提です。

実装で外せない点は4つあります。1行1件のJSONLに`key`を付けて結果を元データへ戻せるようにすること。ジョブ作成は冪等ではないので、二重投入を自分で防ぐこと。結果ファイルは48時間で消えるので、完了後すぐに取り込むこと。失敗した行は`key`で拾い、その分だけで新しいジョブを作ることです。

数分以内に結果が要る処理は、同じ半額のFlexか通常の呼び出しのままにします。Gemini APIの全体像は親記事の[Gemini APIとは？できること・料金・始め方とInteractions APIの最小コード](https://www.issoh.co.jp/tech/details/18759/)にまとめています。

## Gemini Batch APIの仕様：半額の条件・24時間目標・48時間の期限切れ

コードを書く前に、料金と時間の約束事を押さえます。ここを誤解すると、夜間処理が朝に終わっていない、無料枠のキーで投入できない、といった詰まり方をします。

### 通常料金の50%になるモデルと無料枠ではBatchを使えない条件

[Batch APIの公式ドキュメント](https://aistudio.google.com/docs/batch-api)によると、料金は同じモデルの通常料金の50%です。[料金ページ](https://aistudio.google.com/docs/pricing)では、gemini-3.8-flashのBatchは入力0.375ドル・出力1.875ドルで、2027年1月1日から通常料金とともに2倍になります。

| モデル                     | 通常（入力／出力）  | Batch（入力／出力） | 無料枠のBatch |
| ----------------------- | ---------- | ------------ | --------- |
| gemini-3.8-flash        | 0.75／3.75  | 0.375／1.875  | 不可        |
| gemini-3.5-flash-lite   | 0.30／2.50  | 0.15／1.25    | 可         |
| gemini-4-argon（12月1日から） | 4.00／20.00 | 2.00／10.00   | 不可        |

単位は100万トークンあたりのドルで、2026年11月7日時点の表示です。料金ページは3.8 FlashのBatchの無料枠を「Not available」としています。無料枠のキーで試すなら3.5 Flash-Liteを指定してください。モデルごとの単価と課金の設定は[Gemini APIの料金と無料枠：モデル別単価・課金設定・費用の計算コード](https://www.issoh.co.jp/tech/details/18813/)で詳しく扱っています。

### ジョブ状態6種とPENDING・RUNNINGが48時間を超えたときの扱い

ジョブは次の6つの状態をとります。24時間は目標であって保証ではありません。待機中または実行中のまま48時間を超えると`JOB_STATE_EXPIRED`になり、結果は1件も受け取れません。

- `JOB_STATE_PENDING`：作成済みで処理待ち
- `JOB_STATE_RUNNING`：処理中
- `JOB_STATE_SUCCEEDED`：完了。結果を取得できる
- `JOB_STATE_FAILED`：ジョブ全体が失敗
- `JOB_STATE_CANCELLED`：利用者が取り消した
- `JOB_STATE_EXPIRED`：48時間を超えて期限切れ

実務で注意したいのは`SUCCEEDED`の意味です。ジョブが完了しただけで、全行が成功したとは限りません。個々の行の失敗は結果ファイルの中に残るので、後述の手順で行ごとに確かめます。

### Interactions API未対応時のgenerate\_content形式

2026年6月に正式版になったInteractions APIは、[公式の概要ページ](https://aistudio.google.com/docs/interactions-overview)でBatchを「まだ使えない」機能に挙げています。そのためBatchの各行は、従来の`generate_content`と同じ`contents`・`generation_config`の形で書きます。

同期の処理をInteractions APIで書いている場合も、Batchに回す分だけはこの形式に合わせます。両者の形式を照合するための対応表は[Gemini APIの使い方（Python）：google-genai 3.0で生成・ストリーミング・関数呼び出し](https://www.issoh.co.jp/tech/details/19441/)にまとめられた資料です。OpenAIのSDKから`batches`で投入する互換の経路もありますが、[互換レイヤーのページ](https://aistudio.google.com/docs/openai)によるとファイルの送受信は互換レイヤーに対応しておらず、結局google-genaiが要ります。

## JSONL入力ファイルを作りFile APIへ上げてバッチジョブを投入する手順

ここからはgoogle-genai 3.0.0（[PyPI](https://pypi.org/project/google-genai/)・Python 3.11以上）を前提にします。APIキーは環境変数`GEMINI_API_KEY`に入れておきます。

### keyとrequestで1行1件を書くJSONLの形式とインライン20MBの境界

入力の渡し方は2通りです。リクエストをリストで直接渡すインライン方式は、合計が20MB未満の小さなバッチ向けです。それより大きいときは、1行に1件を書いたJSONLファイルを使います。入力ファイルの上限は1ファイル2GBです。

```
{"key": "inq-0001", "request": {"contents": [{"role": "user", "parts": [{"text": "次の問い合わせを分類してください。\n請求書の再発行をお願いします。"}]}]}}
{"key": "inq-0002", "request": {"contents": [{"role": "user", "parts": [{"text": "次の問い合わせを分類してください。\nログインできません。"}]}]}}
```

`key`は自由に決められる識別子で、結果の行にも同じ値が付きます。結果ファイルの行の順番を当てにせず、元データの主キー（問い合わせ番号など）を必ず入れてください。`request`の中には、行ごとに`generation_config`やシステム指示、ツールの指定も書けます。

### client.files.uploadとbatches.createによるジョブ投入

次のコードは、元データからJSONLを作り、File APIに上げてジョブを作成するまでを1本にまとめたものです。分類の指示文とデータは例なので、自社の処理に置き換えます。

```
import json
from google import genai
from google.genai import types

PROMPT = "次の問い合わせを「請求」「アカウント」「その他」のどれかに分類し、分類名だけを返してください。\n"
records = [  # 元データ。id は問い合わせ番号などの主キー
    {"id": "inq-0001", "text": "請求書の再発行をお願いします。"},
    {"id": "inq-0002", "text": "ログインできません。"},
]
run_id = "inquiry-2026-11-07"   # 実行単位の名前。二重投入の確認にも使う

with open(f"{run_id}.jsonl", "w", encoding="utf-8") as f:
    for r in records:
        line = {
            "key": r["id"],
            "request": {
                "contents": [{"role": "user", "parts": [{"text": PROMPT + r["text"]}]}],
                "generation_config": {"max_output_tokens": 200},
            },
        }
        f.write(json.dumps(line, ensure_ascii=False) + "\n")

with genai.Client() as client:
    uploaded = client.files.upload(
        file=f"{run_id}.jsonl",
        config=types.UploadFileConfig(display_name=run_id, mime_type="jsonl"),
    )
    job = client.batches.create(
        model="gemini-3.8-flash",
        src=uploaded.name,                    # アップロードしたファイルを入力にする
        config={"display_name": run_id},
    )
    print(job.name)   # 「batches/123456789」の形式。DBに保存しておく
```

返ってきた`job.name`は、完了の確認と結果の取得に使う唯一の手がかりです。標準出力に流すだけにせず、実行単位の名前と組にしてDBか設定ファイルに保存します。[File APIのページ](https://aistudio.google.com/docs/files)によると、アップロードしたファイルは48時間で消え、1プロジェクトで合計20GBまでです。

### ジョブ作成が冪等でない点と実行単位で二重投入を防ぐ確認の入れ方

公式ドキュメントのベストプラクティスには、ジョブの作成は冪等ではないと書かれています。同じ作成リクエストを2回送ると、ジョブが2つでき、料金も2回分かかります。夜間処理のスクリプトがタイムアウトで再実行された、という場面で起きやすい事故です。

防ぎ方は単純で、作成の前に同じ`display_name`のジョブが既にあるかを`client.batches.list()`で確かめます。

```
def find_job(client, run_id):
    """同じ実行単位の名前を持つジョブが既にあれば返す"""
    for j in client.batches.list():
        if j.display_name == run_id and j.state.name != "JOB_STATE_CANCELLED":
            return j
    return None

existing = find_job(client, run_id)
if existing:
    print("作成済みのため投入しない:", existing.name)
```

一覧を毎回すべて読むのが重い場合は、`job.name`を保存したDBの側で「この実行単位は投入済み」を記録し、そちらを先に見ます。どちらの方法でも、作成と記録の間で落ちたときに備え、一覧での確認は残しておくのが安全です。

## 完了待ちと結果の取得：ポーリング・Webhook・48時間の保存期限

投入したジョブは、終わったかどうかを自分で確かめるか、終わったら通知を受けるかのどちらかで待ちます。件数が少ない社内ツールならポーリングで十分です。

### batches.getを30秒間隔で回し4つの終了状態で抜けるループ

公式の例は、`client.batches.get()`を30秒ごとに呼び、終了を表す4つの状態のどれかになったらループを抜けます。

```
import time

DONE = {"JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED",
        "JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED"}

job = client.batches.get(name=job_name)   # job_name は保存しておいた batches/...
while job.state.name not in DONE:
    time.sleep(30)
    job = client.batches.get(name=job_name)
print("終了:", job.state.name)
```

このループは、最長で48時間プロセスを抱え続けます。cronなどで動かすなら、ループの代わりに15分おきの定期実行で1回だけ`get`し、終わっていなければそのまま抜ける形にします。サーバーの再起動でも待ちが途切れません。

### Webhookのbatch.succeededで受けるときの署名検証と重複排除

ポーリングをやめたいときは、[Webhookの公式ページ](https://aistudio.google.com/docs/webhooks)の仕組みで完了通知を受けます。2026年11月時点ではプレビューです。`client.webhooks.create()`に`uri`と`subscribed_events`（`batch.succeeded`・`batch.failed`など）を渡して登録し、作成時に一度だけ返る署名用のシークレットを保管します。

```
import os
from flask import Flask, request
from standardwebhooks.webhooks import Webhook, WebhookVerificationError

app = Flask(__name__)
wh = Webhook(os.environ["GEMINI_WEBHOOK_SECRET"])   # 登録時の new_signing_secret
seen = set()   # 本番はDBなど再起動で消えない場所に持つ

@app.post("/gemini-callback")
def callback():
    payload = request.get_data(as_text=True)   # 加工前の本文で検証する
    try:
        event = wh.verify(payload, request.headers)
    except WebhookVerificationError:
        return "", 400
    msg_id = request.headers.get("webhook-id")
    if msg_id in seen:          # 同じ通知が複数回届くことがある
        return "", 200
    seen.add(msg_id)
    if event["type"] == "batch.succeeded":
        queue_download(event["data"]["id"])   # 取り込みは別の処理に回し、すぐ200を返す
    return "", 200
```

公式ページによると、2xxを数秒以内に返さないと、指数バックオフで24時間まで再送されます。配信は少なくとも1回（at-least-once）なので、`webhook-id`での重複排除が必要です。結果の取り込みのような重い処理を受け口の中で行わないのはこのためです。公式ページは、`webhook-timestamp`が5分より古い通知を拒否することも勧めています。

### 結果ファイルをfiles.downloadで取りkeyで元データへ戻す処理

完了したジョブの結果は、`job.dest.file_name`が指すJSONLファイルに入っています。各行は応答（`response`）かエラー（`error`）のどちらかを持ちます。

```
results, failed = {}, []
if job.state.name == "JOB_STATE_SUCCEEDED":
    raw = client.files.download(file=job.dest.file_name).decode("utf-8")
    for line in raw.splitlines():
        row = json.loads(line)
        key = row.get("key")
        if "response" in row:
            cands = row["response"].get("candidates") or []
            parts = cands[0].get("content", {}).get("parts", []) if cands else []
            text = "".join(p.get("text", "") for p in parts).strip()
            if text:
                results[key] = text
            else:
                failed.append(key)   # 応答はあるが本文が空（安全フィルタ等）
        else:
            failed.append(key)       # error を持つ行
print("成功", len(results), "件 / 失敗", len(failed), "件")
```

応答があっても本文が空の行は、成功に数えずに失敗の側へ入れています。安全フィルタで止まった行や出力の上限に達した行を、空の分類結果としてDBに書き込まないためです。インライン方式で投入した場合は、`job.dest.inlined_responses`を順に読みます。

## 失敗行の検出と再投入、投入上限に合わせたジョブ分割のPython実装例

数万件を流すと、一部の行が失敗するのは珍しくありません。全件を流し直すと費用が二重にかかるため、失敗した行だけを拾って次のジョブにします。

### failedRequestCountと行ごとのerrorで失敗分だけ作り直す方法

公式ドキュメントは、完了後に`batchStats`の`failedRequestCount`を確かめ、結果ファイルを行ごとに解析するよう勧めています。件数で全体の様子をつかみ、行の`key`で対象を特定する二段構えです。

```
all_ids = {r["id"] for r in records}
retry_ids = set(failed) | (all_ids - set(results))   # 結果に現れなかった行も含める
retry_records = [r for r in records if r["id"] in retry_ids]

if retry_records:
    retry_run_id = run_id + "-retry1"
    # 前の手順と同じ方法で retry_records から JSONL を作り、
    # find_job() で未投入を確かめてから batches.create する
    print("再投入:", len(retry_records), "件 ->", retry_run_id)
```

再投入は1〜2回で打ち切ります。同じ行が続けて失敗するなら、入力が長すぎる、画像が壊れているなど、データの側に原因があることが多いからです。残った行は人が確かめる一覧に回します。ジョブ全体が`EXPIRED`や`FAILED`で終わったときは、全行が再投入の対象です。

### Tier別の投入トークン上限と同時100ジョブに合わせた分割の目安

[レート制限のページ](https://aistudio.google.com/docs/rate-limits)によると、Batchには同時に有効なジョブ100件、ファイル保存20GBの上限があります。さらにモデルごとに、有効なジョブ全体へ投入できるトークン数の上限があります。

| モデル                   | Tier 1 | Tier 2 | Tier 3 |
| --------------------- | ------ | ------ | ------ |
| Gemini 3.8 Flash      | 300万   | 4億     | 10億    |
| Gemini 3.5 Flash-Lite | 1,000万 | 5億     | 10億    |
| Gemini 4 Argon        | 500万   | 5億     | 10億    |

Tier 1の3.8 Flashでは、1件の入力が2,000トークンなら、同時に抱えられるのは約1,500件分です。10万件を一度に流すには、累計100ドルの支払いと初回支払いから3日の経過で上がるTier 2への移行が必要です。Tier 1のままなら、1,000件ずつのジョブに分け、前のジョブが終わってから次を投入する順番待ちを組みます。分割と再実行の一般的な設計は[バッチ推論とは？ジョブ分割・並列度と再実行設計を実装目線で解説](https://www.issoh.co.jp/tech/details/16349/)で整理しています。

## Gemini Batch APIを採用する業務と同期・Flexのままにする場面の判断

半額という数字だけで移すと、待ち時間の扱いで後から困ります。処理の性質で振り分けます。

### 夜間の分類・要約でBatchを選ぶ条件と月10万件の費用の試算

Batchに移してよいのは、結果を翌営業日の朝までに受け取れればよく、1回に数千件以上をまとめて流せる処理です。問い合わせの分類、議事録や日報の要約、商品説明文の下書き、過去文書へのタグ付けがこれに当たります。

費用の差は件数に比例します。1件あたり入力2,000トークン・出力500トークンを10万件、gemini-3.8-flashで処理すると、通常の呼び出しでは337.5ドル、Batchでは168.75ドルです。2027年1月1日の値上げ後は、Batchでも337.5ドルになります。[Flexの公式ページ](https://aistudio.google.com/docs/flex-inference)のとおり、Flexも同じ50%引きですが、目標は1〜15分の同期処理で、容量が足りないと503や429を返します。前の応答を受けて次を呼ぶ連続した処理はFlex、互いに独立した大量の処理はBatch、と分けるのが基本です。同じ考え方は[Azure OpenAI Batch APIとは？半額運用の条件とJSONL実装手順](https://www.issoh.co.jp/tech/details/16627/)でも扱っています。

### Batchを見送るべき場面と業務システムへ組み込む体制の判断

次のどれかに当たる処理は、Batchに移しません。利用者が画面の前で結果を待っている。締め時刻が決まっていて、48時間の期限切れが起きたら業務が止まる。件数が1日数百件で、半額にしても月数ドルしか変わらない。最後の場合は、二重投入の防止や再投入の仕組みを作る手間の方が高くつきます。

顧客の個人情報を含むデータを流すなら、送る前にデータの扱いを確かめてください。有料枠と無料枠の規約の違いは[Gemini APIに学習させない設定：無料枠と有料枠の規約差・store=False・ゼロデータ保持](https://www.issoh.co.jp/tech/details/19445/)、処理場所やSLAを契約で約束する場合のVertex AIとの違いは[Vertex AIとGemini APIの違い：認証・SLA・データ保存・所在地・料金を比較](https://www.issoh.co.jp/tech/details/18410/)で整理しています。夜間処理への組み込みは、投入・待機・取り込み・再投入・監視の5つを一式で作ることが、夜間処理を運用に乗せるための条件です。どの業務をBatchに回すかの整理から相談したい場合は、一創の[生成AI導入支援](https://www.issoh.co.jp/service/ai/genai/)で受け付けています。

## Gemini Batch APIの利用条件と結果取得に関するよくある質問と回答

Gemini Batch APIを使い始めるときに検索の多い質問に、公式ドキュメントの記述をもとに答えます。

### Gemini Batch APIは無料枠で使えますか？

モデルによります。料金ページでは、gemini-3.8-flashとgemini-4-argonのBatchは無料枠で使えません。gemini-3.5-flash-liteはBatchも無料枠の対象と表示されています。無料枠の入力はGoogleの製品改善に使われる扱いなので、試すのは公開情報やダミーのデータに限り、本番は課金を設定したプロジェクトで動かしてください。

### バッチジョブはどのくらいで終わりますか？

公式の目標は24時間以内で、多くはそれより早く終わるとされています。ただし保証ではなく、待機中または実行中のまま48時間を超えると期限切れになり、結果は返りません。中間の結果を早く受け取りたいときは、公式ドキュメントのとおり大きなジョブを小さく分けて投入してください。

### JSONLの結果ファイルはいつまで取得できますか？

File APIの公式ページでは、ファイルの保存期間は48時間です。夜間に投入したジョブの結果を週明けに取りに行く運用では、取得前に消える恐れがあります。完了を検知したらすぐに結果を自社のDBやストレージへ取り込み、そちらを正本にしてください。

### Interactions APIでBatchを使えますか？

2026年11月時点では使えません。Interactions APIの概要ページは、Batchを未対応の機能に挙げています。Batchの各行は`generate_content`と同じ`contents`の形式で書きます。同期の処理はInteractions API、Batchの処理は`generate_content`形式、と分けて持つのが現実的です。

### 埋め込み（Embedding）もBatchで処理できますか？

できます。Batch APIの公式ドキュメントには、`client.batches.create_embeddings()`で`gemini-embedding-2`を指定する例があります。投入トークン上限は通常の生成モデルより低く、Tier 1で50万トークンなので、検索用の索引を一度に作り直すときは分割の前提で計画してください。

## 関連記事

- [Gemini APIとは？できること・料金・始め方とInteractions APIの最小コード【2026年10月】](https://www.issoh.co.jp/tech/details/18759/)：本記事の親。APIの全体像とできること。
- [Gemini APIの料金と無料枠：モデル別単価・課金設定・費用の計算コード【2026年10月】](https://www.issoh.co.jp/tech/details/18813/)：Batchを含む処理区分ごとの単価と課金設定。
- [Gemini APIの使い方（Python）：google-genai 3.0で生成・ストリーミング・関数呼び出し【2026年11月】](https://www.issoh.co.jp/tech/details/19441/)：同期の呼び出しをPythonで書く方法。
- [Gemini APIのモデル一覧：モデルIDと安定版・プレビュー・廃止予定の見分け方【2026年11月】](https://www.issoh.co.jp/tech/details/19443/)：Batchで指定するモデルIDの選び方。
- [Azure OpenAI Batch APIとは？半額運用の条件とJSONL実装手順](https://www.issoh.co.jp/tech/details/16627/)：Azure OpenAIで同じ半額運用をする場合。

---

出典: [Gemini Batch APIの使い方：JSONLで半額実行・完了待ち・失敗行の再投入をPythonで実装【2026年11月】](<https://www.issoh.co.jp/tech/details/19447/>)（株式会社一創）