マルチモーダルRAG実践：クロスモーダル検索システムを構築する5つのコア技術
はじめに
こんなシーンを想像してみてください：ECサイトのカスタマーサポートシステムで、ユーザーが商品画像をアップロードして「これ、別の色はありますか？」と尋ねます。しかし、従来のRAGシステムはテキストのナレッジベースしか検索できず、画像の内容を全く理解できません。サポート担当者は手動で商品カタログを探すしかなく、効率は極めて低いです。これが純テキストRAGの致命的な弱点——モーダルの壁を越えられないということです。
マルチモーダルRAG（Multimodal Retrieval-Augmented Generation）はまさにこの問題を解決するために生まれました。検索システムがテキスト、画像、動画など複数のモーダルを同時に理解できるようにし、「テキストで画像を検索、画像でテキストを検索、動画でテキストを検索」というクロスモーダル検索能力を実現します。2026年、CLIPやSigLIPなどの視覚言語モデルの成熟に伴い、マルチモーダルRAGは実験から本番環境へと移行しました。
本記事では、5つのコア技術を出発点として、ゼロから本番級のクロスモーダル検索システムを構築する方法を解説します。
コア概念クイックリファレンス
| 概念 | 英語 | 説明 |
|---|---|---|
| マルチモーダルRAG | Multimodal RAG | テキスト、画像、動画など複数のモーダルを同時に処理する検索拡張生成システム |
| CLIP | Contrastive Language-Image Pre-training | OpenAIが提案した対比学習による画像テキスト事前学習モデル。画像とテキストを同じベクトル空間にマッピング |
| クロスモーダル検索 | Cross-Modal Retrieval | あるモーダルのクエリで別のモーダルのコンテンツを検索。例：テキストで画像を検索 |
| マルチモーダルEmbedding | Multimodal Embedding | 異なるモーダルのデータを統一されたベクトル表現空間にマッピング |
| 視覚言語モデル | Vision-Language Model | 視覚と言語情報を同時に理解するディープラーニングモデル |
| マルチモーダルチャンキング | Multimodal Chunking | 画像とテキストを含むドキュメントをモーダルごとにインテリジェントに分割・関連付け |
| レイトインタラクションモデル | Late Interaction Model | ColBERTなど、トークンレベルでインタラクションマッチングを行う検索モデル |
問題分析：マルチモーダルRAGの5つの課題
1. モーダルアライメントの難しさ
異なるモーダルのデータ分布は大きく異なります——テキストは離散記号列、画像は連続ピクセル行列、動画にはさらに時間次元が加わります。これらを同じベクトル空間で意味的にアライメントする方法が、マルチモーダルRAGの第一の課題です。単純な結合融合は情報損失を招きやすく、深いアライメントには大量のペア訓練データが必要です。
2. 計算リソースの消費
マルチモーダルモデルのパラメータ数は純テキストモデルを遥かに超えます。CLIP ViT-L/14は約428Mパラメータ、1枚の画像処理に約50ms（GPU）を要します。動画理解はフレームごとの処理が必要で、5分の動画は900フレームの処理が必要となり、計算コストが指数関数的に増大します。
3. チャンキング戦略の複雑さ
純テキストドキュメントは段落ごとに分割すれば済みますが、PDFドキュメント内の画像とテキストの関連性をどう維持するか？画像とテキストが混在するテーブルを分割後、セマンティクスは完全か？マルチモーダルチャンキングはレイアウトとセマンティックな一貫性の両方を考慮する必要があります。
4. 検索精度のボトルネック
クロスモーダル検索には本質的にセマンティックギャップが存在します——ユーザーが自然言語で記述したクエリと画像の視覚的特徴の間には大きな隔たりがあります。「赤いワンピース」というテキストベクトルと実際の赤いワンピース画像のベクトルの類似度は高くない可能性があり、より精緻なリランキング戦略が必要です。
5. レイテンシ制御の困難さ
マルチモーダルRAGのエンドツーエンドレイテンシには、画像エンコーディング（50-200ms）、ベクトル検索（10-50ms）、リランキング（100-300ms）、LLM生成（500-2000ms）が含まれます。リアルタイム応答が求められるシナリオでは、精度とレイテンシのバランスが本番デプロイメントの鍵となります。
技術1：CLIPマルチモーダルEmbedding
CLIPはマルチモーダルRAGの基盤モデルです。対比学習により画像とテキストを同じベクトル空間にマッピングし、「赤いワンピース」のテキストベクトルと赤いワンピース画像のベクトルが空間内で近くなるようにします。
import torch
from transformers import CLIPModel, CLIPProcessor
from PIL import Image
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
image = Image.open("product.jpg")
inputs = processor(
text=["赤いワンピース", "青いジーンズ"],
images=image,
return_tensors="pt",
padding=True
)
outputs = model(**inputs)
logits = outputs.logits_per_image
probs = logits.softmax(dim=1)
print(f"マッチ確率: {probs}")
image_embedding = outputs.image_embeds
text_embedding = outputs.text_embeds
print(f"画像ベクトル次元: {image_embedding.shape}")
print(f"テキストベクトル次元: {text_embedding.shape}")
def get_image_embedding(image_path: str) -> list[float]:
image = Image.open(image_path)
inputs = processor(images=image, return_tensors="pt")
with torch.no_grad():
embedding = model.get_image_features(**inputs)
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
return embedding.squeeze().tolist()
def get_text_embedding(text: str) -> list[float]:
inputs = processor(text=[text], return_tensors="pt", padding=True)
with torch.no_grad():
embedding = model.get_text_features(**inputs)
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
return embedding.squeeze().tolist()
重要ポイント：必ずベクトルをL2正規化（embedding / embedding.norm()）してください。これによりコサイン類似度が内積と等価になり、検索効率が大幅に向上します。
技術2：画像ドキュメントのチャンキングとインデキシング
実際のシナリオでは、ナレッジベースはPDFやPPTなどの画像テキスト混在ドキュメントであることが多く、まずマルチモーダルコンテンツを抽出してからインデックスを構築する必要があります。
import fitz
from PIL import Image
from io import BytesIO
from dataclasses import dataclass
from typing import Optional
@dataclass
class MultimodalChunk:
chunkId: str
chunkType: str
content: str
imageBytes: Optional[bytes] = None
pageNumber: int = 0
bbox: Optional[list[float]] = None
def extract_multimodal_chunks(pdf_path: str) -> list[MultimodalChunk]:
doc = fitz.open(pdf_path)
chunks: list[MultimodalChunk] = []
chunkCounter = 0
for pageNum in range(len(doc)):
page = doc[pageNum]
textBlocks = page.get_text("blocks")
imageList = page.get_images(full=True)
for block in textBlocks:
if block[6] == 0:
chunkCounter += 1
chunks.append(MultimodalChunk(
chunkId=f"chunk_{chunkCounter}",
chunkType="text",
content=block[4].strip(),
pageNumber=pageNum + 1,
bbox=list(block[:4])
))
for imgIndex, imgInfo in enumerate(imageList):
xref = imgInfo[0]
baseImage = doc.extract_image(xref)
imageBytes = baseImage["image"]
if len(imageBytes) < 1024:
continue
chunkCounter += 1
chunks.append(MultimodalChunk(
chunkId=f"chunk_{chunkCounter}",
chunkType="image",
content=f"Page {pageNum + 1} Image {imgIndex + 1}",
imageBytes=imageBytes,
pageNumber=pageNum + 1
))
doc.close()
return chunks
def build_multimodal_index(chunks: list[MultimodalChunk]) -> list[dict]:
indexEntries: list[dict] = []
for chunk in chunks:
if chunk.chunkType == "text":
embedding = get_text_embedding(chunk.content[:512])
elif chunk.chunkType == "image" and chunk.imageBytes:
image = Image.open(BytesIO(chunk.imageBytes))
inputs = processor(images=image, return_tensors="pt")
with torch.no_grad():
embedding = model.get_image_features(**inputs)
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
embedding = embedding.squeeze().tolist()
else:
continue
indexEntries.append({
"id": chunk.chunkId,
"vector": embedding,
"payload": {
"type": chunk.chunkType,
"content": chunk.content[:200],
"page": chunk.pageNumber
}
})
return indexEntries
重要ポイント：1KB未満の画像は通常アイコンや装飾なのでフィルタリングしてください。テキストはCLIPの入力制限に合わせて512トークンに切り詰めます。
技術3：クロスモーダル検索とリランキング
Bi-Encoder（CLIPなど）は粗いフィルタリングに適していますが、精度には限界があります。Cross-Encoderリランキングにより検索精度を大幅に向上できます。
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance
client = QdrantClient(":memory:")
client.create_collection(
collection_name="multimodal",
vectors_config=VectorParams(size=512, distance=Distance.COSINE)
)
client.upsert("multimodal", points=[
PointStruct(
id=1,
vector=image_embedding.tolist(),
payload={"type": "image", "src": "img1.jpg", "content": "赤いワンピース商品画像"}
),
PointStruct(
id=2,
vector=text_embedding.tolist(),
payload={"type": "text", "content": "赤いワンピース、サイズS/M/L、価格4990円"}
)
])
results = client.search(
collection_name="multimodal",
query_vector=text_query_vector,
limit=10
)
from sentence_transformers import CrossEncoder
crossEncoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def cross_modal_rerank(
query: str,
candidates: list[dict],
topK: int = 5
) -> list[dict]:
pairs = []
for candidate in candidates:
content = candidate["payload"].get("content", "")
pairs.append([query, content])
scores = crossEncoder.predict(pairs)
for i, candidate in enumerate(candidates):
candidate["rerankScore"] = float(scores[i])
candidates.sort(key=lambda x: x["rerankScore"], reverse=True)
return candidates[:topK]
rerankedResults = cross_modal_rerank("赤いワンピースはいくらですか", [r.dict() for r in results])
for result in rerankedResults:
print(f"タイプ: {result['payload']['type']}, "
f"内容: {result['payload']['content']}, "
f"リランクスコア: {result['rerankScore']:.4f}")
重要ポイント：Cross-Encoderリランキングの候補セットは100件を超えないようにしてください。超えるとレイテンシが高くなりすぎます。本番環境では粗いフィルタリングでtop-50を取得し、リランキングでtop-5を取得することを推奨します。
技術4：動画理解と検索
動画検索のコアはキーフレーム抽出と時系列モデリングです——すべてのフレームが重要なわけではなく、インテリジェントなサンプリングが必要です。
import cv2
import numpy as np
from dataclasses import dataclass
@dataclass
class KeyFrame:
frameIndex: int
timestamp: float
image: Image.Image
similarity: float = 0.0
def extract_key_frames(
videoPath: str,
threshold: float = 0.85,
maxFrames: int = 30
) -> list[KeyFrame]:
cap = cv2.VideoCapture(videoPath)
fps = cap.get(cv2.CAP_PROP_FPS)
totalFrames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
keyFrames: list[KeyFrame] = []
prevHash = None
frameIndex = 0
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
if frameIndex % max(1, int(fps)) != 0:
frameIndex += 1
continue
currentHash = compute_frame_hash(frame)
if prevHash is None or hamming_similarity(prevHash, currentHash) < threshold:
rgbFrame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
pilImage = Image.fromarray(rgbFrame)
keyFrames.append(KeyFrame(
frameIndex=frameIndex,
timestamp=frameIndex / fps,
image=pilImage
))
prevHash = currentHash
if len(keyFrames) >= maxFrames:
break
frameIndex += 1
cap.release()
return keyFrames
def compute_frame_hash(frame: np.ndarray) -> str:
resized = cv2.resize(frame, (16, 16))
gray = cv2.cvtColor(resized, cv2.COLOR_BGR2GRAY)
meanVal = gray.mean()
return "".join(["1" if p > meanVal else "0" for p in gray.flatten()])
def hamming_similarity(hash1: str, hash2: str) -> float:
if len(hash1) != len(hash2):
return 0.0
same = sum(c1 == c2 for c1, c2 in zip(hash1, hash2))
return same / len(hash1)
def index_video_keyframes(keyFrames: list[KeyFrame]) -> list[dict]:
entries: list[dict] = []
for i, kf in enumerate(keyFrames):
inputs = processor(images=kf.image, return_tensors="pt")
with torch.no_grad():
embedding = model.get_image_features(**inputs)
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
entries.append({
"id": i + 1,
"vector": embedding.squeeze().tolist(),
"payload": {
"type": "video_frame",
"timestamp": kf.timestamp,
"frameIndex": kf.frameIndex
}
})
return entries
def search_video(
query: str,
videoCollection: str,
topK: int = 5
) -> list[dict]:
queryVector = get_text_embedding(query)
results = client.search(
collection_name=videoCollection,
query_vector=queryVector,
limit=topK
)
return [
{
"timestamp": r.payload["timestamp"],
"frameIndex": r.payload["frameIndex"],
"score": r.score
}
for r in results
]
重要ポイント：1秒あたり1フレームのサンプリングが最もコストパフォーマンスの高い戦略です。知覚ハッシュの重複排除閾値0.85で90%の冗長フレームをフィルタリングできます。キーフレームの上限を30フレームにすることでインデックスサイズを制御できます。
技術5：マルチモーダルRAG本番デプロイメント
上記の技術をFastAPI + Qdrantの本番級サービスに統合します。
from fastapi import FastAPI, UploadFile, File, Query
from pydantic import BaseModel
app = FastAPI(title="Multimodal RAG Service")
class SearchRequest(BaseModel):
query: str
topK: int = 5
modalityFilter: Optional[str] = None
class SearchResult(BaseModel):
content: str
modality: str
score: float
metadata: dict
@app.post("/index/document")
async def index_document(file: UploadFile = File(...)):
content = await file.read()
if file.filename.endswith(".pdf"):
chunks = extract_multimodal_chunks_from_bytes(content)
else:
return {"error": "Unsupported format"}
entries = build_multimodal_index(chunks)
points = [
PointStruct(id=i, vector=e["vector"], payload=e["payload"])
for i, e in enumerate(entries)
]
client.upsert("multimodal", points=points)
return {"indexed": len(points)}
@app.post("/index/image")
async def index_image(file: UploadFile = File(...)):
image = Image.open(BytesIO(await file.read()))
inputs = processor(images=image, return_tensors="pt")
with torch.no_grad():
embedding = model.get_image_features(**inputs)
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
vector = embedding.squeeze().tolist()
client.upsert("multimodal", points=[
PointStruct(
id=hash(file.filename) % (10 ** 8),
vector=vector,
payload={"type": "image", "src": file.filename}
)
])
return {"status": "indexed"}
@app.post("/search", response_model=list[SearchResult])
async def search(request: SearchRequest):
queryVector = get_text_embedding(request.query)
filterCondition = None
if request.modalityFilter:
from qdrant_client.models import FieldCondition, Filter, MatchValue
filterCondition = Filter(must=[
FieldCondition(key="type", match=MatchValue(value=request.modalityFilter))
])
results = client.search(
collection_name="multimodal",
query_vector=queryVector,
query_filter=filterCondition,
limit=request.topK * 3
)
reranked = cross_modal_rerank(
request.query,
[r.dict() for r in results],
topK=request.topK
)
return [
SearchResult(
content=r["payload"].get("content", ""),
modality=r["payload"].get("type", "unknown"),
score=r["rerankScore"],
metadata=r["payload"]
)
for r in reranked
]
@app.get("/health")
async def health():
return {"status": "healthy", "collection": "multimodal"}
重要ポイント：本番環境では必ずモーダルフィルタリング（modalityFilter）を追加し、画像クエリがテキスト結果を返す（またはその逆）のを防いでください。粗いフィルタリングで3倍の候補を取得してからリランキングするのが、精度とレイテンシの最適なバランスです。
落とし穴ガイド：5つのよくある罠
-
ベクトル正規化の忽略：CLIPの出力ベクトルは正規化されていません。ユークリッド距離をそのまま使うと精度が急落します。必ずL2正規化してコサイン類似度を使用してください。
-
画像前処理の不一致：インデキシング時にPILでRGB変換し、クエリ時にOpenCVでBGR形式で読み込むと、ベクトル空間が不整合になります。前処理パイプラインの統一が鍵です。
-
動画の全フレームインデキシング：5分の動画は9000フレームあります。全量インデキシングはストレージの無駄遣いであり、検索ノイズも極めて大きくなります。キーフレーム抽出が必須です。
-
粗いフィルタリング段階でのCross-Encoder使用：Cross-Encoderはクエリとドキュメントを同時にエンコードする必要があり、計算量はBi-EncoderのN倍です。リランキングにのみ使用し、粗いフィルタリングには使用しないでください。
-
マルチモーダルチャンクの関連性の忽略：PDF内の画像の横にあるキャプションテキストは画像と強く関連しています。チャンキング時には独立して処理するのではなく、関連チャンクとして一緒にインデックスすべきです。
エラートラブルシューティング：10のよくあるエラー
| エラーメッセージ | 原因 | 解決策 |
|---|---|---|
RuntimeError: CUDA out of memory |
バッチ画像エンコーディング時のGPUメモリ不足 | batch_sizeを減らす、またはtorch.no_grad()で計算グラフを解放 |
ValueError: expected 3D tensor, got 4D |
CLIP入力次元エラー、バッチ次元の不一致 | processorの出力を確認、textとimagesの次元が一致しているか確認 |
PIL.UnidentifiedImageError |
画像形式が破損または非対応 | try-exceptを追加、Image.open().verify()で事前検証 |
qdrant_client.http.exceptions.UnexpectedResponse |
ベクトル次元がコレクション設定と不一致 | CLIP出力次元がVectorParams.sizeと一致することを確認（ViT-B/32は512） |
TypeError: expected str, got list |
processorのtextパラメータには文字列リストが必要 | text=["クエリテキスト"]を渡す（text="クエリテキスト"ではなく） |
torch.jit.ScriptModule object has no attribute |
誤ったモデル重みのロード | CLIPModelを使用していることを確認（JITバージョンではないこと） |
ConnectionRefusedError: Qdrant not reachable |
Qdrantサービスが未起動またはポートエラー | docker psでコンテナの動作を確認、デフォルトポート6333 |
UnicodeDecodeError in PDF extraction |
PDFに非UTF-8エンコードテキストが含まれる | fitzのget_text("text")を使用（生バイト読み込みではなく） |
RecursionError in video frame extraction |
動画ファイルの破損による無限読み取り | maxFrames上限とcap.isOpened()の二重チェックを追加 |
Slow query: >5s latency |
ベクトルインデックスでHNSWが未有効 | Qdrantのhnsw_configを設定、m=16、ef_construct=100 |
高度な最適化のヒント
-
ハイブリッド検索戦略：スパース検索（BM25）とデンス検索（CLIPベクトル）を組み合わせ、Reciprocal Rank Fusion（RRF）でランキングを統合することで、単一検索より15-25%のRecall向上が可能です。
-
マルチスケール画像エンコーディング：同じ画像に対して複数のスケール（グローバル+ローカルクロップ）のEmbeddingを生成し、それぞれインデックスして検索時に結果をマージすることで、細粒度検索精度を大幅に向上できます。
-
クエリ拡張：LLMを使ってユーザークエリを複数の記述に拡張（例：「赤いワンピース」→「赤いスカート、赤いレディース服、赤いアパレル」）、それぞれ検索後にマージすることで再現率を向上。
-
非同期エンコーディングパイプライン：画像エンコーディング、テキストエンコーディング、ベクトル書き込みをメッセージキューに入れて非同期処理することで、スループットを3-5倍向上できます。
-
ホットクエリのキャッシュ：高頻度クエリの検索結果をLRUキャッシュし、5分間のTTLを設定することで、80%の重複計算を削減できます。
比較分析：CLIP vs SigLIP vs Jina CLIP vs Cohere Multimodal
| 特徴 | CLIP ViT-B/32 | SigLIP ViT-B/16 | Jina CLIP v2 | Cohere Multimodal v3 |
|---|---|---|---|---|
| ベクトル次元 | 512 | 768 | 1024 | 1024 |
| 画像テキストアライメント | 対比学習 | Sigmoid損失 | 対比学習+ハードネガティブマイニング | 対比学習 |
| 日本語サポート | 普通 | 良好 | 優秀 | 優秀 |
| 推論速度 | 高速（30ms/画像） | 中速（50ms/画像） | 中速（60ms/画像） | 低速（API呼び出し） |
| 長テキストサポート | 77トークン | 64トークン | 8192トークン | 不明 |
| デプロイ方法 | ローカル | ローカル | ローカル/API | APIのみ |
| 適用シーン | 汎用画像テキスト検索 | 高精度画像テキストマッチング | 長文書マルチモーダル検索 | クイック統合 |
| オープンソース | はい | はい | はい | いいえ |
| ライセンス | MIT | Apache 2.0 | Apache 2.0 | 商用 |
選択のアドバイス：日本語シナリオではJina CLIP v2が最適（8192トークンの長テキスト+優秀な日本語サポート）。速度重視ならCLIP ViT-B/32、最高精度ならSigLIP、迅速なプロトタイピングならCohere APIを選択してください。
オンラインツールのおすすめ
-
JSONフォーマッター — マルチモーダルRAGのインデックスデータを扱う際、JSON構造のフォーマットとデバッグが頻繁に必要です。このツールでベクトルインデックスのpayload形式を素早く確認できます。
-
画像圧縮ツール — 画像ナレッジベースを構築する前に、このツールで画像を一括圧縮することで、50-80%のストレージ容量を削減でき、CLIPの検索精度には影響しません。
-
cURL→コード変換ツール — QdrantやCLIPモデルのAPIエンドポイントをデバッグする際、このツールでcURLコマンドをPython/JavaScriptコードに変換し、プロジェクトに素早く統合できます。
まとめと展望
マルチモーダルRAGは情報検索の境界を再構築しています。2026年、純テキスト検索からクロスモーダル理解への飛躍はもはや実験ではなく、本番システムの標準機能となりました。CLIP Embedding、マルチモーダルチャンキング、クロスモーダルリランキング、動画理解、本番デプロイメントの5つのコア技術を習得することで、次世代のインテリジェント検索システムを構築する能力が身につきます。将来、GPT-5レベルの視覚言語モデルの普及に伴い、マルチモーダルRAGは「検索拡張」から「知覚拡張」へと進化し、AIが真に世界を見て理解できるようになるでしょう。
関連記事
ブラウザローカルツールを無料で試す →