Gemini API で実現するマルチモーダル文書解析システム — 画像・PDF・動画を統合処理する実践アーキテクチャ

取り組みの背景

業務で扱うドキュメントは、テキストだけではありません。請求書のスキャン画像、契約書のPDF、会議の録画動画——こうした多様な形式のファイルから必要な情報を効率的に抽出するのは、これまで非常に手間のかかる作業でしました。

Gemini API のマルチモーダル機能を使えば、画像・PDF・動画といった異なる形式のファイルを1つのAPIで統合的に処理できます。ここではこうしたマルチモーダル文書解析システムの設計から実装、本番運用まで体系的に解説します。

Gemini API のマルチモーダル入力を理解する

サポートされるファイル形式

Gemini API は以下の形式をネイティブにサポートしています。

画像: JPEG, PNG, GIF, WebP, HEIC, HEIF
PDF: 複数ページ対応（最大3,600ページ）
動画: MP4, MPEG, AVI, FLV, MKV, MOV, WebM 等
音声: MP3, WAV, FLAC, AAC, OGG 等

特筆すべきは、PDF をネイティブに理解できる点です。OCR を別途実行する必要がなく、テキストと画像が混在するPDFも1回のAPI呼び出しで処理できます。

2つのファイル入力方式

Gemini API では、ファイルサイズに応じて2つの入力方式を使い分けます。

インラインデータ（20MB以下）: Base64 エンコードしてリクエストボディに含める方式。小さなファイルに最適です。

import google.generativeai as genai
import base64
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro")
 
# 画像をインラインで送信
image_data = Path("receipt.jpg").read_bytes()
response = model.generate_content([
    "この領収書から、日付、店舗名、合計金額を抽出してJSON形式で返してください。",
    {
        "mime_type": "image/jpeg",
        "data": base64.b64encode(image_data).decode("utf-8")
    }
])
print(response.text)

File API（20MB超）: サーバーにファイルをアップロードし、URIで参照する方式。PDFや動画など大容量ファイルに使用します。

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro")
 
# PDFをFile APIでアップロード
uploaded_file = genai.upload_file(
    path="contract.pdf",
    display_name="契約書",
    mime_type="application/pdf"
)
 
# アップロード完了を待機
import time
while uploaded_file.state.name == "PROCESSING":
    time.sleep(2)
    uploaded_file = genai.get_file(uploaded_file.name)
 
# PDFの内容を解析
response = model.generate_content([
    "この契約書から以下の情報をJSON形式で抽出してください: "
    "契約当事者、契約日、契約期間、主要な条件、解約条件",
    uploaded_file
])
print(response.text)

マルチモーダル文書解析パイプラインの設計

アーキテクチャ概要

実用的な文書解析システムは、以下の4段階で構成します。

ファイル受付: アップロードされたファイルの形式判定と前処理
解析実行: Gemini API へのリクエスト（形式に応じたプロンプト選択）
構造化出力: JSON Schema を指定して構造化データを取得
後処理: バリデーション、データベース保存、通知

統合解析クラスの実装

import google.generativeai as genai
from pathlib import Path
from dataclasses import dataclass
from enum import Enum
import json
import mimetypes
 
class FileType(Enum):
    IMAGE = "image"
    PDF = "pdf"
    VIDEO = "video"
    AUDIO = "audio"
 
@dataclass
class AnalysisResult:
    file_name: str
    file_type: FileType
    extracted_data: dict
    confidence: float
    raw_response: str
 
