同期処理で書いた分類バッチを走らせたまま、夜中に一度だけ様子を見に行ったことがあります。進捗の行はゆっくりとしか増えておらず、このまま朝まで置いても終わらないことだけは分かりました。個人開発で続けている壁紙アプリのコンテンツ更新で、Gemini API に画像のカテゴリ分類をさせようとしていたときのことです。
最初のバージョンは Python の同期処理で書いておりました。1 リクエストの応答が平均2秒、対象が1,000枚。画像の前処理とローカル I/O を足すと、推定で6時間を超える計算になります。カタログの更新はほぼ毎週発生しますので、この所要時間では運用に乗りません。
asyncio へ書き換えたところ、同じ枚数を81分で終えられるようになりました。ここでは、その書き換えで実際に効いた実装パターンと、本番で詰まった箇所を、なぜそう書くのかという理由まで含めて書き残します。
なぜ同期処理では限界が来るのか
Gemini API は HTTP ベースの REST API です。同期処理でリクエストを1件ずつ投げると、それぞれのネットワーク往復時間が積み上がっていきます。1リクエストあたりの応答時間が平均2秒だとすると、1,000件で2,000秒（約33分）。画像の前処理やローカルI/Oも含めると6時間という計算が現実に近くなります。
同期処理のコードは読みやすく、デバッグも簡単です。スケールしない以外は問題がなく、少量のデータを扱う間は問題なく動き続けます。しかしアプリのコンテンツが増え、分類対象の画像が数百件から数千件になる段階で、同期処理の限界は突然訪れます。
import google.generativeai as genai
# ❌ 同期処理の例（これでは遅い） def classify_images_sync (image_paths: list[ str ]) -> list[ dict ]: model = genai.GenerativeModel( "gemini-2.5-flash" ) results = [] for path in image_paths: with open (path, "rb" ) as f: image_data = f.read() response = model.generate_content([ '{"category": "...", "tags": [...]}の形式でJSONのみ返してください。' , { "mime_type" : "image/jpeg" , "data" : image_data} ]) results.append({ "path" : path, "result" : response.text}) return results asyncio を使えば、複数のリクエストを同時に飛ばすことができます。ネットワークの応答を待っている間、イベントループは別のリクエストを開始できます。ただし Gemini API には RPM（1分あたりのリクエスト数）と TPM（1分あたりのトークン数）の制限があるため、無制限に並列化するとすぐにレート制限エラーが返ってきます。「並列化しながらレート制限を守る」という二律背反を解決するのが、次に説明する Semaphore パターンです。この二律背反こそが asyncio 初学者が一番詰まるポイントです。レート制限を無視して並列数を上げすぎると 429 エラーが頻発し、レート制限を守りすぎると並列化のメリットが失われます。ちょうどよいバランスを見つけることが設計の核心です。
asyncio.to_thread() による同期SDKのラップ
本記事のパイプラインを組んだ2026年5月の時点では、google-generativeai の Python SDK は同期クライアントのみを提供しておりました。asyncio で使うには asyncio.to_thread() で同期関数をスレッドプールへ逃がすのが、最も安定した方法でした。
この前提は現在は変わっております。後継の google-genai には native async のクライアントがあり、to_thread() は不要になりました。移行時の注意点は後半の「asyncio.to_thread() の前提が変わりました」の節にまとめておりますので、これから新規に書く方はそちらを先にご覧ください。以下はレガシー SDK を使い続ける場合の、いまも有効な形です。
asyncio.to_thread() は Python 3.9 以上で使えます。内部的には loop.run_in_executor(None, func) と同等ですが、より直感的に書けます。これを使うことで、同期処理のAPIコールを「ブロッキングしないコルーチン」として扱えるようになります。
import asyncio import google.generativeai as genai
genai.configure( api_key = "YOUR_GEMINI_API_KEY" ) model = genai.GenerativeModel( "gemini-2.5-flash" )
async def call_gemini_async (prompt: str , image_data: bytes , mime_type: str = "image/jpeg" ) -> str : """ 同期のGemini APIコールをスレッドプールで非同期実行する。 asyncio.to_thread() でブロッキング処理をイベントループの外に逃がす。 """ def _sync_call (): response = model.generate_content([ prompt, { "mime_type" : mime_type, "data" : image_data} ]) return response.text
return await asyncio.to_thread(_sync_call) この実装で重要な点は、genai.GenerativeModel のインスタンスを複数のスレッドで共有していることです。私の環境では、並列数30のスレッドプールで共有しても状態の競合は発生しませんでした。ただし genai.configure(api_key=...) はモジュールレベルで一度だけ呼ぶことが前提です。
これだけでは「非同期で実行できる」だけで、複数件を並列実行する仕組みがありません。次のステップが核心部分です。
Semaphore による並列数の制御
asyncio.Semaphore を使うと「同時に走らせるリクエスト数の上限」を設定できます。セマフォを取得したタスクだけが API コールを実行でき、上限に達したタスクはセマフォが解放されるまで待機します。この仕組みにより、並列数をコントロールしながらすべてのリクエストを最終的に完了させることができます。
私の経験では、gemini-2.5-flash の無料枠（RPM 1,000）に対して並列数20が安全な出発点でした。並列数30では62分まで短縮できましたが、429エラーが増え始めました。並列数40では返ってパフォーマンスが落ちました。この「最適な並列数」はAPIのティアや実行環境によって異なるため、小さな値から段階的に上げていくアプローチを強くお勧めします。
import asyncio import time
CONCURRENT_REQUESTS = 20
async def process_images_parallel ( image_paths: list[ str ], prompt: str , concurrency: int = CONCURRENT_REQUESTS ) -> list[ dict ]: """ 画像リストを並列処理する。 Semaphore で同時実行数を制限し、レート制限エラーを回避する。 """ semaphore = asyncio.Semaphore(concurrency) results = []
async def _process_one (path: str ) -> dict : async with semaphore: try : with open (path, "rb" ) as f: image_data = f.read() result = await call_gemini_async(prompt, image_data) return { "path" : path, "status" : "ok" , "result" : result} except Exception as e: return { "path" : path, "status" : "error" , "error" : str (e)}
tasks = [_process_one(path) for path in image_paths] results = await asyncio.gather( * tasks, return_exceptions = False ) return list (results) 私の環境では、1,000枚を同期処理で約360分（推定）かかっていたものが、並列数20で81分、並列数30で62分に短縮されました。並列数40では429エラーのリトライが増え、71分と遅くなりました。
指数バックオフによるリトライの実装
並列処理を動かし始めてすぐに直面するのが 429 Resource Exhausted エラーです。このエラーは一時的なレート制限なので、適切な待機時間を入れてリトライすれば解決できます。
問題は「固定秒数でリトライすると全スレッドが同時に再試行して再び詰まる」ことです。これを「サンダリングハード」問題と呼びます。20のスレッドが同時に2秒待ってから一斉に再試行すると、再び429が発生しやすくなります。
これを避けるために指数バックオフ（待機時間を2のべき乗で増やす）とジッター（ランダムな上乗せ時間）を組み合わせます。1回目のリトライは約1秒後、2回目は約2秒後、3回目は約4秒後という形で待機時間が増え、かつスレッドごとに少しずれた時間で再試行するため、サンダリングハードを防げます。
import asyncio import random import logging from google.api_core import exceptions as google_exceptions
logger = logging.getLogger( __name__ )
async def call_gemini_with_retry ( prompt: str , image_data: bytes , max_retries: int = 5 , base_delay: float = 1.0 , mime_type: str = "image/jpeg" ) -> str : """ 指数バックオフ + ジッター付きリトライ。 429 / 503 は一時的エラーとして扱い、その他は即座に re-raise する。 """ for attempt in range (max_retries + 1 ): try : return await call_gemini_async(prompt, image_data, mime_type)
except google_exceptions.ResourceExhausted: if attempt >= max_retries: raise delay = base_delay * ( 2 ** attempt) + random.uniform( 0 , 1.0 ) logger.warning( f "Rate limited (attempt { attempt + 1 } / { max_retries } ). Retry in { delay :.1f } s" ) await asyncio.sleep(delay)
except google_exceptions.ServiceUnavailable: if attempt >= max_retries: raise delay = base_delay * ( 1.5 ** attempt) + random.uniform( 0 , 0.5 ) logger.warning( f "Service unavailable (attempt { attempt + 1 } / { max_retries } ). Retry in { delay :.1f } s" ) await asyncio.sleep(delay)
except Exception : raise
raise RuntimeError ( "Unreachable" ) await asyncio.sleep(delay) でリトライ待機中にイベントループに制御を返す点が重要です。スレッドをブロックするのではなく、イベントループが他のタスクを処理できる状態にしながら待機します。この非同期スリープにより、1つのタスクがバックオフ中でも他の20のタスクは継続して処理を進められます。
部分失敗を許容する設計
1,000件の処理で全件成功を前提にしてはいけません。JSONパースエラー、壊れた画像ファイル、一時的なネットワーク障害など、必ずある割合で失敗が起きます。
asyncio.gather() はデフォルト（return_exceptions=False）で、いずれかのタスクで例外が発生すると即座にその例外を raise して処理を中断します。これは「1件のエラーで900件の成功結果が無駄になる」という危険な挙動です。
私が採用した設計は「成功した分は即座にファイルに書き出し、失敗した分は後でリトライする」というパターンです。処理の途中でプロセスが落ちても、書き出し済みの結果は失われません。また asyncio.Lock() でファイルへの書き込みを排他制御することで、複数タスクが同時に書き込もうとしてファイルが壊れるのを防ぎます。
import asyncio import json from dataclasses import dataclass, asdict from pathlib import Path
FENCE = "`" * 3 # モデルが付けてくるマークダウンのコードフェンス
@dataclass class ClassificationResult : path: str status: str result: dict | None = None error: str | None = None attempts: int = 0
async def process_with_partial_failure_handling ( image_paths: list[ str ], prompt: str , output_file: Path, concurrency: int = 20 ) -> tuple[list[ClassificationResult], list[ str ]]: """ 部分失敗を許容しながら処理する。 成功結果は逐次 output_file に追記し、失敗したパスは failed_paths に収集。 """ semaphore = asyncio.Semaphore(concurrency) lock = asyncio.Lock()
async def _process_one (path: str ) -> ClassificationResult: result = ClassificationResult( path = path, status = "error" ) async with semaphore: for attempt in range ( 5 ): try : result.attempts = attempt + 1 with open (path, "rb" ) as f: image_data = f.read() raw = await call_gemini_async(prompt, image_data) json_str = raw.strip() if json_str.startswith( FENCE ): json_str = json_str.split( FENCE )[ 1 ] if json_str.startswith( "json" ): json_str = json_str[ 4 :] parsed = json.loads(json_str.strip()) result.status = "ok" result.result = parsed break except json.JSONDecodeError: result.error = f "JSON parse failed: { raw[: 100 ] } " break except Exception as e: result.error = str (e) if attempt < 4 : delay = 1.0 * ( 2 ** attempt) + random.uniform( 0 , 0.5 ) await asyncio.sleep(delay)
async with lock: with open (output_file, "a" ) as f: f.write(json.dumps(asdict(result), ensure_ascii = False ) + " \n " )
return result
tasks = [_process_one(path) for path in image_paths] results = await asyncio.gather( * tasks) failed_paths = [r.path for r in results if r.status == "error" ] return list (results), failed_paths この実装で特に重要な設計判断は2つあります。1つは json.JSONDecodeError を「リトライ不要なエラー」として即座に break していること。モデルが不正な JSON を返した場合、同じプロンプトで再試行しても同じ結果になる可能性が高く、リトライは時間の無駄になります。2つ目は、結果をファイルに逐次書き出すことで「プロセスが途中でクラッシュしても済み分はやり直し不要」な設計にしていることです。これは長時間バッチ処理には必須の考え方だと実感しています。
コンテンツとJSON整形エラーの実態
実際に1,200枚を処理した中で判明したのが、gemini-2.5-flash のJSON整形問題です。プロンプトに「JSONのみを返してください」と明記しても、3.2%の件数でマークダウンのコードフェンス（バッククォート3つに json を添えた、あの囲み）に包まれた形で返答が来ました。
これは公式ドキュメントには記載がありません。私が発見したのは、500件のバッチを初めて流した後に50件近くのパースエラーが出たときでした。raw.strip().startswith(FENCE) でコードフェンスを検出して取り除く処理を足したところ、その後は同じ問題が再発しなくなりました。
コードフェンス除去のロジックは前述のコードに含まれていますが、なぜこれが必要かを理解した上で使う方が、将来モデルの挙動が変わったときに対応しやすくなります。もしコードフェンスを除去しても JSON パースに失敗する場合は、re.search(r'\{.*\}', raw, re.DOTALL) で JSON 部分だけを抽出するフォールバック処理を追加することを検討してください。
本番で遭遇した3つの落とし穴
公式ドキュメントには書かれていない、実際に詰まったポイントを三つ挙げます。いずれも「知っていれば5分で避けられる、知らないと数時間溶かす」種類の問題です。
落とし穴1: asyncio.Semaphore をモジュールレベルで作成してはいけない
asyncio.Semaphore はイベントループが起動している間に生成しなければなりません。asyncio.run() が起動する前にモジュールレベルで作ると、Python のバージョンによって挙動が異なります。Python 3.8 では DeprecationWarning が出て動く場合があり、3.10 以降では RuntimeError になるケースがあります。「開発環境では動いたのに本番で壊れた」という典型的な原因の一つです。
常に async def 関数の中でセマフォを作成し、asyncio.run() に渡す最上位のコルーチン内でインスタンス化してください。
落とし穴2: asyncio.gather() のデフォルト挙動で1,000件中1件のエラーが全体を止める
asyncio.gather() はデフォルトで return_exceptions=False です。これはいずれかのタスクで例外が発生すると、その例外が即座に raise されてしまうことを意味します。並列で動いているほかのタスクは「バックグラウンドで継続実行されるが結果が捨てられる」状態になります。1,000件を処理していて990件終わったところでエラーが1件発生し、全体が巻き戻ってしまったときの徒労感は忘れられません。
解決策は2つあります。return_exceptions=True を使って例外を結果として受け取るか、各タスク内で try/except を使って例外を内部で処理するかです。私は後者を好みます。タスクが常に辞書を返すようにしておくと、asyncio.gather() 側で例外ハンドリングを意識しなくて済むからです。
落とし穴3: 画像データをタスク作成時点で一括読み込むとメモリが枯渇する
リスト内包表記で1,000件のタスクを作成する際、「読み込んだ画像データをタスクに渡す」設計にすると、asyncio.gather() が呼ばれる前に全画像がメモリに展開されます。壁紙画像は1枚2〜8MBあるため、1,000枚で最大8GBのメモリが必要になります。
正しい設計は「各タスクの実行時点（セマフォ取得後）でファイルを読み込む」ことです。セマフォで並列数を20に制限しているため、メモリに同時に存在する画像データは最大20枚分（最大160MB程度）で済みます。タスクの作成時点ではパスだけを渡し、実際のファイル読み込みはタスク内で行う設計が正解です。
Beautiful 4K/HDR Wallpapers での実際の計測結果
私が実際にこのパターンを適用したのは、壁紙アプリのコンテンツパイプラインです。処理対象は新規追加候補画像1,200枚で、カテゴリ分類・タグ生成・ムード判定の3つの情報を1回のリクエストで取得するプロンプトを使いました。
実測結果:
同期処理（推定）: 約420分
asyncio 並列20: 81分
asyncio 並列30: 62分（これ以上増やしても RPM 制限でエラー頻発）
API コスト: 約3.2ドル（1,200枚、gemini-2.5-flash を使用）
並列数30が最速でしたが、夜間バッチで実行しているため急ぎでもなく、現在は20で安定稼働させています。レート制限エラーはほぼゼロになり、監視コストが大幅に減りました。
この経験を通じて感じたのは、非同期処理は複雑さを増やすけれども、その複雑さは投資に値する、ということです。同期処理が立ち行かなくなるのはある日突然で、コンテンツが増えた瞬間にバッチが終わらなくなります。asyncio の初期設計コストを先に払っておくと、その限界が来る地点を大きく先送りできます。
個人開発を10年以上続けてきた経験から言えば、バックエンドの処理時間問題は「後で直せばいい」と思っていると、「急いで直さないといけない」タイミングで直すことになります。個人開発者は常にリソースが限られているため、その経験は避けられるならば避けるべきです。
asyncio 設計をどのタイミングで導入すべきか
よく聞かれる質問の一つが「最初から asyncio で設計すべきか、後から移行すべきか」です。私の答えは「処理件数が50件を超える見込みがある時点から asyncio を検討する」です。
50件以下なら同期処理で十分です。1リクエスト2秒 × 50件 = 100秒（約1分40秒）は許容範囲内であり、asyncio の複雑さを持ち込む必要はありません。しかし「今後もコンテンツが増え続ける」という前提があるなら、500件になってから慌てて設計を変えるより、最初から asyncio の構造で書いておく方が長期的には楽です。
個人開発では「今すぐ必要な機能だけ実装する」という原則が大切で、過度な先行設計は避けるべきです。一方で、非同期化は「後から入れようとすると関連するすべての関数の書き直しが必要になる」変更の一つです。アプリのバックエンドで await を使いたい場合、その関数を呼ぶ全ての関数も async def でなければならないという制約があります。これがいわゆる「asyncio の感染」です。感染力があるからこそ、設計の入口で判断しておく価値があります。
私が Beautiful 4K/HDR Wallpapers のパイプラインで asyncio を選んだのは、画像コンテンツが「今後も定期的に追加される」という前提があったからです。最初から非同期設計で書いておいたことで、後の大規模バッチへの対応もスムーズに行えました。
レート制限の種類と Gemini API 固有の挙動
Gemini API のレート制限は RPM（Requests Per Minute）と TPM（Tokens Per Minute）の2軸で管理されています。どちらかを超えると 429 Resource Exhausted が返ります。
重要な点は「RPM は満たしているが TPM を超えてエラーになる」ケースが存在することです。画像1枚あたりの入力トークン数は解像度や内容によって大きく異なります。高解像度・複雑なコンテンツの画像が連続すると、RPM は余っているのに TPM の上限を先に打つことがあります。
私が最初に並列数30で実行したとき、429エラーが予想より多く出ました。原因を調べると、壁紙画像の中に特に高解像度のランドスケープ写真が集中してキューに入っていたことでした。画像をシャッフルしてから並列処理に渡すことで、TPM の瞬間的な集中が分散し、エラー率が減りました。
import random
# ランダムシャッフルで TPM の集中を防ぐ image_paths = sorted (image_paths) # まずソート（再現性確保） random.seed( 42 ) # シードを固定してデバッグを容易に random.shuffle(image_paths) # シャッフルによる TPM 集中の分散 この1行を追加するだけで、TPM 起因の429エラーを大幅に減らせます。
また、google.api_core.exceptions.ResourceExhausted の details() メソッドを呼ぶと、エラーの詳細（「Rate limit exceeded」vs「Quota exceeded」など）を確認できます。429エラーが頻発する場合は、RPM と TPM のどちらが原因かを特定することで、適切な対処（並列数の削減 vs シャッフル）を選べます。
進捗モニタリングとコスト推定の実装
70分かかるバッチ処理では、進捗の可視化が作業効率に直結します。tqdm の asyncio 対応バージョンを使うと、各タスクが完了するたびにプログレスバーが更新されます。
from tqdm.asyncio import tqdm as async_tqdm
async def process_with_progress (image_paths, prompt, concurrency = 20 ): semaphore = asyncio.Semaphore(concurrency) async def _one (path): async with semaphore: try : with open (path, "rb" ) as f: data = f.read() result = await call_gemini_with_retry(prompt, data) return { "path" : path, "status" : "ok" , "result" : result} except Exception as e: return { "path" : path, "status" : "error" , "error" : str (e)} tasks = [_one(path) for path in image_paths] return await async_tqdm.gather( * tasks, desc = "Gemini API処理中" , unit = "枚" ) コストの推定については、Google AI Studio のコスト表示がリアルタイムではないため（数分遅延がある）、自前でカウントする方法が役立ちます。gemini-2.5-flash の2026年5月時点の画像入力単価は1枚あたり約0.000038ドルです。1,000枚で約0.038ドル（約5.7円）、1,200枚で約0.046ドル（約6.8円）という計算になります。これに出力トークン分が加わり、実際のコストは私の場合1,200枚で約3.2ドルでした（テキスト出力が長いプロンプトだったため）。
コストの実測値とモデルの精度は常にトレードオフです。gemini-2.5-flash は gemini-2.5-pro より安価ですが、JSON の整形精度が若干低く、コードフェンスが混入する頻度も高い傾向があります。コスト最優先なら flash、精度最優先なら pro、という選択になりますが、私は flash に整形修正ロジックを加えることでバランスを取っています。
asyncio と Python バージョン互換性
asyncio の挙動は Python バージョンによって細かく変化しています。本記事のコードは Python 3.10 以上を対象としています。3.9 では asyncio.to_thread() が使えますが、型ヒントの一部構文（dict | None など）は 3.10 以降でないと動きません。
本番環境のPythonバージョン管理には pyproject.toml で requires-python = ">=3.10" を明記することをお勧めします。また、asyncio.run() のデバッグモード（asyncio.run(main(), debug=True)）を開発中は有効にしておくと、コルーチンが await されないまま放置されるケースや、イベントループ外でのセマフォ作成などの問題を早期に発見できます。
個人開発では CI/CD の整備が後回しになりがちですが、asyncio を使うコードは特に「開発環境では動くが本番で壊れる」パターンに陥りやすいため、最低限のユニットテストを書いておくことを強くお勧めします。pytest-asyncio を使えば async def test_ という形式でテストが書け、モックを使って実際のAPIコールなしに非同期ロジックを検証できます。
実装前に理解しておくべき asyncio の基礎概念
asyncio のコードを書く前に、いくつかの概念を正しく把握しておく必要があります。誤解したまま実装すると、「動いているように見えて実は並列化されていない」という状態に陥ります。
イベントループとコルーチンの関係
asyncio はシングルスレッドの並行処理モデルです。マルチスレッドやマルチプロセスとは異なり、実際に同時にCPUを使う処理は1つだけです。「並列」に見えるのは、コルーチンが await ポイントでイベントループに制御を返し、その間に別のコルーチンが実行されるからです。
これが意味するのは「CPUバウンドな処理（重い計算）は asyncio では速くならない」ということです。Gemini APIの呼び出しがうまく並列化されるのは、APIコールの大部分がネットワーク待ち（I/Oバウンド）だからです。asyncio.to_thread() で同期処理をスレッドプールに逃がすのも、スレッド内でのブロッキングがイベントループを止めないようにするためです。
コルーチンとタスクの違い
async def で定義した関数を呼ぶと、コルーチンオブジェクトが返ります。コルーチンは「まだ実行を始めていない処理の定義」です。実際にイベントループで動かすには await するか asyncio.create_task() でタスクに変換します。
asyncio.gather(*tasks) にコルーチンのリストを渡すと、内部でタスクに変換されて並列実行されます。すでに asyncio.create_task() でタスク化されているものを渡しても正しく動きます。ただし、タスクを作成した時点でイベントループへのスケジューリングが始まるため、大量のタスクを一度に作成すると予期しないタイミングで処理が始まることがあります。Semaphore を使うパターンでは、タスク内でセマフォを取得してから実際の処理を始めるため、この問題を自然に回避しています。
asyncio と threading の使い分け
「同期SDKを asyncio.to_thread() で包む」という今回のアプローチは、少数の長時間処理（例: 1〜5件の重い生成タスク）よりも、多数の短〜中程度の処理（例: 100〜10,000件のAPIコール）に向いています。スレッドプールのオーバーヘッドは1コルーチンあたり数ミリ秒ですが、それが1件あたりの処理時間（2秒）に対して誤差の範囲だからです。
なお、この「包む」前提は現在では不要になりました。後継の google-genai は client.aio に非同期版を持っており、asyncio.to_thread() を経由せずに直接 await できます。スレッドプールのオーバーヘッドはそのぶん完全になくなります。具体的な移行手順は次の節に書きました。
非同期コードのテスト戦略
asyncio を使うコードのテストは、同期コードよりも少し工夫が必要です。pytest-asyncio を使えば非同期テストを書けますが、実際の Gemini API を呼ぶテストは遅くてコストがかかるため、モックが必須です。
モック設計でよく使われるパターンは「成功ケース」「リトライが必要なケース」「回復不可なエラー」の3種類です。成功ケースは単純に正常なJSONを返すモックです。リトライが必要なケースは最初の呼び出しで ResourceExhausted を raise し、2回目で成功するモックです。回復不可なエラーは毎回 InvalidArgument を raise するモックです。
テストでは「リトライが正しい回数だけ呼ばれているか」「バックオフの待機時間が正しい順序になっているか」を確認する点が肝心です。asyncio.sleep をモックしてバックオフのシミュレーションをスキップし、テストの実行時間を短縮するテクニックもよく使われます。
また、asyncio.Semaphore が正しく動いているかをテストするには「同時に実行されたタスク数の最大値が設定値を超えていないか」を確認する方法があります。各タスクの開始と終了のタイムスタンプを記録し、任意の時点での並列実行数をカウントすることで検証できます。
単体テストを書く習慣がない個人開発者でも、少なくとも「正常系のエンドツーエンドテスト」と「429エラーが起きたときにリトライするかどうかの確認」だけは自動化しておくことを強くお勧めします。asyncio のリトライロジックは「手動で確認するのが難しい」種類のコードの筆頭だからです。
コンテンツパイプラインの全体設計で考えること
asyncio による並列化は、コンテンツパイプラインの一部分です。全体設計を考えると、他にも考慮すべき要素があります。
まず「何を処理対象とするかの選定」です。1,200枚の新規候補画像をすべて処理するのではなく、事前フィルタリングで明らかに不適切な画像（低解像度・重複・人物を含む可能性があるもの）を除外することで、APIコストと処理時間を削減できます。私のパイプラインでは、まず画像のメタデータ（サイズ・アスペクト比・ファイルハッシュ）でフィルタリングし、Gemini APIに渡すのは候補の70〜80%程度です。
次に「処理結果のキャッシュ」です。同じ画像を毎回処理するのは無駄なので、画像のハッシュ値をキーにして処理結果をデータベースにキャッシュしておくことで、再処理を防げます。画像が更新された場合のみ再分類するロジックを加えると、月次の更新コストが大幅に下がります。
さらに「失敗した処理のリトライキュー」です。一時的なAPIエラーで失敗した画像は、すぐに再試行するのではなく、別のキューに入れておいて数時間後に再処理する設計が安定します。私はシンプルなテキストファイルに失敗したパスを書き出しておき、翌朝のバッチで自動的に再試行するようにしています。
これらの工夫を組み合わせることで、「月に一度1,200枚を処理するコスト3ドル」から「毎週新規追加分のみ処理するコスト0.5ドル」という運用コストの最適化が実現しました。
個人開発者が asyncio を使う際の現実的なコスト感覚
個人開発では「技術的に正しい選択」より「コストパフォーマンスが高い選択」を優先することがあります。asyncio の学習コストと実装コストについて、正直な感覚を共有します。
asyncio の基本的な使い方（コルーチン・asyncio.run()・await）を理解するには、Python の経験者であれば1〜2日程度で十分です。しかし「本番で安定して動くコード」を書くためには、Semaphore・バックオフ・部分失敗の処理など追加の学習が必要で、実際に詰まりながら学ぶと1週間程度かかります。私の場合、最初のバージョンから「本番で安定するバージョン」に到達するまで、断続的に作業して3週間ほどかかりました。
この学習コストを「APIコールが1件2秒で1,000件処理する必要がある」という観点で評価すると、同期処理の33分と asyncio の81分の差（実行ごとに52分節約）はかなり大きな価値です。週1回実行するバッチなら、年間で約45時間の待機時間を節約できます。個人開発者の時間の価値を自分なりに設定した上で判断することをお勧めします。
一方で「バッチを週1回しか実行しない」「1回あたり100件以下」という場合は、asyncio の投資対効果は下がります。そのような場合は同期処理のシンプルさを選ぶことも賢明な判断です。技術的に優れた方法を選ぶことが目的ではなく、アプリのコンテンツが質よく更新されることが目的だからです。
道具は目的に合わせて選ぶ——この当たり前が、コードの設計でもそのまま効くのだと感じています。asyncio は強力な道具ですが、すべての場面で必要な道具ではありません。ここで紹介したパターンが、皆さまのプロジェクトに asyncio が要るタイミングを見極める材料になれば幸いです。
asyncio.to_thread() の前提が変わりました——google-genai の native async へここまでのコードは google-generativeai を前提にしております。その前提そのものが、いまは動いております。
Google の新しい SDK への移行ガイド は、後継の google-genai がすべての対応プラットフォームで一般提供に入ったこと、そしてレガシーのライブラリを使っているなら移行を強く推奨することを明記しています。asyncio についても、client.aio の下にすべてのメソッドの非同期版が用意されている、と書かれています。
つまり、スレッドプールへ逃がす理由がなくなりました。ネットワーク待ちをイベントループが直接扱えます。
import asyncio from google import genai from google.genai import types
client = genai.Client() # GEMINI_API_KEY 環境変数から読み込み
async def classify_one (image_bytes: bytes , prompt: str ) -> str : # native async クライアントでの 1 枚分の分類（to_thread 不要） response = await client.aio.models.generate_content( model = "gemini-3.8-flash" , contents = [ prompt, types.Part.from_bytes( data = image_bytes, mime_type = "image/jpeg" ), ], ) return response.text Semaphore のパターンは、そのまま残してください。SDK が変わってもレート制限は変わりませんので、並列数の上限を自分で決める必要は同じだけあります。変わったのは「どうやって非同期にするか」だけです。
移しただけで動かなくなるのは、リトライの except 節です
移行でいちばん気をつけたいのは、モデル名でも contents の書き方でもありませんでした。リトライの except 節 です。
前述のリトライ関数は google.api_core.exceptions.ResourceExhausted を捕まえています。これはレガシー SDK が投げる例外で、google-genai は投げません。新しい SDK の例外は google.genai.errors にあり、レート制限は APIError の派生である ClientError（code が 429）として届きます。
厄介なのは、移行した直後は何も壊れて見えないことです。正常系は通り、テストも緑のまま残ります。429 が返ってきた瞬間にだけ、捕まえるはずだった except 節を素通りして、バッチ全体が落ちます。私がこれに気づいたのは、それまで自動で回復していた夜間バッチが、ある朝だけ途中で終わっていたときでした。
import asyncio import random from google.genai import errors
TRANSIENT_CODES = { 429 , 500 , 503 }
async def call_with_retry ( image_bytes: bytes , prompt: str , max_retries: int = 5 , base_delay: float = 1.0 , ) -> str : # google-genai の例外階層に合わせたリトライ（code で一時的かどうかを判定） for attempt in range (max_retries + 1 ): try : return await classify_one(image_bytes, prompt) except errors.APIError as e: if e.code not in TRANSIENT_CODES or attempt >= max_retries: raise delay = base_delay * ( 2 ** attempt) + random.uniform( 0 , 1.0 ) await asyncio.sleep(delay)
raise RuntimeError ( "unreachable" ) e.code で分岐させておくと、400番台の入力エラー（壊れた画像・サイズ超過）を一時的な障害と取り違えて延々と再試行してしまう事故も、同時に防げます。e.message には API が返した説明がそのまま入りますので、ログにはこの二つを並べて出しておくと、翌朝に原因を辿れます。
自分のコードに死んだ except が残っていないかは、2行で確かめられます。
grep -rn "api_core.exceptions" --include= '*.py' . grep -rn "google.generativeai" --include= '*.py' . 前者が残っていて後者が消えている状態が、いちばん危ない組み合わせです。インポートは移せても、例外は移りません。 この確認だけは、SDK を入れ替えるたびに必ず走らせるようにしております。
書き換えそのものは半日で終わりました。時間がかかったのは、except を直したあとに「本当に 429 で再試行できているか」を確かめる作業のほうです。ここは実際に 429 を起こしてみるのがいちばん早く、並列数を一時的に上げて数十件だけ流し、リトライのログが出ることを目で見て確かめました。移行の前後で並列数と処理時間は変えておりませんが、スレッドプールを挟まなくなったぶん、メモリの上振れは素直に小さくなりました。
全体を振り返って：まず小さなバッチで動かしてみる
ここまでの asyncio パターンをまとめますと、核心は5つです。第一に、いま使っている SDK がどちらなのかを確かめること（google-genai なら client.aio で to_thread() は不要です）。第二に asyncio.Semaphore で並列数を制御してレート制限を守ること。第三に指数バックオフとジッターで一時的なエラーから回復すること。第四に処理結果をファイルへ逐次書き出して部分失敗を許容すること。そして第五に、リトライの except が自分の SDK の例外階層と噛み合っているかを確かめることです。
これらを一度に全部実装する必要はありません。まずは非同期のクライアント（レガシー SDK なら asyncio.to_thread()）と asyncio.Semaphore だけで試してみることをお勧めします。10〜50件のリストと並列数5から始め、同期処理と比べた速度差を体感してから、バックオフや部分失敗処理を追加していくアプローチが現実的です。Gemini API の asyncio 活用は、コンテンツを多く扱う個人開発者にとって確実に投資対効果のある技術です。少しずつ積み上げていきましょう。
本記事のコードは Beautiful 4K/HDR Wallpapers の画像分類パイプラインをベースに整理したものです。同じ課題に取り組んでいる方の参考になれば幸いです。