Gemini API × Cloud Storage 大容量ファイル処理パイプライン — 100MBアップロードとプライベートDB連携の本番実装ガイド

取り組みの背景 — ファイルサイズ制限の壁が消えた

Gemini API を本番環境で活用するうえで、これまで最大のボトルネックの一つだったのが 20MBのファイルアップロード制限 です。長時間の会議録音、数百ページの技術ドキュメント、高解像度の製品画像カタログ——こうした実務レベルのファイルは20MBを簡単に超えてしまい、事前にチャンク分割や圧縮といった前処理が不可欠でしました。

2026年4月、Google はこの制限を 一気に100MBへ引き上げ るとともに、Cloud Storage（GCS）バケットおよびプライベートDB署名付きURL をデータ入力ソースとしてネイティブサポートしました。この変更により、ファイルをローカルからアップロードする必要すらなくなり、GCS上のデータを直接 Gemini に渡せるようになっています。

対象読者:

Gemini API で大容量ファイルの処理に課題を抱えている開発者
GCS や S3 との連携パイプラインを構築したいバックエンドエンジニア
マルチモーダル AI をプロダクション環境で運用している技術リーダー

Gemini API のマルチモーダル機能の基本については「Gemini API で実現するマルチモーダル文書解析システム」で詳しく解説しています。本記事はその発展編として、Cloud Storage 連携と大容量ファイル対応に焦点を当てます。

Cloud Storage 統合の全体像

アーキテクチャの3つの入力パターン

Gemini API の新しいファイル入力は、大きく3つのパターンに分類できます。

パターン1: GCSバケット直接参照

Google Cloud Storage のバケットに格納されたファイルを、gs:// URI で直接指定する方法です。ファイルのダウンロード→再アップロードが不要になるため、ネットワーク転送量とレイテンシを大幅に削減できます。

パターン2: 署名付きURL（Pre-signed URL）

GCS だけでなく、AWS S3 や Azure Blob Storage、あるいはプライベートデータベースから発行される署名付きURLを入力として渡せます。マルチクラウド環境やオンプレミスのデータソースとの統合に最適です。

パターン3: 従来のファイルアップロード（100MBまで拡大）

Files API 経由の直接アップロードも、上限が100MBに拡大されましました。小〜中規模のファイルや、クラウドストレージを使わない構成では引き続き有効です。

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

100MBまで対応するファイル形式は、既存のマルチモーダル対応と同じです。

ドキュメント: PDF（数百ページ対応）、テキストファイル
画像: JPEG、PNG、WebP、GIF
音声: MP3、WAV、FLAC、OGG（最大約9.5時間相当）
動画: MP4、MOV、MPEG（最大約1時間相当）

GCS バケット連携の実装

Python SDK での基本実装

import google.generativeai as genai
from google.cloud import storage
import os
 
# APIキー設定
genai.configure(api_key=os.environ["YOUR_GEMINI_API_KEY"])
 
# GCS バケットからファイルを直接参照
model = genai.GenerativeModel("gemini-2.5-pro")
 
# GCS URI を直接指定してファイルを処理
response = model.generate_content([
    "この技術ドキュメントの要点を日本語で構造化してまとめてください。"
    "各セクションの重要度を5段階で評価し、実装時の注意点も付記してください。",
    {"file_uri": "gs://my-project-bucket/docs/technical-spec-v3.pdf",
     "mime_type": "application/pdf"}
])
 
print(response.text)
# 出力例:
# ## 技術ドキュメント要約
# ### セクション1: アーキテクチャ概要 (重要度: ★★★★★)
# - マイクロサービス構成で...
# ### セクション2: API設計 (重要度: ★★★★☆)
# ...

署名付きURL を使ったプライベートDB連携

import google.generativeai as genai
from google.cloud import storage
from datetime import timedelta
 
def generate_signed_url(bucket_name: str, blob_name: str, 
                         expiration_minutes: int = 15) -> str:
    """GCS署名付きURLを生成"""
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    
    url = blob.generate_signed_url(
        version="v4",
        expiration=timedelta(minutes=expiration_minutes),
        method="GET"
    )
    return url
 
# 署名付きURLでプライベートファイルを処理
signed_url = generate_signed_url(
    "my-private-bucket", 
    "reports/quarterly-analysis-q1-2026.pdf"
)
 
model = genai.GenerativeModel("gemini-2.5-pro")
response = model.generate_content([
    "この四半期レポートから、売上トレンドとリスク要因を抽出してください。",
    {"file_uri": signed_url,
     "mime_type": "application/pdf"}
])
 
