◈ API / SDK/2026-03-30上級

Gemini APIマルチモーダルRAGパイプライン構築ガイド — 画像・PDF・動画を横断検索するAIシステムの設計と実装

Gemini 2.5 Pro でテキスト・画像・PDF・動画を横断検索する RAG パイプラインを設計します。ドキュメント処理、Embeddings API によるベクトル化、ベクトルストアと検索エンジン、クエリエンジンの実装から本番のコスト最適化まで体系的に解説します。

gemini-api²⁹³ multimodal²⁴ rag²¹ embeddings¹¹ production¹⁰⁵ advanced¹²

✦ プレミアム記事

検索できない知識は、無いのと同じ

従来のRAG（Retrieval-Augmented Generation）システムはテキストのみを対象としていましたが、実世界のナレッジは多様な形式で存在します。設計書のPDF、ホワイトボードの写真、会議録画の動画、スプレッドシートの図表 — これらを統合的に検索できなければ、AIアシスタントの実用性は限定的です。

Gemini 2.5 Proは、テキスト・画像・PDF・動画・音声を1つのモデルで処理できるマルチモーダルAPIを提供しています。この能力をEmbeddings APIと組み合わせることで、あらゆる形式のドキュメントを統一ベクトル空間で検索できるマルチモーダルRAGパイプラインを構築できます。

ドキュメント処理からベクトルインデックス構築、検索・生成パイプラインまでを、Python の実装コードとともに追っていきます。Function Callingの基本を理解している前提で進めますので、初めての方はそちらを先にご覧ください。

アーキテクチャ設計

マルチモーダルRAGパイプラインは、以下の4つのフェーズで構成します。

Ingest（取り込み）: 各種ファイルを受け取り、処理可能なチャンクに分割する
Embed（ベクトル化）: Gemini Embeddings APIで各チャンクをベクトルに変換する
Index（インデックス）: ベクトルデータベースに格納し、高速検索を可能にする
Query（検索・生成）: ユーザーのクエリに関連するチャンクを検索し、Geminiで回答を生成する

# パイプライン全体の概要
# DocumentProcessor → EmbeddingService → VectorStore → QueryEngine
 
from dataclasses import dataclass
from enum import Enum
 
class DocumentType(Enum):
    TEXT = "text"
    PDF = "pdf"
    IMAGE = "image"
    VIDEO = "video"
 
@dataclass
class DocumentChunk:
    """処理済みドキュメントチャンク"""
    chunk_id: str
    source_file: str
    doc_type: DocumentType
    content_text: str          # テキスト表現（検索用）
    content_description: str   # Geminiによる説明（画像・動画用）
    metadata: dict             # ページ番号、タイムスタンプ等
    embedding: list[float] | None = None

✦

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

この記事の続きを読む

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

この記事で得られること

✦Gemini Embeddings APIを使ったマルチモーダルベクトル検索の設計と実装パターンを習得できる

✦PDF・画像・動画を統一インデックスで管理するドキュメント処理パイプラインの構築方法を理解できる

✦本番環境でのキャッシュ戦略・コスト最適化・スケーリングパターンを具体的に学べる

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

✦

この記事を購入する

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

または

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

ドキュメント処理パイプラインの実装

各ファイル形式に対応したドキュメント処理クラスを実装します。

テキストとPDFの処理

import google.generativeai as genai
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
class DocumentProcessor:
    """マルチモーダルドキュメント処理"""
 
    def __init__(self, model_name: str = "gemini-2.5-pro-preview-05-06"):
        self.model = genai.GenerativeModel(model_name)
        self.chunk_size = 1000  # テキストチャンクのサイズ（文字数）
        self.chunk_overlap = 200
 
    def process_text(self, text: str, source: str) -> list[DocumentChunk]:
        """テキストをオーバーラップ付きチャンクに分割"""
        chunks = []
        start = 0
        chunk_idx = 0
 
        while start < len(text):
            end = min(start + self.chunk_size, len(text))
            chunk_text = text[start:end]
 
            chunks.append(DocumentChunk(
                chunk_id=f"{source}__chunk_{chunk_idx}",
                source_file=source,
                doc_type=DocumentType.TEXT,
                content_text=chunk_text,
                content_description=chunk_text[:200],
                metadata={"chunk_index": chunk_idx, "char_start": start},
            ))
 
            start += self.chunk_size - self.chunk_overlap
            chunk_idx += 1
 
        return chunks
 
    async def process_pdf(self, pdf_path: str) -> list[DocumentChunk]:
        """PDFをGemini Visionで解析してチャンクに分割"""
        pdf_file = genai.upload_file(pdf_path)
 
        # Geminiにページごとの内容を抽出させる
        response = await self.model.generate_content_async([
            pdf_file,
            "このPDFの各ページの内容を、ページ番号付きで詳細に書き起こしてください。"
            "図表がある場合はその内容も説明してください。"
            "各ページは '--- Page N ---' で区切ってください。"
        ])
 
        # ページごとにチャンクを作成
        pages = response.text.split("--- Page ")
        chunks = []
 
        for page in pages:
            if not page.strip():
                continue
 
            # ページ番号を抽出
            lines = page.strip().split("\n", 1)
            page_num = lines[0].replace("---", "").strip()
            page_text = lines[1] if len(lines) > 1 else ""
 
            if page_text.strip():
                chunks.append(DocumentChunk(
                    chunk_id=f"{pdf_path}__page_{page_num}",
                    source_file=pdf_path,
                    doc_type=DocumentType.PDF,
                    content_text=page_text,
                    content_description=page_text[:200],
                    metadata={"page_number": page_num},
                ))
 
        return chunks

