CLIPによるマルチモーダルRAGを構築してみた
はじめに
こんにちは、Timelabで Lynxというカレンダーサービスを開発している諸岡(@hakoten)です。
この記事は、RAGの種類の一つである「マルチモーダルRAG」を具体的なサンプルを用いて試したものになります。
精度や運用の話には触れておらず、あくまでアプローチの入門記事であるころご理解ください。
マルチモーダルRAGとは
従来のベクトルデータベースを使ったRAGでは、テキストベースを中心に検索を行いますが、実際のユースケースでは画像、PDF、音声など、さまざまな形式のデータを扱う必要があります。
この際に、画像やPDFなどを統合的に検索できる仕組みを、マルチモーダルRAGと呼びます。
マルチモーダルRAGの2つのアプローチ
現在のマルチモーダルRAGを実装する方法は、大きく分けて2つのアプローチがあります。
今回は、次で紹介する共有ベクトル空間方式で試してみました。
A. 共有ベクトル空間方式 (今回採用した方式)
テキストと画像などの異なるモーダリティを、同一のベクトル空間に埋め込む方式です。
メリット:
- テキストと画像を直接比較できる
- "cat" というテキストで猫の画像を検索可能
- モダリティ間の意味的な関連性を保持
デメリット:
- 専用のマルチモーダルモデル(CLIP等)が必要
- モデルの事前学習データに依存
B. 単一モーダリティ(テキスト)変換方式
画像などの複数モーダリティのデータを、一旦テキストに変換してから処理する方式です。
メリット:
- 既存のテキストRAGの仕組みをそのまま使える
- 実装がシンプル
- テキストベースの検索技術(BM25等)と組み合わせやすい
デメリット:
- 画像→テキスト変換時に情報が失われる可能性
- 細かい視覚的特徴が失われる
環境
役割 | 技術 | 説明 |
---|---|---|
画像・テキスト埋め込み | CLIP (ViT-B-32) | OpenAI提供の事前学習済みモデル。画像とテキストを同じベクトル空間へ保存するために使用。 |
ベクトルデータベース | Milvus | 画像、テキストのベクトル格納用データベース。 |
回答生成モデル | GPT-5-nano | RAGの検索結果を用いた回答生成LLMモデル |
言語・フレームワーク | Python + open-clip | CLIPを柔軟に扱うためのライブラリであるOpenCLIPを使用。 Python: v3.12, open-clip-torch: 2.20.0 |
共有ベクトルを作成するためのエンベッドモデルには、CLIPというモデルを使っています。
ベクトルデータベースについては、このサンプルにおいて特に制限や選定理由などはありませんが、オープンソースのベクトルデータベースであり、スケーラビリティに優れているMilvusを採用しました(今回のサンプルであれば、ChromaDBやQdrantなどでも問題ありません)。
回答生成については、今回は特に重視していないため、OpenAIの小さいモデルを使用しています。
CLIPとは
今回は、CLIPというモデルを使って、画像とテキストを同一ベクトル空間に保存する方式を試しています。
CLIP(Contrastive Language-Image Pre-training)は、OpenAIが2021年に発表したマルチモーダル学習モデルです。
CLIPの特徴は、画像と自然言語テキストを同時に学習することで、両者を同じ「意味空間」にマッピングできる点です。つまり、「犬の写真」と「a photo of a dog」というテキストを同じように理解し、両者を関連付けて扱えるモデルになります。
CLIPについての簡単な図
CLIPは、画像とテキストを同じベクトル空間に埋め込むように学習されています。
事前準備
ここからは、CLIPを使ったらRAGの実装に入っていきます。まずは、事前準備としてベクトルデータベースのDocker構築と、サンプル画像を準備します。
ベクトルデータベース(Milvus)の構築
まずは、ベクトルデータベースの環境をDockerで用意し、docker composeで立ち上げておきます。
compose.yaml
services:
etcd:
container_name: milvus-etcd
image: quay.io/coreos/etcd:v3.5.5
environment:
- ETCD_AUTO_COMPACTION_MODE=revision
- ETCD_AUTO_COMPACTION_RETENTION=1000
- ETCD_QUOTA_BACKEND_BYTES=4294967296
- ETCD_SNAPSHOT_COUNT=50000
volumes:
- ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd
command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
healthcheck:
test: ["CMD", "etcdctl", "endpoint", "health"]
interval: 30s
timeout: 20s
retries: 3
minio:
container_name: milvus-minio
image: minio/minio:RELEASE.2023-03-20T20-16-18Z
environment:
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
ports:
- "9001:9001"
- "9000:9000"
volumes:
- ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data
command: minio server /minio_data --console-address ":9001"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
interval: 30s
timeout: 20s
retries: 3
milvus:
container_name: milvus-standalone
image: milvusdb/milvus:v2.3.3
command: ["milvus", "run", "standalone"]
security_opt:
- seccomp:unconfined
environment:
ETCD_ENDPOINTS: etcd:2379
MINIO_ADDRESS: minio:9000
volumes:
- ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
interval: 30s
start_period: 90s
timeout: 20s
retries: 3
ports:
- "19530:19530"
- "9091:9091"
depends_on:
- "etcd"
- "minio"
attu:
container_name: milvus-attu
image: zilliz/attu:latest
environment:
MILVUS_URL: milvus-standalone:19530
ports:
- "8000:3000"
depends_on:
- "milvus"
networks:
default:
name: milvus
サンプル画像の準備
ベクトルの埋め込みに使う画像のサンプルは、Unsplashからいくつか用意しました。
画像をベクトルデータベースに格納する
CLIPモデルの初期化
import open_clip
import torch
model_name = "ViT-B-32" # Vision Transformer Base、パッチサイズ32
device = "cuda" if torch.cuda.is_available() else "cpu" # GPU利用可能ならGPU、なければCPU
model, _, preprocess = open_clip.create_model_and_transforms(
model_name, pretrained="openai", device=device
) # モデル本体、前処理関数を作成(OpenAI学習済み重み使用)
# テキストを数値ID列に変換するトークナイザー
tokenizer = open_clip.get_tokenizer(model_name)
# 評価モードに設定(Dropout無効化、決定論的動作)
model.eval()
まずは、画像をベクトル化するためのCLIPモデルを初期化します。
前述の通り、CLIPモデルを利用するために open_clip
ライブラリを使用しています。
モデルは、今回は学習ではなく評価のみで使うため、eval()
で評価モードに設定します。
ベクターデータベース(Milvus)へ画像のベクトルを保存
まず、Milvusのスキーマ定義とコレクションを作成します。
# Milvusサーバーに接続(ローカルホストのデフォルトポート19530を使用)
connections.connect("default", host="localhost", port="19530")
# コレクションのスキーマを定義(データベースのテーブル構造に相当)
fields = [
# id: 各レコードの一意識別子(自動採番される主キー)
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
# image_path: 画像ファイルのパス(最大500文字の文字列)
FieldSchema(name="image_path", dtype=DataType.VARCHAR, max_length=500),
# data_type: データの種類を示すラベル(例: "image", "text"など、最大50文字)
FieldSchema(name="data_type", dtype=DataType.VARCHAR, max_length=50),
# image_embedding: CLIPモデルで生成された画像の特徴ベクトル
# CLIP ViT-B-32モデルは512次元のベクトルを出力するため、dim=512を指定
FieldSchema(name="image_embedding", dtype=DataType.FLOAT_VECTOR, dim=512),
]
# スキーマとコレクション作成
# descriptionには用途を記載
schema = CollectionSchema(fields, description="Multimodal RAG image collection schema")
collection = Collection("test_rag_image_collection", schema)
# ベクトルインデックスの作成
# IVF_FLAT: Inverted File Systemを使用した近似最近傍探索
# COSINE距離(コサイン類似度)で類似度を測定
index_params = {
"metric_type": "COSINE", # COSINE距離(コサイン類似度)
"index_type": "IVF_FLAT", # IVFインデックス
"params": {"nlist": 128}, # クラスタ数(データ量に応じて調整)
}
collection.create_index("image_embedding", index_params)
Milvusに格納する画像ベクトルに対してインデックスを作成します。このとき、IVFというインデックス手法を使い、コサイン距離で128個のクラスタに分割し、検索を高速化します。
次に、前述した画像ファイルをベクトルに変換します。
# assets/配下の画像ファイルを取得
assets_dir = Path(__file__).parent.parent / "assets"
image_files = list(assets_dir.glob("*.jpg"))
# 画像データとembeddingを格納するリスト
documents_data = []
# 勾配計算を無効化(推論時は不要)
with torch.no_grad():
# 各画像ファイルを処理
for image_path in image_files:
# 画像を読み込んでRGBモードに変換
image = Image.open(image_path).convert("RGB")
# 画像を前処理してテンソルに変換し、デバイス(CPU/GPU)に転送
# この後のencode_imageは、バッチ次元が必要なのでunsqueeze(0)で次元を追加
# [バッチサイズ, チャンネル数(RGB), 高さ, 幅] ->(shape: [1, 3, 224, 224])
image_tensor = preprocess(image).unsqueeze(0).to(device)
# CLIPモデルで画像をエンコードして512次元のベクトルに変換
# cpu().numpy()[0]でCPUに戻してNumPy配列に変換し、バッチ次元を削除
image_embedding = model.encode_image(image_tensor).cpu().numpy()[0]
# 画像情報をリストに追加
documents_data.append(
{
"image_path": str(image_path), # 画像ファイルのパス
"image_embedding": image_embedding.tolist(), # numpy配列をリストに変換
"data_type": "image", # データタイプ
}
)
# Milvusにバッチ挿入
if documents_data:
# カラム形式でデータを準備(Milvusは列指向形式を要求)
# 各フィールドごとに全データをまとめたリストを作成
entities = [
[d["image_path"] for d in documents_data], # 画像パスのリスト
[d["data_type"] for d in documents_data], # データタイプのリスト
[d["image_embedding"] for d in documents_data], # ベクトルのリスト
]
# データを挿入(スキーマの順序: image_path, data_type, image_embedding)
collection.insert(entities)
model.encode_image(image_tensor)
の部分が、CLIPによる画像のベクトル変換処理です。
変換した画像ベクトルをファイルパスとともにMilvusへバッチで挿入します。
RAGによる検索
画像をMilvusに格納できたので、次は実際にRAGで画像を検索してみます。
検索する際は、検索文字列をベクトル化する必要がありますが、このときに使うモデルも先ほどと同じ ViT-B-32
です。
定義処理自体は、「CLIPモデルの初期化」の章を参照してください。
ベクトルデータベースから類似した画像のベクトルを検索
# Milvusサーバーに接続
connections.connect("default", host="localhost", port="19530")
# 既存のコレクションを取得
collection = Collection("test_rag_image_collection")
# コレクションをメモリにロード(検索を高速化するため)
collection.load()
# 勾配計算を無効化(推論時は不要)
with torch.no_grad():
# クエリテキストをトークン化してテンソルに変換
# 例: "cat" → [49406, 2368, 49407] のようなトークンIDの配列
text_tokens = tokenizer(query).to(device)
# CLIPモデルでテキストを512次元のベクトルに変換
# encode_text()の出力: [1, 512] → cpu().numpy()[0]で[512]に変換
query_embedding = model.encode_text(text_tokens).cpu().numpy()[0]
# Milvusでベクトル検索を実行
# 検索パラメータの設定(IVFで作成したINDEXから検索する)
search_params = {
"metric_type": "COSINE", # コサイン類似度で測定
"params": {"nprobe": 10}, # 128個に分割したクラスタから近い10個を検索
}
results = collection.search(
data=[query_embedding.tolist()], # 検索クエリベクトル(リスト形式で渡す)
anns_field="image_embedding", # 検索対象のベクトルフィールド名
param=search_params, # 検索パラメータ
limit=3, # 上位3件を取得
output_fields=["image_path"], # 返却するフィールド(画像パス)
)
# 検索結果を整形してリストで返す
# results[0]: 最初のクエリの結果(今回はクエリ1つなので[0])
search_results = [
{
"image_path": result.entity.get("image_path"), # 画像ファイルのパス
"distance": result.distance, # コサイン類似度(1に近いほど類似)
}
for result in results[0]
]
Milvusサーバーへの接続部分は、格納時と同じです。
検索文字列をベクトル化するには、まず tokenizer(query).to(device)
でトークン化した後に、model.encode_text
を使用します。
ここで変換されたベクトルを使って、Milvusに格納された画像から類似したものを検索しています。
LLMによる回答の生成
最後に、検索結果を元にLLMモデルで質問の回答を作成します。
# 検索結果をテキスト形式に整形(LLMに渡すコンテキストを作成)
context = "検索された関連画像情報:\n"
for i, result in enumerate(search_results, 1):
# enumerate(search_results, 1): 1から始まる番号を付けてループ
# 各結果を「番号. (類似度, パス)」の形式で追加
context += f"{i}. {result['distance'], result['image_path']}\n"
# 回答モデルの初期化
llm = ChatOpenAI(
model="gpt-5-nano",
temperature=0,
)
prompt = f"""
以下の画像情報を参考に、質問に答えてください。
質問: {query}
{context}
回答:
"""
# 例:
# 質問: cat
# 検索された関連画像情報:
# 1. (0.95, /path/to/cat_image.jpg)
# 2. (0.87, /path/to/kitten_image.jpg)
# 3. (0.82, /path/to/animal_image.jpg)
# LLMにプロンプトを送信して回答を生成
response = llm.invoke(prompt)
result = response.content
このサンプルにおいて、回答LLMは大きな処理を行わず、検索結果を渡して回答を生成しているのみです。
コード全体
全体のコードは次のようになります。
環境変数に、OPENAI_API_KEY
でOpenAIのAPIキーを設定する必要があります。
コード
import sys
from pathlib import Path
from typing import Any
# プロジェクトルートをパスに追加(src.embeddings.clip_encoderをインポートするため)
sys.path.insert(0, str(Path(__file__).parent.parent))
import argparse
import open_clip
import torch
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from PIL import Image
from pymilvus import Collection, CollectionSchema, DataType, FieldSchema, connections
# .envファイルから環境変数(OPENAI_API_KEY等)を読み込み
load_dotenv()
"""
uv run python samples/test_rag.py --create # コレクションを作成して画像を登録
uv run python samples/test_rag.py --query "cat" # 画像を検索
"""
def create_collection(model: Any, preprocess: Any, device: Any):
# Milvusサーバーに接続(ローカルホストのデフォルトポート19530を使用)
connections.connect("default", host="localhost", port="19530")
# コレクションのスキーマを定義(データベースのテーブル構造に相当)
fields = [
# id: 各レコードの一意識別子(自動採番される主キー)
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
# image_path: 画像ファイルのパス(最大500文字の文字列)
FieldSchema(name="image_path", dtype=DataType.VARCHAR, max_length=500),
# data_type: データの種類を示すラベル(例: "image", "text"など、最大50文字)
FieldSchema(name="data_type", dtype=DataType.VARCHAR, max_length=50),
# image_embedding: CLIPモデルで生成された画像の特徴ベクトル
# CLIP ViT-B-32モデルは512次元のベクトルを出力するため、dim=512を指定
FieldSchema(name="image_embedding", dtype=DataType.FLOAT_VECTOR, dim=512),
]
# スキーマとコレクション作成
# descriptionには用途を記載
schema = CollectionSchema(fields, description="Multimodal RAG image collection schema")
collection = Collection("test_rag_image_collection", schema)
# ベクトルインデックスの作成
# IVF_FLAT: Inverted File Systemを使用した近似最近傍探索
# COSINE距離(コサイン類似度)で類似度を測定
index_params = {
"metric_type": "COSINE", # COSINE距離(コサイン類似度)
"index_type": "IVF_FLAT", # IVFインデックス
"params": {"nlist": 128}, # クラスタ数(データ量に応じて調整)
}
collection.create_index("image_embedding", index_params)
# assets/配下の画像ファイルを取得
assets_dir = Path(__file__).parent.parent / "assets"
image_files = list(assets_dir.glob("*.jpg"))
# 画像データとembeddingを格納するリスト
documents_data = []
# 勾配計算を無効化(推論時はメモリ節約のため不要)
with torch.no_grad():
# 各画像ファイルを処理
for image_path in image_files:
print(f" 処理中: {image_path.name}")
# 画像を読み込んでRGBモードに変換
image = Image.open(image_path).convert("RGB")
# 画像を前処理してテンソルに変換し、デバイス(CPU/GPU)に転送
# この後のencode_imageは、バッチ次元が必要なのでunsqueeze(0)で次元を追加
# [バッチサイズ, チャンネル数(RGB), 高さ, 幅] ->(shape: [1, 3, 224, 224])
image_tensor = preprocess(image).unsqueeze(0).to(device)
# CLIPモデルで画像をエンコードして512次元のベクトルに変換
# cpu().numpy()[0]でCPUに戻してNumPy配列に変換し、バッチ次元を削除
image_embedding = model.encode_image(image_tensor).cpu().numpy()[0]
print(image_embedding.shape)
# 画像情報をリストに追加
documents_data.append(
{
"image_path": str(image_path), # 画像ファイルのパス
"image_embedding": image_embedding.tolist(), # numpy配列をリストに変換
"data_type": "image", # データタイプ
}
)
# Milvusにバッチ挿入
if documents_data:
# カラム形式でデータを準備(Milvusは列指向形式を要求)
# 各フィールドごとに全データをまとめたリストを作成
entities = [
[d["image_path"] for d in documents_data], # 画像パスのリスト
[d["data_type"] for d in documents_data], # データタイプのリスト
[d["image_embedding"] for d in documents_data], # ベクトルのリスト
]
# データを挿入(スキーマの順序: image_path, data_type, image_embedding)
collection.insert(entities)
print(f"✅ {len(documents_data)}個の画像データを格納しました")
def search_collection(model: Any, tokenizer: Any, device: Any, query: str):
# Milvusサーバーに接続
connections.connect("default", host="localhost", port="19530")
# 既存のコレクションを取得
collection = Collection("test_rag_image_collection")
# コレクションをメモリにロード(検索を高速化するため)
collection.load()
# 勾配計算を無効化(推論時はメモリ節約のため不要)
with torch.no_grad():
# クエリテキストをトークン化してテンソルに変換
# 例: "cat" → [49406, 2368, 49407] のようなトークンIDの配列
text_tokens = tokenizer(query).to(device)
# CLIPモデルでテキストを512次元のベクトルに変換
# encode_text()の出力: [1, 512] → cpu().numpy()[0]で[512]に変換
query_embedding = model.encode_text(text_tokens).cpu().numpy()[0]
# Milvusでベクトル検索を実行
# 検索パラメータの設定(IVFで作成したINDEXから検索する)
search_params = {
"metric_type": "COSINE", # コサイン類似度で測定
"params": {"nprobe": 10}, # 128個に分割したクラスタから近い10個を検索
}
results = collection.search(
data=[query_embedding.tolist()], # 検索クエリベクトル(リスト形式で渡す)
anns_field="image_embedding", # 検索対象のベクトルフィールド名
param=search_params, # 検索パラメータ
limit=3, # 上位3件を取得
output_fields=["image_path"], # 返却するフィールド(画像パス)
)
# 検索結果を整形してリストで返す
# results[0]: 最初のクエリの結果(今回はクエリ1つなので[0])
return [
{
"image_path": result.entity.get("image_path"), # 画像ファイルのパス
"distance": result.distance, # コサイン類似度(1に近いほど類似)
}
for result in results[0]
]
def generate_multimodal_answer(query: str, search_results: list[Any]):
"""
検索結果を基にLLMで回答を生成
【回答生成の流れ】
1. 検索結果から関連情報を抽出
2. コンテキストとして整形
3. GPT-4に質問とコンテキストを送信
4. 自然な日本語で回答を生成
【プロンプトエンジニアリング】
- 検索結果を番号付きリストで提示
- 質問に対して具体的に回答するよう指示
- 関連性の低い情報は除外するよう促す
Args:
query: 元の質問
search_results: 検索結果のリスト
Returns:
str: 生成された回答テキスト
"""
print("\n🤖 回答生成中...")
# 検索結果をテキスト形式に整形(LLMに渡すコンテキストを作成)
context = "検索された関連画像情報:\n"
for i, result in enumerate(search_results, 1):
# enumerate(search_results, 1): 1から始まる番号を付けてループ
# 各結果を「番号. (類似度, パス)」の形式で追加
context += f"{i}. {result['distance'], result['image_path']}\n"
# 回答モデルの初期化
llm = ChatOpenAI(
model="gpt-5-nano",
temperature=0,
)
# プロンプト(指示文)を構築
# LLMに「何をしてほしいか」を明確に伝える
prompt = f"""
以下の画像情報を参考に、質問に答えてください。
質問: {query}
{context}
回答:
"""
# 例:
# 質問: cat
# 検索された関連画像情報:
# 1. (0.95, /path/to/cat_image.jpg)
# 2. (0.87, /path/to/kitten_image.jpg)
# 3. (0.82, /path/to/animal_image.jpg)
# LLMにプロンプトを送信して回答を生成
response = llm.invoke(prompt)
return response.content
def main():
parser = argparse.ArgumentParser(description="マルチモーダルRAG with CLIP デモ")
parser.add_argument("--create", action="store_true", help="コレクションを作成して画像を登録")
parser.add_argument("--query", type=str, default="cat", help="検索クエリ(--search時に使用)")
args = parser.parse_args()
print("🚀 マルチモーダルRAG start")
model_name = "ViT-B-32"
device = "cuda" if torch.cuda.is_available() else "cpu"
model, _, preprocess = open_clip.create_model_and_transforms(
model_name, pretrained="openai", device=device
)
tokenizer = open_clip.get_tokenizer(model_name)
model.eval()
if args.create:
create_collection(model, preprocess, device)
else:
search_results = search_collection(model, tokenizer, device, args.query)
print("📊 検索結果:")
for result in search_results:
print(f" - {result['distance']}, {result['image_path']}")
if search_results:
answer = generate_multimodal_answer(args.query, search_results)
print(f"\n💬 質問: {args.query}")
print(f"📝 回答: {answer}")
else:
print(f"\n💬 質問: {args.query}")
print("❌ 関連する結果が見つかりませんでした")
if __name__ == "__main__":
main()
- ベクトルデータベースのセット:
uv run python samples/test_rag.py
- 検索:
uv run python samples/test_rag.py --query "cat"
実際に検索してみる
uv run python samples/test_rag.py --query "Is there a photo of coffee?"
🚀 マルチモーダルRAG start
📊 検索結果:
- 0.2870102822780609, assets/coffee_image.jpg
- 0.23830248415470123, assets/technology_image.jpg
- 0.2223903238773346, assets/cat_image.jpg
🤖 回答生成中...
💬 質問: Is there a photo of coffee?
📝 回答: はい。コーヒーの写真が含まれています。該当ファイルは coffee_image.jpg で、パスは assets/coffee_image.jpg です。
コーヒーの画像について尋ねたところ、「coffee_image.jpg」が最も類似度の高い画像として検索されていることがわかります。
CLIPモデル自体は、英語と画像で学習されているため、質問も英語である方が精度が高いですが、日本語でも最も近い画像として検索することができました。
uv run python samples/test_rag.py --query "コーヒーの画像はありますか?"
🚀 マルチモーダルRAG start
📊 検索結果:
- 0.24254494905471802, assets/coffee_image.jpg
- 0.21470631659030914, assets/technology_image.jpg
- 0.21066710352897644, assets/city_image.jpg
🤖 回答生成中...
💬 質問: コーヒーの画像はありますか?
📝 回答: はい、コーヒーの画像があります。関連画像のうち1番目が coffee_image.jpg で、ファイルパスは assets/coffee_image.jpg です。
今回のサンプルでは、精度向上については全く考慮していませんが、CLIPを使う上では他言語への対応について別途検討が必要になると考えられます。
Discussion