print(response.text)
# 出力例:
# ## Q1 2026 四半期分析
# ### 売上トレンド
# - 前年同期比 +23% の成長...
# ### 識別されたリスク要因
# 1. サプライチェーンの遅延リスク...

AWS S3 署名付きURLとの連携

マルチクラウド環境では、S3 の Pre-signed URL も同様に利用できます。

import boto3
from botocore.config import Config
import google.generativeai as genai
 
def get_s3_presigned_url(bucket: str, key: str, 
                          expiration: int = 900) -> str:
    """AWS S3 署名付きURLを生成"""
    s3_client = boto3.client(
        's3',
        config=Config(signature_version='s3v4'),
        region_name='ap-northeast-1'
    )
    url = s3_client.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket, 'Key': key},
        ExpiresIn=expiration
    )
    return url
 
# S3上の動画ファイルをGeminiで解析
s3_url = get_s3_presigned_url(
    "my-media-bucket",
    "recordings/meeting-2026-04-01.mp4"
)
 
model = genai.GenerativeModel("gemini-2.5-pro")
response = model.generate_content([
    "この会議録画から議事録を作成してください。"
    "発言者ごとにまとめ、アクションアイテムを抽出してください。",
    {"file_uri": s3_url,
     "mime_type": "video/mp4"}
])
 
print(response.text)

バッチファイル処理パイプラインの設計

アーキテクチャ全体図

大量のファイルを効率的に処理するには、単発のAPI呼び出しではなくパイプライン設計が必要です。以下に、GCS + Cloud Functions + Gemini API を組み合わせた本番パイプラインの構成を示します。

処理フロー:

ファイルがGCSバケットにアップロードされる
Cloud Functions（またはCloud Run）がイベントトリガーで起動
ファイルのメタデータ検証（サイズ・形式チェック）
Gemini API にGCS URIを渡して処理
結果をFirestore/BigQueryに格納
エラー時はDead Letter Queueに退避

Cloud Functions によるイベント駆動処理

import functions_framework
import google.generativeai as genai
from google.cloud import firestore
import json
import logging
 
logger = logging.getLogger(__name__)
 
# Gemini モデルをモジュールレベルで初期化（コールドスタート最適化）
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-flash")
db = firestore.Client()
 
# ファイル形式とMIMEタイプのマッピング
MIME_MAP = {
    ".pdf": "application/pdf",
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".mp4": "video/mp4",
    ".mp3": "audio/mpeg",
    ".wav": "audio/wav",
}
 
@functions_framework.cloud_event
def process_uploaded_file(cloud_event):
    """GCSアップロードイベントで自動起動するファイル処理関数"""
    data = cloud_event.data
    bucket_name = data["bucket"]
    file_name = data["name"]
    file_size = int(data.get("size", 0))
    
    # バリデーション
    if file_size > 100 * 1024 * 1024:  # 100MB
        logger.error(f"File too large: {file_name} ({file_size} bytes)")
        return {"status": "error", "reason": "file_too_large"}
    
    # 拡張子からMIMEタイプを判定
    ext = "." + file_name.rsplit(".", 1)[-1].lower()
    mime_type = MIME_MAP.get(ext)
    if not mime_type:
        logger.warning(f"Unsupported file type: {ext}")
        return {"status": "skipped", "reason": "unsupported_type"}
    
    # GCS URI を構築
    gcs_uri = f"gs://{bucket_name}/{file_name}"
    
    try:
        # Gemini API で処理
        response = model.generate_content([
            "このファイルの内容を分析し、以下のJSON形式で結果を返してください: "
            '{"summary": "概要", "key_points": ["要点1", "要点2"], '
            '"category": "カテゴリ", "language": "検出言語"}',
            {"file_uri": gcs_uri, "mime_type": mime_type}
        ])
        
        # 結果をFirestoreに保存
        result = json.loads(response.text)
        doc_ref = db.collection("processed_files").document()
        doc_ref.set({
            "file_uri": gcs_uri,
            "file_name": file_name,
            "file_size": file_size,
            "mime_type": mime_type,
            "analysis": result,
            "processed_at": firestore.SERVER_TIMESTAMP,
            "status": "completed"
        })
        
        logger.info(f"Successfully processed: {file_name}")
        return {"status": "success", "doc_id": doc_ref.id}
        
    except Exception as e:
        logger.error(f"Processing failed for {file_name}: {e}")
        # Dead Letter Queue に退避
        db.collection("processing_errors").add({
            "file_uri": gcs_uri,
            "error": str(e),
            "timestamp": firestore.SERVER_TIMESTAMP,
            "retry_count": 0
        })
        return {"status": "error", "reason": str(e)}

