Gemini 2.5 Pro Latest API 完全：開発者向け高度活用テクニック集

pip install google-generativeai

import google.generativeai as genai
 
# APIキーを設定
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
# モデルを指定
model = genai.GenerativeModel("gemini-2.5-pro-latest")
 
# テキスト生成
response = model.generate_content(
    "Python で効率的なクイックソートを実装する方法を説明してください。"
)
 
print(response.text)

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-latest")
 
# チャット履歴を管理するオブジェクトを作成
chat = model.start_chat(history=[])
 
# 1 番目のメッセージ
response1 = chat.send_message("Java のメモリ管理について説明してください。")
print("Assistant:", response1.text)
 
# 2 番目のメッセージ（前の会話を参照）
response2 = chat.send_message("ガベージコレクションの仕組みを詳しく教えてください。")
print("Assistant:", response2.text)
 
# 3 番目のメッセージ
response3 = chat.send_message("G1GC と ZGC の違いは？")
print("Assistant:", response3.text)

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-latest")
 
# stream=True オプションでストリーミング有効化
response = model.generate_content(
    "大規模言語モデルの学習プロセスを段階的に説明してください。",
    stream=True
)
 
# トークンごとに結果を受け取る
for chunk in response:
    if chunk.text:
        print(chunk.text, end="", flush=True)
 
print()  # 改行

from flask import Flask, request, jsonify, stream_with_context, Response
import google.generativeai as genai
 
app = Flask(__name__)
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-latest")
 
@app.route("/stream", methods=["POST"])
def stream_response():
    user_input = request.json.get("message", "")
    
    def generate():
        response = model.generate_content(user_input, stream=True)
        for chunk in response:
            if chunk.text:
                yield f"data: {chunk.text}\n\n"
    
    return Response(
        stream_with_context(generate()),
        mimetype="text/event-stream"
    )
 
if __name__ == "__main__":
    app.run(debug=True)

const eventSource = new EventSource("/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "質問内容" })
});
 
eventSource.addEventListener("message", (event) => {
  console.log(event.data);
  // DOMに追加するなど、リアルタイム表示処理
});

import google.generativeai as genai
import json
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel(
    "gemini-2.5-pro-latest",
    tools=[
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "指定した都市の天気を取得します",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "都市名（例：Tokyo, New York）"
                        },
                        "units": {
                            "type": "string",
                            "enum": ["celsius", "fahrenheit"],
                            "description": "温度単位"
                        }
                    },
                    "required": ["city"]
                }
            }
        }
    ]
)
 
# ユーザーのリクエスト
user_message = "東京の天気を教えてください。"
 
# API 呼び出し
response = model.generate_content(user_message)
 
# Function Call を検出して実行
if response.candidates and response.candidates[0].content.parts:
    for part in response.candidates[0].content.parts:
        if hasattr(part, "function_call"):
            func_call = part.function_call
            print(f"関数: {func_call.name}")
            print(f"引数: {func_call.args}")
            
            # 実装例：ここで実際の天気APIを呼び出す
            if func_call.name == "get_weather":
                city = func_call.args.get("city")
                # weather_api_call(city) のような実装

# ユーザー: 「東京の天気と気温、それに明日のニューヨークの予報も」
# → get_weather を複数回呼び出す必要がある
 
response = model.generate_content(
    "東京と明日のニューヨークの天気を比較してください。"
)
 
function_calls = []
for part in response.candidates[0].content.parts:
    if hasattr(part, "function_call"):
        function_calls.append(part.function_call)
 
# 全ての Function Call を実行
results = []
for func_call in function_calls:
    if func_call.name == "get_weather":
        city = func_call.args.get("city")
        result = get_weather_from_api(city)  # 実装例
        results.append(result)
 
# 結果を AI に返す
chat = model.start_chat()
chat.send_message(user_message)
chat.send_message({
    "role": "user",
    "parts": [{"function_response": {"name": "get_weather", "response": r}} for r in results]
})

import google.generativeai as genai
from PIL import Image
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-latest")
 
# ローカル画像ファイルの読み込み
image = Image.open("screenshot.png")
 
# テキストと画像を合わせて送信
response = model.generate_content([
    "このスクリーンショットに写っているエラーメッセージを読み取り、対処方法を教えてください。",
    image
])
 
print(response.text)

from PIL import Image
import requests
from io import BytesIO
 
# URL から画像をダウンロード
image_url = "https://example.com/image.jpg"
response = requests.get(image_url)
image = Image.open(BytesIO(response.content))
 
# Gemini に送信
model_response = model.generate_content([
    "この画像について分析してください。",
    image
])

# PDFファイルをアップロード
pdf_file = genai.upload_file(path="document.pdf")
 
response = model.generate_content([
    "このPDFドキュメントから、主要なキーポイントを3つ抽出してください。",
    pdf_file
])
 
print(response.text)

# 動画ファイルをアップロード（MP4, WebM など）
video_file = genai.upload_file(path="demo_video.mp4")
 
response = model.generate_content([
    "この動画の内容を要約し、重要なシーンを時間順に列挙してください。",
    video_file
])
 
print(response.text)

genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
model = genai.GenerativeModel(
    "gemini-2.5-pro-latest",
    system_instruction="""
    あなたは顧客サポートエージェントです。以下のルールを守ってください：
    1. 常に日本語で回答する
    2. 顧客に敬意を持って接する
    3. 自社製品の問題と判断された場合は、自動的にサポートチームにエスカレートするよう指示する
    4. 3回の回答で問題が解決しない場合は、チケットを自動発行することを提案する
    5. 個人情報は絶対に保存しない
    """
)
 
response = model.generate_content("製品 X が起動しません。")
print(response.text)

from google.generativeai.types import HarmCategory, HarmBlockThreshold
 
model = genai.GenerativeModel(
    "gemini-2.5-pro-latest",
    safety_settings=[
        {
            "category": HarmCategory.HARM_CATEGORY_HARASSMENT,
            "threshold": HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
        },
        {
            "category": HarmCategory.HARM_CATEGORY_HATE_SPEECH,
            "threshold": HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
        },
    ]
)

# 実際の API 呼び出し前に、トークン数を確認
input_text = "このテキストのトークン数は？" * 100
 
token_count = model.count_tokens(input_text)
print(f"入力トークン数: {token_count.total_tokens}")
 
# 費用計算（概算）
cost_per_mtok = 1.25  # Pro モデルの入力コスト（ドル/百万トークン）
estimated_cost = (token_count.total_tokens / 1_000_000) * cost_per_mtok
print(f"推定コスト: ${estimated_cost:.4f}")

import anthropic
 
# キャッシングするコンテンツを準備
large_document = open("long_document.txt").read()
 
response = model.generate_content(
    [
        # キャッシュするコンテンツにはメタデータを含める
        {
            "type": "text",
            "text": large_document,
            "cache_control": {"type": "ephemeral"}
        },
        {
            "type": "text",
            "text": "このドキュメントから、セクション3を要約してください。"
        }
    ]
)

# 複数のリクエストをバッチで送信
batch_requests = [
    {"role": "user", "content": "Python の async/await の説明"},
    {"role": "user", "content": "Go の並行処理の説明"},
    {"role": "user", "content": "Rust のオーナーシップの説明"},
]
 
# バッチ API を使う（詳細は公式ドキュメント参照）
# batch_response = genai.batch_generate_content(batch_requests, model="gemini-2.5-pro-latest")

import google.generativeai as genai
import time
from google.api_core import retry
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-latest")
 
# リトライロジック付きの実装
@retry.Retry(deadline=300, initial=1, multiplier=2, maximum=60)
def call_gemini_with_retry(prompt):
    try:
        response = model.generate_content(prompt)
        return response.text
    except Exception as e:
        print(f"エラー: {e}")
        raise
 
try:
    result = call_gemini_with_retry("質問内容")
    print(result)
except Exception as e:
    print(f"最終的に失敗しました: {e}")

import time
import random
 
def call_with_exponential_backoff(prompt, max_retries=5):
    base_wait = 1  # 初期待機時間（秒）
    
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            # 指数バックオフ + ジッタ
            wait_time = base_wait * (2 ** attempt) + random.uniform(0, 1)
            print(f"レート制限に達しました。{wait_time:.1f}秒待機します...")
            time.sleep(wait_time)
    
    return None
 
# 使用例
result = call_with_exponential_backoff("複雑な質問")

import socket
 
# タイムアウトを明示的に設定
socket.setdefaulttimeout(30)  # 30秒
 
try:
    response = model.generate_content(
        "質問",
        timeout=30  # リクエストごとのタイムアウト
    )
except socket.timeout:
    print("リクエストがタイムアウトしました")

import time
import random
from google import genai
 
client = genai.Client()
 
def call_with_retry(prompt, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = client.models.generate_content(
                model="gemini-2.5-pro",
                contents=prompt,
            )
            return response
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                # 指数バックオフ + ジッター
                base_wait = 2 ** attempt
                jitter = random.uniform(0, 0.5 * base_wait)
                wait_time = base_wait + jitter
                time.sleep(wait_time)
                continue
            raise

from google import genai
from google.genai import types
 
client = genai.Client()
 
# 1. キャッシュを作成
cache = client.caches.create(
    model="gemini-2.5-pro",
    config=types.CreateCachedContentConfig(
        contents=[
            types.Content(
                role="user",
                parts=[types.Part(text=LARGE_SYSTEM_PROMPT)]
            )
        ],
        system_instruction="You are a helpful assistant.",
        ttl="3600s",
    ),
)
 
# 2. キャッシュを使ってリクエスト
response = client.models.generate_content(
    model="gemini-2.5-pro",
    contents="ユーザーの質問",
    config=types.GenerateContentConfig(
        cached_content=cache.name,
    ),
)

# batch_requests.jsonl
# {"key": "req1", "request": {"contents": [...]}}
# {"key": "req2", "request": {"contents": [...]}}
 
batch_job = client.batches.create(
    model="gemini-2.5-pro",
    src="batch_requests.jsonl",
)
 
# 完了を待つ
while batch_job.state != "SUCCEEDED":
    time.sleep(60)
    batch_job = client.batches.get(name=batch_job.name)
 
# 結果を取得
results = client.batches.list_output(name=batch_job.name)