Gemini 2.5 Pro × Python 非同期処理：高スループットAPIシステムの本番構築

# ❌ 悪い例：同期的な逐次処理
import google.generativeai as genai
import time
 
def process_documents_sync(documents: list[str]) -> list[str]:
    model = genai.GenerativeModel("gemini-2.5-pro")
    results = []
    for doc in documents:
        # 1リクエストずつ実行 → APIの待ち時間が積み重なる
        response = model.generate_content(doc)
        results.append(response.text)
    return results
 
# 100件のドキュメント → 約200秒（2秒/リクエスト × 100件）

# ✅ 良い例：asyncioによる並列処理
import asyncio
import google.generativeai as genai
from google.generativeai.types import GenerateContentResponse
 
async def process_single_document(
    model: genai.GenerativeModel,
    document: str,
    semaphore: asyncio.Semaphore
) -> str:
    """
    セマフォで同時実行数を制限しながら1件処理する
    semaphore: 同時実行数の上限を管理するセマフォ
    """
    async with semaphore:
        # asyncio.to_thread でブロッキングAPIを非同期化
        response = await asyncio.to_thread(
            model.generate_content, document
        )
        return response.text
 
async def process_documents_async(
    documents: list[str],
    max_concurrent: int = 10  # 同時実行数の上限
) -> list[str]:
    model = genai.GenerativeModel("gemini-2.5-pro")
    semaphore = asyncio.Semaphore(max_concurrent)
    
    tasks = [
        process_single_document(model, doc, semaphore)
        for doc in documents
    ]
    
    # 全タスクを並列実行し、結果を順序を保って返す
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # エラーをフィルタリング
    return [r if not isinstance(r, Exception) else f"ERROR: {r}" for r in results]
 
# 100件のドキュメント → 約20秒（10並列 × 2秒/リクエスト × 100/10件）
if __name__ == "__main__":
    docs = ["ドキュメント内容{}".format(i) for i in range(100)]
    results = asyncio.run(process_documents_async(docs))
    print(f"処理完了: {len(results)}件")

import asyncio
import google.generativeai as genai
 
async def generate_async_native(prompt: str) -> str:
    """
    google-generativeai の非同期ネイティブAPIを使用
    asyncio.to_thread が不要になり、より効率的
    """
    model = genai.GenerativeModel("gemini-2.5-pro")
    
    # generate_content_async: ネイティブ非同期メソッド
    response = await model.generate_content_async(
        prompt,
        generation_config=genai.GenerationConfig(
            temperature=0.7,
            max_output_tokens=2048,
        )
    )
    return response.text
 
async def batch_generate(prompts: list[str]) -> list[str]:
    """バッチで非同期生成"""
    semaphore = asyncio.Semaphore(15)  # 同時15リクエストまで
    
    async def _single(prompt: str) -> str:
        async with semaphore:
            return await generate_async_native(prompt)
    
    return await asyncio.gather(*[_single(p) for p in prompts])
 
# 実行例
if __name__ == "__main__":
    prompts = [f"Pythonの{topic}を説明してください" 
               for topic in ["asyncio", "マルチスレッド", "プロセスプール"]]
    results = asyncio.run(batch_generate(prompts))
    for p, r in zip(prompts, results):
        print(f"Q: {p[:30]}...\nA: {r[:100]}...\n")

import asyncio
import random
import time
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
)
import logging
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
 
logger = logging.getLogger(__name__)
 
class GeminiRateLimitError(Exception):
    """レート制限エラーのカスタム例外"""
    pass
 