エラーハンドリングとリトライ戦略

大容量ファイル特有のエラーパターン

100MBファイルの処理では、従来の小さなファイルでは遭遇しなかったエラーパターンが発生します。

タイムアウトエラー: 大容量PDFや長時間動画の解析は処理時間が長くなります。デフォルトのタイムアウト（60秒）では不足する場合があるため、適切な値に設定する必要があります。

レート制限: 大容量ファイルはトークン消費量が大きく、RPM/TPM制限に到達しやすくなります。

部分的な処理失敗: ファイルの一部が破損している場合、全体の処理が失敗するのではなく、解析可能な部分のみが返されることがあります。

Exponential Backoff 付きリトライ実装

import time
import random
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
 
def process_with_retry(model, contents, max_retries=5, 
                        base_delay=2.0, max_delay=60.0):
    """Exponential Backoff + Jitter によるリトライ処理"""
    
    for attempt in range(max_retries):
        try:
            response = model.generate_content(
                contents,
                request_options={"timeout": 300}  # 5分タイムアウト
            )
            return response
            
        except google_exceptions.ResourceExhausted as e:
            # レート制限 — 必ずリトライ
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            wait_time = delay + jitter
            print(f"Rate limited (attempt {attempt+1}/{max_retries}). "
                  f"Waiting {wait_time:.1f}s...")
            time.sleep(wait_time)
            
        except google_exceptions.DeadlineExceeded as e:
            # タイムアウト — ファイルが大きすぎる可能性
            if attempt < max_retries - 1:
                print(f"Timeout (attempt {attempt+1}). "
                      f"Consider using a faster model or smaller file.")
                time.sleep(base_delay * (2 ** attempt))
            else:
                raise
                
        except google_exceptions.InvalidArgument as e:
            # ファイル形式エラーなど — リトライ不要
            raise
    
    raise Exception(f"Max retries ({max_retries}) exceeded")
 
# 使用例
model = genai.GenerativeModel("gemini-2.5-pro")
response = process_with_retry(model, [
    "この100ページのPDFを章ごとに要約してください。",
    {"file_uri": "gs://my-bucket/large-document.pdf",
     "mime_type": "application/pdf"}
])

コスト最適化戦略

モデル選択によるコスト制御

大容量ファイルの処理はトークン消費が大きいため、モデル選択がコストに直結します。コスト最適化の基本戦略については「Gemini APIコスト完全最適化ガイド」も併せてご覧ください。

Gemini 2.5 Flash: 高速・低コスト。定型的なファイル分類、メタデータ抽出、簡易要約に最適
Gemini 2.5 Pro: 高精度。複雑な分析、多言語ドキュメント処理、詳細なレポート生成に使用
Flex Tier: 非リアルタイムのバッチ処理にはFlexティアを使うことで、最大50%のコスト削減が可能

2段階処理パターン

コストを最適化するベストプラクティスは、2段階処理です。

import google.generativeai as genai
 
# Stage 1: Flash で高速トリアージ（低コスト）
flash_model = genai.GenerativeModel("gemini-2.5-flash")
 
triage_response = flash_model.generate_content([
    "このファイルを分析し、以下を判定してください: "
    '{"complexity": "low/medium/high", "language": "言語", '
    '"estimated_pages": 数値, "requires_deep_analysis": true/false}',
    {"file_uri": "gs://my-bucket/report.pdf",
     "mime_type": "application/pdf"}
])
 
import json
triage = json.loads(triage_response.text)
 
# Stage 2: 複雑なファイルのみ Pro で深掘り分析
if triage.get("requires_deep_analysis"):
    pro_model = genai.GenerativeModel("gemini-2.5-pro")
    detailed_response = pro_model.generate_content([
        "この文書の詳細な分析レポートを作成してください。"
        "法的リスク、財務上の注意点、技術的課題を網羅的に抽出してください。",
        {"file_uri": "gs://my-bucket/report.pdf",
         "mime_type": "application/pdf"}
    ])
    print(detailed_response.text)
else:
    # Flash の結果で十分
    print(f"Simple file — Flash result used. Language: {triage['language']}")

Context Caching との併用

同じファイルに対して複数の質問を投げる場合、Context Caching を使うとキャッシュヒット分のトークンコストを大幅に削減できます。

