◈ API / SDK/2026-04-10上級

Gemini 2.5 Pro API 実践トラブルシューティング

generativelanguage.googleapis.com のエラーに悩む開発者向け。Gemini 2.5 Pro API のエラーコード別解決策、本番環境での安定稼働パターン、コスト最適化まで、実装レベルのトラブルシューティング完全版。

gemini-api²⁸⁰ gemini-2.5-pro¹³ troubleshooting⁵⁷ generativelanguage³ api-error

✦ プレミアム記事

Gemini 2.5 Pro API は、Google の最新 AI モデルへのアクセスを提供する強力なツールです。しかし、generativelanguage.googleapis.com エンドポイントでの開発時に、認証エラー、レート制限、モデル見つからない問題など、様々なトラブルに遭遇することは珍しくありません。ここで整理するのはAPI レベルでのトラブルシューティングから本番環境での安定稼働まで、実践的な解決策をお伝えします。

Gemini 2.5 Pro API の基本構成

まず、Gemini 2.5 Pro API の全体像を理解する点が肝心です。

generativelanguage.googleapis.com エンドポイント

Gemini API は generativelanguage.googleapis.com をベースに、以下の主要エンドポイントで構成されています：

https://generativelanguage.googleapis.com/v1beta/models/{model}/generateContent

ここで {model} には、お使いのモデル名が入ります。

Gemini 2.5 Pro のモデル名

Gemini 2.5 Pro には、複数の指定方法があります：

モデル名	説明	推奨用途
`gemini-2.5-pro`	安定版（定期更新）	本番環境・安定性重視
`gemini-2.5-pro-latest`	最新バージョン（リアルタイム更新）	開発・検証環境
`gemini-2.5-pro-exp`	実験的バージョン	最新機能のテスト

重要: モデル名を誤ると「Model not found」エラーが発生します。本番環境ではgemini-2.5-proを、開発環境ではgemini-2.5-pro-latestを推奨します。

API 認証方式

Gemini API は以下の認証をサポート：

API Key 認証 （簡単、開発向け）
OAuth 2.0 認証 （セキュア、サーバー/バックエンド向け）

# API Key 認証の例（シェル）
curl https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro:generateContent?key=YOUR_GEMINI_API_KEY \
  -H "Content-Type: application/json" \
  -d '{
    "contents": [{
      "parts": [{"text": "Hello, Gemini!"}]
    }]
  }'

認証エラーのトラブルシューティング

認証に関する問題は、API リクエストの最初のハードルです。

エラー: 401 Unauthorized / Invalid API Key

原因:

API Key が設定されていない、または空文字列
API Key が無効（期限切れ、削除済み）
API Key が異なるプロジェクトのもの
API Key をリクエストに含めていない

確認手順:

Google Cloud Console にログイン
左メニュー → 「API とサービス」→「認証情報」
該当する API Key を確認（有効状態か？）
Key の制限設定を確認（API 制限が Generative Language API に限定されているか？）

解決コード例（Python）:

import google.generativeai as genai
 
# 環境変数から安全に取得
api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
    raise ValueError("GEMINI_API_KEY 環境変数が設定されていません")
 
genai.configure(api_key=api_key)
 
# これでリクエスト可能
model = genai.GenerativeModel('gemini-2.5-pro')
response = model.generate_content("Hello, Gemini!")

エラー: 403 Forbidden / Permission Denied

原因:

プロジェクトで Generative Language API が有効化されていない
サービスアカウントに必要な IAM ロール（roles/aiplatform.serviceAgent）が割り当てられていない
OAuth 認証で必要なスコープが不足

解決手順:

Google Cloud Console で該当プロジェクトを選択
「API とサービス」→「ライブラリ」を開く
「Generative Language API」を検索し、「有効にする」をクリック
有効化から反映まで数分かかることがあります

IAM ロール設定（OAuth の場合）:

「IAM」メニューを開く
サービスアカウント / ユーザーを選択
以下のロールを付与：
- roles/aiplatform.serviceAgent （推奨）
- または roles/editor （本番環境では避ける）

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦generativelanguage.googleapis.comのエラーに悩んでいた人が、エラーコード別の解決策を素早く見つけられるようになる

