Gemini マルチモーダル API で画像・音声・動画を統合処理する有料サービスの実装

Gemini APIの最大の差別化ポイントは、テキストだけでなく画像・音声・動画・PDFを単一のAPIで処理できるマルチモーダル能力です。そしてこのマルチモーダル処理こそ、競合APIサービスとの差別化が図りやすく、高い付加価値を設定できる領域です。

ここではマルチモーダル処理を中核とした有料サービスを実際に構築する観点から、APIの使い方から課金フローまでを一通り解説します。

マルチモーダル処理が「有料化しやすい」理由

テキスト生成SaaSは参入障壁が低く、競合が多い。一方、マルチモーダル処理（画像分析・音声文字起こし・動画解析）は、エンドユーザーが「自分でできない」と感じやすい処理です。

この「自分でできない感」が、高い価格受容性につながります。

たとえば、以下のようなユースケースは「AI SaaSに月額$50払っても安い」と感じてもらいやすいです。

不動産写真を自動でAI採点・改善提案する
会議録音を文字起こし＋要約＋アクションアイテム抽出する
動画コンテンツを自動でハイライト抽出する
PDFマニュアルに対して自然言語でQ&Aする

それぞれをGeminiのマルチモーダルAPIで実装する方法と、課金設計を解説します。

Gemini File API：メディアファイルのアップロードと処理

マルチモーダル処理の起点は、Gemini File APIへのファイルアップロードです。

# file_upload.py
import google.generativeai as genai
import time
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
def upload_file(file_path: str) -> genai.types.File:
    """ファイルをGemini File APIにアップロードし、処理完了を待機する"""
    
    path = Path(file_path)
    
    # MIMEタイプを拡張子から判定
    mime_map = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".webp": "image/webp",
        ".mp3": "audio/mp3",
        ".wav": "audio/wav",
        ".m4a": "audio/m4a",
        ".mp4": "video/mp4",
        ".mov": "video/quicktime",
        ".pdf": "application/pdf",
    }
    mime_type = mime_map.get(path.suffix.lower(), "application/octet-stream")
    
    print(f"Uploading {path.name} ({mime_type})...")
    
    uploaded_file = genai.upload_file(
        path=str(path),
        mime_type=mime_type,
        display_name=path.name,
    )
    
    # 動画・音声はサーバー側処理に時間がかかる場合がある
    while uploaded_file.state.name == "PROCESSING":
        time.sleep(2)
        uploaded_file = genai.get_file(uploaded_file.name)
    
    if uploaded_file.state.name == "FAILED":
        raise ValueError(f"File upload failed: {uploaded_file.name}")
    
    print(f"Upload complete: {uploaded_file.uri}")
    return uploaded_file
 
def delete_file_after_processing(file: genai.types.File):
    """処理完了後にFile APIのファイルを削除（ストレージ節約）"""
    genai.delete_file(file.name)
    print(f"Deleted: {file.name}")

File APIにアップロードされたファイルは最大48時間保持されます。処理後は明示的に削除することで、ストレージを節約できます。

ユースケース別の実装パターン

1. 不動産写真のAI採点サービス

# real_estate_scorer.py
from dataclasses import dataclass
from typing import List
import json
 
@dataclass
class PhotoScore:
    score: int          # 1-100
    strengths: List[str]
    improvements: List[str]
    estimated_impact: str
 
def score_real_estate_photo(image_path: str) -> PhotoScore:
    """不動産写真をAIが採点し、改善点を提案する"""
    
    uploaded_file = upload_file(image_path)
    model = genai.GenerativeModel("gemini-2.5-pro")
    
    prompt = """
    この不動産物件の写真を専門家の視点で評価してください。
 
    以下のJSON形式で回答してください:
    {
      "score": 1〜100の整数,
      "strengths": ["強みを3つ以内で"],
      "improvements": ["改善点を3つ以内で"],
      "estimated_impact": "改善した場合の反響への期待効果（1文）"
    }
 
    評価基準:
    - 明るさ・コントラスト（20点）
    - 構図・アングル（20点）
    - 整理整頓・清潔感（20点）
    - 設備・内装の見せ方（20点）
    - 全体的な魅力度（20点）
    """
    
    response = model.generate_content([prompt, uploaded_file])
    delete_file_after_processing(uploaded_file)
    
    # JSON部分を抽出してパース
    text = response.text
    start = text.find("{")
    end = text.rfind("}") + 1
    data = json.loads(text[start:end])
    
    return PhotoScore(**data)
 
# 使用例
score = score_real_estate_photo("living_room.jpg")
print(f"スコア: {score.score}/100")
print(f"強み: {', '.join(score.strengths)}")
print(f"改善点: {', '.join(score.improvements)}")