画像と動画の処理

    async def process_image(self, image_path: str) -> list[DocumentChunk]:
        """画像をGemini Visionで分析してテキスト表現を生成"""
        image_file = genai.upload_file(image_path)
 
        response = await self.model.generate_content_async([
            image_file,
            "この画像の内容を詳細に説明してください。"
            "テキストが含まれている場合はOCR結果も含めてください。"
            "図表やグラフの場合はデータの要約も行ってください。"
        ])
 
        return [DocumentChunk(
            chunk_id=f"{image_path}__full",
            source_file=image_path,
            doc_type=DocumentType.IMAGE,
            content_text=response.text,
            content_description=response.text[:200],
            metadata={"image_path": image_path},
        )]
 
    async def process_video(
        self, video_path: str, interval_seconds: int = 30
    ) -> list[DocumentChunk]:
        """動画をセグメントごとに分析"""
        video_file = genai.upload_file(video_path)
 
        # 動画全体の要約 + タイムスタンプ付きセグメント分析
        response = await self.model.generate_content_async([
            video_file,
            f"この動画を{interval_seconds}秒ごとのセグメントに分けて、"
            "各セグメントの内容を詳細に説明してください。"
            "形式: '[MM:SS - MM:SS] 説明' で各セグメントを記述してください。"
        ])
 
        # セグメントごとにチャンクを作成
        chunks = []
        segments = response.text.strip().split("\n")
 
        for idx, segment in enumerate(segments):
            if segment.strip():
                chunks.append(DocumentChunk(
                    chunk_id=f"{video_path}__segment_{idx}",
                    source_file=video_path,
                    doc_type=DocumentType.VIDEO,
                    content_text=segment,
                    content_description=segment[:200],
                    metadata={"segment_index": idx},
                ))
 
        return chunks

Embeddings APIによるベクトル化

Gemini Embeddings APIを使って、すべてのチャンクをベクトル空間に変換します。

class EmbeddingService:
    """Gemini Embeddings APIラッパー"""
 
    def __init__(self, model: str = "models/text-embedding-004"):
        self.model = model
        self.batch_size = 100  # APIのバッチ制限
 
    async def embed_chunks(
        self, chunks: list[DocumentChunk]
    ) -> list[DocumentChunk]:
        """チャンクリストにエンベディングを付与"""
        texts = [chunk.content_text for chunk in chunks]
 
        # バッチ処理
        all_embeddings = []
        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i + self.batch_size]
            result = genai.embed_content(
                model=self.model,
                content=batch,
                task_type="retrieval_document",
            )
            all_embeddings.extend(result["embedding"])
 
        # チャンクにエンベディングを付与
        for chunk, embedding in zip(chunks, all_embeddings):
            chunk.embedding = embedding
 
        return chunks
 
    async def embed_query(self, query: str) -> list[float]:
        """検索クエリをエンベディングに変換"""
        result = genai.embed_content(
            model=self.model,
            content=query,
            task_type="retrieval_query",  # クエリ用のタスクタイプ
        )
        return result["embedding"]

ベクトルストアと検索エンジン

コサイン類似度によるベクトル検索を実装します。本番環境ではPinecone、Weaviate、pgvector等の専用ベクトルDBを使用しますが、ここでは原理を理解するためにNumPyで実装します。

import numpy as np
from typing import Optional
 
class VectorStore:
    """インメモリベクトルストア（本番ではpgvector等を使用）"""
 
    def __init__(self):
        self.chunks: list[DocumentChunk] = []
        self.embeddings: np.ndarray | None = None
 
    def add(self, chunks: list[DocumentChunk]):
        """チャンクをストアに追加"""
        self.chunks.extend(chunks)
        vectors = [c.embedding for c in self.chunks if c.embedding]
        self.embeddings = np.array(vectors)
 
    def search(
        self,
        query_embedding: list[float],
        top_k: int = 5,
        doc_type_filter: Optional[DocumentType] = None,
    ) -> list[tuple[DocumentChunk, float]]:
        """コサイン類似度で検索"""
        if self.embeddings is None or len(self.embeddings) == 0:
            return []
 
        query_vec = np.array(query_embedding)
        # コサイン類似度の計算
        similarities = np.dot(self.embeddings, query_vec) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_vec)
        )
 
        # フィルタリング
        results = []
        for idx in np.argsort(similarities)[::-1]:
            chunk = self.chunks[idx]
            if doc_type_filter and chunk.doc_type != doc_type_filter:
                continue
            results.append((chunk, float(similarities[idx])))
            if len(results) >= top_k:
                break
 
        return results