✦Gemini 2.5 Proのモデル名や設定で詰まっていた人が、本番環境で安定稼働させるための実装パターンを手に入れられる

✦APIコストと性能のバランスに迷っていた人が、ユースケース別の最適設定を自分で判断できるようになる

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

よくある API エラーとその解決策

エラー: 400 Bad Request

原因: リクエスト本体の形式が不正

よくあるパターン:

(a) JSON フォーマットが不正

# ❌ 悪い例
response = model.generate_content({
    "contents": [
        "parts": [{"text": "Hello"}]  # 括弧の対応がおかしい
    ]
})
 
# ✓ 良い例
response = model.generate_content({
    "contents": [{
        "parts": [{"text": "Hello"}]
    }]
})

(b) required フィールドが欠けている

# ❌ 悪い例（parts が空）
response = model.generate_content({
    "contents": [{
        "parts": []  # parts は必須
    }]
})
 
# ✓ 良い例
response = model.generate_content({
    "contents": [{
        "parts": [{"text": "Hello"}]
    }]
})

(c) 不正な生成設定

# ❌ 悪い例（temperature が範囲外）
response = model.generate_content(
    "Hello",
    generation_config={
        "temperature": 2.5  # 範囲は 0.0～2.0
    }
)
 
# ✓ 良い例
response = model.generate_content(
    "Hello",
    generation_config={
        "temperature": 1.0
    }
)

デバッグコード:

import json
 
try:
    response = model.generate_content(prompt)
except Exception as e:
    print(f"エラー: {e}")
    print(f"レスポンス本体: {e.response.text if hasattr(e, 'response') else 'N/A'}")

エラー: 429 Too Many Requests（レート制限）

原因: API リクエストの送信頻度が上限を超えた

Gemini API の無料ティアでは、1分あたりのリクエスト数に制限があります：

Free : 60 リクエスト/分
Pro : 500 リクエスト/分～（プランにより異なる）

解決策：エクスポーネンシャルバックオフの実装

import time
import random
from google.api_core.exceptions import ResourceExhausted
 
def generate_with_retry(model, prompt, max_retries=3):
    """429 エラーのリトライロジック"""
    for attempt in range(max_retries):
        try:
            return model.generate_content(prompt)
        except ResourceExhausted:
            if attempt == max_retries - 1:
                raise
            # バックオフ時間を指数関数的に増加（＋ジッターで衝突回避）
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"レート制限発生。{wait_time:.1f}秒待機...")
            time.sleep(wait_time)

エラー: 500 Internal Server Error / 503 Service Unavailable

原因: Google 側のサーバー問題（一時的）

対応:

数分待機してリトライ
本番環境ではサーキットブレーカーパターンを導入

from circuitbreaker import circuit
 
@circuit(failure_threshold=5, recovery_timeout=60)
def call_gemini_api(prompt):
    """サーキットブレーカーパターン"""
    model = genai.GenerativeModel('gemini-2.5-pro')
    return model.generate_content(prompt)
 
# 失敗が連続すると、自動的に呼び出しをスキップ
try:
    response = call_gemini_api("Hello")
except Exception as e:
    print(f"API 呼び出し失敗: {e}")

エラー: Model not found / INVALID_ARGUMENT

原因: モデル名が誤り、または廃止されたバージョン

確認方法:

# 利用可能なモデル一覧を取得
for model in genai.list_models():
    print(f"- {model.name}")
 
# Gemini 2.5 Pro の正式な名前を確認
model_name = "gemini-2.5-pro"  # または "gemini-2.5-pro-latest"
try:
    model = genai.GenerativeModel(model_name)
    print(f"✓ モデル '{model_name}' は利用可能")
except Exception as e:
    print(f"✗ モデル '{model_name}' は利用不可: {e}")

Gemini 2.5 Pro 特有の設定ポイント

Thinking Mode の有効化と制御

Gemini 2.5 Pro の強力な機能が「thinking mode」（推論モード）です。これにより、複雑な問題をより深く思考することが可能になります。

from google.generativeai.types import content_types
 
