Gemini API を asyncio で並列化したら、壁紙アプリの多言語説明文生成が12倍速になった

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分）

import asyncio
import google.generativeai as genai
import time
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
LANGUAGES = [
    "Japanese", "English", "Chinese (Simplified)", "Chinese (Traditional)",
    "Korean", "French", "German", "Spanish", "Portuguese", "Italian",
    "Arabic", "Russian"
]
 
async def generate_description_async(
    model: genai.GenerativeModel,
    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.
        """
        # generate_content_async は非同期版
        response = await model.generate_content_async(prompt)
        return lang, response.text
 
async def generate_all_languages(app_name: str) -> dict[str, str]:
    model = genai.GenerativeModel("gemini-2.5-flash")
    # 同時リクエスト数を5に制限（RPM超過を防ぐ）
    semaphore = asyncio.Semaphore(5)
 
    tasks = [
        generate_description_async(model, 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秒）

import asyncio
import random
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
 
async def generate_with_retry(
    model: genai.GenerativeModel,
    prompt: str,
    semaphore: asyncio.Semaphore,
    max_retries: int = 3
) -> str:
    """
    レート制限時に指数バックオフでリトライする。
    """
    async with semaphore:
        for attempt in range(max_retries):
            try:
                response = await model.generate_content_async(prompt)
                return response.text
            except google_exceptions.ResourceExhausted as e:
                if attempt == max_retries - 1:
                    raise
                # ジッターを加えた指数バックオフ
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"⚠️ レート制限: {wait:.1f}秒待機後リトライ")
                await asyncio.sleep(wait)
            except Exception as e:
                raise

# ❌ 危険：一つのエラーで全部止まる
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

APPS = [
    "Beautiful HD Wallpapers",
    "Ukiyo-e Wallpapers",
    "Relaxing Healing",
    "Law of Attraction Everyday"
]
 
async def generate_all_apps():
    model = genai.GenerativeModel("gemini-2.5-flash")
    semaphore = asyncio.Semaphore(5)  # 全アプリ合計で5並列
 
    all_tasks = []
    for app in APPS:
        for lang in LANGUAGES:
            all_tasks.append(
                generate_description_async(model, app, lang, semaphore)
            )
 
    results = await asyncio.gather(*all_tasks, return_exceptions=True)
    # ... 結果を {app: {lang: text}} 形式の辞書に整理して返す