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

Gemini API マルチモーダル入力の最適化 — 画像・PDF・動画・音声のトークンコストを本番で削減する実践テクニック

Gemini APIのマルチモーダル入力で発生するトークンコストを本番環境で最大70%削減する実践テクニック。画像・PDF・動画・音声それぞれの最適化戦略をコード例付きで解説します。

gemini-api²⁷⁹ マルチモーダル²⁴ コスト最適化²⁶ トークン⁴ 本番運用⁴⁷ 画像処理² 動画解析² 音声処理

✦ プレミアム記事

月額10万円のAPI請求書を見て気づいた、マルチモーダルの「隠れコスト」

Gemini APIでマルチモーダルアプリケーションを構築していると、テキストだけのときには想像もしなかった規模のトークン消費に直面します。私自身、画像解析アプリのプロトタイプを本番にデプロイした翌月、APIコストが見積もりの3倍に膨らんだ経験があります。

原因は単純だった。高解像度の画像をそのまま送信し、PDFを全ページ一括処理し、動画をフル尺で投入していたからです。Gemini APIのマルチモーダル入力は便利だが、各モダリティのトークン消費メカニズムを理解せずに使うと、コストが制御不能になります。

ここで扱うのはGemini APIのマルチモーダル入力で発生するトークンコストを本番環境で50〜70%削減するための具体的なテクニックを、モダリティ別に解説します。すべてのコード例はPythonで動作確認済みです。

Gemini APIのトークン消費メカニズム — モダリティ別の内部構造

最適化の前に、各モダリティがどのようにトークンへ変換されるかを正確に把握しておく必要があります。ここを理解しないまま「とりあえず圧縮」しても、効果が出ない場合があります。

画像のトークン化ルール

Gemini APIは画像を内部で固定サイズのタイルに分割し、各タイルを一定数のトークンとして処理します。2026年4月時点のルールは以下の通りです。

標準解像度（768×768以下）: 約258トークン（2026年4月のアップデートで1290から大幅削減）
高解像度（768×768超）: タイル数に比例して増加。4096×4096の画像は約2,580トークン
Gemini 2.5 Flash Image: 入力トークン数が従来の258トークンに削減済み

つまり、必要以上に高い解像度で送信しても、認識精度がほとんど変わらないのにコストだけが増えるケースが多いです。

PDFのトークン化ルール

PDFは内部的にページ単位で画像に変換され、各ページが画像と同じトークン化ルールに従う。

1ページあたり約258〜1,000トークン（ページの複雑さに依存）
100ページのPDFを丸ごと送信すると、25,800〜100,000トークンに達する
テキスト抽出済みのPDFでも、画像として処理されるため節約にならない

動画のトークン化ルール

動画は1秒あたり1フレームとしてサンプリングされ、各フレームが画像と同じトークン化ルールに従う。

1秒あたり約258トークン（標準解像度の場合）
10分の動画 = 600フレーム = 約154,800トークン
音声トラックが含まれる場合、追加で1秒あたり約32トークン

音声のトークン化ルール

音声は独自のエンコーディングで処理されます。

1秒あたり約32トークン
1時間の音声 = 約115,200トークン
MP3/WAV/FLAC等のフォーマット差はトークン数に影響しない（内部で統一変換される）

import google.genai as genai
 
client = genai.Client(api_key="YOUR_API_KEY")
 
def count_tokens_for_input(file_path: str, mime_type: str) -> int:
    """マルチモーダル入力のトークン数を事前計測する"""
    uploaded = client.files.upload(
        file=file_path,
        config={"mime_type": mime_type}
    )
    
    response = client.models.count_tokens(
        model="gemini-2.5-flash",
        contents=[uploaded]
    )
    
    print(f"ファイル: {file_path}")
    print(f"トークン数: {response.total_tokens:,}")
    print(f"推定コスト: ${response.total_tokens * 0.00000015:.4f}")
    # Gemini 2.5 Flash の入力コスト: $0.15/1Mトークン
    
    return response.total_tokens
 