@retry(
    stop=stop_after_attempt(5),               # 最大5回リトライ
    wait=wait_exponential(multiplier=1, min=2, max=60),  # 指数バックオフ（2〜60秒）
    retry=retry_if_exception_type((
        GeminiRateLimitError,
        google_exceptions.ResourceExhausted,   # 429エラー
        google_exceptions.ServiceUnavailable,  # 503エラー
    )),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
async def generate_with_retry(
    model: genai.GenerativeModel,
    prompt: str,
    jitter: bool = True
) -> str:
    """
    指数バックオフ＋ジッター付きでGemini APIを呼び出す
    
    jitter=True: ランダムな遅延を追加して同時リトライの衝突を防ぐ
    """
    try:
        # ジッターを追加（0〜500msのランダム遅延）
        if jitter:
            await asyncio.sleep(random.uniform(0, 0.5))
        
        response = await model.generate_content_async(prompt)
        return response.text
        
    except google_exceptions.ResourceExhausted as e:
        logger.warning(f"レート制限エラー: {e}")
        raise GeminiRateLimitError(str(e)) from e
    except Exception as e:
        logger.error(f"予期しないエラー: {e}")
        raise
 
# 使用例
async def main():
    model = genai.GenerativeModel("gemini-2.5-pro")
    prompts = [f"プロンプト{i}" for i in range(50)]
    
    semaphore = asyncio.Semaphore(10)
    
    async def process_one(prompt: str) -> str:
        async with semaphore:
            return await generate_with_retry(model, prompt)
    
    results = await asyncio.gather(
        *[process_one(p) for p in prompts],
        return_exceptions=True
    )
    
    success = sum(1 for r in results if not isinstance(r, Exception))
    failed = len(results) - success
    print(f"成功: {success}件 / 失敗: {failed}件")
 
asyncio.run(main())

import asyncio
import time
from dataclasses import dataclass, field
from collections import deque
 
@dataclass
class AdaptiveRateLimiter:
    """
    成功/失敗率に応じてレートを動的に調整するレート制限器
    """
    max_rpm: int = 60              # 1分あたり最大リクエスト数
    window_seconds: int = 60       # 計測ウィンドウ（秒）
    _request_times: deque = field(default_factory=deque)
    _error_count: int = 0
    _success_count: int = 0
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    
    async def acquire(self) -> None:
        """
        レート制限内でリクエストを取得
        制限に近い場合は自動的に待機
        """
        async with self._lock:
            now = time.monotonic()
            
            # ウィンドウ外の古いリクエストを削除
            while self._request_times and \
                  self._request_times[0] < now - self.window_seconds:
                self._request_times.popleft()
            
            # レート制限チェック
            current_rpm = len(self._request_times)
            effective_limit = self._get_effective_limit()
            
            if current_rpm >= effective_limit:
                # 次のスロットが空くまで待機
                wait_time = self._request_times[0] + self.window_seconds - now
                if wait_time > 0:
                    await asyncio.sleep(wait_time + 0.1)
            
            self._request_times.append(time.monotonic())
    
    def _get_effective_limit(self) -> int:
        """
        エラー率に基づいて実効レートを計算
        エラーが多い場合は保守的なレートに自動調整
        """
        total = self._error_count + self._success_count
        if total < 10:
            return self.max_rpm
        
        error_rate = self._error_count / total
        if error_rate > 0.2:  # 20%以上エラーの場合
            return int(self.max_rpm * 0.5)  # 50%に制限
        elif error_rate > 0.1:  # 10%以上
            return int(self.max_rpm * 0.7)  # 70%に制限
        else:
            return self.max_rpm
    
    def record_success(self) -> None:
        self._success_count += 1
    
    def record_error(self) -> None:
        self._error_count += 1

import asyncio
from typing import AsyncGenerator
import google.generativeai as genai
 
async def stream_response(
    prompt: str,
    model_name: str = "gemini-2.5-pro"
) -> AsyncGenerator[str, None]:
    """
    Gemini APIのストリーミングレスポンスを非同期ジェネレータとして返す
    
    使用例:
        async for chunk in stream_response("長い文章を生成してください"):
            print(chunk, end="", flush=True)
    """
    model = genai.GenerativeModel(model_name)
    
    # ストリーミングモードで生成
    response = await model.generate_content_async(
        prompt,
        stream=True,
        generation_config=genai.GenerationConfig(
            temperature=0.7,
            max_output_tokens=4096,
        )
    )
    
    full_text = ""
    async for chunk in response:
        if chunk.text:
            full_text += chunk.text
            yield chunk.text
    
    # ストリーミング完了後の検証
    if not full_text:
        raise ValueError("空のレスポンスが返されました")
 
# FastAPIとの統合例
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
 
app = FastAPI()
 
@app.post("/generate/stream")
async def generate_stream(request: dict):
    prompt = request.get("prompt", "")
    
    async def event_generator():
        async for chunk in stream_response(prompt):
            # Server-Sent Events形式で送信
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginxバッファリング無効化
        }
    )

import asyncio
import json
import logging
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional
import google.generativeai as genai
from tenacity import retry, stop_after_attempt, wait_exponential
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
@dataclass
class ProcessingResult:
    doc_id: str
    status: str          # "success" | "failed" | "skipped"
    result: Optional[str] = None
    error: Optional[str] = None
    tokens_used: int = 0
 
