マルチモーダルRAG実装ガイド：画像・図表検索の仕組みとPythonコード

import logging
import os
import shutil
import tempfile
from pathlib import Path
from typing import List, Optional, Tuple

import chromadb
from pdf2image import convert_from_path
from PIL import Image
from sentence_transformers import SentenceTransformer, util

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class MultimodalRAG:
    def __init__(self, collection_name: str = "multimodal_docs"):
        """
        マルチモーダルRAGシステムの初期化。
        CLIPモデルとChromaDBをセットアップします。
        """
        try:
            logger.info("モデルとデータベースを初期化中...")
            # 画像とテキストの両方を扱えるCLIPモデルのロード
            # 日本語対応が必要な場合は 'paraphrase-multilingual-clip-ViT-B-32' などを検討
            self.model = SentenceTransformer('clip-ViT-B-32')
            logger.info(f"モデルロード完了: {self.model}")
            
            # ChromaDBのセットアップ（永続化の場合はパスを指定）
            self.chroma_client = chromadb.Client()
            self.collection = self.chroma_client.get_or_create_collection(
                name=collection_name,
                metadata={"hnsw:space": "cosine"}
            )
            logger.info("データベース接続完了")
            
        except Exception as e:
            logger.error(f"初期化中にエラーが発生しました: {e}")
            raise

    def _extract_images_from_pdf(self, pdf_path: str) -> List[Tuple[Image.Image, int]]:
        """
        PDFから画像を抽出するヘルパーメソッド。
        pdf2imageを使用してPDFをページごとに画像化します。
        ※本来はレイアウト解析ライブラリ（Unstructuredなど）を使って図表部分のみ切り出すのが理想ですが、
           ここでは簡易的にページ全体を画像として扱います。
        """
        images = []
        try:
            logger.info(f"PDFから画像を抽出中: {pdf_path}")
            # PDFを画像リストに変換
            pil_images = convert_from_path(pdf_path)
            
            for i, img in enumerate(pil_images):
                images.append((img, i + 1)) # (画像オブジェクト, ページ番号)
            
            logger.info(f"{len(images)} ページ分の画像を抽出しました")
            return images
        except Exception as e:
            logger.error(f"画像抽出エラー: {e}")
            return []

    def index_document(self, pdf_path: str, doc_id: str):
        """
        ドキュメントをインデックス化する。
        テキストと画像を抽出し、それぞれベクトル化してDBに保存します。
        """
        if not os.path.exists(pdf_path):
            logger.error(f"ファイルが見つかりません: {pdf_path}")
            return

        try:
            # 1. 画像の抽出とエンベディング
            images = self._extract_images_from_pdf(pdf_path)
            
            image_embeddings = []
            image_ids = []
            image_metadatas = []
            
            for img, page_num in images:
                # 画像を一時保存してパスを取得（メタデータ用）
                # 実際運用ではS3等のストレージパスを格納します
                temp_dir = tempfile.mkdtemp()
                img_path = os.path.join(temp_dir, f"page_{page_num}.png")
                img.save(img_path)
                
                # 画像のエンベディング生成
                emb = self.model.encode(img)
                image_embeddings.append(emb.tolist())
                image_ids.append(f"{doc_id}_img_page_{page_num}")
                image_metadatas.append({
                    "type": "image",
                    "source": pdf_path,
                    "page": page_num,
                    "doc_id": doc_id
                })
                
                # クリーンアップ
                shutil.rmtree(temp_dir)

            # 2. テキストの抽出とエンベディング（簡易化のためダミーテキストを生成）
            # 実際にはPyPDF2などでテキスト抽出を行います
            text_chunks = [
                f"This is a text summary from page {i+1} of {pdf_path}."
                for i in range(len(images))
            ]
            
            text_embeddings = self.model.encode(text_chunks)
            text_ids = [f"{doc_id}_text_page_{i+1}" for i in range(len(text_chunks))]
            text_metadatas = [
                {"type": "text", "source": pdf_path, "page": i+1, "doc_id": doc_id}
                for i in range(len(text_chunks))
            ]

            # 3. データベースへの追加
            if image_embeddings:
                self.collection.add(
                    ids=image_ids,
                    embeddings=image_embeddings,
                    metadatas=image_metadatas
                )
            
            if text_embeddings:
                self.collection.add(
                    ids=text_ids,
                    embeddings=text_embeddings,
                    metadatas=text_metadatas
                )
            
            logger.info(f"ドキュメント {doc_id} のインデックス化が完了しました")

        except Exception as e:
            logger.error(f"インデックス化中に予期せぬエラーが発生: {e}")
            raise

    def search(self, query: str, n_results: int = 3) -> dict:
        """
        クエリに関連するテキストおよび画像を検索する。
        """
        try:
            logger.info(f"検索クエリ実行: {query}")
            query_embedding = self.model.encode(query).tolist()
            
            results = self.collection.query(
                query_embeddings=[query_embedding],
                n_results=n_results
            )
            
            return results
        except Exception as e:
            logger.error(f"検索中にエラーが発生: {e}")
            return {}

# 実行例
if __name__ == "__main__":
    # 初期化
    rag = MultimodalRAG()
    
    # ダミーのPDFファイルパス（実際には存在するファイルを指定してください）
    # ここではエラーハンドリングの動作確認のため、存在しないパスを指定した場合の挙動も想定
    dummy_pdf_path = "sample_report.pdf"
    
    # PDFが存在しない場合のフォールバック処理（デモ用）
    if not os.path.exists(dummy_pdf_path):
        logger.warning("サンプルPDFが見つかりません。処理をスキップします。")
        # 本来はここでサンプルデータを作成するなどの処理を入れる
    else:
        # ドキュメントの登録
        rag.index_document(dummy_pdf_path, doc_id="doc_001")
        
        # 検索の実行
        query = "グラフが上昇しているページはどれですか？"
        search_results = rag.search(query)
        
        logger.info(f"検索結果: {search_results}")