個人開発でアプリやストア素材を扱っていると、画像のメタデータ整備という地味な作業がいつのまにか時間を食っていることに気づきます。私自身、数百枚単位で増えていく素材に対して、タグ付けと説明文づくりを手で回していた時期がありました。一枚あたり数十秒でも、まとまれば半日が消えます。
そこで 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%で、カジュアルな # 日常着として適しています。 この方法でも十分な情報が得られますが、出力が自由形式テキストのため、プログラムで処理するには不便です。次のステップで構造化出力を導入します。
構造化出力で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リクエストです。
ここから読み取れる実務的な結論は、全部を上位モデルで回す必要はない ということです。明快な商品（単色の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 の一致率とコストを自分のデータで測ってみてください。その数字が、全件をどう振り分けるかの判断材料になります。私自身、この「まず小さく測る」を挟むようになってから、見積もりの外れが目に見えて減りました。お読みいただきありがとうございました。