class DocumentProcessor:
    """
    大量ドキュメントを非同期並列処理するパイプライン
    
    特徴:
    - チェックポイント保存（中断後に再開可能）
    - 動的な同時実行数制御
    - 詳細な処理統計
    """
    
    def __init__(
        self,
        max_concurrent: int = 10,
        checkpoint_path: str = "checkpoint.jsonl",
        max_retries: int = 3
    ):
        self.model = genai.GenerativeModel("gemini-2.5-pro")
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.checkpoint_path = Path(checkpoint_path)
        self.max_retries = max_retries
        self._processed_ids: set[str] = set()
        self._results: list[ProcessingResult] = []
        
    def _load_checkpoint(self) -> None:
        """既存のチェックポイントを読み込み、処理済みIDを復元"""
        if self.checkpoint_path.exists():
            with open(self.checkpoint_path) as f:
                for line in f:
                    if line.strip():
                        result = json.loads(line)
                        if result["status"] == "success":
                            self._processed_ids.add(result["doc_id"])
            logger.info(f"チェックポイント読み込み: {len(self._processed_ids)}件の処理済みドキュメント")
    
    def _save_checkpoint(self, result: ProcessingResult) -> None:
        """処理結果をチェックポイントに追記"""
        with open(self.checkpoint_path, "a") as f:
            f.write(json.dumps(asdict(result), ensure_ascii=False) + "\n")
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=4, max=30)
    )
    async def _process_single(
        self,
        doc_id: str,
        content: str,
        prompt_template: str
    ) -> ProcessingResult:
        """1件のドキュメントを処理（リトライ付き）"""
        async with self.semaphore:
            try:
                prompt = prompt_template.format(content=content)
                response = await self.model.generate_content_async(
                    prompt,
                    generation_config=genai.GenerationConfig(
                        max_output_tokens=1024,
                        temperature=0.3,  # 一貫性を重視
                    )
                )
                
                # トークン使用量の記録
                tokens = getattr(response.usage_metadata, "total_token_count", 0)
                
                return ProcessingResult(
                    doc_id=doc_id,
                    status="success",
                    result=response.text,
                    tokens_used=tokens
                )
                
            except Exception as e:
                logger.error(f"ドキュメント {doc_id} の処理失敗: {e}")
                return ProcessingResult(
                    doc_id=doc_id,
                    status="failed",
                    error=str(e)
                )
    
    async def process_all(
        self,
        documents: dict[str, str],  # {doc_id: content}
        prompt_template: str = "以下のテキストを要約してください:\n\n{content}"
    ) -> list[ProcessingResult]:
        """
        全ドキュメントを並列処理し、結果を返す
        
        Args:
            documents: ドキュメントIDと内容の辞書
            prompt_template: {content}プレースホルダーを含むプロンプトテンプレート
        
        Returns:
            処理結果のリスト
        """
        self._load_checkpoint()
        
        # 未処理のドキュメントのみ処理対象に
        pending_docs = {
            doc_id: content 
            for doc_id, content in documents.items()
            if doc_id not in self._processed_ids
        }
        
        logger.info(f"処理対象: {len(pending_docs)}件 / 全体: {len(documents)}件")
        
        tasks = [
            self._process_single(doc_id, content, prompt_template)
            for doc_id, content in pending_docs.items()
        ]
        
        completed = 0
        results = []
        
        for coro in asyncio.as_completed(tasks):
            result = await coro
            results.append(result)
            self._save_checkpoint(result)
            
            completed += 1
            if completed % 10 == 0:
                success_rate = sum(1 for r in results if r.status == "success") / completed * 100
                total_tokens = sum(r.tokens_used for r in results)
                logger.info(
                    f"進捗: {completed}/{len(tasks)} "
                    f"({success_rate:.1f}%成功, "
                    f"累計{total_tokens:,}トークン)"
                )
        
        return results
 
# 使用例
async def main():
    # テスト用ドキュメント生成
    documents = {
        f"doc_{i:04d}": f"これはドキュメント{i}の内容です。" * 20
        for i in range(100)
    }
    
    processor = DocumentProcessor(
        max_concurrent=10,
        checkpoint_path="/tmp/doc_checkpoint.jsonl"
    )
    
    results = await processor.process_all(
        documents,
        prompt_template="以下の文書を3行で要約してください:\n\n{content}"
    )
    
    # 結果の集計
    success = [r for r in results if r.status == "success"]
    failed = [r for r in results if r.status == "failed"]
    total_tokens = sum(r.tokens_used for r in success)
    
    print(f"\n=== 処理結果 ===")
    print(f"成功: {len(success)}件")
    print(f"失敗: {len(failed)}件")
    print(f"合計トークン: {total_tokens:,}")
    print(f"推定コスト: ${total_tokens / 1_000_000 * 7.0:.3f} (Gemini 2.5 Pro Input Price参考)")
 