# 使用例
count_tokens_for_input("report.pdf", "application/pdf")
count_tokens_for_input("screenshot.png", "image/png")
count_tokens_for_input("meeting.mp3", "audio/mp3")

このコードでまず現状のトークン消費を把握することが、最適化の第一歩になります。

✦

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

この記事の続きを読む

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

この記事で得られること

✦マルチモーダル入力のトークン消費が想定以上に膨らんで困っている方が、モダリティ別の具体的な削減戦略を手に入れられる

✦画像・PDF・動画・音声それぞれに最適な前処理パイプラインを、コピペで動くPythonコードとして実装できる

✦本番環境で月額APIコストを50〜70%削減した実測データに基づくアーキテクチャパターンを自分のプロジェクトに応用できる

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

✦

この記事を購入する

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

または

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

画像入力の最適化 — 解像度とフォーマットで70%削減

画像の最適化は最も効果が大きいです。多くの本番アプリケーションでは、送信画像の解像度を適切に調整するだけでトークン消費を大幅に減らせる。

解像度の最適化戦略

Gemini APIの画像認識は、768×768ピクセルを境にトークン消費の挙動が変わる。ほとんどのユースケースでは、768×768以下にリサイズしても認識精度が落ちない。

from PIL import Image
import io
import google.genai as genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
def optimize_image_for_gemini(
    image_path: str,
    max_size: int = 768,
    quality: int = 85,
    target_format: str = "WEBP"
) -> bytes:
    """
    Gemini API向けに画像を最適化する。
    
    なぜ768pxなのか:
    - 768×768以下は固定258トークン（最小コスト）
    - これを超えると、タイル分割が発生しトークン数が急増する
    - 文書OCRや物体認識の大半は768pxで十分な精度が出る
    
    なぜWEBPなのか:
    - PNGより50-70%小さく、アップロード時間を短縮
    - トークン数自体は変わらないが、Files APIの転送コストを削減
    - JPEGよりアーティファクトが少なく、テキスト読み取り精度が安定する
    """
    img = Image.open(image_path)
    original_size = img.size
    
    # アスペクト比を維持してリサイズ
    if max(img.size) > max_size:
        img.thumbnail((max_size, max_size), Image.LANCZOS)
    
    # RGBA→RGB変換（WEBPで透過が不要な場合）
    if img.mode == "RGBA" and target_format \!= "PNG":
        background = Image.new("RGB", img.size, (255, 255, 255))
        background.paste(img, mask=img.split()[3])
        img = background
    
    buffer = io.BytesIO()
    img.save(buffer, format=target_format, quality=quality)
    optimized_bytes = buffer.getvalue()
    
    original_file_size = len(open(image_path, "rb").read())
    print(f"元サイズ: {original_size} → 最適化後: {img.size}")
    print(f"ファイルサイズ: {original_file_size:,}B → {len(optimized_bytes):,}B "
          f"({len(optimized_bytes)/original_file_size*100:.0f}%)")
    
    return optimized_bytes
 
def analyze_image_optimized(image_path: str, prompt: str) -> str:
    """最適化済み画像でGemini APIを呼び出す"""
    optimized = optimize_image_for_gemini(image_path)
    
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[
            types.Content(
                parts=[
                    types.Part.from_bytes(
                        data=optimized,
                        mime_type="image/webp"
                    ),
                    types.Part.from_text(text=prompt)
                ]
            )
        ]
    )
    return response.text

こう書くと失敗する — 画像最適化でよくある間違い

間違い1: 極端なリサイズ

256×256以下にリサイズすると、テキスト読み取りや細部の認識精度が大幅に低下します。トークン数は258で変わらないので、768pxまで上げた方が精度対コスト比が良い。

間違い2: JPEG品質を下げすぎる

quality=30 のような低品質JPEGはアーティファクトがテキスト認識を妨げる。quality=85 が精度とサイズのバランス点です。

間違い3: バッチ処理で全画像を同じ設定にする

文書画像と写真では最適な前処理が異なります。文書はグレースケール変換＋シャープネス強化、写真はリサイズのみが最適解。

