Gemini 2.5 Pro × FastAPI: プロダクション対応AIバックエンドの実装ガイド

取り組みの背景：なぜ Gemini × FastAPI なのか

Gemini 2.5 Pro APIは、1Mトークンのコンテキストウィンドウ、ネイティブマルチモーダル対応、強力なFunction Callingを備えた現時点で最も汎用性の高いLLM APIの一つです。一方、FastAPIはPythonのWebフレームワークの中で最も高速かつ非同期処理に優れており、AIバックエンドの構築に理想的な選択肢です。

この2つを組み合わせることで、本番環境で動く実用的なAIバックエンドを構築できます。ここでは単純なチャットボットの実装にとどまらず、ストリーミングレスポンス、レート制限、リトライロジック、Function Calling、コスト最適化、Dockerデプロイまで、プロダクションに必要なすべての要素を段階的に解説します。

この記事を読み終わると、以下のことができるようになります：

FastAPI + Gemini API を使ったストリーミング対応エンドポイントの実装
エクスポネンシャルバックオフによるレート制限の適切な処理
Function Calling を活用したツール連携バックエンドの構築
Dockerを使ったプロダクションデプロイ
APIコストを最小化するための最適化テクニック

対象読者: PythonとFastAPIの基礎知識があり、Gemini APIを本番環境で活用したいエンジニア

前提知識・環境準備

必要なもの

Python 3.11以上
Gemini API キー（Google AI Studio で取得）
Docker（プロダクションデプロイ時）
基本的なasync/await の理解

環境セットアップ

# プロジェクトディレクトリを作成
mkdir gemini-fastapi-backend && cd gemini-fastapi-backend
 
# 仮想環境の作成と有効化
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
 
# 依存パッケージのインストール
pip install fastapi uvicorn google-generativeai tenacity python-dotenv pydantic redis

.env ファイルを作成してAPIキーを管理します：

# .env
GEMINI_API_KEY=your_api_key_here
REDIS_URL=redis://localhost:6379
ENVIRONMENT=production
MAX_RETRIES=3

プロジェクト構成

gemini-fastapi-backend/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPIアプリのエントリポイント
│   ├── config.py         # 設定管理
│   ├── models.py         # Pydanticモデル
│   ├── services/
│   │   ├── gemini.py     # Gemini APIクライアント
│   │   └── tools.py      # Function Callingツール定義
│   └── routers/
│       ├── chat.py       # チャットエンドポイント
│       └── stream.py     # ストリーミングエンドポイント
├── Dockerfile
├── docker-compose.yml
└── requirements.txt

概念解説：アーキテクチャの設計思想

なぜ非同期処理が重要か

Gemini APIへのリクエストはネットワーク待ちが発生するI/Oバウンドな処理です。FastAPIのasync/awaitを正しく使うことで、1つのワーカープロセスで数百の同時リクエストを処理できます。同期的な実装と比べて、スループットが10〜50倍改善されることも珍しくありません。

ストリーミングの重要性

大規模言語モデルはトークンを逐次生成するため、ストリーミングを使わないと最後のトークンが生成されるまでレスポンスを返せません。Gemini 2.5 Proのような高性能モデルでは応答が長くなりがちで、ユーザー体験に大きく影響します。Server-Sent Events（SSE）を使ったストリーミングにより、最初のトークンが生成された瞬間からユーザーに届けることができます。

ステップバイステップ実装

Step 1: 設定管理とGeminiクライアント

まず設定とGeminiクライアントを実装します：

# app/config.py
from pydantic_settings import BaseSettings
 
class Settings(BaseSettings):
    gemini_api_key: str
    environment: str = "development"
    max_retries: int = 3
    redis_url: str = "redis://localhost:6379"
 
    # Geminiモデル設定
    default_model: str = "gemini-2.5-pro"
    max_output_tokens: int = 8192
    temperature: float = 0.7
 
    class Config:
        env_file = ".env"
 
settings = Settings()

# app/services/gemini.py
import asyncio
import google.generativeai as genai
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
from google.api_core.exceptions import ResourceExhausted, ServiceUnavailable
from app.config import settings
import logging
 
logger = logging.getLogger(__name__)
 