# Thinking mode の有効化
response = model.generate_content(
    "複雑な数学問題を解いてください",
    generation_config={
        "thinking": {
            "type": "ENABLED",
            "budget_tokens": 10000  # 思考トークン数の上限
        }
    }
)
 
print(response.text)

thinking_budget_tokens は、思考フェーズに消費できるトークン数を制限します。大きいほど深い思考が可能ですが、コストと時間も増加します。

Safety Settings の調整

不適切なコンテンツフィルタリングは、合法的なユースケースまでブロックする可能性があります。

from google.generativeai.types import safety_types
 
safety_settings = [
    safety_types.SafetySetting(
        category=safety_types.HarmCategory.HARM_CATEGORY_UNSPECIFIED,
        threshold=safety_types.HarmBlockThreshold.BLOCK_NONE
    ),
    safety_types.SafetySetting(
        category=safety_types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
        threshold=safety_types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE
    ),
]
 
response = model.generate_content(
    prompt,
    safety_settings=safety_settings
)

Generation Config の最適化

top_p, top_k, temperature の組み合わせで、出力の多様性と一貫性をコントロール：

generation_config = {
    "temperature": 0.7,      # 0～2.0。低いほど決定的、高いほど創造的
    "top_p": 0.95,          # Nucleus sampling。0.95 が標準
    "top_k": 40,            # 上位 40 候補から選択
    "max_output_tokens": 2048,
    "candidate_count": 1     # 生成する候補数（1 推奨）
}
 
response = model.generate_content(prompt, generation_config=generation_config)

実践的な Python コード例

基本的な実装（エラーハンドリング付き）

import google.generativeai as genai
import os
import time
from google.api_core.exceptions import (
    ResourceExhausted,
    InvalidArgument,
    PermissionDenied,
)
 
class GeminiClient:
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("GEMINI_API_KEY")
        if not self.api_key:
            raise ValueError("API Key が設定されていません")
        genai.configure(api_key=self.api_key)
        self.model = genai.GenerativeModel('gemini-2.5-pro')
 
    def generate_content(self, prompt: str, max_retries: int = 3) -> str:
        """エラーハンドリングとリトライ付きのコンテンツ生成"""
        for attempt in range(max_retries):
            try:
                response = self.model.generate_content(prompt)
                return response.text
            except ResourceExhausted:
                if attempt == max_retries - 1:
                    raise
                wait_time = (2 ** attempt) + (attempt * 0.1)
                print(f"[試行 {attempt + 1}/{max_retries}] レート制限. {wait_time:.1f}秒待機...")
                time.sleep(wait_time)
            except (InvalidArgument, PermissionDenied) as e:
                print(f"API 設定エラー: {e}")
                raise
            except Exception as e:
                print(f"予期しないエラー: {e}")
                raise
 
# 使用例
client = GeminiClient()
result = client.generate_content("今日の天気を教えてください")
print(result)

非同期処理

import asyncio
import google.generativeai as genai
 
async def generate_async(prompt: str) -> str:
    """非同期でコンテンツを生成"""
    model = genai.GenerativeModel('gemini-2.5-pro')
    # google-generativeai はネイティブ非同期をサポートしていないため、
    # ThreadPoolExecutor で実行
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(
        None, 
        lambda: model.generate_content(prompt)
    )
    return response.text
 
# 複数リクエストを並行実行
async def batch_generate(prompts: list[str]) -> list[str]:
    tasks = [generate_async(p) for p in prompts]
    return await asyncio.gather(*tasks)
 
# 実行例
prompts = ["Hello", "How are you?", "Tell me a joke"]
results = asyncio.run(batch_generate(prompts))
for prompt, result in zip(prompts, results):
    print(f"Q: {prompt}\nA: {result}\n")

ストリーミング API のトラブルシューティング

ストリーミングレスポンスでは、接続切断や部分的な応答処理が課題になります。

ストリーミング中の接続切断対応