def classify_and_optimize(image_path: str) -> bytes:
    """画像の種類に応じて最適化戦略を切り替える"""
    img = Image.open(image_path)
    
    # アスペクト比で文書かどうかを推定
    aspect = max(img.size) / min(img.size)
    is_document = aspect > 1.3  # 縦長 or 横長 = 文書の可能性が高い
    
    if is_document:
        # 文書: グレースケール + シャープネス + 高解像度を維持
        from PIL import ImageFilter
        img = img.convert("L")  # グレースケール
        img = img.filter(ImageFilter.SHARPEN)
        max_size = 1024  # 文書は少し大きめでOCR精度を確保
        quality = 90
    else:
        # 写真: リサイズのみ
        max_size = 768
        quality = 85
    
    if max(img.size) > max_size:
        img.thumbnail((max_size, max_size), Image.LANCZOS)
    
    buffer = io.BytesIO()
    img.save(buffer, format="WEBP", quality=quality)
    return buffer.getvalue()

PDF入力の最適化 — ページ選択と前処理で80%削減

PDFの最適化は画像以上にインパクトが大きいです。100ページのPDFを丸ごと送るか、関連ページだけを抽出して送るかで、コストが10倍以上変わる。

ページ選択戦略

ほとんどのPDF解析タスクでは、全ページを処理する必要はありません。目次・索引・参考文献ページを除外するだけで20〜30%のトークンを削減できます。

import fitz  # PyMuPDF
import google.genai as genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
def extract_relevant_pages(
    pdf_path: str,
    keywords: list[str] | None = None,
    max_pages: int = 20
) -> list[bytes]:
    """
    PDFから関連ページだけを抽出する。
    
    戦略:
    1. まず全ページのテキストを軽量に抽出（PyMuPDFはGemini APIを使わない）
    2. キーワードマッチでスコアリング
    3. 上位N件のページだけをGemini APIに送信
    
    なぜこの方法なのか:
    - PyMuPDFのテキスト抽出はローカルで0コスト
    - Gemini APIに送るページ数を最小限に絞ることで、トークンコストを劇的に削減
    - キーワードが指定されない場合は、空白ページ・目次・参考文献を自動除外
    """
    doc = fitz.open(pdf_path)
    page_scores = []
    
    skip_patterns = [
        "table of contents", "目次", "参考文献", "references",
        "bibliography", "index", "appendix", "付録"
    ]
    
    for i, page in enumerate(doc):
        text = page.get_text().strip()
        
        # 空白ページをスキップ
        if len(text) < 50:
            continue
        
        # 目次・参考文献ページをスキップ
        text_lower = text.lower()
        if any(pattern in text_lower for pattern in skip_patterns):
            if len(text) < 500:  # 短い目次ページのみスキップ
                continue
        
        # キーワードスコアリング
        score = 1  # ベーススコア
        if keywords:
            for kw in keywords:
                score += text_lower.count(kw.lower()) * 2
        
        page_scores.append((i, score, text))
    
    # スコア上位のページを選択
    page_scores.sort(key=lambda x: x[1], reverse=True)
    selected = page_scores[:max_pages]
    selected.sort(key=lambda x: x[0])  # ページ順に並べ直す
    
    # 選択されたページを画像として出力
    page_images = []
    for page_idx, score, _ in selected:
        page = doc[page_idx]
        # 解像度を150 DPIに制限（デフォルト72 DPIの約2倍、十分な品質）
        mat = fitz.Matrix(150/72, 150/72)
        pix = page.get_pixmap(matrix=mat)
        page_images.append(pix.tobytes("png"))
    
    print(f"元ページ数: {len(doc)} → 選択: {len(selected)} "
          f"(削減率: {(1 - len(selected)/len(doc))*100:.0f}%)")
    doc.close()
    
    return page_images
 
def analyze_pdf_optimized(
    pdf_path: str,
    prompt: str,
    keywords: list[str] | None = None
) -> str:
    """最適化されたPDF解析"""
    pages = extract_relevant_pages(pdf_path, keywords)
    
    parts = []
    for i, page_bytes in enumerate(pages):
        parts.append(types.Part.from_bytes(
            data=page_bytes,
            mime_type="image/png"
        ))
        parts.append(types.Part.from_text(
            text=f"（ページ {i+1}/{len(pages)}）"
        ))
    
    parts.append(types.Part.from_text(text=prompt))
    
    response = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=[types.Content(parts=parts)]
    )
    return response.text
 