検索・生成パイプライン（RAGクエリエンジン）

検索結果をコンテキストとしてGeminiに渡し、回答を生成します。

class RAGQueryEngine:
    """マルチモーダルRAGクエリエンジン"""
 
    def __init__(
        self,
        embedding_service: EmbeddingService,
        vector_store: VectorStore,
        model_name: str = "gemini-2.5-pro-preview-05-06",
    ):
        self.embedding_service = embedding_service
        self.vector_store = vector_store
        self.model = genai.GenerativeModel(model_name)
 
    async def query(
        self,
        question: str,
        top_k: int = 5,
        doc_type_filter: Optional[DocumentType] = None,
    ) -> dict:
        """質問に対してRAG検索＋生成を実行"""
 
        # Step 1: クエリのエンベディング
        query_embedding = await self.embedding_service.embed_query(question)
 
        # Step 2: ベクトル検索
        results = self.vector_store.search(
            query_embedding, top_k=top_k,
            doc_type_filter=doc_type_filter,
        )
 
        if not results:
            return {
                "answer": "関連するドキュメントが見つかりませんでした。",
                "sources": [],
            }
 
        # Step 3: コンテキスト構築
        context_parts = []
        sources = []
        for chunk, score in results:
            context_parts.append(
                f"[Source: {chunk.source_file} | Type: {chunk.doc_type.value} | "
                f"Score: {score:.3f}]\n{chunk.content_text}"
            )
            sources.append({
                "file": chunk.source_file,
                "type": chunk.doc_type.value,
                "relevance": round(score, 3),
                "excerpt": chunk.content_description,
            })
 
        context = "\n\n---\n\n".join(context_parts)
 
        # Step 4: Geminiで回答生成
        prompt = f"""以下の検索結果を基に、ユーザーの質問に正確に回答してください。
回答に使用した情報源を明記してください。検索結果に含まれない情報で推測する場合は、
その旨を明示してください。
 
【検索結果】
{context}
 
【質問】
{question}"""
 
        response = await self.model.generate_content_async(prompt)
 
        return {
            "answer": response.text,
            "sources": sources,
            "tokens_used": response.usage_metadata.total_token_count,
        }
 
# 使用例
# engine = RAGQueryEngine(embedding_service, vector_store)
# result = await engine.query("先月の売上レポートのグラフで最も成長した地域は？")
# print(result["answer"])
# print(f"参照元: {result['sources']}")

コンテキストキャッシュの活用についてはContext Caching ガイドも参考にしてください。

コスト最適化とスケーリング

バッチ処理とキャッシュ戦略

Embeddings のキャッシュ: 一度生成したエンベディングはDBに永続化し、ドキュメント更新時のみ再計算する
Context Caching の活用: 大量のドキュメントチャンクをコンテキストに含める場合、Gemini の Context Caching API でキャッシュし、繰り返しクエリのコストを削減する
モデルの使い分け: エンベディング生成にはtext-embedding-004、回答生成の初回フィルタリングにはgemini-2.5-flash、最終回答にはgemini-2.5-proと段階的に使い分ける

# コスト最適化: Flash で初回フィルタリング → Pro で最終回答
async def cost_optimized_query(engine: RAGQueryEngine, question: str):
    # Step 1: Flashで関連性の高い結果だけ絞り込み
    flash_model = genai.GenerativeModel("gemini-2.5-flash-preview-04-17")
    results = engine.vector_store.search(
        await engine.embedding_service.embed_query(question),
        top_k=10  # 多めに取得
    )
 
    # Step 2: Flashで関連性を再評価（安価）
    rerank_prompt = f"以下のドキュメント断片から、質問「{question}」に最も関連するものを上位3つ選んでください。"
    # ... 省略
 
    # Step 3: Proで最終回答生成（高品質・高コスト）
    # 絞り込んだ3件のみをコンテキストとして使用

個人開発者の視点から（実体験メモ）

ここまでの要点

Gemini APIのマルチモーダル能力を活用すれば、テキスト・画像・PDF・動画を統合的に検索・分析できるRAGパイプラインを構築できます。本記事で解説した4フェーズのアーキテクチャ（Ingest → Embed → Index → Query）と、コスト最適化のためのモデル段階使い分け戦略を組み合わせることで、実用的かつスケーラブルなシステムを実現できます。

まずはテキストとPDFだけの小規模なパイプラインから始め、画像・動画と段階的に対応ファイル形式を拡張していくアプローチがおすすめです。Gemini 2.5 Pro Extended Thinkingの概要もあわせてご参照ください。

マルチモーダルAIの基盤技術についてさらに深く

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

Gemini Lab は広告なしで運営しており、サーバー費用などの運営コストはメンバーシップのご支援で賄っています。実装コード・ベンチマーク・本番設計パターンなど、実務でお役立ていただける記事を毎日更新しています。もし読んでよかったと感じていただけましたら、ぜひご覧ください。

✦コピー&ペーストで使える実装コード付き
✦毎日新しい上級ガイドを追加
✦¥580/月または ¥1,480 の永久アクセス

メンバーシップを見る →