def stream_content_with_timeout(prompt: str, timeout: int = 30) -> str:
    """タイムアウト付きストリーミング"""
    model = genai.GenerativeModel('gemini-2.5-pro')
    response = model.generate_content(prompt, stream=True)
    
    accumulated_text = ""
    try:
        for chunk in response:
            if chunk.text:
                accumulated_text += chunk.text
                print(chunk.text, end='', flush=True)
    except Exception as e:
        print(f"\nストリーミング中断: {e}")
        print(f"ここまでの応答: {accumulated_text}")
        return accumulated_text
    
    print()  # 改行
    return accumulated_text

部分的応答の処理

def parse_streaming_json(prompt: str) -> dict:
    """ストリーミング中に JSON を段階的にパース"""
    import json
    
    model = genai.GenerativeModel('gemini-2.5-pro')
    response = model.generate_content(prompt, stream=True)
    
    buffer = ""
    for chunk in response:
        if chunk.text:
            buffer += chunk.text
            
            # 完全な JSON オブジェクトが揃ったか確認
            try:
                parsed = json.loads(buffer)
                print(f"✓ 部分的 JSON 取得: {parsed}")
            except json.JSONDecodeError:
                # まだ不完全
                pass
    
    # 最終的な解析
    return json.loads(buffer)

コスト最適化とクォータ管理

トークン数の計算方法

def estimate_tokens(text: str) -> dict:
    """テキストのトークン数を見積もり"""
    model = genai.GenerativeModel('gemini-2.5-pro')
    
    # 正式なトークンカウント API
    response = model.count_tokens(text)
    return {
        "text": text[:50] + "...",
        "total_tokens": response.total_tokens
    }
 
# テスト
print(estimate_tokens("This is a test prompt"))

バッチ処理によるコスト削減

def batch_process(prompts: list[str], batch_size: int = 5) -> list[str]:
    """バッチ処理で複数のリクエストを効率的に処理"""
    model = genai.GenerativeModel('gemini-2.5-pro')
    results = []
    
    for i in range(0, len(prompts), batch_size):
        batch = prompts[i:i+batch_size]
        print(f"処理中: バッチ {i//batch_size + 1} ({len(batch)}件)")
        
        for prompt in batch:
            try:
                response = model.generate_content(prompt)
                results.append(response.text)
            except Exception as e:
                print(f"エラー: {e}")
                results.append(None)
    
    return results
 
# 使用例
prompts = [f"Explain {topic}" for topic in ["AI", "ML", "NLP"]]
responses = batch_process(prompts)

クォータ上限の確認と増加申請

# Google Cloud Console → Quotas で確認可能
# 制限値を超える場合の申請フロー:
# 1. Cloud Console → IAM と管理 → クォータ
# 2. 「Generative Language API」検索
# 3. 制限項目を選択 → 「クォータを編集」
# 4. 新しい上限値を入力 → 申請

個人開発者の視点から（実体験メモ）

本番環境での安定稼働パターン

ロードバランシング

from collections import deque
import random
 
class LoadBalancedGeminiClient:
    """複数リージョン・複数 API Key でのロードバランシング"""
    
    def __init__(self, api_keys: list[str]):
        self.api_keys = deque(api_keys)
        self.models = [
            genai.GenerativeModel('gemini-2.5-pro')
            for _ in api_keys
        ]
    
    def generate_content(self, prompt: str) -> str:
        """ラウンドロビンで Key を切り替え"""
        api_key = self.api_keys[0]
        model = self.models[0]
        
        try:
            response = model.generate_content(prompt)
            return response.text
        except Exception as e:
            # キーをローテーション
            self.api_keys.rotate(-1)
            raise
    
    def rotate_keys(self):
        """手動でキーをローテーション"""
        self.api_keys.rotate(-1)
 
# 使用例
client = LoadBalancedGeminiClient([
    os.getenv("GEMINI_API_KEY_1"),
    os.getenv("GEMINI_API_KEY_2"),
    os.getenv("GEMINI_API_KEY_3"),
])

エラーモニタリング

import logging
from datetime import datetime
 
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('gemini_api.log'),
        logging.StreamHandler()
    ]
)
 
logger = logging.getLogger(__name__)
 