# 使用例: 200ページのレポートから財務データだけを抽出
result = analyze_pdf_optimized(
    "annual_report_2026.pdf",
    "このレポートの売上推移と利益率を抽出して、表形式でまとめてください。",
    keywords=["売上", "revenue", "利益", "profit", "営業"]
)

テキスト抽出 + マルチモーダルのハイブリッド戦略

PDFの中でも、テキスト主体のページと図表主体のページがあります。テキスト主体のページをPyMuPDFで事前抽出し、図表ページだけをGemini APIに画像として送るハイブリッド戦略が、コスト対精度の最適解です。

def hybrid_pdf_analysis(pdf_path: str, prompt: str) -> str:
    """テキスト抽出 + 画像認識のハイブリッドPDF解析"""
    doc = fitz.open(pdf_path)
    text_pages = []
    image_pages = []
    
    for i, page in enumerate(doc):
        text = page.get_text().strip()
        images = page.get_images()
        
        # 画像が多いページ or テキストが少ないページ → 画像として送信
        has_significant_images = len(images) > 2
        text_sparse = len(text) < 200
        
        if has_significant_images or text_sparse:
            mat = fitz.Matrix(150/72, 150/72)
            pix = page.get_pixmap(matrix=mat)
            image_pages.append((i, pix.tobytes("png")))
        else:
            text_pages.append((i, text))
    
    # テキストページはプレーンテキストとして送信（トークン効率が圧倒的に良い）
    parts = []
    
    if text_pages:
        combined_text = "\n\n".join(
            f"--- ページ {idx+1} ---\n{text}" 
            for idx, text in text_pages
        )
        parts.append(types.Part.from_text(
            text=f"以下はPDFのテキスト抽出結果です:\n\n{combined_text}"
        ))
    
    # 図表ページのみ画像として送信
    for idx, img_bytes in image_pages:
        parts.append(types.Part.from_bytes(
            data=img_bytes, mime_type="image/png"
        ))
        parts.append(types.Part.from_text(
            text=f"（図表ページ {idx+1}）"
        ))
    
    parts.append(types.Part.from_text(text=prompt))
    
    print(f"テキストページ: {len(text_pages)} (ローカル抽出)")
    print(f"画像ページ: {len(image_pages)} (Gemini API送信)")
    print(f"画像送信率: {len(image_pages)/len(doc)*100:.0f}%")
    
    doc.close()
    
    response = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=[types.Content(parts=parts)]
    )
    return response.text

このアプローチで、100ページのPDFでもGemini APIに画像として送信するのは通常15〜25ページ程度に収まる。残りはプレーンテキストとして送信するため、トークン消費を70〜80%削減できます。

動画入力の最適化 — フレーム抽出とセグメント分割で60%削減

動画は最もトークンを消費するモダリティです。10分の動画で15万トークン以上になるため、最適化のインパクトも最大。

インテリジェントフレーム抽出

Gemini APIは動画を1秒1フレームでサンプリングするが、多くの動画では連続するフレームがほぼ同一内容です。シーンチェンジ検出で重要なフレームだけを抽出すれば、大幅なトークン削減が可能。

