Python で Gemini を扱う実装ガイド — テキスト生成・Embedding・Vision・モデル切替

🎯 はじめに：4 種の典型ユースケースを Python テンプレで揃える

本記事はシリーズ 「Gemini AI プラットフォーム活用」の第 4 回 です。前回までで 課金構造・モデル選定・Batch API を扱いました。今回は Python から Gemini を呼び出す具体的なコードを、実務で頻出する 4 ユースケースに沿ってテンプレ化します。

そのうえで、ストリーミング・構造化出力（JSON）・エラーハンドリングといったプロダクション化に必須な要素も組み込みます。

📦 1. インストールと認証

SDK のインストール

Gemini API には新旧 2 系統の Python SDK があります。

# 新 SDK（推奨。Vertex AI 移行も同じ SDK で済む）
pip install google-genai

# 旧 SDK（既存プロジェクト互換用）
pip install google-generativeai

⚠️ 新規プロジェクトは必ず新 SDK の google-genai を選んでください。Part 5 の Vertex AI 移行は新 SDK 前提で、設定 1 行（vertexai=True）で同じコードを Vertex AI に切り替えられます。

APIキーの設定

# Linux / macOS
export GOOGLE_API_KEY="your-api-key-here"

# Windows PowerShell
$env:GOOGLE_API_KEY = "your-api-key-here"

# .env で管理する場合（python-dotenv）
echo 'GOOGLE_API_KEY=your-api-key-here' >> .env

クライアント初期化テンプレート

# src/llm/gemini_client.py
import os
from google import genai

# 環境変数 GOOGLE_API_KEY を自動で読む
client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])

💡 業務コードでは API キーを直接渡すのではなく、Secret Manager / Parameter Store から読み込む層を 1 段挟むのが定石。os.environ[...] でも最低限の OK。

📝 2. テキスト生成（チャット・応答）

一発呼び出し（ノンストリーミング）

from google import genai

client = genai.Client()

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="プロダクトマネージャー向けに『ユビキタス言語』を 3 行で説明して。",
)

print(response.text)
print(f"使用トークン: 入力={response.usage_metadata.prompt_token_count}, "
      f"出力={response.usage_metadata.candidates_token_count}")

ストリーミング応答（チャット UI 向け）

stream = client.models.generate_content_stream(
    model="gemini-2.5-flash",
    contents="DDD のレポジトリパターンを 200 字で説明して。",
)

for chunk in stream:
    print(chunk.text, end="", flush=True)

システム指示と複数ターン

from google.genai import types

response = client.models.generate_content(
    model="gemini-2.5-flash",
    config=types.GenerateContentConfig(
        system_instruction="あなたは BtoB SaaS のシニアアーキテクトです。回答は箇条書き・日本語。",
        temperature=0.3,
    ),
    contents=[
        {"role": "user", "parts": [{"text": "マルチテナントの認可設計でハマりがちな点は？"}]},
        {"role": "model", "parts": [{"text": "主に以下の 3 点です..."}]},
        {"role": "user", "parts": [{"text": "その中で最も多い事故例を教えて。"}]},
    ],
)
print(response.text)

構造化出力（JSON モード）

LLM 出力を確実にパース可能な JSON で受け取る運用は、業務コードでは必須に近いものです。

from pydantic import BaseModel
from google.genai import types

class ProductTag(BaseModel):
    category: str
    tags: list[str]
    confidence: float

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="商品: 防水機能付きスマートウォッチ、心拍数計、GPS 内蔵",
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=ProductTag,
    ),
)

parsed: ProductTag = response.parsed
print(parsed.category, parsed.tags, parsed.confidence)

response_schema に Pydantic モデルを渡すと、Gemini 側が スキーマ準拠の JSON を必ず返してくれるようになり、json.loads の失敗率が激減します。

🧮 3. テキスト埋め込み（RAG 用）

gemini-embedding-001（推奨）

from google import genai

client = genai.Client()

# 単発
result = client.models.embed_content(
    model="gemini-embedding-001",
    contents="埋め込みたいテキスト",
)
embedding = result.embeddings[0].values
print(f"次元数: {len(embedding)}")  # 3072

次元削減（768 次元で運用するケース）

from google.genai import types

result = client.models.embed_content(
    model="gemini-embedding-001",
    contents="埋め込みたいテキスト",
    config=types.EmbedContentConfig(
        output_dimensionality=768,  # 768 / 1536 / 3072
    ),
)

💡 768 次元ならベクトル DB 容量が 3072 次元の 1/4、検索速度も上がります。Part 2 で扱ったとおり、RAG 用途なら 768 次元から始めるのが定石。

⚠ 保存先 DB のインデックス制約: pgvector の HNSW / IVFFlat は 2000 次元上限で 3072 次元はインデックス不可。halfvec(3072) または 768 / 1536 へのダウンサンプリングが必要です。詳細は Part 2: Embedding モデルの選び方 を参照。

旧 SDK での実装（互換性）

既存プロジェクトが google-generativeai を使っている場合：

import google.generativeai as genai
import os

genai.configure(api_key=os.environ["GOOGLE_API_KEY"])

result = genai.embed_content(
    model="models/text-embedding-004",
    content="埋め込みたいテキスト",
    task_type="retrieval_document",  # または "retrieval_query"
)
embedding = result["embedding"]  # 768 次元

⚠️ text-embedding-004 は 2025 年 8 月に廃止予定。新規は必ず gemini-embedding-001 を使ってください。