# Gemini APIクライアントの初期化
genai.configure(api_key=settings.gemini_api_key)
 
class GeminiClient:
    def __init__(self, model_name: str = None):
        self.model_name = model_name or settings.default_model
        self.generation_config = genai.types.GenerationConfig(
            max_output_tokens=settings.max_output_tokens,
            temperature=settings.temperature,
        )
        self.model = genai.GenerativeModel(
            model_name=self.model_name,
            generation_config=self.generation_config,
        )
 
    @retry(
        stop=stop_after_attempt(settings.max_retries),
        wait=wait_exponential(multiplier=1, min=4, max=60),
        retry=retry_if_exception_type((ResourceExhausted, ServiceUnavailable)),
        before_sleep=lambda retry_state: logger.warning(
            f"Rate limited. Retrying in {retry_state.next_action.sleep}s "
            f"(attempt {retry_state.attempt_number}/{settings.max_retries})"
        ),
    )
    async def generate(self, prompt: str, system_instruction: str = None) -> str:
        """
        エクスポネンシャルバックオフ付きのコンテンツ生成。
        ResourceExhausted（429）やServiceUnavailable（503）の場合に自動リトライ。
        """
        model = self.model
        if system_instruction:
            model = genai.GenerativeModel(
                model_name=self.model_name,
                generation_config=self.generation_config,
                system_instruction=system_instruction,
            )
 
        # asyncio.to_thread でブロッキングAPIを非同期化
        response = await asyncio.to_thread(
            model.generate_content, prompt
        )
        return response.text
 
    async def stream_generate(self, prompt: str, system_instruction: str = None):
        """
        ストリーミング生成。各チャンクをasync generatorとして返す。
        """
        model = self.model
        if system_instruction:
            model = genai.GenerativeModel(
                model_name=self.model_name,
                generation_config=self.generation_config,
                system_instruction=system_instruction,
            )
 
        # ストリーミングレスポンスを取得（同期APIをスレッドで実行）
        response = await asyncio.to_thread(
            model.generate_content, prompt, stream=True
        )
 
        for chunk in response:
            if chunk.text:
                yield chunk.text
 
# シングルトンインスタンス
gemini_client = GeminiClient()

Step 2: ストリーミングエンドポイントの実装

# app/routers/stream.py
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from app.services.gemini import gemini_client
import json
import asyncio
 
router = APIRouter(prefix="/api/v1", tags=["streaming"])
 
class StreamRequest(BaseModel):
    message: str
    system_instruction: str | None = None
    session_id: str | None = None
 
async def event_generator(request: StreamRequest):
    """
    Server-Sent Events (SSE) 形式でストリーミングレスポンスを生成する。
    各チャンクはJSON形式で {"type": "chunk", "content": "..."} として送信。
    エラー時は {"type": "error", "message": "..."} を送信。
    """
    try:
        async for chunk in gemini_client.stream_generate(
            prompt=request.message,
            system_instruction=request.system_instruction,
        ):
            data = json.dumps({"type": "chunk", "content": chunk})
            yield f"data: {data}\n\n"
            # バックプレッシャー対策: バッファが詰まらないように
            await asyncio.sleep(0)
 
        # 完了イベントを送信
        yield f"data: {json.dumps({'type': 'done'})}\n\n"
 
    except Exception as e:
        error_data = json.dumps({"type": "error", "message": str(e)})
        yield f"data: {error_data}\n\n"
 
@router.post("/stream")
async def stream_chat(request: StreamRequest):
    """
    ストリーミングチャットエンドポイント。
    Server-Sent Events (SSE) でリアルタイムにトークンを返す。
 
    クライアント側での受信例（JavaScript）:
    const eventSource = new EventSource('/api/v1/stream');
    eventSource.onmessage = (e) => {
        const data = JSON.parse(e.data);
        if (data.type === 'chunk') console.log(data.content);
    };
    """
    if not request.message.strip():
        raise HTTPException(status_code=400, detail="メッセージを入力してください")
 
    return StreamingResponse(
        event_generator(request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx のバッファリングを無効化
        },
    )

Step 3: Function Calling の実装

Function Calling を使って外部APIと連携するバックエンドを実装します：

