レビューが返事を待っている、という感覚があります。個人開発で iOS / Android の壁紙アプリと癒し系アプリを公開し続け、いまは6アプリを並行運用しています。App Store Connect と Google Play Console のレビュー欄を毎朝開くたびに、星1〜2のレビューや具体的な要望のコメントが少しずつ溜まっていきます。1件ずつ目を通すのは大切ですが、6アプリ分を3ヶ月放置したら8,000件近く積み上がっていた、というのが今回の話の入口でした。
通常の Gemini API で1件ずつ処理するのは確かに動きます。けれどコスト、所要時間、そして「夜中にレートリミットで止まって翌朝やり直し」という運用負荷を考えると、Batch API に寄せたほうが圧倒的に楽でした。
本記事は、実際に8,000件を一晩で片付けた実装と実測値に加えて、2026年6月に Gemini API へ入ったイベント駆動の Webhooks で「翌朝のポーリング」そのものを消す設計までを、動くコードとともに残すものです。1人で複数アプリを回す現実のなかで、夜間処理を「動かしてから寝る」だけでなく「完了を勝手に受け取る」段階まで持っていく記録です。
なぜ通常APIではなく Batch API を選んだのか
最初は通常APIでループを回していました。gemini-2.5-flash で1件あたりプロンプトとレビュー本文を渡し、{ "category": "...", "severity": "...", "needs_reply": true|false } の構造化出力を受け取るだけのシンプルな構成です。500件ほどでテストしているうちは何の問題もありませんでした。
8,000件を流し始めて、最初に詰まったのは1分あたりのリクエスト上限でした。Tier 1 で動かしていた時期だったため、429 RESOURCE_EXHAUSTED が頻発します。asyncio で並列度を絞っても、所要時間が読めません。深夜に流し始めて朝起きたら半分しか終わっていない、という日が2回続いた時点で別の手段を考え始めました。
Batch API に切り替えた理由は次の3点です。1つ目は料金が 通常APIの50% であること。2つ目は 24時間以内 に処理が完了することが SLA で示されていて、レートリミットの心配をしなくていいこと。3つ目は JSONL を1ファイル投げて結果を待つだけなので、夜中に進捗を気にする必要がなくなることでした。「動かしてから寝る」と「動かしながら起きている」では、翌日の消耗度が全然違うのです。
実装：JSONL 生成からアップロードまで
実装はとてもシンプルでした。レビューを Firestore から引っ張ってきて、JSONL を1本作り、google-genai SDK で投げます。
import json from pathlib import Path from google import genai
client = genai.Client( api_key = "YOUR_GEMINI_API_KEY" )
# Firestore などから引っ張ったレビュー配列 reviews = load_reviews_from_last_quarter() # [{ id, app_id, locale, body }, ...]
schema = { "type" : "object" , "properties" : { "category" : { "type" : "string" , "enum" : [ "bug" , "feature_request" , "praise" , "billing" , "other" ], }, "severity" : { "type" : "string" , "enum" : [ "low" , "medium" , "high" ]}, "needs_reply" : { "type" : "boolean" }, "summary_ja" : { "type" : "string" }, }, "required" : [ "category" , "severity" , "needs_reply" , "summary_ja" ], }
# JSONL を1行ずつ書き出す。key は必ず複合キーにする（理由は後述） jsonl_path = Path( "reviews_batch.jsonl" ) with jsonl_path.open( "w" , encoding = "utf-8" ) as f: for r in reviews: request = { "key" : f " { r[ 'app_id' ] } - { r[ 'id' ] } " , "request" : { "contents" : [ { "role" : "user" , "parts" : [ { "text" : f "アプリ: { r[ 'app_id' ] } / 言語: { r[ 'locale' ] }\n\n{ r[ 'body' ] } " } ], } ], "generationConfig" : { "responseMimeType" : "application/json" , "responseSchema" : schema, "temperature" : 0.1 , }, "systemInstruction" : { "parts" : [{ "text" : "あなたはアプリレビューを分類する補助役です。" }] }, }, } f.write(json.dumps(request, ensure_ascii = False ) + " \n " ) そのまま Batch にアップロードして、ジョブを作るだけです。
uploaded = client.files.upload( file = str (jsonl_path), config = { "display_name" : "reviews-2026q1-q2" , "mime_type" : "jsonl" }, )
batch_job = client.batches.create( model = "gemini-2.5-flash" , src = uploaded.name, config = { "display_name" : "review-classification-overnight" }, ) print (batch_job.name) # batches/xxxx ここで PC を閉じて寝ます。翌朝、client.batches.get(name=batch_job.name) を叩いて state == "JOB_STATE_SUCCEEDED" を確認できれば、結果ファイルを client.files.download() でダウンロードし、JSONL を読み戻して Firestore に書き戻すだけです。
通常APIと並べた数字
同じ8,000件を、通常APIで1日かけて流したときと、Batch API で一晩で流したときの実測です（gemini-2.5-flash・東京リージョン・2026年5月時点の自分の環境）。
項目 通常API Batch API
所要時間 約11時間 約6時間20分 失敗リトライ率 約3.4% 0件 コスト 約 US$2.10 約 US$1.05
コストは半額になり、所要時間は約4割短縮されました。私が一番嬉しかったのは「失敗 0件」で、夜中に起きてレートリミット復旧を待つ必要がなくなったことです。料金差よりも、この精神的な負荷の軽さの方が、6アプリを1人で回している身としては助かりました。
ちなみに通常APIのコストには、失敗リトライ分の重複課金は含めていません。実運用では 429 でもトークンは消費されるため、もう少し膨らみます。Batch API はその意味でも素直で、見積もりが読みやすいのが好きです。
ポーリングを待たず、完了を「受け取る」——Webhooks へ寄せる
ここからが、2026年6月の変更を受けて書き足した部分です。
「動かしてから寝る」までは到達しましたが、翌朝の client.batches.get(...) を叩くために、結局は朝の cron か手動の確認が残っていました。完了が午前3時でも午前9時でも、私のスクリプトが見にいくまで結果は宙ぶらりんのままです。
2026年6月、Gemini API にイベント駆動の Webhooks が追加され、Batch API や長時間オペレーションのポーリングを置き換える 仕組みが入りました。考え方はシンプルで、こちらが状態を取りにいくのではなく、ジョブが終わった瞬間に Google 側がこちらのエンドポイントへ通知を投げてくれます。完了通知を受け取ったら、その流れでダウンロードと書き戻しまで自動で走らせられます。
受信側は、署名検証つきの薄い HTTP エンドポイントを1つ立てるだけです。私は既存の通知用 FastAPI に相乗りさせました。
import hashlib import hmac import os from fastapi import FastAPI, Request, HTTPException from google import genai
app = FastAPI() client = genai.Client( api_key = os.environ[ "GEMINI_API_KEY" ]) WEBHOOK_SECRET = os.environ[ "GEMINI_WEBHOOK_SECRET" ].encode()
def verify_signature (body: bytes , signature: str ) -> bool : expected = hmac.new( WEBHOOK_SECRET , body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature)
@app.post ( "/gemini/batch-callback" ) async def batch_callback (request: Request): raw = await request.body() sig = request.headers.get( "X-Goog-Signature" , "" ) if not verify_signature(raw, sig): raise HTTPException( status_code = 401 , detail = "bad signature" )
event = await request.json() # 完了イベントのみ処理。それ以外（pending 更新等）は 200 で握って終える if event.get( "state" ) != "JOB_STATE_SUCCEEDED" : return { "ok" : True , "skipped" : event.get( "state" )}
job_name = event[ "name" ] # batches/xxxx if already_processed(job_name): # 冪等性ガード（後述） return { "ok" : True , "dedup" : True }
job = client.batches.get( name = job_name) result_blob = client.files.download( file = job.dest.file_name) write_back_to_firestore(result_blob) mark_processed(job_name) return { "ok" : True } ジョブを作るときに、この受信 URL を通知先として渡します。フィールド名は提供開始直後で変わり得るため、config に通知設定を載せる形にして、実際のキー名は changelog で確認しながら合わせるのが安全です。
batch_job = client.batches.create( model = "gemini-2.5-flash" , src = uploaded.name, config = { "display_name" : "review-classification-overnight" , # 2026-06 追加の通知設定。キー名は提供初期のため要確認 "notification" : { "webhook_url" : "https://api.example.com/gemini/batch-callback" , }, }, ) ここで効いてくるのが、JSONL の key を複合キーにしておく設計です。Webhook は再送される前提の仕組みで、同じ完了イベントが2回届くことがあります。受信側の already_processed(job_name) による冪等性ガードに加えて、結果行を元レビューへマージするキーが安定していれば、二重通知が来ても書き戻しの結果は変わりません。「取りに行く」から「受け取る」へ変えると、再送と順序の前提が変わるので、冪等性は後付けではなく最初から仕込んでおくのが安全です。
私の環境では、この置き換えで朝の cron を1本まるごと消せました。完了が深夜でも、起きたときには分類結果が Firestore に入っていて、bug + severity=high の通知だけが手元に届いている、という状態になりました。常駐して待つプロセスがなくなったぶん、Cloud Run の最小インスタンスも 0 に戻せています。
ポーリングを残したほうがいい場合
ただし、Webhooks に全面移行するのが常に正解とは限りません。私が併用しているのは次の判断です。
受信エンドポイントを常時公開できない環境（ローカルのみ・社内ネットワーク内）では、素直に batches.get のポーリングを残す
通知が届かなかったときの保険として、1日1回だけ「未完了として記録されているジョブ」を batches.get で照合するリコンサイル処理を別に持つ
通知の受信ログと、実際の JOB_STATE を週次で突き合わせ、取りこぼしの有無を確認する
Webhook は速くて静かですが、「届かなかったことに気づけない」のが弱点です。完了を受け取る設計にしても、最後の安全網として薄いポーリングを1本だけ残しておくと、夜間処理の信頼性が一段上がります。
運用してわかった注意点
実装は単純でしたが、運用してみると小さな段差がいくつかありました。ここで触れておきたいのは3つです。
JSONL の key を必ずユニークにすること 。私は最初うっかり Firestore のドキュメント ID をそのまま key にしていましたが、過去に同じレビューを誤って二重登録していたケースで key が衝突し、結果ファイルの紐付けで一部レビューが消えました。key は処理後に結果と元レビューをマージするキーになるため、{app_id}-{review_id} のように複合キーにしておくと事故が減ります。前述のとおり、Webhook の再送に耐えるうえでも複合キーが効いてきます。
Batch ジョブがハングしたときの諦めどころ 。Gemini Batch API は SLA 24時間ですが、たまに JOB_STATE_PENDING のまま12時間以上動かないことがあります。その場合は素直に新規ジョブを作り直したほうが早いです。古いジョブは client.batches.cancel(name=...) で止めるだけにしました。Gemini はキャンセルしても料金がほとんど発生しないので、この見切りが軽いのも助かります。
構造化出力のスキーマを temperature: 0.1 でも油断しないこと 。responseSchema を渡しても、ごくまれに summary_ja が空文字で返ることがあります。Firestore に書き戻す前に必ず if not result["summary_ja"]: continue のようなガードを入れておくと、後段の通知メールで「タイトルが空のメール」が飛ぶ事故を回避できます。
期限つきの非推奨を、運用カレンダーに載せる
もう1つ、複数アプリを1人で回していると痛感するのが、期限つきの非推奨（deprecation）を見落とさない仕組み の大切さです。
たとえば2026年6月には、画像系のプレビューモデル gemini-3.1-flash-image-preview と gemini-3-pro-image-preview が 6/25 に停止 するアナウンスが出ました。私のように OGP 画像や壁紙の試作にプレビューモデルを挟んでいると、停止日を過ぎた瞬間に夜間パイプラインが静かに 404 で落ちます。Webhooks で「完了を受け取る」設計にしても、そもそも投げるモデルが消えていては元も子もありません。
私は対策として、参照しているモデル ID を1か所の設定ファイルに集約し、ai.google.dev/gemini-api/docs/deprecations の停止日をカレンダーへ手で転記するようにしました。地味ですが、複数の夜間ジョブが同じモデル定数を共有していれば、差し替えは1行で済みます。「速く受け取る」仕組みと「期限を見落とさない」規律は、自動運用ではセットで効いてくると感じています。
次に取り組むこと
8,000件の分類結果を眺めていたら、bug + severity=high + needs_reply=true だけで200件近く残っていました。これを gemini-2.5-pro に投げて、返信ドラフト案を生成させるパイプラインを準備しています。Batch API は本気で書いた長文を投げるとそれなりに時間がかかるので、返信ドラフト生成は通常APIに戻すか、Batch のままにするかをまだ迷っていて、近日中にもう一度実測する予定です。
1人で6アプリを回す現実の前では、Gemini Batch API のような道具を入れて、空いた時間で「もっと深く読むべきレビュー」に集中するほうが、結果的にユーザーへの返事の質が上がりました。そこにさらに Webhooks を重ねて「完了を受け取る」段階まで来たことで、夜間処理に向ける注意がもう一段軽くなった気がしています。同じ規模で似た悩みを抱えている方の参考になれば嬉しいです。