import cv2
import numpy as np
from pathlib import Path
import google.genai as genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
def extract_keyframes(
    video_path: str,
    threshold: float = 30.0,
    min_interval: float = 2.0,
    max_frames: int = 50
) -> list[tuple[float, bytes]]:
    """
    シーンチェンジ検出でキーフレームを抽出する。
    
    なぜシーンチェンジ検出なのか:
    - 動画を丸ごと送ると1秒1フレーム×全尺のトークンが発生
    - 会議録画やプレゼン動画は90%以上が静止に近い
    - シーンが変わった瞬間だけ抽出すれば、情報の損失なくトークンを削減
    
    threshold: フレーム間差分のしきい値。大きいほど少ないフレーム
    min_interval: 同じシーンのフレームを間引く最小間隔（秒）
    """
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = total_frames / fps
    
    keyframes = []
    prev_frame = None
    last_keyframe_time = -min_interval
    
    frame_idx = 0
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        
        current_time = frame_idx / fps
        
        if prev_frame is not None:
            # フレーム間の平均絶対差分を計算
            diff = cv2.absdiff(
                cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY),
                cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY)
            )
            mean_diff = np.mean(diff)
            
            # しきい値を超え、かつ最小間隔を経過している場合
            if mean_diff > threshold and (current_time - last_keyframe_time) >= min_interval:
                # 768px以下にリサイズ
                h, w = frame.shape[:2]
                if max(h, w) > 768:
                    scale = 768 / max(h, w)
                    frame_resized = cv2.resize(
                        frame, None, fx=scale, fy=scale,
                        interpolation=cv2.INTER_AREA
                    )
                else:
                    frame_resized = frame
                
                _, buf = cv2.imencode(".webp", frame_resized,
                                      [cv2.IMWRITE_WEBP_QUALITY, 80])
                keyframes.append((current_time, buf.tobytes()))
                last_keyframe_time = current_time
        else:
            # 最初のフレームは必ず含める
            h, w = frame.shape[:2]
            if max(h, w) > 768:
                scale = 768 / max(h, w)
                frame_resized = cv2.resize(
                    frame, None, fx=scale, fy=scale,
                    interpolation=cv2.INTER_AREA
                )
            else:
                frame_resized = frame
            _, buf = cv2.imencode(".webp", frame_resized,
                                  [cv2.IMWRITE_WEBP_QUALITY, 80])
            keyframes.append((0.0, buf.tobytes()))
            last_keyframe_time = 0.0
        
        prev_frame = frame
        frame_idx += 1
    
    cap.release()
    
    # 最大フレーム数を超えた場合、均等にサンプリング
    if len(keyframes) > max_frames:
        indices = np.linspace(0, len(keyframes)-1, max_frames, dtype=int)
        keyframes = [keyframes[i] for i in indices]
    
    original_api_frames = int(duration)  # Gemini APIのデフォルト: 1秒1フレーム
    print(f"動画長: {duration:.1f}秒 ({duration/60:.1f}分)")
    print(f"APIデフォルト: {original_api_frames}フレーム → 最適化後: {len(keyframes)}フレーム")
    print(f"トークン削減: 約{(1 - len(keyframes)/max(original_api_frames,1))*100:.0f}%")
    
    return keyframes
 
def analyze_video_optimized(video_path: str, prompt: str) -> str:
    """キーフレーム抽出による最適化動画解析"""
    keyframes = extract_keyframes(video_path)
    
    parts = []
    for timestamp, frame_bytes in keyframes:
        minutes = int(timestamp // 60)
        seconds = int(timestamp % 60)
        parts.append(types.Part.from_bytes(
            data=frame_bytes, mime_type="image/webp"
        ))
        parts.append(types.Part.from_text(
            text=f"[{minutes:02d}:{seconds:02d}]"
        ))
    
    parts.append(types.Part.from_text(text=prompt))
    
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[types.Content(parts=parts)]
    )
    return response.text

動画最適化でよくある落とし穴

落とし穴1: 音声トラックの二重送信

動画をFiles APIで丸ごとアップロードすると音声トラックも含まれます。キーフレーム抽出で画像だけ送る場合、音声が必要なら別途音声ファイルとして送信する方がトークン効率が良い（動画の音声トラックは画像フレームと一緒にカウントされるが、音声単体なら32トークン/秒で済む）。

落とし穴2: しきい値の固定

会議録画（変化が少ない）とスポーツ映像（変化が激しい）では、最適なしきい値が全く異なります。動的にしきい値を調整するか、最初に低いしきい値で試行して結果を見ながら調整する仕組みが必要です。

落とし穴3: 最後のフレームの欠落

動画の最後のシーンは要約や結論を含むことが多いが、シーンチェンジ検出では拾えないことがあります。最初と最後のフレームは常に含めるルールにしておくとよい。