# app/services/tools.py
import google.generativeai as genai
import httpx
import asyncio
from datetime import datetime
 
# ツール定義（Gemini APIに渡す関数スキーマ）
TOOLS = [
    genai.protos.Tool(
        function_declarations=[
            genai.protos.FunctionDeclaration(
                name="get_current_weather",
                description="指定した都市の現在の天気情報を取得する",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={
                        "city": genai.protos.Schema(
                            type=genai.protos.Type.STRING,
                            description="都市名（例: Tokyo, Osaka）",
                        ),
                        "unit": genai.protos.Schema(
                            type=genai.protos.Type.STRING,
                            enum=["celsius", "fahrenheit"],
                            description="温度の単位",
                        ),
                    },
                    required=["city"],
                ),
            ),
            genai.protos.FunctionDeclaration(
                name="search_articles",
                description="Gemini Labの記事を検索する",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={
                        "query": genai.protos.Schema(
                            type=genai.protos.Type.STRING,
                            description="検索クエリ",
                        ),
                        "limit": genai.protos.Schema(
                            type=genai.protos.Type.INTEGER,
                            description="返す記事数（デフォルト: 5）",
                        ),
                    },
                    required=["query"],
                ),
            ),
        ]
    )
]
 
# ツールの実行関数
async def execute_tool(function_name: str, args: dict) -> dict:
    """Function Callで呼び出されたツールを実行する"""
    if function_name == "get_current_weather":
        # 実際の天気APIへのリクエスト（ここではモック）
        city = args.get("city", "Tokyo")
        unit = args.get("unit", "celsius")
        # 実際はOpen-Meteo等の無料APIを使用
        return {
            "city": city,
            "temperature": 22,
            "unit": unit,
            "condition": "晴れ",
            "humidity": 55,
        }
 
    elif function_name == "search_articles":
        query = args.get("query", "")
        limit = args.get("limit", 5)
        # 実際はElasticsearchやAlgoliaを使用
        return {
            "results": [
                {"title": f"Gemini {query}入門", "url": f"/articles/gemini-basics/{query.lower().replace(' ', '-')}"},
            ],
            "total": 1,
        }
 
    return {"error": f"Unknown function: {function_name}"}

# app/routers/chat.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import google.generativeai as genai
import asyncio
from app.config import settings
from app.services.tools import TOOLS, execute_tool
import logging
 
router = APIRouter(prefix="/api/v1", tags=["chat"])
logger = logging.getLogger(__name__)
 
class ChatRequest(BaseModel):
    message: str
    history: list[dict] = []
    use_tools: bool = True
 
class ChatResponse(BaseModel):
    response: str
    tool_calls: list[dict] = []
    model: str
    tokens_used: int | None = None
 
@router.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """
    Function Calling対応のチャットエンドポイント。
    ツールを呼び出しながら複数ターンの推論を実行する。
    """
    genai.configure(api_key=settings.gemini_api_key)
 
    model = genai.GenerativeModel(
        model_name=settings.default_model,
        tools=TOOLS if request.use_tools else None,
    )
 
    # チャット履歴を変換
    chat_history = []
    for msg in request.history:
        chat_history.append({
            "role": msg["role"],
            "parts": [msg["content"]],
        })
 
    chat_session = model.start_chat(history=chat_history)
 
    tool_calls_log = []
 
    # Function Calling ループ
    response = await asyncio.to_thread(chat_session.send_message, request.message)
 
    # ツール呼び出しが必要な場合はループ
    while response.candidates[0].content.parts[0].function_call.name if (
        response.candidates and
        response.candidates[0].content.parts and
        hasattr(response.candidates[0].content.parts[0], 'function_call') and
        response.candidates[0].content.parts[0].function_call.name
    ) else False:
        fc = response.candidates[0].content.parts[0].function_call
        func_name = fc.name
        func_args = dict(fc.args)
 
        logger.info(f"Tool call: {func_name}({func_args})")
        tool_calls_log.append({"function": func_name, "args": func_args})
 
        # ツールを実行
        tool_result = await execute_tool(func_name, func_args)
 
        # 結果をGeminiに返す
        response = await asyncio.to_thread(
            chat_session.send_message,
            genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name=func_name,
                    response={"result": tool_result},
                )
            )
        )
 
    # 最終テキストレスポンスを取得
    final_text = response.text
 
    # トークン使用量を取得（利用可能な場合）
    tokens_used = None
    if hasattr(response, 'usage_metadata'):
        tokens_used = response.usage_metadata.total_token_count
 
    return ChatResponse(
        response=final_text,
        tool_calls=tool_calls_log,
        model=settings.default_model,
        tokens_used=tokens_used,
    )