def monitored_generate(prompt: str, model) -> str:
    """ロギング付きの実行"""
    start_time = datetime.now()
    try:
        response = model.generate_content(prompt)
        duration = (datetime.now() - start_time).total_seconds()
        logger.info(f"✓ リクエスト成功（{duration:.2f}秒）")
        return response.text
    except Exception as e:
        logger.error(f"✗ リクエスト失敗: {type(e).__name__} - {e}")
        raise

API バージョン管理

class VersionedGeminiClient:
    """複数バージョンの並行実行管理"""
    
    AVAILABLE_MODELS = {
        "stable": "gemini-2.5-pro",
        "latest": "gemini-2.5-pro-latest",
        "experimental": "gemini-2.5-pro-exp",
    }
    
    def __init__(self, version: str = "stable"):
        self.version = version
        self.model = genai.GenerativeModel(self.AVAILABLE_MODELS[version])
    
    def switch_version(self, version: str):
        """バージョンを切り替え"""
        if version not in self.AVAILABLE_MODELS:
            raise ValueError(f"不正なバージョン: {version}")
        self.version = version
        self.model = genai.GenerativeModel(self.AVAILABLE_MODELS[version])
    
    def generate(self, prompt: str) -> str:
        return self.model.generate_content(prompt).text
 
# 使用例：段階的なロールアウト
client = VersionedGeminiClient(version="stable")
# テスト後
client.switch_version("latest")

Gemini 2.5 Pro API は、適切なトラブルシューティングと設定により、極めて堅牢で高性能なアプリケーションの基盤となります。このガイドを参考に、本番環境での安定稼働と継続的な改善を実現してください。

Gemini 2.5 Proとは：2.0からの進化と特徴

Gemini 2.5 Proは、2024年後半から2025年初期に登場した、Google AIの最新世代モデルです。

Gemini 2.0との主な違い：

項目	Gemini 2.0	Gemini 2.5 Pro
コンテキストウィンドウ	100万トークン	200万トークン（推定）
思考モード	基本実装	Advanced Thinking Mode（最大32K思考トークン）
マルチモーダル入力	画像・PDF・動画	同上 + 音声対応拡張
APIレスポンス速度	平均2〜4秒	平均1.5〜3秒（Flex Priority対応）
コスト効率	ベース価格	50%削減（キャッシング活用時）

Gemini 2.5 Proの革新性：

Advanced Thinking Mode: 複雑な推論タスクで精度向上
Context Caching: API呼び出しコストの大幅削減
Flexible Routing: リアルタイムトラフィック最適化
Multimodal Grounding: ビジョンと言語の高度な統合

gemini-2.5-pro-latest モデルの仕様と正しい使い方

エンドポイント構造：generativelanguage.googleapis.com

gemini-2.5-pro-latest へのアクセスは、次のエンドポイントを通じて行われます：

https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro-latest:generateContent

重要な注意点:

エンドポイントURLのバージョンは v1beta （本番環境ではstableの v1 が利用可能な場合もありますが、最新機能は v1beta のみ）
gemini-2.5-pro-latest は Google API Keyベースの認証が必須（Service Accountでも可能だが制限がある）
レート制限：デフォルトは1分あたり100リクエスト（有償プランで拡張可能）

クライアントの正しい初期化：Python & TypeScript

Python での初期化：

import google.generativeai as genai
 
API_KEY = "YOUR_GEMINI_API_KEY"
genai.configure(api_key=API_KEY)
 
# gemini-2.5-pro-latest モデルを正式に指定
model = genai.GenerativeModel(
    model_name="gemini-2.5-pro-latest",
    generation_config={
        "temperature": 0.7,
        "max_output_tokens": 2048,
        "top_p": 0.95,
        "top_k": 40,
    }
)
 
# テスト用の簡単なリクエスト
response = model.generate_content("Hello, Gemini 2.5 Pro!")
print(response.text)

TypeScript での初期化：

import { GoogleGenerativeAI } from "@google/generative-ai";
 
const API_KEY = "YOUR_GEMINI_API_KEY";
const client = new GoogleGenerativeAI({ apiKey: API_KEY });
 
const model = client.getGenerativeModel({
  model: "gemini-2.5-pro-latest",
});
 
