◈ API / SDK/2026-06-22上級

Gemini APIで商品画像を構造化分析する — 数千枚を回して固めた本番パイプライン

商品画像から自動でタグ・説明文・カテゴリを生成するツールを、単発の試作から数千枚を安定処理する本番パイプラインへ。構造化出力・再開可能なバッチ・実測コスト・モデルルーティングまで、個人開発の運用で固めた知見をまとめます。

Gemini API²⁴⁵ Vision⁴ structured-output²⁶ python¹⁰⁵batchコスト最適化²⁵ 個人開発¹²¹

✦ プレミアム記事

個人開発でアプリやストア素材を扱っていると、画像のメタデータ整備という地味な作業がいつのまにか時間を食っていることに気づきます。私自身、数百枚単位で増えていく素材に対して、タグ付けと説明文づくりを手で回していた時期がありました。一枚あたり数十秒でも、まとまれば半日が消えます。

そこで Gemini API のマルチモーダル分析に切り替えたのですが、最初に書いた「一枚送って結果を受け取る」だけのスクリプトは、試作としては動いても、数千枚を一気に流すと脆さが一気に表に出ました。途中で一度落ちると最初からやり直し、コストの見積もりが甘く、たまに返ってくるカテゴリが微妙に揺れる。Dolice Labs の運用で何度か作り直すうちに、本番で回り続ける形に落ち着いてきました。

この記事は、その「単発の試作から、落ちても続く本番パイプラインへ」の差分を中心にまとめます。基本のコードから始めて、実測したコスト、再開可能なバッチ、そして公式ドキュメントには書かれていない運用上の勘どころまで順に置いていきます。

前提知識と環境準備

必要なもの

Python 3.10以上
Google AI Studio のAPIキー（Google AI Studioから取得可能）
google-genai パッケージ

環境セットアップ

# 仮想環境の作成と有効化
python -m venv gemini-image-env
source gemini-image-env/bin/activate  # Windows: gemini-image-env\Scripts\activate
 
# パッケージのインストール
pip install google-genai Pillow

APIキーは環境変数に設定しておきます。

export GEMINI_API_KEY="your-api-key-here"

Gemini APIの基本的なセットアップがまだの方は、Gemini APIクイックスタートガイドを先にご覧ください。

基本的な画像分析 — 1枚の商品画像から情報を抽出する

まずは最もシンプルなケースとして、1枚の商品画像をGemini APIに送信し、自然言語で商品情報を取得してみましょう。

# basic_image_analysis.py
import os
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def analyze_product_image(image_path: str) -> str:
    """商品画像を分析し、説明文を生成する"""
    # 画像ファイルを読み込む
    with open(image_path, "rb") as f:
        image_data = f.read()
 
    response = client.models.generate_content(
        model="gemini-3.1-pro",
        contents=[
            types.Content(
                role="user",
                parts=[
                    types.Part.from_bytes(data=image_data, mime_type="image/jpeg"),
                    types.Part.from_text(
                        "この商品画像を分析してください。"
                        "商品名、カテゴリ、色、素材、特徴を日本語で詳しく説明してください。"
                    ),
                ],
            )
        ],
    )
    return response.text
 
# 実行例
result = analyze_product_image("product_sample.jpg")
print(result)
 
# 期待する出力例:
# この商品は白いコットン素材のTシャツです。
# クルーネックのシンプルなデザインで、胸元に小さなブランドロゴが
# 刺繍されています。素材は柔らかい綿100%で、カジュアルな
# 日常着として適しています。

この方法でも十分な情報が得られますが、出力が自由形式テキストのため、プログラムで処理するには不便です。次のステップで構造化出力を導入します。

✦

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

この記事の続きを読む

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

この記事で得られること

✦数千枚を回して実測したコスト・レイテンシ・精度の内訳と、Flash と Pro を使い分けるルーティングの損益分岐

✦途中で落ちても続きから再開できるチェックポイント方式のバッチ実装（コピーして動くコード付き）

✦enum ドリフト・スキーマ厳格化・アップロード前の縮小など、公式ドキュメントに載っていない運用上の落とし穴と対処

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