asyncio.run(main())

import asyncio
import time
import google.generativeai as genai
 
async def benchmark_concurrency(
    concurrency_levels: list[int] = [1, 5, 10, 15, 20],
    test_count: int = 10
) -> dict[int, float]:
    """
    各同時実行数でのスループットを測定し最適値を返す
    
    Returns:
        {同時実行数: リクエスト/秒} の辞書
    """
    model = genai.GenerativeModel("gemini-2.5-flash")  # 測定にはFlashを使用
    test_prompt = "1+1は？"  # 短いプロンプトでAPIレイテンシを測定
    
    results = {}
    
    for concurrency in concurrency_levels:
        semaphore = asyncio.Semaphore(concurrency)
        error_count = 0
        
        async def single_request():
            nonlocal error_count
            async with semaphore:
                try:
                    await model.generate_content_async(test_prompt)
                    return True
                except Exception:
                    error_count += 1
                    return False
        
        start_time = time.monotonic()
        await asyncio.gather(*[single_request() for _ in range(test_count)])
        elapsed = time.monotonic() - start_time
        
        rps = test_count / elapsed
        error_rate = error_count / test_count
        
        print(f"同時実行数 {concurrency:3d}: {rps:.2f} req/s, エラー率 {error_rate:.0%}")
        
        if error_rate > 0.2:  # エラー率20%超なら中断
            print(f"→ 同時実行数 {concurrency} は過剰です。前の値を使用してください。")
            break
        
        results[concurrency] = rps
    
    optimal = max(results, key=results.get)
    print(f"\n最適な同時実行数: {optimal} ({results[optimal]:.2f} req/s)")
    return results
 
asyncio.run(benchmark_concurrency())

import asyncio
import time
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.metrics import MeterProvider
import google.generativeai as genai
 
# トレーサーとメーターの初期化
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer("gemini_client")
 
meter_provider = MeterProvider()
metrics.set_meter_provider(meter_provider)
meter = metrics.get_meter("gemini_client")
 
# カスタムメトリクスの定義
request_counter = meter.create_counter(
    "gemini_requests_total",
    description="Gemini APIリクエストの総数"
)
latency_histogram = meter.create_histogram(
    "gemini_request_duration_seconds",
    description="Gemini APIリクエストのレイテンシ（秒）"
)
token_counter = meter.create_counter(
    "gemini_tokens_total",
    description="使用トークンの総数"
)
 
async def instrumented_generate(
    model: genai.GenerativeModel,
    prompt: str,
    **kwargs
) -> str:
    """
    OpenTelemetryトレーシングとメトリクスを統合したAPI呼び出し
    """
    with tracer.start_as_current_span("gemini.generate") as span:
        span.set_attribute("gemini.model", model.model_name)
        span.set_attribute("gemini.prompt_length", len(prompt))
        
        start = time.monotonic()
        status = "success"
        
        try:
            response = await model.generate_content_async(prompt, **kwargs)
            
            # トークン使用量の記録
            tokens = getattr(response.usage_metadata, "total_token_count", 0)
            span.set_attribute("gemini.tokens_used", tokens)
            token_counter.add(tokens, {"model": model.model_name})
            
            return response.text
            
        except Exception as e:
            status = "error"
            span.record_exception(e)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            raise
            
        finally:
            duration = time.monotonic() - start
            latency_histogram.record(duration, {"model": model.model_name, "status": status})
            request_counter.add(1, {"model": model.model_name, "status": status})

import google.api_core.exceptions as exceptions
from google.api_core.retry import AsyncRetry
 
# タイムアウト付きリクエスト
response = await model.generate_content_async(
    prompt,
    request_options={"timeout": 120}  # 120秒タイムアウト
)

def chunk_text(text: str, max_tokens: int = 500_000) -> list[str]:
    """テキストを最大トークン数以下に分割"""
    # 日本語: 約1.5文字/トークン を想定
    max_chars = max_tokens * 1
    
    if len(text) <= max_chars:
        return [text]
    
    chunks = []
    while text:
        chunk = text[:max_chars]
        # 段落境界で分割
        last_newline = chunk.rfind("\n\n")
        if last_newline > max_chars * 0.8:
            chunk = chunk[:last_newline]
        chunks.append(chunk)
        text = text[len(chunk):]
    
    return chunks