const generationConfig = {
  temperature: 0.7,
  maxOutputTokens: 2048,
  topP: 0.95,
  topK: 40,
};
 
const response = await model.generateContent({
  contents: [
    {
      role: "user",
      parts: [{ text: "Hello, Gemini 2.5 Pro!" }],
    },
  ],
  generationConfig,
});
 
console.log(response.response.text());

generativelanguage.googleapis.com への直接HTTP呼び出し

REST APIを直接使う場合：

curl -X POST \
  "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro-latest:generateContent?key=YOUR_GEMINI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "contents": [{
      "role": "user",
      "parts": [{"text": "Explain quantum computing in simple terms"}]
    }],
    "generationConfig": {
      "temperature": 0.7,
      "maxOutputTokens": 2048
    }
  }'

基本的なテキスト生成リクエスト

シンプルなテキスト入力から複雑なマルチターンまで、基本パターンを解説します。

シンプルなテキスト生成

response = model.generate_content("Gemini 2.5 Proの特徴を箇条書きで教えて")
print(response.text)

マルチターン会話（Chat History）

chat = model.start_chat(
    history=[
        {"role": "user", "parts": "あなたは何ができますか？"},
        {"role": "model", "parts": "私はテキスト生成、翻訳、コード作成等ができます"},
    ]
)
 
# 続きの会話
response = chat.send_message("APIでの使用方法を詳しく教えて")
print(response.text)

システムプロンプト（System Instruction）の設定

model = genai.GenerativeModel(
    model_name="gemini-2.5-pro-latest",
    system_instruction="""
    あなたはシニアソフトウェアエンジニアです。
    コード例は常にPythonで、エラーハンドリングを含めてください。
    回答は正確かつ簡潔に。
    """
)
 
response = model.generate_content("非同期処理の実装パターンを教えて")

思考モード（Thinking Mode）の活用法

思考モードは、Gemini 2.5 Proの最大の強みです。複雑な推論タスクで精度を大幅に向上させます。

思考モードの有効化

response = model.generate_content(
    "与えられた3つの異なるマーケティング戦略を、5つの評価軸で比較分析してください。戦略A: SNS中心、戦略B: インフルエンサー協力、戦略C: データドリブン広告",
    generation_config={
        "temperature": 1.0,  # 思考モードは temperature=1.0 を推奨
        "thinking": {
            "type": "enabled",
            "budget_tokens": 10000  # 思考に割く最大トークン数
        }
    }
)
 
# 思考プロセスと最終回答の両方を確認
for part in response.parts:
    if part.thought:
        print(f"思考プロセス:\n{part.thought}")
    elif part.text:
        print(f"最終回答:\n{part.text}")

思考トークン予算の最適化

思考トークンを多く割くほど精度が向上しますが、コストと処理時間も増加します。

推奨設定ガイド：

シンプルな質問: budget_tokens=1000〜2000
中程度の推論タスク: budget_tokens=5000〜10000
複雑なビジネス分析: budget_tokens=15000〜20000
数学・論理的証明: budget_tokens=20000〜32000

# 複雑な分析タスクの例
response = model.generate_content(
    "マーケティングROI向上のための統合戦略案を立案してください。データ: 過去12ヶ月のコンバージョン率、顧客獲得コスト、LTV分布",
    generation_config={
        "thinking": {
            "type": "enabled",
            "budget_tokens": 24000  # 複雑タスク向け
        }
    }
)

マルチモーダル入力（画像・動画・音声）

Gemini 2.5 Proはマルチモーダル対応です。複数のメディアタイプを同時処理できます。

画像処理

import base64
from pathlib import Path
 
# ローカルファイルから画像を読み込み
image_path = "example.jpg"
image_data = base64.standard_b64encode(Path(image_path).read_bytes()).decode("utf-8")
 
response = model.generate_content([
    "この画像に何が写っていますか？詳しく説明してください。",
    {
        "mime_type": "image/jpeg",
        "data": image_data,
    }
])
print(response.text)

PDF分析

# PDFファイルのアップロードと分析
pdf_path = "report.pdf"
pdf_data = base64.standard_b64encode(Path(pdf_path).read_bytes()).decode("utf-8")
 
