Antigravity × Python asyncio で Google AI API を並列処理する — 複数プロンプトを同時に捌く実践ガイド

100件の商品説明を AI に送って品質チェックしてもらうとき、1件ずつ順番に処理していたら何時間かかるか計算したことはありますか？私がこの問題に直面したのは、Antigravity で ECサイト向けのコンテンツ自動生成ツールを作っていたときです。同期処理では 100件で約15分かかっていたのが、asyncio を使った並列処理に切り替えたら3分以内に収まりました。

Antigravity IDE 上で Python の asyncio と google-genai SDK を組み合わせ、複数の AI リクエストを効率よく並列実行する方法を順番にご紹介します。単純な並列化だけでなく、レート制限に引っかからないための Semaphore 制御まで、実際に使えるパターンでお伝えします。

なぜ同期処理のままだとつらいのか

Google AI API（Gemini API）は1回のリクエストで応答が返るまで数秒かかります。これを同期処理でループすると、前のリクエストが終わるまで次が始まらないため、件数が増えるほど待ち時間が積み上がります。

# ❌ 同期処理の例 — 100件で単純に100倍の時間がかかる
import google.generativeai as genai
 
def analyze_reviews_sync(reviews: list[str]) -> list[str]:
    model = genai.GenerativeModel("gemini-2.0-flash")
    results = []
    for review in reviews:
        response = model.generate_content(f"このレビューの感情を分析してください：{review}")
        results.append(response.text)
    return results

100件 × 平均2秒 = 200秒（約3分半）。さらに件数が増えれば比例して増え続けます。

非同期処理に切り替えると、API の応答を待っている間に別のリクエストを送れるため、ネットワーク待ち時間を大幅に削減できます。

Antigravity でプロジェクトをセットアップする

まず Antigravity IDE で新しい Python プロジェクトを作成します。ターミナルを開いて以下を実行してください。

# 仮想環境の作成と有効化
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
 
# 必要なパッケージのインストール
pip install google-genai aiohttp python-dotenv

.env ファイルを作成してAPIキーを設定します。

GEMINI_API_KEY=YOUR_GEMINI_API_KEY

Antigravity の AI アシストで非同期コードを書くとき、プロジェクトルートに .antigravity/rules.md を置くと AI の提案精度が上がります。

# .antigravity/rules.md
このプロジェクトでは Python asyncio を使った非同期処理を実装しています。
- google-genai SDK の AsyncClient を優先的に使用すること
- 全ての非同期関数には型ヒントを付けること
- エラーハンドリングは aiohttp.ClientResponseError と google.api_core.exceptions で行うこと

このルールがあると、Antigravity の Inline Chat（⌘I）でコード補完や修正を依頼したときに、プロジェクトのスタイルに合った提案が返ってきます。

asyncio.gather() で複数リクエストを並列実行する

google-genai SDK の最新版（v0.8以降）は非同期クライアントをサポートしています。基本的な並列処理パターンを見てみましょう。

import asyncio
import os
from google import genai
from google.genai import types
from dotenv import load_dotenv
 
load_dotenv()
 
# 非同期クライアントの初期化
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
async def analyze_single(review: str, index: int) -> tuple[int, str]:
    """1件のレビューを非同期で分析する"""
    try:
        response = await client.aio.models.generate_content(
            model="gemini-2.0-flash",
            contents=f"このレビューの感情を分析し、ポジティブ/ネガティブ/中立のいずれかで答えてください：{review}",
            config=types.GenerateContentConfig(
                max_output_tokens=50,
                temperature=0.1,
            ),
        )
        return index, response.text
    except Exception as e:
        # エラー時はインデックスと共にエラー内容を返す（処理を止めない）
        return index, f"ERROR: {str(e)}"
 