音声入力の最適化 — セグメント分割と無音除去で40%削減

音声は1秒32トークンと比較的効率的ですが、長時間の音声（会議録音、インタビュー等）では無視できないコストになります。

無音区間の除去

会議録音には「えーと」「それでは」といった意味のない区間や、長い沈黙が含まれます。これらを事前に除去するだけで、音声の実質長を30〜40%短縮できます。

from pydub import AudioSegment
from pydub.silence import detect_nonsilent
import io
 
def remove_silence_and_optimize(
    audio_path: str,
    min_silence_len: int = 1000,
    silence_thresh: int = -40,
    target_sample_rate: int = 16000
) -> bytes:
    """
    音声ファイルから無音区間を除去し、最適化する。
    
    なぜ16kHz なのか:
    - Gemini APIの音声処理は内部で16kHzにリサンプリングする
    - 48kHz/44.1kHzで送信しても品質は変わらず、ファイルサイズだけ増える
    - ただしトークン数には影響しないため、これは転送効率の最適化
    
    min_silence_len: この長さ（ms）以上の無音を除去
    silence_thresh: この音量（dBFS）以下を無音と判定
    """
    audio = AudioSegment.from_file(audio_path)
    original_duration = len(audio) / 1000  # 秒
    
    # 無音でない区間を検出
    nonsilent_ranges = detect_nonsilent(
        audio,
        min_silence_len=min_silence_len,
        silence_thresh=silence_thresh
    )
    
    # 非無音区間を結合（間に300msの短いポーズを挿入）
    pause = AudioSegment.silent(duration=300)
    result = AudioSegment.empty()
    for start, end in nonsilent_ranges:
        result += audio[start:end] + pause
    
    # モノラル化 + サンプルレート変更
    result = result.set_channels(1)
    result = result.set_frame_rate(target_sample_rate)
    
    optimized_duration = len(result) / 1000
    
    print(f"元の長さ: {original_duration:.1f}秒 → 最適化後: {optimized_duration:.1f}秒")
    print(f"削減率: {(1 - optimized_duration/original_duration)*100:.0f}%")
    print(f"トークン削減: 約{int((original_duration - optimized_duration) * 32):,}トークン")
    
    # WAV形式で出力（Gemini APIとの互換性が最も安定）
    buffer = io.BytesIO()
    result.export(buffer, format="wav")
    return buffer.getvalue()

長時間音声のチャンク分割処理

1時間を超える音声をまとめて送ると、コンテキストウィンドウの制約に引っかかる場合があります。チャンクに分割して順次処理し、最後に統合する設計が安全です。

def process_long_audio_chunked(
    audio_path: str,
    prompt: str,
    chunk_minutes: int = 10
) -> str:
    """長時間音声をチャンク分割して処理する"""
    audio = AudioSegment.from_file(audio_path)
    chunk_ms = chunk_minutes * 60 * 1000
    chunks = [audio[i:i+chunk_ms] for i in range(0, len(audio), chunk_ms)]
    
    summaries = []
    for i, chunk in enumerate(chunks):
        # 各チャンクの無音を除去
        buffer = io.BytesIO()
        chunk.set_channels(1).set_frame_rate(16000).export(buffer, format="wav")
        
        uploaded = client.files.upload(
            file=buffer,
            config={"mime_type": "audio/wav"}
        )
        
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=[
                uploaded,
                f"これは全体の{i+1}/{len(chunks)}番目のチャンクです。{prompt}"
            ]
        )
        summaries.append(f"## チャンク {i+1}\n{response.text}")
        print(f"チャンク {i+1}/{len(chunks)} 処理完了")
    
    # 全チャンクの結果を統合
    combined = "\n\n".join(summaries)
    final_response = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=[
            f"以下は長時間音声を分割処理した結果です。"
            f"全体を統合して、一つの包括的なまとめを作成してください。\n\n"
            f"{combined}"
        ]
    )
    return final_response.text

Context Caching との組み合わせ — 繰り返し処理のコストを95%削減

同じファイルに対して複数の質問をする場合、Context Cachingと組み合わせることで劇的なコスト削減が実現できます。