このサービスの課金設計としては、「1回の採点あたり$2」の従量制か、「月額$79で100回」のサブスクリプションが現実的です。

2. 会議音声の文字起こし＋要約サービス

# meeting_analyzer.py
from dataclasses import dataclass
from typing import List
 
@dataclass
class MeetingAnalysis:
    transcript: str
    summary: str
    action_items: List[str]
    key_decisions: List[str]
    duration_minutes: float
 
def analyze_meeting_audio(audio_path: str) -> MeetingAnalysis:
    """会議音声を文字起こし・要約・アクションアイテム抽出する"""
    
    uploaded_file = upload_file(audio_path)
    
    # 音声文字起こしはFlash（高速・低コスト）で可能
    flash_model = genai.GenerativeModel("gemini-2.5-flash")
    
    # Step 1: 文字起こし（Flash）
    transcript_response = flash_model.generate_content([
        "この音声を正確に文字起こししてください。話者が複数いる場合は [話者A]: [話者B]: で区別してください。",
        uploaded_file,
    ])
    transcript = transcript_response.text
    
    # Step 2: 分析はProで（高品質な要約・アクションアイテム抽出）
    pro_model = genai.GenerativeModel("gemini-2.5-pro")
    
    analysis_response = pro_model.generate_content(f"""
    以下の会議文字起こしを分析してください。
 
    【文字起こし】
    {transcript}
 
    以下の形式で回答してください:
    
    ## 要約（3文以内）
    [要約]
    
    ## アクションアイテム
    - [担当者があれば名前付きで]: [タスク内容]
    
    ## 主要決定事項
    - [決定した内容]
    """)
    
    delete_file_after_processing(uploaded_file)
    
    # 結果をパース
    analysis_text = analysis_response.text
    # 実際の実装ではより堅牢なパーサーを使用
    
    return MeetingAnalysis(
        transcript=transcript,
        summary=extract_section(analysis_text, "要約"),
        action_items=extract_list(analysis_text, "アクションアイテム"),
        key_decisions=extract_list(analysis_text, "主要決定事項"),
        duration_minutes=estimate_duration(audio_path),
    )
 
def extract_section(text: str, header: str) -> str:
    """マークダウンセクションを抽出するヘルパー"""
    lines = text.split("\n")
    in_section = False
    section_lines = []
    
    for line in lines:
        if header in line and line.startswith("##"):
            in_section = True
            continue
        if in_section and line.startswith("##"):
            break
        if in_section:
            section_lines.append(line)
    
    return "\n".join(section_lines).strip()
 
def extract_list(text: str, header: str) -> list:
    section = extract_section(text, header)
    return [line.lstrip("- ").strip() for line in section.split("\n") if line.strip().startswith("-")]
 
def estimate_duration(audio_path: str) -> float:
    """音声ファイルの長さを推定（実際はmutagen等を使用）"""
    import os
    size_mb = os.path.getsize(audio_path) / (1024 * 1024)
    # MP3 128kbps: 約1MB/分
    return size_mb

この「文字起こしはFlash・分析はPro」のハイブリッド構成がコスト最適です。1時間の会議音声処理コストは、Flash文字起こし($0.02程度) + Pro分析($0.05程度) = 合計$0.07前後。これを$5/会議で請求できれば粗利98.6%です。

3. PDF文書への自然言語Q&Aサービス

# pdf_qa.py
from dataclasses import dataclass
 
@dataclass
class QAResult:
    answer: str
    confidence: str   # "high" / "medium" / "low"
    source_hints: list[str]  # 参照箇所のヒント
 
class PDFQAService:
    def __init__(self):
        self.model = genai.GenerativeModel("gemini-2.5-pro")
        self._cached_files: dict[str, genai.types.File] = {}
    
    def load_pdf(self, pdf_path: str, doc_id: str):
        """PDFをアップロードしてキャッシュする"""
        if doc_id not in self._cached_files:
            self._cached_files[doc_id] = upload_file(pdf_path)
        return self._cached_files[doc_id]
    
    def ask(self, doc_id: str, question: str) -> QAResult:
        """PDFに対して質問する"""
        
        if doc_id not in self._cached_files:
            raise ValueError(f"Document {doc_id} not loaded. Call load_pdf() first.")
        
        pdf_file = self._cached_files[doc_id]
        
        response = self.model.generate_content([
            f"""以下のPDF文書の内容に基づいて、質問に回答してください。
 
            【重要な指示】
            - 文書に記載されていない内容は「この文書には記載されていません」と回答する
            - 回答の確実度を "high" / "medium" / "low" で示す
            - 参照した箇所のヒント（ページ番号や見出し）があれば教える
 
            【質問】
            {question}
 
            以下のJSON形式で回答:
            {{
              "answer": "回答内容",
              "confidence": "high/medium/low",
              "source_hints": ["参照箇所のヒント"]
            }}
            """,
            pdf_file,
        ])
        
        import json
        text = response.text
        start = text.find("{")
        end = text.rfind("}") + 1
        data = json.loads(text[start:end])
        
        return QAResult(**data)
    
    def cleanup(self, doc_id: str):
        """セッション終了後にファイルを削除"""
        if doc_id in self._cached_files:
            delete_file_after_processing(self._cached_files[doc_id])
            del self._cached_files[doc_id]