Step 4: FastAPIアプリの組み立て

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import logging
from app.routers import chat, stream
from app.config import settings
 
# ロギング設定
logging.basicConfig(
    level=logging.INFO if settings.environment == "production" else logging.DEBUG,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """アプリケーションのライフサイクル管理"""
    logging.info(f"🚀 Starting Gemini FastAPI Backend (env: {settings.environment})")
    yield
    logging.info("🛑 Shutting down Gemini FastAPI Backend")
 
app = FastAPI(
    title="Gemini FastAPI Backend",
    description="Gemini 2.5 Pro × FastAPIによるプロダクション対応AIバックエンド",
    version="1.0.0",
    lifespan=lifespan,
)
 
# CORS設定（本番環境では適切なオリジンを設定すること）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"] if settings.environment == "production" else ["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
# ルーターを登録
app.include_router(chat.router)
app.include_router(stream.router)
 
@app.get("/health")
async def health_check():
    """ヘルスチェックエンドポイント（ロードバランサー用）"""
    return {"status": "ok", "model": settings.default_model}
 
# 期待する出力（curl http://localhost:8000/health）:
# {"status": "ok", "model": "gemini-2.5-pro"}

応用パターン

レート制限の高度な処理

プロダクション環境では、複数のクライアントからの同時リクエストに対して、より精緻なレート制限が必要です：

# app/middleware/rate_limit.py
import asyncio
from collections import deque
from datetime import datetime, timedelta
import time
 
class AdaptiveRateLimiter:
    """
    適応型レート制限器。
    429エラーが増えると自動的にリクエスト間隔を広げ、
    成功が続くと間隔を縮める。
    """
    def __init__(self, initial_rps: float = 10.0):
        self.min_interval = 1.0 / initial_rps
        self.current_interval = self.min_interval
        self.last_request_time = 0.0
        self.lock = asyncio.Lock()
        self._consecutive_errors = 0
        self._consecutive_successes = 0
 
    async def acquire(self):
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_request_time
            if elapsed < self.current_interval:
                await asyncio.sleep(self.current_interval - elapsed)
            self.last_request_time = time.monotonic()
 
    def record_success(self):
        self._consecutive_successes += 1
        self._consecutive_errors = 0
        # 10回連続成功したらレートを上げる
        if self._consecutive_successes >= 10:
            self.current_interval = max(
                self.min_interval,
                self.current_interval * 0.9
            )
            self._consecutive_successes = 0
 
    def record_rate_limit(self):
        self._consecutive_errors += 1
        self._consecutive_successes = 0
        # レート制限されたら間隔を2倍に
        self.current_interval = min(60.0, self.current_interval * 2.0)
 
rate_limiter = AdaptiveRateLimiter(initial_rps=5.0)

コスト最適化：レスポンスキャッシュ

同一プロンプトへの重複リクエストをキャッシュすることで、コストを大幅に削減できます：

# app/services/cache.py
import hashlib
import json
from datetime import timedelta
import redis.asyncio as redis
from app.config import settings
 
class ResponseCache:
    def __init__(self):
        self.client = redis.from_url(settings.redis_url)
        self.ttl = timedelta(hours=24)  # 24時間キャッシュ
 
    def _make_key(self, prompt: str, model: str) -> str:
        """プロンプトとモデル名からキャッシュキーを生成"""
        content = f"{model}:{prompt}"
        return f"gemini:cache:{hashlib.sha256(content.encode()).hexdigest()}"
 
    async def get(self, prompt: str, model: str) -> str | None:
        key = self._make_key(prompt, model)
        cached = await self.client.get(key)
        if cached:
            return json.loads(cached)
        return None
 
    async def set(self, prompt: str, model: str, response: str):
        key = self._make_key(prompt, model)
        await self.client.setex(
            key,
            int(self.ttl.total_seconds()),
            json.dumps(response)
        )
 
cache = ResponseCache()

トラブルシューティング

よくあるエラーと対処法

ResourceExhausted（429 Too Many Requests）

最も頻繁に発生するエラーです。Gemini APIには無料枠と有料枠でそれぞれレート制限があります。tenacityライブラリを使ったリトライロジックで自動的に対処できます（Step 1のコード参照）。

# 現在のレート制限を確認する方法
# 無料枠: gemini-2.5-pro は 60 RPM / 1,000 RPD
# 有料枠: プランによって異なる（Google AI Studioで確認）

ValueError: Invalid operation: The response.text quick accessor

ストリーミングレスポンスで response.text を直接アクセスしようとすると発生します。ストリーミング時は chunk.text を使用してください。

asyncio.to_thread でのタイムアウト

Gemini APIのリクエストが長い場合、デフォルトのタイムアウトで失敗することがあります：

import asyncio
 
async def generate_with_timeout(prompt: str, timeout: float = 120.0) -> str:
    try:
        return await asyncio.wait_for(
            asyncio.to_thread(model.generate_content, prompt),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        raise HTTPException(status_code=504, detail="APIタイムアウト（120秒）")

Dockerデプロイ

Dockerfile

# Dockerfile
FROM python:3.11-slim
 
WORKDIR /app
 
# セキュリティ: 非rootユーザーで実行
RUN addgroup --system appgroup && adduser --system --ingroup appgroup appuser
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY app/ ./app/
 
# 非rootユーザーに切り替え
USER appuser
 
# ヘルスチェック設定
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1
 
EXPOSE 8000
 
# 本番環境: workers数 = (CPU数 × 2) + 1 が推奨
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

docker-compose.yml

# docker-compose.yml
version: "3.9"
 
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - GEMINI_API_KEY=${GEMINI_API_KEY}
      - REDIS_URL=redis://redis:6379
      - ENVIRONMENT=production
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped
 
  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5
    restart: unless-stopped
 
# 起動方法:
# GEMINI_API_KEY=your_key docker-compose up -d

コスト・パフォーマンス考慮事項

Gemini 2.5 Pro のコスト計算（2026年3月時点）

用途	モデル	入力コスト	出力コスト
長いコンテキスト	Gemini 2.5 Pro (>200K tokens)	$2.50/1M tokens	$15.00/1M tokens
標準	Gemini 2.5 Pro (≤200K tokens)	$1.25/1M tokens	$10.00/1M tokens
コスト効率重視	Gemini 2.5 Flash	$0.075/1M tokens	$0.30/1M tokens

コスト削減のベストプラクティス

キャッシュの活用: 同一または類似のクエリに対してRedisキャッシュを使用（実装例はStep 3参照）
適切なモデル選択: 複雑な推論が不要なタスクにはGemini 2.5 Flashを使用
Context Cachingの活用: 長いSystem Instructionや共通コンテキストはContext Cachingで効率化
バッチ処理: リアルタイムでない処理はBatch APIを使用してコストを半減
プロンプト最適化: 不要なプロンプトの冗長性を除去してトークン数を削減

まとめと次のステップ

ここではGemini 2.5 Pro × FastAPIによるプロダクション対応AIバックエンドの構築を、以下の観点から解説しました：

非同期処理: asyncio.to_thread で同期SDKをFastAPIに統合
ストリーミング: SSEによるリアルタイムトークン配信
レート制限: tenacity と適応型レート制限による安定性確保
Function Calling: ツール連携によるエージェント的な動作の実現
コスト最適化: キャッシュ・適切なモデル選択・バッチ処理
Dockerデプロイ: 本番環境に対応したコンテナ化

次のステップとして、以下の記事も参考にしてください：

Gemini API Function Calling 実践ガイド — より高度なFunction Callingパターン
Vertex AI × Gemini プロダクションデプロイ — エンタープライズ向けデプロイ戦略
Gemini APIコスト最適化完全ガイド — コスト削減の詳細な手法