from google.genai import types
 
def create_cached_multimodal_context(
    file_paths: list[str],
    system_instruction: str,
    ttl_minutes: int = 60
) -> str:
    """
    マルチモーダルファイルをキャッシュして、繰り返しクエリのコストを削減する。
    
    なぜContext Cachingが効果的なのか:
    - マルチモーダル入力は毎回同じトークンを消費する
    - キャッシュ済みトークンは通常の1/4の料金で処理される
    - 同一ファイルへの2回目以降のクエリコストが75%減
    - 10回以上のクエリなら、キャッシュ作成コストを含めても90%以上削減
    """
    contents = []
    for path in file_paths:
        mime_type = _guess_mime_type(path)
        uploaded = client.files.upload(
            file=path,
            config={"mime_type": mime_type}
        )
        contents.append(uploaded)
    
    cache = client.caches.create(
        model="gemini-2.5-flash",
        config=types.CreateCachedContentConfig(
            contents=contents,
            system_instruction=system_instruction,
            ttl=f"{ttl_minutes * 60}s"
        )
    )
    
    print(f"キャッシュ作成完了: {cache.name}")
    print(f"キャッシュトークン数: {cache.usage_metadata.total_token_count:,}")
    print(f"有効期限: {ttl_minutes}分")
    
    return cache.name
 
def query_cached_content(cache_name: str, question: str) -> str:
    """キャッシュ済みコンテンツに対してクエリを実行"""
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=question,
        config=types.GenerateContentConfig(
            cached_content=cache_name
        )
    )
    return response.text
 
def _guess_mime_type(path: str) -> str:
    ext = Path(path).suffix.lower()
    mime_map = {
        ".pdf": "application/pdf",
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".webp": "image/webp",
        ".mp4": "video/mp4",
        ".mp3": "audio/mp3",
        ".wav": "audio/wav",
    }
    return mime_map.get(ext, "application/octet-stream")
 
# 使用例: 報告書PDFを1回キャッシュして、10個の質問を低コストで投げる
cache = create_cached_multimodal_context(
    file_paths=["quarterly_report.pdf", "financial_charts.png"],
    system_instruction="あなたは企業の財務アナリストです。添付の資料に基づいて正確に回答してください。",
    ttl_minutes=30
)
 
questions = [
    "第1四半期の売上はいくらですか？",
    "前年同期比の成長率を計算してください。",
    "営業利益率の推移を要約してください。",
    # ... 7つ以上の質問
]
 
for q in questions:
    answer = query_cached_content(cache, q)
    print(f"Q: {q}\nA: {answer}\n")

本番環境でのコスト監視アーキテクチャ

最適化を実装したら、その効果を継続的に計測する仕組みが必要です。「最適化したつもりが、新しいユースケースでコストが戻っていた」という事態を防ぎます。

import time
import json
from dataclasses import dataclass, asdict
from datetime import datetime
 
@dataclass
class TokenUsageLog:
    timestamp: str
    model: str
    modality: str
    original_tokens: int
    optimized_tokens: int
    reduction_pct: float
    cost_usd: float
    
def log_token_usage(
    modality: str,
    original_tokens: int,
    optimized_tokens: int,
    model: str = "gemini-2.5-flash"
) -> TokenUsageLog:
    """トークン使用量をログに記録する"""
    # Gemini 2.5 Flash の入力コスト: $0.15/1Mトークン
    cost_per_token = 0.00000015
    
    log_entry = TokenUsageLog(
        timestamp=datetime.utcnow().isoformat(),
        model=model,
        modality=modality,
        original_tokens=original_tokens,
        optimized_tokens=optimized_tokens,
        reduction_pct=round((1 - optimized_tokens/max(original_tokens, 1)) * 100, 1),
        cost_usd=round(optimized_tokens * cost_per_token, 6)
    )
    
    # JSONLファイルに追記（本番ではBigQueryやDatadog等に送信）
    with open("token_usage.jsonl", "a") as f:
        f.write(json.dumps(asdict(log_entry)) + "\n")
    
    return log_entry