response = model.generate_content([
    "このPDFの要点を500語以内でまとめてください。",
    {
        "mime_type": "application/pdf",
        "data": pdf_data,
    }
])

動画分析

# URLベースの動画入力（推奨）
response = model.generate_content([
    "このYouTube動画のハイライトシーンを3つ特定し、各シーンのタイムスタンプを教えてください。",
    {
        "mime_type": "video/mp4",
        "file_data": {
            "file_uri": "https://example.com/video.mp4"
        }
    }
])

ストリーミングレスポンスの実装

長い出力の場合、ストリーミングで段階的に受け取る方がUX向上です。

Pythonでのストリーミング実装

response = model.generate_content(
    "ソフトウェアアーキテクチャの詳細ガイドを書いてください。",
    stream=True
)
 
for chunk in response:
    if chunk.text:
        print(chunk.text, end="", flush=True)

TypeScriptでのストリーミング実装

const stream = await model.generateContentStream({
  contents: [
    {
      role: "user",
      parts: [{ text: "ソフトウェアアーキテクチャの詳細ガイドを書いてください。" }],
    },
  ],
});
 
for await (const chunk of stream.stream) {
  if (chunk.candidates?.[0]?.content?.parts?.[0]?.text) {
    process.stdout.write(chunk.candidates[0].content.parts[0].text);
  }
}

コスト最適化：トークン数管理とキャッシュ活用

Gemini 2.5 Proはコストがかかるため、最適化が重要です。

Context Caching によるコスト削減

同じプロンプト（例：長いドキュメント）を繰り返し利用する場合、キャッシングで最大90%コスト削減が可能です。

# キャッシュの有効化（v1beta必須）
response = model.generate_content(
    contents=[
        {
            "role": "user",
            "parts": [
                {
                    "text": "以下のドキュメントを分析してください：" + 
                            open("large_doc.txt").read()  # 大きなテキスト
                }
            ]
        },
        {
            "role": "user",
            "parts": [
                {
                    "text": "質問1：主な論点は何ですか？"
                }
            ]
        }
    ],
    generation_config={
        "cache_control": {
            "type": "ephemeral"  # キャッシュ有効期間：5分
        }
    }
)

トークン数の事前確認

# countTokens API を使用してコストを事前予測
from google.generativeai import types
 
prompt = "長いプロンプト例"
token_count = model.count_tokens(prompt)
print(f"入力トークン数: {token_count.input_tokens}")
print(f"予想出力トークン数: {token_count.output_tokens}")
 
# 推定コスト計算
INPUT_COST_PER_1M = 1.25  # USD
OUTPUT_COST_PER_1M = 2.50  # USD
 
input_cost = (token_count.input_tokens / 1_000_000) * INPUT_COST_PER_1M
output_cost = (token_count.output_tokens / 1_000_000) * OUTPUT_COST_PER_1M
total_cost = input_cost + output_cost
 
print(f"推定費用: ${total_cost:.6f}")

エラーハンドリングとレート制限対策

本番環境では、エラーハンドリングは必須です。

基本的なエラーハンドリング

import time
from google.api_core import exceptions
 