import google.generativeai as genai
from google.generativeai import caching
import datetime
 
# キャッシュを作成（最低32,768トークン以上のコンテンツが必要）
cache = caching.CachedContent.create(
    model="models/gemini-2.5-flash",
    display_name="quarterly-report-q1",
    contents=[
        {"file_uri": "gs://my-bucket/q1-report-2026.pdf",
         "mime_type": "application/pdf"}
    ],
    ttl=datetime.timedelta(hours=2)  # 2時間有効
)
 
# キャッシュを使って複数の質問を低コストで実行
cached_model = genai.GenerativeModel.from_cached_content(cache)
 
questions = [
    "売上の前年比成長率を教えてください",
    "最もリスクの高い事業領域はどこですか",
    "次四半期の予測に影響する外部要因は何ですか"
]
 
for q in questions:
    response = cached_model.generate_content(q)
    print(f"Q: {q}")
    print(f"A: {response.text}\n")
 
# 完了後にキャッシュを削除
cache.delete()

セキュリティとアクセス制御

署名付きURLのベストプラクティス

署名付きURLを使う際のセキュリティ上の注意点をまとめます。

有効期限を最短に設定する: ファイル処理に必要な最低限の時間（通常15〜30分）に設定し、不要なアクセスリスクを最小化します。

IAMロールの最小権限原則: Cloud Functions のサービスアカウントには storage.objects.get のみを付与し、書き込み権限は分離します。

VPC Service Controls: 本番環境ではVPCサービスコントロールを設定し、Gemini API へのアクセスを許可されたネットワークからに限定します。

# セキュアな署名付きURL生成の例
from google.cloud import storage
from datetime import timedelta
 
def create_secure_signed_url(bucket_name: str, blob_name: str) -> str:
    """最小権限の署名付きURLを生成"""
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    
    url = blob.generate_signed_url(
        version="v4",
        expiration=timedelta(minutes=15),  # 最短の有効期限
        method="GET",
        # コンテンツタイプを制限
        response_type="application/pdf",
        # IP制限（オプション）
        # headers={"X-Goog-Content-SHA256": file_hash}
    )
    return url

本番運用でのモニタリング

処理メトリクスの収集

大容量ファイルパイプラインの安定運用には、以下のメトリクスを常時監視する点が肝心です。

import time
from google.cloud import monitoring_v3
from google.protobuf import timestamp_pb2
 
class PipelineMetrics:
    """パイプラインメトリクスの収集・送信"""
    
    def __init__(self, project_id: str):
        self.client = monitoring_v3.MetricServiceClient()
        self.project_name = f"projects/{project_id}"
    
    def record_processing(self, file_size_mb: float, 
                           processing_time_s: float,
                           model: str, success: bool):
        """処理結果のメトリクスを記録"""
        series = monitoring_v3.TimeSeries()
        series.metric.type = "custom.googleapis.com/gemini/file_processing"
        series.metric.labels["model"] = model
        series.metric.labels["success"] = str(success).lower()
        series.resource.type = "global"
        
        point = monitoring_v3.Point()
        point.value.double_value = processing_time_s
        now = time.time()
        point.interval.end_time.seconds = int(now)
        series.points.append(point)
        
        self.client.create_time_series(
            name=self.project_name,
            time_series=[series]
        )
 
# 使用例
metrics = PipelineMetrics("my-gcp-project")
 
start = time.time()
response = model.generate_content([...])
elapsed = time.time() - start
 
metrics.record_processing(
    file_size_mb=85.2,
    processing_time_s=elapsed,
    model="gemini-2.5-flash",
    success=True
)

まとめ

Gemini API の Cloud Storage 統合と100MBファイル対応は、マルチモーダルAIパイプラインの実用性を大きく引き上げる変更です。GCS URI による直接参照、署名付きURLによるマルチクラウド連携、2段階処理によるコスト最適化、そしてイベント駆動のバッチ処理——これらを組み合わせることで、従来は前処理に多大な工数がかかっていた大容量ファイルのAI処理を、シンプルかつ堅牢なパイプラインとして実現できます。

特に、FlexティアとContext Cachingの併用は、大量のドキュメント処理コストを劇的に削減する強力なパターンです。まずは小さなプロトタイプから始めて、本番環境に段階的にスケールしていくことをおすすめします。

GCS・BigQuery・Cloud Functions を組み合わせたデータパイプラインの設計思想を体系的に学べます。