class MultimodalAnalyzer:
    """マルチモーダル文書解析エンジン"""
 
    # 形式別のシステムプロンプト
    PROMPTS = {
        FileType.IMAGE: (
            "この画像を詳細に分析してください。"
            "テキスト、数値、表、図表を全て抽出し、"
            "構造化されたJSON形式で返してください。"
        ),
        FileType.PDF: (
            "このPDF文書を全ページ分析してください。"
            "タイトル、セクション見出し、本文、表、"
            "図表のキャプションを構造化JSON形式で返してください。"
        ),
        FileType.VIDEO: (
            "この動画の内容を分析してください。"
            "主要なシーン、テキスト表示、発話内容の要約を"
            "タイムスタンプ付きのJSON形式で返してください。"
        ),
    }
 
    def __init__(self, api_key: str, model_name: str = "gemini-2.5-pro"):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel(
            model_name,
            generation_config={
                "response_mime_type": "application/json",
                "temperature": 0.1,  # 解析精度を重視
            }
        )
 
    def detect_file_type(self, file_path: str) -> FileType:
        """ファイル形式を自動判定"""
        mime_type, _ = mimetypes.guess_type(file_path)
        if mime_type:
            if mime_type.startswith("image/"):
                return FileType.IMAGE
            elif mime_type == "application/pdf":
                return FileType.PDF
            elif mime_type.startswith("video/"):
                return FileType.VIDEO
            elif mime_type.startswith("audio/"):
                return FileType.AUDIO
        raise ValueError(f"サポートされていないファイル形式: {file_path}")
 
    def analyze(
        self,
        file_path: str,
        custom_prompt: str | None = None
    ) -> AnalysisResult:
        """ファイルを解析し、構造化データを返す"""
        path = Path(file_path)
        file_type = self.detect_file_type(file_path)
        file_size = path.stat().st_size
 
        # ファイルサイズに応じてアップロード方式を選択
        if file_size > 20 * 1024 * 1024:  # 20MB超
            uploaded = genai.upload_file(path=str(path))
            import time
            while uploaded.state.name == "PROCESSING":
                time.sleep(2)
                uploaded = genai.get_file(uploaded.name)
            file_input = uploaded
        else:
            import base64
            mime_type, _ = mimetypes.guess_type(str(path))
            file_input = {
                "mime_type": mime_type,
                "data": base64.b64encode(
                    path.read_bytes()
                ).decode("utf-8")
            }
 
        # プロンプトの選択
        prompt = custom_prompt or self.PROMPTS.get(
            file_type, self.PROMPTS[FileType.IMAGE]
        )
 
        # API呼び出し
        response = self.model.generate_content([prompt, file_input])
        raw_text = response.text
 
        # JSON パース
        try:
            extracted = json.loads(raw_text)
        except json.JSONDecodeError:
            extracted = {"raw_text": raw_text}
 
        return AnalysisResult(
            file_name=path.name,
            file_type=file_type,
            extracted_data=extracted,
            confidence=0.95,  # 実装例
            raw_response=raw_text,
        )

構造化出力で解析精度を高める

JSON Schema の活用

Gemini API の response_mime_type と response_schema を組み合わせることで、出力のJSON構造を厳密に制御できます。

import google.generativeai as genai
from google.generativeai import types
 
# 請求書解析用のスキーマ定義
invoice_schema = types.Schema(
    type=types.Type.OBJECT,
    properties={
        "invoice_number": types.Schema(type=types.Type.STRING),
        "date": types.Schema(type=types.Type.STRING),
        "vendor": types.Schema(
            type=types.Type.OBJECT,
            properties={
                "name": types.Schema(type=types.Type.STRING),
                "address": types.Schema(type=types.Type.STRING),
            },
        ),
        "items": types.Schema(
            type=types.Type.ARRAY,
            items=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "description": types.Schema(type=types.Type.STRING),
                    "quantity": types.Schema(type=types.Type.INTEGER),
                    "unit_price": types.Schema(type=types.Type.NUMBER),
                    "amount": types.Schema(type=types.Type.NUMBER),
                },
            ),
        ),
        "subtotal": types.Schema(type=types.Type.NUMBER),
        "tax": types.Schema(type=types.Type.NUMBER),
        "total": types.Schema(type=types.Type.NUMBER),
    },
)
 
model = genai.GenerativeModel(
    "gemini-2.5-pro",
    generation_config={
        "response_mime_type": "application/json",
        "response_schema": invoice_schema,
    },
)
 
# 請求書画像を解析
response = model.generate_content([
    "この請求書の内容を正確に抽出してください。",
    uploaded_invoice_image,
])
 
# 型安全なJSONが返る
invoice_data = json.loads(response.text)
print(f"請求額: ¥{invoice_data['total']:,.0f}")

複数ファイルの同時処理

Gemini API は1回のリクエストで複数のファイルを処理できます。関連する文書をまとめて送信し、横断的な分析を行うことも可能です。

# 複数の請求書を一括処理
files = [
    genai.upload_file("invoice_2026_01.pdf"),
    genai.upload_file("invoice_2026_02.pdf"),
    genai.upload_file("invoice_2026_03.pdf"),
]
 
response = model.generate_content([
    "これら3件の請求書を比較分析してください。"
    "月ごとの合計金額の推移、共通の費目、"
    "異常値があれば指摘してください。",
    *files,
])

バッチ処理パイプラインの構築

非同期バッチ処理の実装

大量のファイルを処理する場合、非同期処理とリトライ機構が不可欠です。

import asyncio
import aiofiles
from typing import List
from dataclasses import dataclass
 
