2026年5月、iOS 向けの4アプリ（Beautiful HD Wallpapers、浮世絵壁紙、Relaxing Healing、Law of Attraction Everyday）の大型アップデートを同時進行させていました。新しい iPhone モデルへの解像度対応、AdMob メディエーション拡張、StoreKit 2 への移行と、やることが山積みだったのですが、その中でひとつ想定外に時間を取られたのが「多言語 App Store 説明文の再生成」です。
各ストアのメタデータは12言語で管理しています。モデル交代やコンセプト変更のたびに全言語版を更新するのですが、Gemini API を使った順次処理のスクリプトが遅くて困っていました。asyncio で並列化したところ、処理時間が約12分の1になりました。今回はその実装と、実際にハマった落とし穴をまとめます。
なお本記事は公開後に一度書き直しています。当時使っていた google-generativeai は非推奨となり、現在の推奨は統合版の google-genai です。並列化の考え方そのものは変わりませんが、コードの書き方は変わりました。旧版と新版の両方を残し、どこが変わったのかがわかる形にしてあります。
順次処理の限界：12言語 × 4アプリで13分かかっていた
当初のスクリプトは単純な for ループで各言語の説明文を生成していました。12言語の App Store 説明文を順番にリクエストするだけのシンプルな実装です。
以下は2026年5月当時のコードで、旧 SDK（google-generativeai）で書かれています。現在このまま新規に書くことは勧めませんが、遅さの原因が構造にあることを示すために当時のまま載せておきます。
# ⚠️ 旧SDK（google-generativeai・非推奨）での当時の実装
import google.generativeai as genai
import time
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-flash")
LANGUAGES = [
"Japanese", "English", "Chinese (Simplified)", "Chinese (Traditional)",
"Korean", "French", "German", "Spanish", "Portuguese", "Italian",
"Arabic", "Russian"
]
def generate_description(app_name: str, lang: str) -> str:
prompt = f"""
Write an App Store description for "{app_name}" in {lang}.
Keep it under 4000 characters. Focus on beauty, relaxation, and daily use.
"""
response = model.generate_content(prompt)
return response.text
# 順次処理（遅い）
results = {}
start = time.time()
for lang in LANGUAGES:
results[lang] = generate_description("Beautiful HD Wallpapers", lang)
print(f"✓ {lang} 完了")
elapsed = time.time() - start
print(f"\n合計: {elapsed:.1f}秒") # 実測: 約780秒（13分）12言語で約780秒（13分）かかっていました。4アプリ分だと1時間近くになります。アップデートのたびにこれを待つのは現実的ではなく、「面倒だから後で」という先送りの温床になっていました。
待ち時間の内訳を見ると、CPU はほとんど遊んでいます。1リクエストあたり60秒前後のうち、実際に手元の Python が仕事をしているのは数十ミリ秒です。残りは Google のサーバーからの応答をただ待っている時間でした。
asyncio と threading の違い：なぜ asyncio を選んだか
並列化の方法として threading も選択肢に入ります。ただ、Gemini API の呼び出しは CPU ではなくネットワーク I/O の待ち時間が支配的です。こういったケースでは threading よりも asyncio の方が軽量で扱いやすいと感じています。
threading はスレッドごとにメモリを消費し、GIL（グローバルインタープリタロック）の影響もあります。一方 asyncio はシングルスレッドで動作し、I/O 待ちの間に別のタスクを進める設計です。Gemini API のようなネットワーク往復が多い処理には、asyncio の方がシンプルに書けて管理もしやすいと感じています。
具体的に言うと、threading で12スレッドを立てた場合、それぞれのスレッドが独立したスタックを持ち、スレッド間の同期（Lock や Queue）を手動で管理する必要があります。asyncio では Semaphore ひとつで同時実行数を制御でき、エラー処理も try/except の通常のパターンで書けます。個人開発でメンテナンスコストを低く抑えたい場合には、asyncio の方が合っていると感じています。
もう一点、Gemini API の Python SDK が非同期呼び出しを公式にサポートしていることも理由です。旧 SDK では generate_content_async() がその役割を担っていました。現行の google-genai では client.aio 以下に全メソッドの非同期版が用意されています。どちらの世代でも、同期コードからの移行は構造を変えずに済みます。
multiprocessing については、このケースでは不要です。CPU バウンドの処理（画像処理や数値計算など）には multiprocessing が有効ですが、API 呼び出しは I/O バウンドなので asyncio で十分です。
asyncio.gather() で並列化する
asyncio と Semaphore を使って、複数のリクエストを同時に送る実装です。Semaphore で同時リクエスト数を制限することで、レート制限への対策も同時に行います。以下は現行 SDK（google-genai）で書き直した版です。
# 現行SDK: pip install -U google-genai
import asyncio
import time
from google import genai
# GEMINI_API_KEY 環境変数から読まれる
client = genai.Client()
MODEL = "gemini-3.7-flash"
LANGUAGES = [
"Japanese", "English", "Chinese (Simplified)", "Chinese (Traditional)",
"Korean", "French", "German", "Spanish", "Portuguese", "Italian",
"Arabic", "Russian"
]
async def generate_description_async(
app_name: str,
lang: str,
semaphore: asyncio.Semaphore
) -> tuple[str, str]:
"""
semaphore で同時リクエスト数を制限しながら非同期生成する。
返り値は (言語, 説明文) のタプル。
"""
async with semaphore:
prompt = f"""
Write an App Store description for "{app_name}" in {lang}.
Keep it under 4000 characters. Focus on beauty, relaxation, and daily use.
"""
# client.aio が非同期版のエントリポイント
response = await client.aio.models.generate_content(
model=MODEL,
contents=prompt,
)
return lang, response.text
async def generate_all_languages(app_name: str) -> dict[str, str]:
# 同時リクエスト数を5に制限（RPM超過を防ぐ）
semaphore = asyncio.Semaphore(5)
tasks = [
generate_description_async(app_name, lang, semaphore)
for lang in LANGUAGES
]
# 全タスクを並列実行
results = await asyncio.gather(*tasks, return_exceptions=True)
output = {}
for result in results:
if isinstance(result, Exception):
print(f"❌ エラー: {result}")
else:
lang, text = result
output[lang] = text
print(f"✓ {lang} 完了")
return output
# 実行
start = time.time()
results = asyncio.run(generate_all_languages("Beautiful HD Wallpapers"))
elapsed = time.time() - start
print(f"\n合計: {elapsed:.1f}秒") # 実測: 約65秒（1分5秒）実測で約65秒になりました。780秒 → 65秒、約12倍の高速化です。
旧 SDK 版から変わったのは3行だけです。モデルオブジェクトを引き回す代わりにクライアントをひとつ持ち、await model.generate_content_async(prompt) が await client.aio.models.generate_content(model=..., contents=...) になりました。Semaphore と gather() の構造には一切手を入れていません。
落とし穴1：429 エラー（レート制限）への対処
並列化するとレート制限（429 Too Many Requests）に当たる頻度が上がります。Gemini API の無料枠は RPM（リクエスト/分）の上限が比較的低く、特に注意が必要です。実際に運用していて、Semaphore の値を10に設定したところ、複数言語分の結果がレート制限エラーになる場面がありました。
ここで一点、旧 SDK 時代の記事をそのまま読むと損をする変更があります。現行の google-genai は、429 や 503 のような一時的な失敗に対して自前でリトライを行います。初回の待ち時間はおよそ1秒、最大60秒まで指数的に伸ばしながら、既定で最大4回まで再試行します。つまり以前は必須だった手書きのバックオフは、そのまま同じ形で書くと SDK 内部のリトライと二重になります。
それでも手書きのラッパーが要る場面はあります。SDK の再試行を使い切ってもなお失敗するとき、その言語だけを後で拾い直したい場合です。私は「SDK に一次リトライを任せ、それでも落ちたものを記録して個別に再実行する」という二段構えにしています。
import asyncio
from google import genai
from google.genai import errors
client = genai.Client()
async def generate_with_fallback(
prompt: str,
lang: str,
semaphore: asyncio.Semaphore,
) -> tuple[str, str | None]:
"""
SDK 内蔵のリトライを使い切っても失敗した場合だけ、
None を返して呼び出し側で再実行キューに回す。
"""
async with semaphore:
try:
response = await client.aio.models.generate_content(
model="gemini-3.7-flash",
contents=prompt,
)
return lang, response.text
except errors.ClientError as e:
# 429 RESOURCE_EXHAUSTED はここに来る
if e.code == 429:
print(f"⚠️ {lang}: レート制限。再実行キューへ")
return lang, None
raise
except errors.ServerError as e:
print(f"⚠️ {lang}: サーバー側エラー ({e.code})。再実行キューへ")
return lang, None例外の型も旧 SDK から変わっています。以前は google.api_core.exceptions.ResourceExhausted を捕まえていましたが、現行 SDK では google.genai.errors 配下の ClientError（4xx）と ServerError（5xx）に整理されており、どちらも APIError を継承しています。e.code に HTTP ステータスが入るため、429 だけを取り出す判定はこの形が素直です。
Semaphore の最適値は利用プランによって変わります。無料枠では Semaphore(3) 程度が安定していました。有料プランでは5で問題ありませんでした。
落とし穴2：asyncio.gather() のエラー処理
asyncio.gather() は return_exceptions=True を指定しないと、最初の例外が発生した時点で全タスクが中断されます。12言語のうち1言語が失敗しただけで、残り11言語の結果が全て失われてしまいます。
# ❌ 危険：一つのエラーで全部止まる
results = await asyncio.gather(*tasks)
# ✅ 安全：エラーも返り値として受け取る
results = await asyncio.gather(*tasks, return_exceptions=True)
# 結果を安全に処理
for result in results:
if isinstance(result, Exception):
print(f"スキップ: {result}")
else:
lang, text = result
output[lang] = text失敗した言語は後から個別に再実行する設計にしておくと、運用が楽になります。
4アプリ分を一括処理する
最終的に4アプリ × 12言語 = 48リクエストを一括処理するスクリプトに仕上げました。
APPS = [
"Beautiful HD Wallpapers",
"Ukiyo-e Wallpapers",
"Relaxing Healing",
"Law of Attraction Everyday"
]
async def generate_all_apps():
semaphore = asyncio.Semaphore(5) # 全アプリ合計で5並列
all_tasks = []
for app in APPS:
for lang in LANGUAGES:
all_tasks.append(
generate_description_async(app, lang, semaphore)
)
results = await asyncio.gather(*all_tasks, return_exceptions=True)
# ... 結果を {app: {lang: text}} 形式の辞書に整理して返す48リクエストを一括処理しても実測で約90〜120秒で完了しました。順次処理なら52分かかる計算だったので、体感としては全く別のツールになりました。
Semaphore をアプリごとではなく全体でひとつ持つ点が要点です。アプリ単位に分けると、それぞれが5並列で走って合計20並列になり、レート制限にまっすぐ突っ込みます。並列数の管理は「同時に飛ぶリクエストの総数」で考える必要があります。
旧SDKからの移行で実際に触った箇所
同じスクリプトを現行 SDK へ移した際に手を入れた箇所を、差分として並べておきます。移行作業そのものは30分ほどで終わりました。
| 項目 | 旧: google-generativeai | 現行: google-genai |
|---|---|---|
| インストール | pip install google-generativeai | pip install -U google-genai |
| 初期化 | genai.configure(api_key=...) | client = genai.Client()（GEMINI_API_KEY を自動参照） |
| モデル指定 | genai.GenerativeModel("...") を先に生成 | 呼び出しごとに model= で渡す |
| 非同期呼び出し | await model.generate_content_async(prompt) | await client.aio.models.generate_content(model=..., contents=...) |
| 生成オプション | メソッド引数に個別指定 | config=types.GenerateContentConfig(...) に集約 |
| 例外 | google.api_core.exceptions | google.genai.errors（ClientError / ServerError） |
| リトライ | 自前で実装 | SDK 内蔵（既定で最大4回・指数バックオフ） |
引っかかったのは contents の綴りです。旧 SDK の埋め込み系メソッドが content 単数だったのに対し、現行 SDK は contents 複数で統一されています。移行直後、この1文字でしばらく悩みました。
モデル名も合わせて見直しています。当時の gemini-2.5-flash は世代交代が進んでおり、新規に書くなら現行の Flash 系を指定するのが素直です。モデル ID を定数で1箇所に固めておくと、こうした差し替えが一行で済みます。移行のたびに grep で追いかけていた頃と比べると、ずいぶん楽になりました。詳細な移行手順は Google GenAI SDK への移行ガイド にまとまっています。
実運用での注意：生成結果の品質確認を省略しない
高速化の副作用として「結果を確認せずに使ってしまう」罠があります。順次処理で1言語ずつ確認しながら進めていた頃は、おかしな表現に気づきやすかったのですが、一括生成では画面をスクロールするだけになりがちです。
私の場合、特に以下の言語で細かい確認が必要でした。
アラビア語（Arabic）は右から左に書く言語のため、App Store で表示したときの折り返しが他言語と異なります。生成された説明文が文法的に正しくても、実際のストアページでの見た目の確認は必要です。中国語繁体字（Traditional Chinese）は、簡体字で生成された後に変換されたような表現になることがあり、台湾のストアで使うには少しぎこちない場合があります。
これらの確認作業自体は並列化で高速化できるものではありませんが、「13分待ってやっと出た結果を確認する」のと「65秒で出た結果を確認する」のでは、心理的な余裕が全然違います。速くなったことで、確認に時間をかけられるようになりました。
また、生成結果をファイルに保存しておくことも重要です。Gemini API の応答は毎回微妙に異なるため、一度良い結果が出たらそれをバージョン管理しておくことをお勧めします。私は app_descriptions/{app_name}/{lang}/v{date}.txt という形式で保存し、更新時に前回との diff を確認するフローにしています。
待ち時間を削ると、妥協が減る
13分待つスクリプトは「後でやろう」になり、1分のスクリプトは「今すぐやろう」になります。この違いが、アップデートの品質にそのまま出ました。
並列化してから、各言語の説明文を丁寧に確認・調整できるようになりました。特に Arabic と中国語繁体字は微妙なニュアンスの調整が必要で、以前は「時間がないから今回はこれで」と妥協していました。処理が速くなると、その妥協がなくなります。速さそのものより、速さが生む余白のほうが効いていると感じています。
現行 SDK は client.aio に非同期版が揃っているため、同期コードからの移行は思ったより少ない変更で済みます。まず1〜3言語の小さなバッチで試してから、徐々に並列数を増やしていくのがお勧めです。
まずは手元のスクリプトを Semaphore(3) の小さなバッチで動かしてみてください。所要時間を1回測っておくと、並列数を上げたときの効き方が数字で見えます。asyncio の細部については Gemini API Python SDK のドキュメント に現行の書き方がまとまっています。
私自身、非同期処理はまだ手探りで書いている部分があります。同じところでつまずいた方の参考になれば嬉しいです。お読みいただきありがとうございました。