async def analyze_reviews_async(reviews: list[str]) -> list[str]:
    """複数レビューを並列分析する"""
    tasks = [analyze_single(review, i) for i, review in enumerate(reviews)]
    results = await asyncio.gather(*tasks)
    
    # インデックス順に並べ直して返す
    sorted_results = sorted(results, key=lambda x: x[0])
    return [text for _, text in sorted_results]
 
# 実行例
async def main():
    reviews = [
        "とても使いやすくて満足しています！",
        "梱包が雑で少し傷がありました",
        "普通です。特に不満はないです",
        # ... 実際には100件以上
    ]
    results = await analyze_reviews_async(reviews)
    for review, sentiment in zip(reviews, results):
        print(f"{review[:20]}... → {sentiment}")
 
if __name__ == "__main__":
    asyncio.run(main())

asyncio.gather() に渡したタスクは全て並列で実行されます。ただし、件数が多い場合はこのままだと同時に数百リクエストを送ることになり、レート制限エラーが発生します。

Semaphore でレート制限を安全にコントロールする

Google AI API には無料枠・有料枠それぞれにリクエスト/分（RPM）の制限があります。Semaphore を使うと、同時実行数の上限を設けながら並列処理できます。

import asyncio
from google import genai
from google.genai import types
import os
from dotenv import load_dotenv
 
load_dotenv()
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# Gemini 2.0 Flash の無料枠：15 RPM / 有料枠：2000 RPM
# 安全マージンを持って設定（例: 有料枠で並列10に制限）
MAX_CONCURRENT = 10
 
async def analyze_with_semaphore(
    semaphore: asyncio.Semaphore,
    review: str,
    index: int,
) -> tuple[int, str]:
    """Semaphore で同時実行数を制限しながら分析する"""
    async with semaphore:
        try:
            response = await client.aio.models.generate_content(
                model="gemini-2.0-flash",
                contents=f"感情分析（ポジティブ/ネガティブ/中立）：{review}",
                config=types.GenerateContentConfig(max_output_tokens=50),
            )
            return index, response.text.strip()
        except Exception as e:
            # レート制限エラーの場合は少し待ってリトライ
            if "RESOURCE_EXHAUSTED" in str(e):
                await asyncio.sleep(5)
                return await analyze_with_semaphore(semaphore, review, index)
            return index, f"ERROR: {str(e)}"
 
async def batch_analyze(reviews: list[str]) -> list[str]:
    """大量レビューをバッチで並列分析する"""
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    tasks = [
        analyze_with_semaphore(semaphore, review, i)
        for i, review in enumerate(reviews)
    ]
    results = await asyncio.gather(*tasks)
    return [text for _, text in sorted(results)]

Semaphore の値は実際のAPI制限と余裕を見て調整します。私の経験では、有料プランでも MAX_CONCURRENT = 10〜15 程度に設定しておくと安定します。

実践応用：商品レビュー一括分析パイプライン

実際に動くパイプラインとして、CSV から読み込んだレビューを分析して結果を保存する例を示します。

import asyncio
import csv
import json
from pathlib import Path
from google import genai
from google.genai import types
import os
from dotenv import load_dotenv
 
load_dotenv()
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
MAX_CONCURRENT = 10
 
ANALYSIS_PROMPT = """
以下の商品レビューを分析し、JSON 形式で回答してください：
- sentiment: "positive" / "negative" / "neutral"
- score: 1〜5 の整数（感情の強さ）
- key_points: 主なポイントを最大2つ（リスト）
 
レビュー：{review}
 
JSON のみ出力してください（```json``` は不要）。
"""
 