@dataclass
class BatchResult:
    total: int
    succeeded: int
    failed: int
    results: List[AnalysisResult]
    errors: List[dict]
 
class BatchProcessor:
    """大量ファイルのバッチ解析"""
 
    def __init__(
        self,
        analyzer: MultimodalAnalyzer,
        max_concurrent: int = 5,
        max_retries: int = 3,
    ):
        self.analyzer = analyzer
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.max_retries = max_retries
 
    async def process_file(
        self, file_path: str
    ) -> tuple[AnalysisResult | None, dict | None]:
        """1ファイルをリトライ付きで処理"""
        async with self.semaphore:
            for attempt in range(self.max_retries):
                try:
                    # 同期APIを非同期で実行
                    result = await asyncio.to_thread(
                        self.analyzer.analyze, file_path
                    )
                    return result, None
                except Exception as e:
                    if attempt < self.max_retries - 1:
                        # 指数バックオフで待機
                        wait = 2 ** attempt
                        await asyncio.sleep(wait)
                    else:
                        return None, {
                            "file": file_path,
                            "error": str(e),
                            "attempts": self.max_retries,
                        }
 
    async def process_batch(
        self, file_paths: List[str]
    ) -> BatchResult:
        """複数ファイルを並行処理"""
        tasks = [
            self.process_file(path) for path in file_paths
        ]
        outcomes = await asyncio.gather(*tasks)
 
        results = []
        errors = []
        for result, error in outcomes:
            if result:
                results.append(result)
            if error:
                errors.append(error)
 
        return BatchResult(
            total=len(file_paths),
            succeeded=len(results),
            failed=len(errors),
            results=results,
            errors=errors,
        )

使用例

import asyncio
 
analyzer = MultimodalAnalyzer(api_key="YOUR_GEMINI_API_KEY")
processor = BatchProcessor(analyzer, max_concurrent=3)
 
# フォルダ内の全PDFを一括処理
import glob
pdf_files = glob.glob("invoices/*.pdf")
 
batch_result = asyncio.run(processor.process_batch(pdf_files))
print(f"処理完了: {batch_result.succeeded}/{batch_result.total} 成功")
# 期待出力: 処理完了: 48/50 成功
 
if batch_result.errors:
    print(f"失敗: {batch_result.failed}件")
    for err in batch_result.errors:
        print(f"  - {err['file']}: {err['error']}")

コスト最適化のテクニック

モデルの使い分け

全ての処理に Gemini 2.5 Pro を使う必要はありません。タスクの複雑さに応じてモデルを切り替えることでコストを大幅に削減できます。

Gemini 2.5 Flash: 定型フォーマットの文書（請求書、領収書、名刺）の解析
Gemini 2.5 Pro: 複雑な契約書の条項分析、複数文書の横断分析
Gemini 2.5 Flash Lite: 単純なテキスト抽出、ファイル形式の判定

コンテキストキャッシュの活用

同じシステムプロンプトやスキーマで大量のファイルを処理する場合、コンテキストキャッシュを利用することで入力トークンのコストを最大75%削減できます。

from google.generativeai import caching
import datetime
 
# キャッシュを作成（1時間有効）
cache = caching.CachedContent.create(
    model="models/gemini-2.5-pro",
    display_name="invoice-analysis-cache",
    system_instruction=(
        "あなたは請求書解析の専門家です。"
        "アップロードされたファイルから正確にデータを抽出し、"
        "指定のJSON形式で返してください。"
    ),
    ttl=datetime.timedelta(hours=1),
)
 
# キャッシュを使ってモデルを初期化
cached_model = genai.GenerativeModel.from_cached_content(cache)
 
# 以降のリクエストではシステムプロンプトのトークンが課金されない
for pdf in pdf_files:
    response = cached_model.generate_content([
        "この請求書を解析してください。",
        genai.upload_file(pdf)
    ])

まとめ

Gemini API のマルチモーダル機能は、これまでバラバラのツールで行っていた文書解析ワークフローを一本化できる強力な基盤です。画像・PDF・動画を同じAPIで処理でき、構造化出力やコンテキストキャッシュと組み合わせることで、本番品質の文書解析パイプラインを効率的に構築できます。

まずは1種類のファイル形式（たとえば請求書PDF）の解析から始めて、徐々に対応形式を広げていくアプローチがおすすめです。構造化出力の本番パターンを活用し、RAGパイプラインやプロンプトエンジニアリングと組み合わせることで、さらに高度な文書解析システムを実現できるでしょう。