✦

この記事を購入する

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

または

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

構造化出力でJSON形式のデータを取得する

Gemini APIの構造化出力機能を使えば、レスポンスを指定したJSONスキーマに沿った形式で強制的に返させることができます。これにより、取得したデータをそのままデータベースに保存したり、CSVに変換したりすることが簡単になります。

# structured_image_analysis.py
import os
import json
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# 商品分析の出力スキーマを定義
product_schema = types.Schema(
    type="object",
    properties={
        "product_name": types.Schema(
            type="string",
            description="商品名（日本語）",
        ),
        "category": types.Schema(
            type="string",
            enum=["衣類", "靴", "バッグ", "アクセサリー", "家電", "食品", "その他"],
            description="商品カテゴリ",
        ),
        "tags": types.Schema(
            type="array",
            items=types.Schema(type="string"),
            description="商品の特徴を表すタグ（5〜10個）",
        ),
        "color": types.Schema(
            type="string",
            description="メインの色",
        ),
        "material": types.Schema(
            type="string",
            description="素材（推定）",
        ),
        "description_short": types.Schema(
            type="string",
            description="商品の簡潔な説明（50文字以内）",
        ),
        "description_long": types.Schema(
            type="string",
            description="商品の詳細な説明（200文字程度）",
        ),
        "target_audience": types.Schema(
            type="string",
            description="ターゲット層",
        ),
        "estimated_season": types.Schema(
            type="string",
            enum=["春", "夏", "秋", "冬", "オールシーズン"],
            description="おすすめの季節",
        ),
    },
    required=[
        "product_name", "category", "tags", "color",
        "description_short", "description_long",
    ],
)
 
def analyze_product_structured(image_path: str) -> dict:
    """商品画像を構造化データとして分析する"""
    with open(image_path, "rb") as f:
        image_data = f.read()
 
    response = client.models.generate_content(
        model="gemini-3.1-pro",
        contents=[
            types.Content(
                role="user",
                parts=[
                    types.Part.from_bytes(data=image_data, mime_type="image/jpeg"),
                    types.Part.from_text(
                        "この商品画像を分析し、ECサイト用の商品情報を生成してください。"
                    ),
                ],
            )
        ],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=product_schema,
        ),
    )
    return json.loads(response.text)
 
# 実行例
product_data = analyze_product_structured("product_sample.jpg")
print(json.dumps(product_data, ensure_ascii=False, indent=2))
 
# 期待する出力例:
# {
#   "product_name": "オーガニックコットンTシャツ",
#   "category": "衣類",
#   "tags": ["Tシャツ", "コットン", "ホワイト", "カジュアル", "無地", "ユニセックス"],
#   "color": "ホワイト",
#   "material": "コットン",
#   "description_short": "柔らかなオーガニックコットンのシンプルTシャツ",
#   "description_long": "上質なオーガニックコットン100%を使用した...",
#   "target_audience": "20〜40代の男女",
#   "estimated_season": "オールシーズン"
# }

構造化出力を日英ペアで安定させる工夫については、構造化出力で用語のゆれを抑える実装メモもあわせてご覧ください。

複数画像のバッチ処理に対応する

実際の運用では、大量の商品画像をまとめて処理する必要があります。以下のコードは、指定ディレクトリ内の全画像をバッチ処理し、結果をCSVファイルに出力するスクリプトです。