このPDF Q&Aサービスは、技術マニュアル・法務文書・社内規程への社員向けQ&Aとして高い需要があります。企業向けに月額$199/ドキュメントで提供するだけでも、10社契約で月$1,990の安定収益になります。

Stripe Metered Billing との連携

マルチモーダルサービスの課金は、処理したファイルサイズや処理回数に基づく従量制が自然です。

# stripe_metered_billing.py
import stripe
from dataclasses import dataclass
import time
 
stripe.api_key = "YOUR_STRIPE_SECRET_KEY"
 
@dataclass
class UsageRecord:
    customer_id: str
    subscription_item_id: str
    quantity: int
    unit: str  # "image_analysis" / "audio_minutes" / "pdf_qa"
 
def report_usage_to_stripe(record: UsageRecord) -> stripe.UsageRecord:
    """Stripe Metered Billingに使用量を報告する"""
    
    usage_record = stripe.SubscriptionItem.create_usage_record(
        record.subscription_item_id,
        quantity=record.quantity,
        timestamp=int(time.time()),
        action="increment",
    )
    
    print(f"Usage reported: {record.quantity} {record.unit} for {record.customer_id}")
    return usage_record
 
# マルチモーダル処理と課金を統合したサービスクラス
class BilledMultimodalService:
    def __init__(self):
        self.model_flash = genai.GenerativeModel("gemini-2.5-flash")
        self.model_pro = genai.GenerativeModel("gemini-2.5-pro")
    
    def analyze_image_with_billing(
        self,
        image_path: str,
        prompt: str,
        customer_id: str,
        subscription_item_id: str,
        use_pro: bool = False,
    ) -> dict:
        """画像分析と同時にStripe課金を記録する"""
        
        uploaded = upload_file(image_path)
        model = self.model_pro if use_pro else self.model_flash
        
        response = model.generate_content([prompt, uploaded])
        delete_file_after_processing(uploaded)
        
        # 課金記録（1画像 = 1ユニット）
        report_usage_to_stripe(UsageRecord(
            customer_id=customer_id,
            subscription_item_id=subscription_item_id,
            quantity=1,
            unit="image_analysis",
        ))
        
        return {
            "result": response.text,
            "model": "gemini-2.5-pro" if use_pro else "gemini-2.5-flash",
            "billed": True,
        }
    
    def transcribe_audio_with_billing(
        self,
        audio_path: str,
        duration_minutes: float,
        customer_id: str,
        subscription_item_id: str,
    ) -> dict:
        """音声文字起こしと課金を記録する（分単位課金）"""
        
        uploaded = upload_file(audio_path)
        
        response = self.model_flash.generate_content([
            "この音声を正確に文字起こしして、全文を出力してください。",
            uploaded,
        ])
        
        delete_file_after_processing(uploaded)
        
        # 課金記録（分単位、切り上げ）
        billed_minutes = max(1, round(duration_minutes))
        report_usage_to_stripe(UsageRecord(
            customer_id=customer_id,
            subscription_item_id=subscription_item_id,
            quantity=billed_minutes,
            unit="audio_minutes",
        ))
        
        return {
            "transcript": response.text,
            "billed_minutes": billed_minutes,
            "model": "gemini-2.5-flash",
        }

Stripeのダッシュボードで Metered Billing の Price を作成し、subscription_item_id を保存しておくことで、処理ごとに使用量を積み上げ、月末に自動請求されます。

エラーハンドリングと本番運用

マルチモーダルAPIのエラーは、テキストAPIより種類が多い（サポート外フォーマット・ファイルサイズ制限・処理タイムアウトなど）。本番環境では丁寧なエラーハンドリングが必要です。

# error_handling.py
import google.api_core.exceptions as google_exceptions
from functools import wraps
import logging
 
logger = logging.getLogger(__name__)
 
class MultimodalServiceError(Exception):
    """サービス固有のエラー基底クラス"""
    def __init__(self, message: str, user_message: str, retryable: bool = False):
        super().__init__(message)
        self.user_message = user_message  # エンドユーザーへの表示メッセージ
        self.retryable = retryable
 