async def analyze_review_structured(
    semaphore: asyncio.Semaphore,
    review: str,
    index: int,
) -> tuple[int, dict]:
    """構造化データとしてレビューを分析する"""
    async with semaphore:
        try:
            response = await client.aio.models.generate_content(
                model="gemini-2.0-flash",
                contents=ANALYSIS_PROMPT.format(review=review),
                config=types.GenerateContentConfig(
                    max_output_tokens=200,
                    temperature=0.1,
                    response_mime_type="application/json",  # JSON 出力を強制
                ),
            )
            result = json.loads(response.text)
            return index, {"review": review, **result, "status": "ok"}
        except json.JSONDecodeError:
            return index, {"review": review, "status": "parse_error"}
        except Exception as e:
            if "RESOURCE_EXHAUSTED" in str(e):
                await asyncio.sleep(5)
                return await analyze_review_structured(semaphore, review, index)
            return index, {"review": review, "status": "error", "error": str(e)}
 
async def run_pipeline(input_csv: str, output_json: str):
    """メインパイプライン：CSV読み込み → 並列分析 → JSON保存"""
    # レビューの読み込み
    reviews = []
    with open(input_csv, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            reviews.append(row["review_text"])
    
    print(f"📊 {len(reviews)}件のレビューを処理します...")
    
    # 並列分析の実行
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    tasks = [
        analyze_review_structured(semaphore, review, i)
        for i, review in enumerate(reviews)
    ]
    results = await asyncio.gather(*tasks)
    sorted_results = [data for _, data in sorted(results)]
    
    # 結果の集計と保存
    success = sum(1 for r in sorted_results if r["status"] == "ok")
    errors = len(sorted_results) - success
    
    output = {
        "total": len(reviews),
        "success": success,
        "errors": errors,
        "results": sorted_results,
    }
    
    Path(output_json).write_text(
        json.dumps(output, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(f"✅ 完了: {success}件成功 / {errors}件エラー → {output_json}")
 
if __name__ == "__main__":
    asyncio.run(run_pipeline("reviews.csv", "analysis_results.json"))

response_mime_type="application/json" を指定すると、Gemini が JSON 以外の文字列を返す可能性が減り、パースエラーを大幅に減らせます。これはドキュメントに目立つ形では書かれていない設定ですが、構造化出力を扱うときに非常に効果的です。

Antigravity の AI アシストを非同期コードに活かす

非同期処理のコードは、処理の流れが追いにくく、バグの発見が難しい傾向があります。Antigravity の AI アシストをうまく使うと、この課題を軽減できます。

Inline Chat（⌘I）での活用例:

# このコードを選択して ⌘I → 「このコードにエラーハンドリングを追加して」
async def analyze_single(review: str) -> str:
    response = await client.aio.models.generate_content(...)
    return response.text

Antigravity の AI は選択範囲のコンテキストを読んで、適切な例外処理のパターンを提案してくれます。特に google.api_core.exceptions.ResourceExhausted や google.api_core.exceptions.DeadlineExceeded のような SDK 固有の例外クラスは知らないと対処しにくいため、AI の補完に頼るのが現実的です。

デバッグのコツ:

並列処理のデバッグには Antigravity のターミナルと Python の logging モジュールを組み合わせるのが効果的です。

import logging
 
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(task_id)s] %(message)s",  # task_id はカスタムフィルターで追加
)

ログを見ながら Antigravity の AI Chat（⌘L）に「このエラーログの原因と解決策を教えて」と貼り付けると、非同期特有のデッドロックや競合状態のパターンをすぐに診断してもらえます。

全体を振り返って：まず小さく始めて効果を確認する

今日からできる一歩として、既存の同期処理コードを1つ選んで非同期化を試してみてください。全体を一度に書き換えようとすると大変ですが、1つの関数だけ async def に変えて asyncio.run() で呼ぶところから始めれば、処理速度の改善をすぐに実感できます。

Python SDK の非同期処理についてさらに深く知りたい方には、Antigravity × Python SDK 本番運用マスターガイドも参考になるかと思います。リアルタイムのストリーミング処理についてはPython ストリーミング API リアルタイム表示ガイドでも解説しています。

非同期化は「一度やると元に戻れない」感覚があります。コードが少し複雑になりますが、処理時間が5分の1以下になったときの気持ちよさは格別です。ぜひ試してみてください。