# batch_product_analysis.py
import os
import csv
import json
import time
from pathlib import Path
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def batch_analyze_products(
    image_dir: str,
    output_csv: str,
    max_concurrent: int = 5,
    delay_sec: float = 1.0,
) -> list[dict]:
    """ディレクトリ内の全商品画像をバッチ分析する"""
    image_dir = Path(image_dir)
    supported_extensions = {".jpg", ".jpeg", ".png", ".webp"}
    image_files = sorted(
        f for f in image_dir.iterdir()
        if f.suffix.lower() in supported_extensions
    )
 
    print(f"処理対象: {len(image_files)} 件の画像")
    results = []
    errors = []
 
    for i, image_file in enumerate(image_files):
        print(f"[{i + 1}/{len(image_files)}] {image_file.name} を分析中...")
 
        try:
            product_data = analyze_product_structured(str(image_file))
            product_data["file_name"] = image_file.name
            results.append(product_data)
        except Exception as e:
            print(f"  エラー: {e}")
            errors.append({"file_name": image_file.name, "error": str(e)})
 
        # レートリミット回避のため待機
        if (i + 1) % max_concurrent == 0:
            time.sleep(delay_sec)
 
    # CSV出力
    if results:
        fieldnames = [
            "file_name", "product_name", "category", "tags",
            "color", "material", "description_short",
            "description_long", "target_audience", "estimated_season",
        ]
        with open(output_csv, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for row in results:
                row["tags"] = ", ".join(row.get("tags", []))
                writer.writerow({k: row.get(k, "") for k in fieldnames})
 
        print(f"\n完了: {len(results)} 件成功, {len(errors)} 件エラー")
        print(f"CSVファイル: {output_csv}")
 
    return results
 
# 実行例
if __name__ == "__main__":
    results = batch_analyze_products(
        image_dir="./product_images",
        output_csv="./product_analysis_results.csv",
    )

このスクリプトは小さなディレクトリなら問題なく動きます。ただ、数千枚を一度に流すと、ネットワークの瞬断やレートリミットで途中停止したときに、それまでの結果がメモリ上にしか無いため全部やり直しになります。後半でこの弱点を埋める再開可能な実装に差し替えます。

エラーハンドリングとリトライ機能

本番環境で安定して動作させるためには、適切なエラーハンドリングとリトライ機能が欠かせません。APIのレートリミットや一時的なエラーに対応するためのユーティリティ関数を実装しましょう。

# retry_utils.py
import time
import functools
from google.api_core import exceptions as google_exceptions
 
def retry_on_api_error(max_retries: int = 3, base_delay: float = 2.0):
    """API エラー時にエクスポネンシャルバックオフでリトライするデコレータ"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except google_exceptions.ResourceExhausted:
                    # レートリミット: 長めに待機
                    wait = base_delay * (2 ** attempt) * 2
                    print(f"  レートリミット。{wait:.1f}秒待機（{attempt + 1}/{max_retries}）")
                    time.sleep(wait)
                except google_exceptions.ServiceUnavailable:
                    # サーバーエラー: 短めに待機
                    wait = base_delay * (2 ** attempt)
                    print(f"  サーバーエラー。{wait:.1f}秒待機（{attempt + 1}/{max_retries}）")
                    time.sleep(wait)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    wait = base_delay * (2 ** attempt)
                    print(f"  予期せぬエラー: {e}。{wait:.1f}秒待機")
                    time.sleep(wait)
            raise RuntimeError(f"{max_retries}回のリトライ後も失敗しました")
        return wrapper
    return decorator
 
# 使用例: デコレータを関数に適用
@retry_on_api_error(max_retries=3, base_delay=2.0)
def analyze_with_retry(image_path: str) -> dict:
    return analyze_product_structured(image_path)

数千枚を回して見えた実測コストとスループット

試作の段階では「だいたい安い」で済ませてしまいがちですが、本番に載せる前に一度きちんと測っておくと、後の判断がぶれません。私が手元のカタログ画像を対象に計測したときの、おおよその内訳を共有します。あくまで条件依存の参考値ですが、桁感をつかむには十分です。

計測条件は、長辺1024pxへ縮小したJPEG・構造化出力スキーマ固定・1枚あたり1リクエストです。

項目	gemini-3.5-flash	gemini-3.1-pro
1枚あたりの体感レイテンシ	約1.2〜2.0秒	約2.5〜4.0秒
1,000枚あたりの概算コスト	数十円台	その3〜5倍程度
カテゴリ一致率（手動正解との比較）	約88%	約94%
説明文の手直し率	約2割	約1割弱
スキーマ違反による再試行率	1%未満	1%未満

ここから読み取れる実務的な結論は、全部を上位モデルで回す必要はないということです。明快な商品（単色のTシャツやマグカップなどありふれたSKU）はFlashで十分な品質が出ます。判断が割れるのは、複数の要素が写り込んだ写真や、素材の見分けが難しい商品です。

そこで私は、まずFlashで一次分析し、信頼度が低いものだけPro で再分析する二段構えにしました。全体のコストはおおむね半分以下に下がり、精度はほぼPro 単体に近づきます。閾値の作り方は次節の運用知見で触れます。

なお gemini-3.5-flash は一般提供（GA）となり、速度とコストのバランスからこうした一次パスの主力に据えやすくなりました。モデルルーティングとキャッシュでコストを削る考え方は、キャッシュとモデルルーティングでコストを削った記録にも整理してあります。

公式ドキュメントには書かれていない運用知見

ここからが、試作と本番を分ける細部です。どれも、数千枚を流すうちに実際につまずいて学んだことです。

1. enum ドリフトを正規化で吸収する

スキーマで enum を指定しても、モデルは表記が近い別の値（「バッグ」と「バック」、英語と日本語の混在など）をまれに返します。スキーマ違反として弾くより、受け取り側で正規化したほうが歩留まりが上がります。

# normalize.py
CATEGORY_CANON = {
    "衣類": "衣類", "服": "衣類", "アパレル": "衣類",
    "バッグ": "バッグ", "バック": "バッグ", "鞄": "バッグ",
    "靴": "靴", "シューズ": "靴",
    "アクセサリー": "アクセサリー", "アクセ": "アクセサリー",
}
 
def canon_category(value: str, fallback: str = "その他") -> str:
    """モデルが返したカテゴリ表記を正規の値へ寄せる"""
    v = (value or "").strip()
    return CATEGORY_CANON.get(v, fallback)

2. アップロード前に縮小してコストとレイテンシを同時に下げる

商品写真は数MBある場合も多く、原寸のまま送ると入力コストもアップロード時間も無駄に膨らみます。分類や説明文の品質は、長辺1024px程度に縮めてもほとんど落ちません。前処理として一段挟むだけで、体感速度がはっきり変わります。

# downscale.py
import io
from PIL import Image
 
def downscale_jpeg(path: str, max_side: int = 1024, quality: int = 85) -> bytes:
    """長辺を max_side に収め、JPEG バイト列として返す"""
    img = Image.open(path).convert("RGB")
    w, h = img.size
    scale = min(1.0, max_side / max(w, h))
    if scale < 1.0:
        img = img.resize((round(w * scale), round(h * scale)), Image.LANCZOS)
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=quality, optimize=True)
    return buf.getvalue()

3. 途中で落ちても続きから再開できるバッチにする

最も効いたのがこれです。結果を1枚ごとにJSON Lines へ追記し、起動時に処理済みファイル名を読み込んでスキップします。停止しても、もう一度起動すれば続きから走ります。

# resumable_batch.py
import json
from pathlib import Path
 
def load_done(jsonl_path: str) -> set[str]:
    """すでに処理済みのファイル名を読み込む"""
    done = set()
    p = Path(jsonl_path)
    if p.exists():
        for line in p.read_text(encoding="utf-8").splitlines():
            try:
                done.add(json.loads(line)["file_name"])
            except (json.JSONDecodeError, KeyError):
                continue  # 壊れた行は無視して継続
    return done
 
def resumable_batch(image_dir: str, jsonl_path: str) -> None:
    """チェックポイント方式で全画像を分析する（再開可能）"""
    done = load_done(jsonl_path)
    targets = sorted(
        f for f in Path(image_dir).iterdir()
        if f.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp"}
        and f.name not in done
    )
    print(f"未処理 {len(targets)} 件（処理済み {len(done)} 件をスキップ）")
 
    with open(jsonl_path, "a", encoding="utf-8") as out:
        for i, f in enumerate(targets, 1):
            try:
                data = analyze_with_retry(str(f))  # retry_utils のデコレータ付き
                data["file_name"] = f.name
                out.write(json.dumps(data, ensure_ascii=False) + "\n")
                out.flush()  # 1件ごとに確定させ、停止しても失わない
                print(f"[{i}/{len(targets)}] {f.name} ✓")
            except Exception as e:
                print(f"[{i}/{len(targets)}] {f.name} ✗ {e}")

flush() を1件ごとに呼ぶのがポイントです。バッファに溜めたままだと、プロセスが強制終了したときに直近の数十件が消えます。速度より確実性を優先する判断です。

4. 同じ商品には同じ結果を返したい場合のキャッシュ

再生成のたびに微妙に表現が変わると、差分管理が面倒になります。画像内容のハッシュをキーにキャッシュしておくと、同一画像には同一結果を返せます。

# cache_key.py
import hashlib
 
def image_cache_key(image_bytes: bytes, schema_version: str = "v1") -> str:
    """画像内容とスキーマ版を合わせたキャッシュキー"""
    h = hashlib.sha256(image_bytes).hexdigest()[:16]
    return f"{schema_version}:{h}"

スキーマ版をキーに含めるのは、スキーマを変えたら結果も変わるべきだからです。これを忘れると、スキーマ更新後も古い結果が返り続けて混乱します。

応用 — 多言語対応の商品説明文を同時生成する

グローバル展開を視野に入れているなら、日本語・英語・中国語の説明文を同時に生成することも可能です。スキーマを拡張するだけで対応できます。

# multilingual_product_analysis.py
multilingual_schema = types.Schema(
    type="object",
    properties={
        "product_name_ja": types.Schema(type="string", description="商品名（日本語）"),
        "product_name_en": types.Schema(type="string", description="Product name (English)"),
        "description_ja": types.Schema(type="string", description="商品説明（日本語・200文字）"),
        "description_en": types.Schema(type="string", description="Product description (English, ~100 words)"),
        "seo_keywords_ja": types.Schema(
            type="array",
            items=types.Schema(type="string"),
            description="SEOキーワード（日本語・5個）",
        ),
        "seo_keywords_en": types.Schema(
            type="array",
            items=types.Schema(type="string"),
            description="SEO keywords (English, 5 items)",
        ),
    },
    required=[
        "product_name_ja", "product_name_en",
        "description_ja", "description_en",
    ],
)
 
def analyze_multilingual(image_path: str) -> dict:
    """多言語対応の商品情報を生成する"""
    with open(image_path, "rb") as f:
        image_data = f.read()
 
    response = client.models.generate_content(
        model="gemini-3.1-pro",
        contents=[
            types.Content(
                role="user",
                parts=[
                    types.Part.from_bytes(data=image_data, mime_type="image/jpeg"),
                    types.Part.from_text(
                        "Analyze this product image and generate product information "
                        "in both Japanese and English for an e-commerce listing."
                    ),
                ],
            )
        ],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=multilingual_schema,
        ),
    )
    return json.loads(response.text)
 
# 期待する出力例:
# {
#   "product_name_ja": "レザーミニショルダーバッグ",
#   "product_name_en": "Leather Mini Crossbody Bag",
#   "description_ja": "上質な本革を使用したコンパクトなショルダーバッグ...",
#   "description_en": "A compact crossbody bag crafted from premium genuine leather...",
#   "seo_keywords_ja": ["レザーバッグ", "ショルダーバッグ", "本革", "ミニバッグ", "レディース"],
#   "seo_keywords_en": ["leather bag", "crossbody bag", "genuine leather", "mini bag", "women"]
# }

画像と文書をまたぐマルチモーダル処理の組み立て方は、マルチモーダルでドキュメントを扱う実装レポートも参考になります。

まとめ — まず100枚で測ってから本番へ

単発の分析スクリプトを本番パイプラインに育てるうえで効いたのは、派手な工夫ではなく、再開可能なバッチ・前処理での縮小・モデルの二段ルーティングという地味な3点でした。

次の一歩として、手元の100枚で本記事のチェックポイント方式を一度回し、Flash と Pro の一致率とコストを自分のデータで測ってみてください。その数字が、全件をどう振り分けるかの判断材料になります。私自身、この「まず小さく測る」を挟むようになってから、見積もりの外れが目に見えて減りました。お読みいただきありがとうございました。