Geminiで構築するマルチモーダルRAGシステム：画像・動画・テキストの統合処理
検索拡張生成（RAG）は知識駆動型のAIアプリケーション構築を大きく変えましました。しかし従来のRAGシステムはテキスト空間に限定され、画像と動画に埋め込まれた豊かなセマンティック情報を見落としていましました。Geminiのマルチモーダル機能は新しい地平を開きます。視覚情報とテキスト情報をシームレスに統合したRAGシステムの実現です。
このガイドは、本番環境対応のマルチモーダルRAGシステムをGeminiで構築するための実装パターンをアーキテクチャから運用戦略まで網羅します。
マルチモーダルRAGのパラダイムシフト
従来のRAGシステムは実証済みのパターンに従います：
- ドキュメントをテキストパッセージに分割
- テキストを埋め込みに変換
- 埋め込みをベクトルデータベースに保存
- クエリに関連するパッセージを検索
- 取得したコンテキストを使用して回答を生成
マルチモーダルRAGはこのパラダイムをモダリティ横断的に拡張します。財務報告書は説明テキストと図表の両方を含み、両者が完全なストーリーを語ります。医療ガイドは症状の画像と診断テキストをペアで提供します。ビデオチュートリアルは視覚的なデモンストレーションと音声説明の両方を含みます。
マルチモーダルRAGは、ドメインが本質的にモダリティを組み合わせる場合に最も価値があります。ドキュメント処理（PDFのグラフ），eコマース（商品説明），教育コンテンツはいずれも大きなメリットがあります。純粋にテキストベースのドメインでは、オーバーヘッドが利益を正当化しないかもしれません。
アーキテクチャ：統一された埋め込み空間
主要な建築上の課題は、画像、動画フレーム、テキストをモダリティ間で意味のある類似性をもつ空間に埋め込む方法です。
アプローチ1：共有埋め込みモデル
Geminiの統一的な理解を活用して、多様なモダリティを同じ空間に埋め込みます：
import anthropic
import base64
from typing import Union
import numpy as np
client = anthropic.Anthropic()
def embed_multimodal_content(
text: str = None,
image_path: str = None,
video_frames: list = None
) -> dict:
"""
Geminiのマルチモーダル理解を使用してコンテンツを埋め込みます。
セマンティック要約と構造化埋め込みを返します。
"""
content = []
if text:
content.append({
"type": "text",
"text": f"このコンテンツを分析し、セマンティック検索用にまとめてください：{text}"
})
if image_path:
with open(image_path, "rb") as img:
image_data = base64.standard_b64encode(img.read()).decode("utf-8")
content.append({
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": image_data,
},
})
if video_frames:
for i, frame in enumerate(video_frames[:5]):
with open(frame, "rb") as f:
frame_data = base64.standard_b64encode(f.read()).decode("utf-8")
content.append({
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": frame_data,
},
})
content.append({
"type": "text",
"text": "主要概念、エンティティ、関係性をキャプチャした詳細なセマンティック要約（200-300語）を提供してください。"
})
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=500,
messages=[
{
"role": "user",
"content": content,
}
],
)
summary = response.content[0].text
return {
"summary": summary,
"modalities": {
"has_text": bool(text),
"has_image": bool(image_path),
"has_video": bool(video_frames),
"frame_count": len(video_frames) if video_frames else 0,
},
"tokens_used": response.usage.input_tokens + response.usage.output_tokens,
}
# 使用例
result = embed_multimodal_content(
text="四半期決算レポートは新興市場での前年比23%の成長を示しています",
image_path="chart.png",
video_frames=["frame_1.jpg", "frame_2.jpg", "frame_3.jpg"]
)
print(f"要約: {result['summary']}")
print(f"モダリティ: {result['modalities']}")このアプローチはクロスモーダルコンテキストをキャプチャするセマンティック要約を生成します。要約が埋め込み表現になり、検索に意味のある、セマンティック的に豊かな情報を提供します。
アプローチ2：デュアルストレージとクロスモーダル検索
より高度なシステムでは、各モダリティの独立したストレージを保持しながらクロスモーダル検索を可能にします：
from dataclasses import dataclass
from typing import List
import json
@dataclass
class MultimodalDocument:
doc_id: str
text_chunks: List[str]
image_chunks: List[dict] # {path, caption, description}
video_chunks: List[dict] # {path, timestamp, frame_description}
unified_summary: str
metadata: dict
class MultimodalRAGStore:
def __init__(self, vector_db_client):
self.vector_db = vector_db_client
self.documents = {}
def ingest_multimodal_document(self, doc: MultimodalDocument):
"""
混合モダリティを含むドキュメントを取り込みます。
各チャンクタイプに対して別々の埋め込みを作成します。
"""
# テキストチャンクを処理
text_embeddings = []
for chunk in doc.text_chunks:
summary = self._get_semantic_summary(text=chunk)
embedding = self._embed_summary(summary)
text_embeddings.append({
"chunk": chunk,
"summary": summary,
"embedding": embedding,
"modality": "text",
"doc_id": doc.doc_id,
})
# 画像チャンクを処理
image_embeddings = []
for image_chunk in doc.image_chunks:
summary = self._get_semantic_summary(image_path=image_chunk["path"])
embedding = self._embed_summary(summary)
image_embeddings.append({
"path": image_chunk["path"],
"caption": image_chunk.get("caption", ""),
"summary": summary,
"embedding": embedding,
"modality": "image",
"doc_id": doc.doc_id,
})
# 動画チャンクを処理
video_embeddings = []
for video_chunk in doc.video_chunks:
summary = self._get_semantic_summary(
video_frames=[video_chunk["frames"]]
)
embedding = self._embed_summary(summary)
video_embeddings.append({
"path": video_chunk["path"],
"timestamp": video_chunk.get("timestamp"),
"summary": summary,
"embedding": embedding,
"modality": "video",
"doc_id": doc.doc_id,
})
# すべての埋め込みを保存
all_embeddings = text_embeddings + image_embeddings + video_embeddings
self.vector_db.add_embeddings(all_embeddings)
# ドキュメント参照を保存
self.documents[doc.doc_id] = {
"text_chunks": len(text_embeddings),
"image_chunks": len(image_embeddings),
"video_chunks": len(video_embeddings),
"metadata": doc.metadata,
}
def retrieve_multimodal(
self,
query: str,
top_k: int = 5,
modality_filter: str = None
) -> List[dict]:
"""
モダリティ横断的に関連チャンクを検索します。
クエリはテキストですが、結果はすべてのモダリティにわたります。
"""
query_summary = self._get_semantic_summary(text=query)
query_embedding = self._embed_summary(query_summary)
results = self.vector_db.search(
query_embedding,
top_k=top_k * 2,
metadata_filter={"modality": modality_filter} if modality_filter else None,
)
# 関連性と多様性で再ランク付け
reranked = self._rerank_results(results, top_k)
return reranked
def _get_semantic_summary(self, **kwargs) -> str:
result = embed_multimodal_content(**kwargs)
return result["summary"]
def _embed_summary(self, summary: str) -> List[float]:
# 埋め込みモデルを使用（例：text-embedding-3-small）
pass
def _rerank_results(self, results: List[dict], top_k: int) -> List[dict]:
# モダリティ横断的に多様性を確保
modality_counts = {}
reranked = []
for result in results:
mod = result["modality"]
if modality_counts.get(mod, 0) < top_k // 3:
reranked.append(result)
modality_counts[mod] = modality_counts.get(mod, 0) + 1
return reranked[:top_k]生成：コンテキスト認識の合成
マルチモーダルコンテキストが取得されると、生成フェーズはより微妙になります。視覚的なソースとテキストソースから同時に合成しています。
生成された応答で画像または動画を参照する場合は、ソースについて明確にしてください。漠然とコンテンツを参照するのではなく「添付のチャートに示されているように...」と言いましょう。これは明確さと信頼を維持します。
def generate_multimodal_response(
query: str,
retrieved_chunks: List[dict],
client: anthropic.Anthropic
) -> str:
"""
マルチモーダル検索コンテキストを使用して応答を生成します。
"""
# モダリティごとにコンテキストを整理
text_context = [c for c in retrieved_chunks if c["modality"] == "text"]
image_context = [c for c in retrieved_chunks if c["modality"] == "image"]
video_context = [c for c in retrieved_chunks if c["modality"] == "video"]
# プロンプトを構築
message_content = [
{
"type": "text",
"text": f"ユーザークエリ: {query}\n\n"
}
]
# テキストコンテキストを追加
if text_context:
text_refs = "\n".join([
f"- {chunk['chunk'][:200]}..."
for chunk in text_context
])
message_content.append({
"type": "text",
"text": f"関連テキストソース:\n{text_refs}\n\n"
})
# 画像コンテキストを実画像で追加
for img_chunk in image_context[:2]:
with open(img_chunk["path"], "rb") as f:
image_data = base64.standard_b64encode(f.read()).decode("utf-8")
message_content.append({
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": image_data,
},
})
message_content.append({
"type": "text",
"text": f"画像コンテキスト: {img_chunk['caption']}\n詳細: {img_chunk['summary'][:300]}\n\n"
})
# 動画コンテキスト説明を追加
if video_context:
video_refs = "\n".join([
f"- {chunk['summary'][:200]}... ({chunk['timestamp']}から)"
for chunk in video_context
])
message_content.append({
"type": "text",
"text": f"動画コンテキスト:\n{video_refs}\n\n"
})
message_content.append({
"type": "text",
"text": "上記のマルチモーダルコンテキストに基づいて、包括的な回答を提供してください。"
"適切に特定のソース（テキスト、画像キャプション、動画タイムスタンプ）を参照してください。"
})
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1500,
messages=[
{
"role": "user",
"content": message_content,
}
],
)
return response.content[0].text動画処理：時間軸での抽出
動画は最も複雑なモダリティです。ビデオ全体を一度に処理する（非効率で誤りやすい）のではなく、セマンティック的に意味のあるキーフレームを抽出します。
import cv2
from pathlib import Path
class VideoFrameExtractor:
def __init__(self, client: anthropic.Anthropic):
self.client = client
def extract_semantic_frames(
self,
video_path: str,
max_frames: int = 10
) -> List[dict]:
"""
動画からセマンティック的に重要なフレームを抽出します。
シーン変化検出と時間軸サンプリングを使用します。
"""
cap = cv2.VideoCapture(video_path)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
fps = cap.get(cv2.CAP_PROP_FPS)
frame_indices = self._select_frames(total_frames, max_frames)
extracted_frames = []
frame_descriptions = []
for idx in frame_indices:
cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
ret, frame = cap.read()
if not ret:
continue
# フレームを一時保存
frame_path = f"/tmp/frame_{idx}.jpg"
cv2.imwrite(frame_path, frame)
# セマンティック説明を取得
description = self._describe_frame(frame_path)
timestamp = idx / fps
extracted_frames.append({
"path": frame_path,
"frame_index": idx,
"timestamp": timestamp,
"description": description,
})
frame_descriptions.append(
f"[{timestamp:.1f}秒] {description}"
)
cap.release()
return extracted_frames
def _select_frames(
self,
total_frames: int,
max_frames: int
) -> List[int]:
"""
適応的サンプリングを使用してフレームを選択します：
- 時間軸分布（均等間隔）
- シーン境界（利用可能な場合）
"""
indices = []
# ベース時間軸サンプリング
step = max(1, total_frames // max_frames)
indices.extend(range(0, total_frames, step))
# 常に最初と最後のフレームを含む
if 0 not in indices:
indices.insert(0, 0)
if total_frames - 1 not in indices:
indices.append(total_frames - 1)
return sorted(list(set(indices)))[:max_frames]
def _describe_frame(self, frame_path: str) -> str:
"""Geminiを使用してフレーム内容を説明します。"""
with open(frame_path, "rb") as f:
image_data = base64.standard_b64encode(f.read()).decode("utf-8")
response = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=200,
messages=[
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": image_data,
},
},
{
"type": "text",
"text": "このビデオフレームの主要なコンテンツを2～3文で説明してください。"
"低レベルの詳細ではなく、セマンティック的な意味に焦点を当ててください。"
}
],
}
],
)
return response.content[0].text検索戦略：マルチモーダルクエリ
従来のRAGはテキストクエリを想定しています。マルチモーダルRAGは新しい検索パターンを可能にします：
class MultimodalSearchStrategies:
def __init__(self, rag_store: MultimodalRAGStore):
self.rag = rag_store
def search_by_example_image(
self,
example_image_path: str,
query_description: str = None,
top_k: int = 5
) -> List[dict]:
"""
クエリとして画像を使用して同様のコンテンツを検索します。
オプションでテキスト説明と組み合わせることができます。
"""
content = [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": self._load_image(example_image_path),
},
}
]
if query_description:
content.append({
"type": "text",
"text": f"さらに、次に焦点を当ててください: {query_description}"
})
content.append({
"type": "text",
"text": "あなたが見ているものを説明し、何があなたを視覚的またはセマンティック的に区別されるようにするのか説明してください。"
})
# 画像の意味的な理解を得る
response = self.rag.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=300,
messages=[{"role": "user", "content": content}],
)
image_description = response.content[0].text
# 同様のコンテンツを取得
return self.rag.retrieve_multimodal(image_description, top_k=top_k)
def search_across_modality_gaps(
self,
query: str,
modality_sequence: List[str],
top_k: int = 5
) -> List[dict]:
"""
モダリティギャップを橋渡きするコンテンツを検索します。
例：視覚パターンのテキスト説明を検索します。
"""
results = []
for modality in modality_sequence:
modal_results = self.rag.retrieve_multimodal(
query,
top_k=top_k,
modality_filter=modality
)
results.extend(modal_results)
# モダリティを横断するシーケンスを優先する再ランク付け
return self._cross_modality_rerank(results, top_k)
def _load_image(self, path: str) -> str:
with open(path, "rb") as f:
return base64.standard_b64encode(f.read()).decode("utf-8")
def _cross_modality_rerank(
self,
results: List[dict],
top_k: int
) -> List[dict]:
"""複数のモダリティを組み合わせる結果を優先します。"""
modalities_seen = set()
reranked = []
for result in results:
mod = result["modality"]
# 多様なモダリティを優先
if mod not in modalities_seen or len(reranked) < top_k // 2:
reranked.append(result)
modalities_seen.add(mod)
return reranked[:top_k]本番環境での考慮事項
キャッシングとコスト最適化
マルチモーダル処理は高コストです。積極的にキャッシュします：
from functools import lru_cache
import hashlib
class CachedMultimodalRAG:
def __init__(self, rag_store, cache_backend):
self.rag = rag_store
self.cache = cache_backend
def retrieve_with_cache(
self,
query: str,
use_cache: bool = True
) -> List[dict]:
"""
キャッシング付きで検索します。
キャッシュヒットは高額なセマンティック処理を回避します。
"""
query_hash = hashlib.md5(query.encode()).hexdigest()
cache_key = f"multimodal_retrieve:{query_hash}"
if use_cache:
cached = self.cache.get(cache_key)
if cached:
return cached
results = self.rag.retrieve_multimodal(query)
if use_cache:
self.cache.set(cache_key, results, ttl=86400) # 24時間
return results監視と品質メトリクス
マルチモーダル固有のメトリクスを追跡します：
class MultimodalMetrics:
def __init__(self):
self.metrics = {
"total_queries": 0,
"avg_modality_diversity": 0,
"image_processing_time": [],
"video_processing_time": [],
"cross_modal_relevance": [],
}
def record_retrieval(self, results: List[dict]):
"""取得された結果のメトリクスを記録します。"""
modalities = set(r["modality"] for r in results)
diversity = len(modalities) / 3 # 最大3つのモダリティ
self.metrics["total_queries"] += 1
self.metrics["avg_modality_diversity"] = (
(self.metrics["avg_modality_diversity"] * (self.metrics["total_queries"] - 1)
+ diversity)
/ self.metrics["total_queries"]
)全体を振り返って
マルチモーダルRAGシステムは、視覚的および動画コンテンツから知識を引き出し、より豊かで根拠のあるAI応答を可能にします。シンプルなユースケースではまず統一埋め込みアプローチから始め、システムの規模が拡大するにつれてクロスモーダル検索を備えたデュアルストレージに段階的に移行できます。
成功の鍵は、各モダリティを思慮深く扱うこと。テキストのみのフレームワークに動画を無理やり適合させるのではなく、画像、動画、テキストがどのように一緒に機能するかに特に設計する点が肝心です。
マルチモーダル Embedding の基礎
概要
Gemini のマルチモーダル embedding モデルは、異なるメディアタイプ（テキスト、画像、動画）を同じベクトル空間に射影します。これにより、テキストクエリで画像を検索したり、画像クエリでテキストを検索することが可能になります。
from google.cloud import aiplatform
# マルチモーダル embedding モデルの初期化
def get_multimodal_embeddings():
"""マルチモーダル embedding モデルを取得"""
return aiplatform.gapic.v1.PredictionServiceClient()
# 対応メディアタイプ
supported_types = {
"text": "テキスト文字列",
"image": "JPEG/PNG 形式の画像",
"video": "MP4/WebM 形式の動画"
}マルチモーダル embedding では、すべてのメディアタイプが共通のベクトル空間に埋め込まれるため、異なるメディアタイプ間でセマンティック検索が可能です。
テキスト Embedding の実装
基本的なテキスト Embedding
from google.cloud import aiplatform
import json
class GeminiMultimodalEmbeddings:
"""Gemini マルチモーダル embedding クラス"""
def __init__(self, project_id, region="us-central1"):
self.project_id = project_id
self.region = region
self.client = aiplatform.gapic.v1.PredictionServiceClient(
client_options={"api_endpoint": f"{region}-aiplatform.googleapis.com"}
)
def embed_text(self, texts):
"""テキストをベクトル化"""
endpoint = (
f"projects/{self.project_id}/locations/{self.region}/"
f"publishers/google/models/multimodalembedding@001"
)
embeddings = []
for text in texts:
request = aiplatform.gapic.v1.PredictRequest(
endpoint=endpoint,
instances=[
{
"mimeType": "text/plain",
"text": text
}
]
)
response = self.client.predict(request=request)
embedding = response.predictions[0]["textEmbedding"]
embeddings.append(embedding)
return embeddings
# 使用例
embedder = GeminiMultimodalEmbeddings(project_id="my-project")
texts = [
"Gemini API について学ぶ",
"Vertex AI の使い方",
"機械学習モデルの訓練方法"
]
embeddings = embedder.embed_text(texts)
print(f"生成された embedding: {len(embeddings)} 個")
print(f"ベクトル次元: {len(embeddings[0])}")画像 Embedding の実装
画像ファイルのベクトル化
import base64
from pathlib import Path
def encode_image_to_base64(image_path):
"""画像ファイルを base64 エンコード"""
with open(image_path, "rb") as f:
return base64.standard_b64encode(f.read()).decode("utf-8")
class MultimodalRAGSystem:
"""マルチモーダル RAG システム"""
def __init__(self, project_id, region="us-central1"):
self.embedder = GeminiMultimodalEmbeddings(project_id, region)
def embed_images(self, image_paths):
"""複数の画像をベクトル化"""
embeddings = []
for image_path in image_paths:
# 画像をエンコード
image_data = encode_image_to_base64(image_path)
# MIME タイプを判定
suffix = Path(image_path).suffix.lower()
mime_type = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp"
}.get(suffix, "image/jpeg")
# API 呼び出し
endpoint = (
f"projects/{self.embedder.project_id}/locations/{self.embedder.region}/"
f"publishers/google/models/multimodalembedding@001"
)
request = aiplatform.gapic.v1.PredictRequest(
endpoint=endpoint,
instances=[
{
"mimeType": mime_type,
"image": {
"bytesBase64Encoded": image_data
}
}
]
)
response = self.embedder.client.predict(request=request)
embedding = response.predictions[0]["imageEmbedding"]
embeddings.append({
"path": image_path,
"embedding": embedding,
"mime_type": mime_type
})
return embeddings
# 使用例
rag = MultimodalRAGSystem(project_id="my-project")
image_paths = ["image1.jpg", "image2.png", "chart.webp"]
image_embeddings = rag.embed_images(image_paths)
print(f"画像 embedding を生成しました: {len(image_embeddings)} 個")画像の base64 エンコードは大きなファイルでは時間がかかります。バッチ処理やキャッシングを活用してパフォーマンスを最適化してください。
動画 Embedding の実装
動画フレーム抽出と Embedding
import cv2
import numpy as np
from typing import List, Tuple
class VideoEmbeddingProcessor:
"""動画処理と embedding 生成"""
def __init__(self, rag_system: MultimodalRAGSystem):
self.rag = rag_system
def extract_frames(self, video_path, fps=1):
"""動画からフレームを抽出"""
cap = cv2.VideoCapture(video_path)
frames = []
frame_count = 0
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
video_fps = cap.get(cv2.CAP_PROP_FPS)
# フレーム抽出間隔を計算
frame_interval = int(video_fps / fps)
while True:
ret, frame = cap.read()
if not ret:
break
if frame_count % frame_interval == 0:
# フレームを JPEG で保存（一時的）
temp_path = f"/tmp/frame_{frame_count}.jpg"
cv2.imwrite(temp_path, frame)
frames.append({
"frame_number": frame_count,
"timestamp": frame_count / video_fps,
"path": temp_path
})
frame_count += 1
cap.release()
return frames
def embed_video(self, video_path, fps=1):
"""動画全体を embedding"""
frames = self.extract_frames(video_path, fps=fps)
# フレームをベクトル化
frame_embeddings = []
for frame in frames:
image_emb = self.rag.embed_images([frame["path"]])
frame_embeddings.append({
"timestamp": frame["timestamp"],
"frame_number": frame["frame_number"],
"embedding": image_emb[0]["embedding"]
})
return {
"video_path": video_path,
"total_frames": len(frames),
"frame_embeddings": frame_embeddings
}
# 使用例
rag = MultimodalRAGSystem(project_id="my-project")
processor = VideoEmbeddingProcessor(rag)
video_embeddings = processor.embed_video(
"presentation.mp4",
fps=1 # 1秒ごとにフレーム抽出
)
print(f"動画フレーム embedding を生成しました: {video_embeddings['total_frames']} フレーム")ベクトルデータベースへの保存
Vertex AI Vector Search への統合
from google.cloud import aiplatform
import json
class VectorSearchStorage:
"""Vertex AI Vector Search への保存"""
def __init__(self, project_id, region="us-central1"):
self.project_id = project_id
self.region = region
self.client = aiplatform.gapic.v1.MatchServiceClient(
client_options={"api_endpoint": f"{region}-aiplatform.googleapis.com"}
)
def create_index(self, index_name, embedding_dimension=1408):
"""ベクトルインデックスを作成"""
# Vertex AI Vector Search で新しいインデックスを作成
config = {
"indexDisplayName": index_name,
"dimensions": embedding_dimension,
"matchingLowLevelConfig": {
"linearSearchLowLevelConfig": {}
}
}
# 実装の詳細は Google Cloud SDK を参照
return config
def insert_embeddings(
self,
index_name,
embeddings_list: List[dict]
):
"""embedding をインデックスに挿入"""
"""
embeddings_list 形式:
[
{
"id": "doc_1",
"embedding": [0.1, 0.2, ...],
"metadata": {
"type": "text",
"source": "document.txt",
"content": "..."
}
},
...
]
"""
# バッチ処理で効率化
batch_size = 100
for i in range(0, len(embeddings_list), batch_size):
batch = embeddings_list[i:i+batch_size]
# API へのアップロード処理
pass
# ChromaDB での軽量実装例
from chromadb import Client
from chromadb.config import Settings
class ChromaVectorStorage:
"""ChromaDB でのベクトル保存（開発用）"""
def __init__(self, persist_dir="./chroma_db"):
settings = Settings(
chroma_db_impl="duckdb+parquet",
persist_directory=persist_dir,
anonymized_telemetry=False
)
self.client = Client(settings)
self.collection = self.client.get_or_create_collection(
name="multimodal_rag",
metadata={"hnsw:space": "cosine"}
)
def add_documents(self, documents):
"""ドキュメントを追加"""
"""
documents 形式:
[
{
"id": "doc_1",
"embedding": [0.1, 0.2, ...],
"metadata": {"type": "text", "source": "..."},
"document": "テキスト内容"
},
...
]
"""
ids = []
embeddings = []
metadatas = []
documents = []
for doc in documents:
ids.append(doc["id"])
embeddings.append(doc["embedding"])
metadatas.append(doc["metadata"])
documents.append(doc.get("document", ""))
self.collection.add(
ids=ids,
embeddings=embeddings,
metadatas=metadatas,
documents=documents
)
def search(self, query_embedding, n_results=5):
"""ベクトル検索"""
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results
)
return results統合検索パイプライン
ハイブリッド検索の実装
import numpy as np
from typing import List, Dict
class HybridSearchPipeline:
"""テキスト + セマンティック検索のハイブリッドパイプライン"""
def __init__(self, rag_system, storage):
self.rag = rag_system
self.storage = storage
def search(self, query, query_type="text", n_results=5):
"""
統合検索実行
query_type: 'text', 'image', または 'multimodal'
"""
if query_type == "text":
query_embedding = self.rag.embed_text([query])[0]
elif query_type == "image":
query_embedding = self.rag.embed_images([query])[0]["embedding"]
else:
raise ValueError(f"Unsupported query type: {query_type}")
# ベクトル検索
results = self.storage.search(query_embedding, n_results=n_results)
# 結果をスコアでソート
scored_results = []
for idx, (doc_id, metadata) in enumerate(
zip(results["ids"][0], results["metadatas"][0])
):
distance = results["distances"][0][idx]
# コサイン距離を類似度に変換
similarity = 1 - distance / 2
scored_results.append({
"id": doc_id,
"similarity_score": similarity,
"metadata": metadata,
"document": results["documents"][0][idx] if results.get("documents") else ""
})
# スコアで降順ソート
scored_results.sort(key=lambda x: x["similarity_score"], reverse=True)
return scored_results
def multimodal_search(self, queries_dict, weights=None):
"""複数モーダルの統合検索"""
"""
queries_dict: {
'text': 'テキストクエリ',
'image': 'image_path.jpg'
}
"""
if weights is None:
weights = {k: 1.0 for k in queries_dict.keys()}
results_list = []
# 各クエリで検索
for query_type, query in queries_dict.items():
results = self.search(query, query_type=query_type)
for r in results:
r["weight"] = weights.get(query_type, 1.0)
results_list.extend(results)
# ID ごとにスコアを集計
aggregated = {}
for r in results_list:
doc_id = r["id"]
if doc_id not in aggregated:
aggregated[doc_id] = {
"id": doc_id,
"total_score": 0,
"metadata": r["metadata"],
"document": r["document"],
"source_count": 0
}
aggregated[doc_id]["total_score"] += r["similarity_score"] * r["weight"]
aggregated[doc_id]["source_count"] += 1
# 集計スコアで降順ソート
final_results = sorted(
aggregated.values(),
key=lambda x: x["total_score"],
reverse=True
)
return final_results[:5] # Top 5 を返す
# 使用例
rag = MultimodalRAGSystem(project_id="my-project")
storage = ChromaVectorStorage()
search_pipeline = HybridSearchPipeline(rag, storage)
# テキスト検索
text_results = search_pipeline.search(
query="Python プログラミング",
query_type="text"
)
# マルチモーダル検索
multimodal_results = search_pipeline.multimodal_search({
'text': 'データ分析',
'image': 'chart.png'
}, weights={'text': 1.0, 'image': 0.8})マルチモーダル検索では、異なるメディアタイプから得られた結果をスコア正規化して集計する点が肝心です。
Gemini による回答生成
RAG との統合
from anthropic import Anthropic
class MultimodalRAGWithGeneration:
"""マルチモーダル RAG と生成統合"""
def __init__(self, search_pipeline, api_key=None):
self.search = search_pipeline
self.client = Anthropic(api_key=api_key)
def generate_answer(self, query, query_type="text", context_limit=3):
"""クエリに対して検索結果に基づく回答を生成"""
# 検索実行
search_results = self.search.search(
query=query,
query_type=query_type,
n_results=context_limit
)
# コンテキスト構築
context_text = self._build_context(search_results)
# Gemini で回答生成
prompt = f"""以下の検索結果に基づいて、ユーザーの質問に答えてください。
検索結果:
{context_text}
質問: {query}
回答:"""
response = self.client.messages.create(
model="gemini-2.5-pro",
max_tokens=1024,
messages=[{
"role": "user",
"content": prompt
}]
)
return {
"answer": response.content[0].text,
"sources": search_results,
"query": query
}
def _build_context(self, search_results):
"""検索結果をコンテキストテキストに変換"""
context_parts = []
for i, result in enumerate(search_results, 1):
source = result["metadata"].get("source", "Unknown")
score = result["similarity_score"]
document = result["document"]
context_parts.append(
f"[ソース {i}: {source} (スコア: {score:.2f})]\n{document}\n"
)
return "\n".join(context_parts)
# 使用例
rag_gen = MultimodalRAGWithGeneration(search_pipeline)
result = rag_gen.generate_answer(
query="データ分析の最適実践方法は何ですか？",
query_type="text"
)
print("回答:", result["answer"])
print("\nソース:")
for source in result["sources"]:
print(f" - {source['metadata']['source']}")パフォーマンス最適化
Embedding キャッシング
import hashlib
from datetime import datetime, timedelta
class EmbeddingCache:
"""Embedding 結果のキャッシング"""
def __init__(self, ttl_hours=24):
self.cache = {}
self.ttl = timedelta(hours=ttl_hours)
def _get_key(self, content, content_type):
"""コンテンツのハッシュキーを生成"""
if content_type == "file":
with open(content, "rb") as f:
content_bytes = f.read()
else:
content_bytes = content.encode()
return hashlib.sha256(content_bytes).hexdigest()
def get(self, content, content_type):
"""キャッシュから embedding を取得"""
key = self._get_key(content, content_type)
if key in self.cache:
cached_embedding, cached_time = self.cache[key]
if datetime.now() - cached_time < self.ttl:
return cached_embedding
return None
def set(self, content, content_type, embedding):
"""embedding をキャッシュに保存"""
key = self._get_key(content, content_type)
self.cache[key] = (embedding, datetime.now())
# 使用例
cache = EmbeddingCache(ttl_hours=24)
# キャッシュから取得を試みる
cached = cache.get("query_text", "text")
if cached is not None:
embedding = cached
else:
# embedding を生成してキャッシュに保存
embedding = rag.embed_text(["query_text"])[0]
cache.set("query_text", "text", embedding)本番運用のベストプラクティス
定期的な Embedding 更新
import schedule
import time
from pathlib import Path
class DocumentIndexer:
"""定期的なドキュメント索引更新"""
def __init__(self, rag_system, storage, doc_dir):
self.rag = rag_system
self.storage = storage
self.doc_dir = Path(doc_dir)
def index_documents(self):
"""ディレクトリ内のドキュメントをインデックス"""
documents = []
# テキストファイル
for txt_file in self.doc_dir.glob("*.txt"):
with open(txt_file) as f:
content = f.read()
embeddings = self.rag.embed_text([content])
documents.append({
"id": str(txt_file),
"embedding": embeddings[0],
"metadata": {"type": "text", "source": txt_file.name},
"document": content
})
# 画像ファイル
image_files = list(self.doc_dir.glob("*.jpg")) + \
list(self.doc_dir.glob("*.png"))
image_embeddings = self.rag.embed_images([str(f) for f in image_files])
for img_emb in image_embeddings:
documents.append({
"id": img_emb["path"],
"embedding": img_emb["embedding"],
"metadata": {"type": "image", "source": Path(img_emb["path"]).name},
"document": f"Image: {Path(img_emb['path']).name}"
})
# ストレージに保存
self.storage.add_documents(documents)
def schedule_updates(self, interval_hours=24):
"""定期更新のスケジューリング"""
schedule.every(interval_hours).hours.do(self.index_documents)
while True:
schedule.run_pending()
time.sleep(60)
# 使用例
indexer = DocumentIndexer(rag, storage, "./documents")
# バックグラウンドで定期更新を実行
indexer.schedule_updates(interval_hours=24)