def call_with_retry(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except exceptions.ResourceExhausted:
            # レート制限: 指数バックオフで再試行
            wait_time = 2 ** attempt
            print(f"レート制限。{wait_time}秒待機します...")
            time.sleep(wait_time)
        except exceptions.InvalidArgument as e:
            # 不正な入力: リトライ不可
            print(f"入力エラー: {e}")
            raise
        except Exception as e:
            # その他のエラー
            print(f"予期しないエラー: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
            else:
                raise
    
    raise Exception("最大リトライ回数に達しました")
 
# 使用例
result = call_with_retry("質問内容")

429エラー（Too Many Requests）への対応

import asyncio
from datetime import datetime, timedelta
 
class RateLimiter:
    def __init__(self, calls_per_minute=100):
        self.calls_per_minute = calls_per_minute
        self.call_times = []
    
    async def acquire(self):
        now = datetime.now()
        # 1分以上古いコールを削除
        self.call_times = [
            t for t in self.call_times 
            if now - t < timedelta(minutes=1)
        ]
        
        if len(self.call_times) >= self.calls_per_minute:
            # 次のスロットまで待機
            sleep_until = self.call_times[0] + timedelta(minutes=1)
            wait_seconds = (sleep_until - now).total_seconds()
            if wait_seconds > 0:
                print(f"レート制限: {wait_seconds:.1f}秒待機")
                await asyncio.sleep(wait_seconds)
        
        self.call_times.append(datetime.now())
 
# 使用例
limiter = RateLimiter(calls_per_minute=60)
 
async def safe_api_call(prompt):
    await limiter.acquire()
    response = model.generate_content(prompt)
    return response.text

実践プロジェクト：チャットボット構築サンプル

ここまでの知識を活用した、フル機能のチャットボット例を紹介します。

要件

ユーザーとの複数ターン会話をサポート
思考モードで複雑な質問に対応
エラーハンドリングとレート制限対応
API呼び出しコストをログ出力

実装例（Python）

import json
import logging
from datetime import datetime
from typing import Optional
 
# ロギング設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
class Gemini25ProChatBot:
    def __init__(self, api_key: str):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel(
            model_name="gemini-2.5-pro-latest",
            system_instruction="""
            あなたはエキスパートアシスタントです。
            ユーザーの質問に正確かつ実践的に答えてください。
            不確実な情報は「わかりません」と正直に答えてください。
            """
        )
        self.chat_history = []
        self.total_input_tokens = 0
        self.total_output_tokens = 0
    
    def add_to_history(self, role: str, text: str):
        """チャット履歴に追加"""
        self.chat_history.append({"role": role, "parts": [{"text": text}]})
    
    def estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
        """API呼び出しコストを計算"""
        INPUT_COST = 1.25 / 1_000_000
        OUTPUT_COST = 2.50 / 1_000_000
        return (input_tokens * INPUT_COST) + (output_tokens * OUTPUT_COST)
    
    def send_message(self, user_input: str, use_thinking: bool = False) -> Optional[str]:
        """メッセージ送信"""
        try:
            self.add_to_history("user", user_input)
            
            config = {
                "temperature": 0.7,
                "max_output_tokens": 2048,
            }
            
            if use_thinking:
                config["thinking"] = {
                    "type": "enabled",
                    "budget_tokens": 8000
                }
            
            response = self.model.generate_content(
                contents=self.chat_history,
                generation_config=config
            )
            
            assistant_response = response.text
            self.add_to_history("model", assistant_response)
            
            # コスト計算とログ
            token_info = self.model.count_tokens(self.chat_history)
            estimated_cost = self.estimate_cost(
                token_info.input_tokens,
                token_info.output_tokens
            )
            
            logger.info(
                f"トークン: 入力={token_info.input_tokens}, "
                f"出力={token_info.output_tokens}, "
                f"推定費用=${estimated_cost:.6f}"
            )
            
            self.total_input_tokens += token_info.input_tokens
            self.total_output_tokens += token_info.output_tokens
            
            return assistant_response
            
        except Exception as e:
            logger.error(f"エラー: {e}")
            return None
    
    def get_session_summary(self) -> dict:
        """セッション統計を取得"""
        total_cost = self.estimate_cost(
            self.total_input_tokens,
            self.total_output_tokens
        )
        return {
            "total_input_tokens": self.total_input_tokens,
            "total_output_tokens": self.total_output_tokens,
            "estimated_total_cost": total_cost,
            "conversation_turns": len(self.chat_history) // 2
        }
 
# 使用例
if __name__ == "__main__":
    bot = Gemini25ProChatBot(api_key="YOUR_GEMINI_API_KEY")
    
    # 会話開始
    response1 = bot.send_message("マーケティング戦略について教えてください")
    print(f"Bot: {response1}\n")
    
    response2 = bot.send_message(
        "より詳細な実装ステップを教えてください。複雑に考えてください。",
        use_thinking=True
    )
    print(f"Bot: {response2}\n")
    
    # セッション統計を表示
    summary = bot.get_session_summary()
    print(json.dumps(summary, indent=2, ensure_ascii=False))