def handle_gemini_errors(func):
    """Gemini API エラーをサービス固有エラーに変換するデコレーター"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        
        except google_exceptions.InvalidArgument as e:
            error_str = str(e)
            
            if "file type" in error_str.lower() or "mime" in error_str.lower():
                raise MultimodalServiceError(
                    message=f"Unsupported file type: {e}",
                    user_message="このファイル形式はサポートされていません。JPG、PNG、MP3、MP4、PDFをお使いください。",
                    retryable=False,
                )
            
            if "file size" in error_str.lower() or "too large" in error_str.lower():
                raise MultimodalServiceError(
                    message=f"File too large: {e}",
                    user_message="ファイルサイズが制限（2GB）を超えています。ファイルを分割してお試しください。",
                    retryable=False,
                )
            
            raise MultimodalServiceError(
                message=f"Invalid request: {e}",
                user_message="リクエストの形式が正しくありません。",
                retryable=False,
            )
        
        except google_exceptions.ResourceExhausted as e:
            logger.warning(f"Rate limited: {e}")
            raise MultimodalServiceError(
                message=f"Rate limited: {e}",
                user_message="ただいまサービスが混雑しています。しばらく時間をおいて再度お試しください。",
                retryable=True,  # リトライ可能
            )
        
        except google_exceptions.DeadlineExceeded as e:
            logger.error(f"Timeout: {e}")
            raise MultimodalServiceError(
                message=f"Timeout: {e}",
                user_message="処理がタイムアウトしました。ファイルが大きい場合は分割してお試しください。",
                retryable=True,
            )
        
        except Exception as e:
            logger.exception(f"Unexpected error in {func.__name__}: {e}")
            raise MultimodalServiceError(
                message=f"Unexpected error: {e}",
                user_message="予期しないエラーが発生しました。しばらくしてから再度お試しください。",
                retryable=False,
            )
    
    return wrapper
 
# 使用例
@handle_gemini_errors
def safe_analyze_image(image_path: str, prompt: str) -> str:
    uploaded = upload_file(image_path)
    model = genai.GenerativeModel("gemini-2.5-pro")
    response = model.generate_content([prompt, uploaded])
    delete_file_after_processing(uploaded)
    return response.text

料金体系の設計：高利益率を維持するには

マルチモーダルSaaSの料金設計で重要なのは、API原価の10〜20倍の価格設定です。

実際の処理コスト（参考値）：

画像分析1件（Flash）: $0.001〜$0.003
画像分析1件（Pro）: $0.01〜$0.03
音声文字起こし1分（Flash）: $0.002〜$0.005
PDF Q&A 1問（Pro）: $0.02〜$0.10（ドキュメントのサイズによる）

これをもとにした推奨料金体系：

不動産写真採点サービス（月額サブスク）:

スターター: $29/月・30枚（API原価 $0.09〜$0.90 → 粗利 96〜99%）
プロ: $79/月・100枚（API原価 $0.30〜$3.00 → 粗利 96〜99%）
エンタープライズ: $299/月・500枚

会議文字起こし＋分析サービス（従量制）:

$0.10/分（Flash文字起こし + Pro分析の合計原価: $0.007〜$0.015/分）
月$100サブスク = 1,000分の処理枠

PDF Q&Aサービス（ドキュメント単位）:

ドキュメント登録: $5（1ドキュメントあたり）
Q&A: $0.50/問（または月額$49で無制限Q&A）

このような価格設定では、サービスの粗利率は90〜99%の範囲に収まります。Stripe手数料（2.9%+$0.30）・インフラコスト（Cloudflare Workers・Supabase等）を差し引いても、利益率70%以上は現実的に維持できます。

全体を振り返って：マルチモーダルで差別化する

Gemini APIのマルチモーダル機能を活かした有料サービスは、テキスト生成SaaSよりも高い価格受容性があります。理由は明確です。「自分ではできない処理をAIがやってくれる」という価値認識が強いからです。

実装のポイントを整理します。

File APIのアップロード・削除フローを丁寧に実装することが安定運用の基盤になります
文字起こしはFlash・高品質分析はProというハイブリッド構成でコストを最適化できます
Stripe Metered Billingとの連携で、処理量に応じた自然な従量課金を実現できます
エラー種別（サポート外フォーマット・レート制限・タイムアウト）をきちんとハンドリングし、ユーザーに分かりやすいメッセージを返すことが信頼性につながります

サービスとして公開するハードルはテキストSaaSよりも高い分、参入障壁も高い。それがマルチモーダルサービスの収益性の高さを支えています。実装の参考になれば幸いです。