🖼 4. 画像理解（Vision）

PIL Image を直接渡す（推奨）

from google import genai
from PIL import Image

client = genai.Client()
image = Image.open("path/to/product.jpg")

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=[
        "この画像に写っている商品の特徴を箇条書きで 5 点挙げてください。",
        image,
    ],
)
print(response.text)

Base64 で送信（Web からアップロード等）

from google import genai
from google.genai import types
import base64

client = genai.Client()

with open("path/to/product.jpg", "rb") as f:
    image_data = base64.b64encode(f.read()).decode("utf-8")

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=[
        types.Part.from_bytes(
            data=base64.b64decode(image_data),
            mime_type="image/jpeg",
        ),
        "この画像を分析してください。",
    ],
)
print(response.text)

File API で大きなファイルを扱う（20MB 超）

PDF・動画・音声で 20MB を超える場合は File API でアップロードしてから参照します。

from google import genai
import time

client = genai.Client()

# 1. アップロード
uploaded = client.files.upload(file="path/to/long_video.mp4")

# 2. PROCESSING → ACTIVE を待つ
while uploaded.state.name == "PROCESSING":
    time.sleep(2)
    uploaded = client.files.get(name=uploaded.name)

# 3. 参照して呼び出し
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=[uploaded, "この動画の主要シーンを時系列で要約してください。"],
)
print(response.text)

構造化出力 × Vision の組み合わせ

from pydantic import BaseModel
from google.genai import types
from PIL import Image

class ColorAnalysis(BaseModel):
    primary_color: str
    secondary_color: str | None
    hex_code: str

image = Image.open("product.jpg")

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=[image, "この商品のメインカラーを分析してください。"],
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=ColorAnalysis,
    ),
)

result: ColorAnalysis = response.parsed
print(result.primary_color, result.hex_code)

CTS-EC 共通商品マスタ のカラー判定もこの形のコードで動いています。

🔀 5. 用途別モデルの切り替え

MODELS = {
    "fastest_cheapest": "gemini-2.5-flash-lite",   # 単純分類・抽出
    "default":           "gemini-2.5-flash",        # 標準ワークホース
    "high_quality":      "gemini-2.5-pro",          # 複雑推論
    "best_in_class":     "gemini-3-pro-preview",    # 最高性能（高単価）
    "embedding":         "gemini-embedding-001",    # ベクトル化
}

def generate(prompt: str, mode: str = "default") -> str:
    response = client.models.generate_content(
        model=MODELS[mode],
        contents=prompt,
    )
    return response.text

# 使い分け
short_summary = generate("...", mode="fastest_cheapest")
deep_analysis = generate("...", mode="high_quality")

💡 モデル名をマジックストリングで散らさないのはすぐに効いてくる工夫。新モデルが出た時の置換が dict 1 箇所で済みます。

🛡 6. プロダクション化の頻出パターン

リトライ（指数バックオフ）

import time
from google.genai import errors

def generate_with_retry(prompt: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            response = client.models.generate_content(
                model="gemini-2.5-flash",
                contents=prompt,
            )
            return response.text
        except errors.APIError as e:
            if e.code in (429, 503) and attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"リトライ {attempt + 1}/{max_retries}（{wait}秒待機）")
                time.sleep(wait)
                continue
            raise

レート制限（429）と一時障害（503）はリトライで解決するケースが多いです。tenacity などのライブラリを使うとさらに簡潔に書けます。

タイムアウトと安全フィルタの確認

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=prompt,
    config=types.GenerateContentConfig(
        max_output_tokens=1000,
    ),
)

# 安全フィルタでブロックされた場合の確認
if not response.candidates:
    print(f"ブロック: {response.prompt_feedback.block_reason}")
elif response.candidates[0].finish_reason.name == "SAFETY":
    print(f"安全フィルタ: {response.candidates[0].safety_ratings}")
else:
    print(response.text)

コスト計測ヘルパ

def estimate_cost(usage, input_price: float, output_price: float) -> float:
    """USD 換算でコストを概算"""
    in_cost  = usage.prompt_token_count     / 1_000_000 * input_price
    out_cost = usage.candidates_token_count / 1_000_000 * output_price
    return in_cost + out_cost

# Gemini 2.5 Flash のコスト
cost = estimate_cost(
    response.usage_metadata,
    input_price=0.15,
    output_price=0.60,
)
print(f"このリクエスト: ${cost:.6f}")

✅ 第 4 回まとめ

• 新規プロジェクトは google-genai（新 SDK）。Vertex AI 移行も同じ SDK で済む
• 認証は GOOGLE_API_KEY 環境変数。Secret Manager 経由が望ましい
• 構造化出力（JSON モード + Pydantic） は業務コードでは必須に近い装備
• 埋め込みは gemini-embedding-001 の 768 次元から始めるのが定石
• 画像は PIL Image を contents に直接渡すのが最短。20MB 超は File API
• モデル名は dict で集中管理して切り替えコストを下げる
• リトライ・安全フィルタ判定・コスト計測を最初から組み込んでおくと運用が楽

次回 Vertex AI 移行ガイド では、ここまで書いてきた AI Studio ベースのコードを Vertex AI へ 1 行で切り替える方法と、エンタープライズ運用に必要な IAM・VPC-SC・Provisioned Throughput を扱います。

📚 シリーズ記事

🔗 関連リソース

• google-genai 公式 SDK ドキュメント
• Gemini API クイックスタート（公式）