Building Multimodal RAG Systems with Gemini: Processing Images, Video, and Text Together
Retrieval-augmented generation (RAG) has transformed how we build knowledge-driven AI applications. But traditional RAG systems operate primarily in text space, missing the rich semantic information embedded in images and video. Gemini's multimodal capabilities unlock a new frontier: RAG systems that seamlessly integrate visual and textual knowledge.
The Multimodal RAG Paradigm Shift
Traditional RAG systems follow a proven pattern:
- Chunk documents into text passages
- Encode text into embeddings
- Store embeddings in vector database
- Retrieve relevant passages on query
- Generate answer using retrieved context
Multimodal RAG extends this paradigm across modalities. A financial report might contain both explanatory text AND charts that together tell the complete story. A medical guide might pair images of symptoms with diagnostic text. Video tutorials contain both visual demonstrations and verbal explanations.
Multimodal RAG is most valuable when your domain inherently combines modalities. Document processing (PDFs with charts), e-commerce (products with descriptions), and educational content benefit significantly. For purely text-based domains, the overhead may not justify the gains.
Architecture: The Unified Embedding Space
The key architectural challenge: how do you embed images, video frames, and text into a space where similarity is meaningful across modalities?
Approach 1: Shared Embedding Model
Use Gemini's unified understanding to embed diverse modalities into the same space:
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:
"""
Embed content using Gemini's multimodal understanding.
Returns structured embeddings and semantic summaries.
"""
content = []
if text:
content.append({
"type": "text",
"text": f"Analyze and summarize this content for semantic search: {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]): # Limit frames
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": "Provide a detailed semantic summary (200-300 words) capturing key concepts, entities, and relationships."
})
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,
}
# Example usage
result = embed_multimodal_content(
text="The quarterly report shows 23% YoY growth in emerging markets",
image_path="chart.png",
video_frames=["frame_1.jpg", "frame_2.jpg", "frame_3.jpg"]
)
print(f"Summary: {result['summary']}")
print(f"Modalities: {result['modalities']}")This approach generates semantic summaries that capture cross-modal context. The summary becomes your embedding representation—semantically rich and meaningful for retrieval.
Approach 2: Dual Storage with Cross-Modal Retrieval
For more sophisticated systems, maintain separate storage for each modality but enable cross-modal retrieval:
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):
"""
Ingest a document with mixed modalities.
Create separate embeddings for each chunk type.
"""
# Process text chunks
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,
})
# Process image chunks
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,
})
# Process video chunks
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,
})
# Store all embeddings
all_embeddings = text_embeddings + image_embeddings + video_embeddings
self.vector_db.add_embeddings(all_embeddings)
# Store document reference
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]:
"""
Retrieve relevant chunks across modalities.
Query is text, but results span all modalities.
"""
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, # Get more, then filter
metadata_filter={"modality": modality_filter} if modality_filter else None,
)
# Rerank by relevance and diversity
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]:
# Use your embedding model (e.g., text-embedding-3-small)
pass
def _rerank_results(self, results: List[dict], top_k: int) -> List[dict]:
# Ensure diversity across modalities
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]Generation: Context-Aware Synthesis
With multimodal context retrieved, the generation phase becomes more nuanced. You're synthesizing from visual and textual sources simultaneously.
When referencing images or video in generated responses, be explicit about the source. Say "As shown in the accompanying chart..." rather than vaguely referencing content. This maintains clarity and trust.
def generate_multimodal_response(
query: str,
retrieved_chunks: List[dict],
client: anthropic.Anthropic
) -> str:
"""
Generate response using multimodal retrieved context.
"""
# Organize context by modality
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"]
# Build prompt with context
message_content = [
{
"type": "text",
"text": f"User Query: {query}\n\n"
}
]
# Add text context
if text_context:
text_refs = "\n".join([
f"- {chunk['chunk'][:200]}..."
for chunk in text_context
])
message_content.append({
"type": "text",
"text": f"Relevant Text Sources:\n{text_refs}\n\n"
})
# Add image context with actual images
for img_chunk in image_context[:2]: # Limit to 2 images per response
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"Image Context: {img_chunk['caption']}\nDetails: {img_chunk['summary'][:300]}\n\n"
})
# Add video context descriptions
if video_context:
video_refs = "\n".join([
f"- {chunk['summary'][:200]}... (from {chunk['timestamp']})"
for chunk in video_context
])
message_content.append({
"type": "text",
"text": f"Video Context:\n{video_refs}\n\n"
})
message_content.append({
"type": "text",
"text": "Based on the above multimodal context, provide a comprehensive answer. "
"Reference specific sources (text, image captions, video timestamps) where relevant."
})
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1500,
messages=[
{
"role": "user",
"content": message_content,
}
],
)
return response.content[0].textVideo Processing: Temporal Extraction
Video is the most complex modality. Rather than processing the entire video at once (inefficient and error-prone), extract semantically meaningful keyframes.
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]:
"""
Extract semantically important frames from video.
Uses scene change detection and temporal sampling.
"""
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
# Save frame temporarily
frame_path = f"/tmp/frame_{idx}.jpg"
cv2.imwrite(frame_path, frame)
# Get semantic description
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}s] {description}"
)
cap.release()
return extracted_frames
def _select_frames(
self,
total_frames: int,
max_frames: int
) -> List[int]:
"""
Select frames using adaptive sampling:
- Temporal distribution (evenly spaced)
- Scene boundaries (when available)
"""
indices = []
# Base temporal sampling
step = max(1, total_frames // max_frames)
indices.extend(range(0, total_frames, step))
# Always include first and last frames
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:
"""Use Gemini to describe frame content."""
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": "Describe the key content of this video frame in 2-3 sentences. "
"Focus on semantic meaning rather than low-level details."
}
],
}
],
)
return response.content[0].textSearch Strategies: Multimodal Queries
Traditional RAG assumes text queries. Multimodal RAG unlocks new search patterns:
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]:
"""
Find similar content using an image as the query.
Optionally combine with text description.
"""
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"Additionally, focus on: {query_description}"
})
content.append({
"type": "text",
"text": "Describe what you see and what makes this visually or semantically distinct."
})
# Get semantic understanding of image
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
# Retrieve similar content
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]:
"""
Retrieve content that bridges modalities.
E.g., find text explanations for visual patterns.
"""
results = []
for modality in modality_sequence:
modal_results = self.rag.retrieve_multimodal(
query,
top_k=top_k,
modality_filter=modality
)
results.extend(modal_results)
# Rerank to prefer sequences that cross modalities
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]:
"""Prefer results that combine multiple modalities."""
modalities_seen = set()
reranked = []
for result in results:
mod = result["modality"]
# Prefer diverse modalities
if mod not in modalities_seen or len(reranked) < top_k // 2:
reranked.append(result)
modalities_seen.add(mod)
return reranked[:top_k]Production Considerations
Caching and Cost Optimization
Multimodal processing is expensive. Cache aggressively:
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]:
"""
Retrieve with query caching.
Cache hits avoid expensive semantic processing.
"""
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) # 24h
return resultsMonitoring and Quality Metrics
Track multimodal-specific metrics:
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]):
"""Record metrics on retrieved results."""
modalities = set(r["modality"] for r in results)
diversity = len(modalities) / 3 # Max 3 modalities
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"]
)Wrapping up
Multimodal RAG systems unlock knowledge from your visual and video content, enabling richer, more grounded AI responses. Start with the unified embedding approach for simpler use cases, then graduate to dual storage with cross-modal retrieval as your system scales.
The key to success is treating each modality thoughtfully—not forcing video into a text-only framework, but designing specifically for the strengths of images, video, and text when they work together.
Multimodal Embedding Fundamentals
Overview
Gemini's multimodal embedding model projects different media types (text, images, video) into a shared vector space. This enables searching images with text queries and vice versa.
from google.cloud import aiplatform
# Initialize multimodal embedding model
def get_multimodal_embeddings():
"""Get multimodal embedding service"""
return aiplatform.gapic.v1.PredictionServiceClient()
# Supported media types
supported_types = {
"text": "Text strings",
"image": "JPEG/PNG format images",
"video": "MP4/WebM format videos"
}In multimodal embeddings, all media types project into a shared vector space, enabling semantic search across different media types.
Text Embedding Implementation
Basic Text Embedding
from google.cloud import aiplatform
class GeminiMultimodalEmbeddings:
"""Gemini multimodal embedding class"""
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):
"""Vectorize 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
# Usage example
embedder = GeminiMultimodalEmbeddings(project_id="my-project")
texts = [
"Learning Gemini API",
"Using Vertex AI",
"Training machine learning models"
]
embeddings = embedder.embed_text(texts)
print(f"Generated {len(embeddings)} embeddings")
print(f"Vector dimension: {len(embeddings[0])}")Image Embedding Implementation
Vectorizing Image Files
import base64
from pathlib import Path
def encode_image_to_base64(image_path):
"""Encode image file to base64"""
with open(image_path, "rb") as f:
return base64.standard_b64encode(f.read()).decode("utf-8")
class MultimodalRAGSystem:
"""Multimodal RAG system"""
def __init__(self, project_id, region="us-central1"):
self.embedder = GeminiMultimodalEmbeddings(project_id, region)
def embed_images(self, image_paths):
"""Vectorize multiple images"""
embeddings = []
for image_path in image_paths:
# Encode image
image_data = encode_image_to_base64(image_path)
# Determine MIME type
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 call
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
# Usage example
rag = MultimodalRAGSystem(project_id="my-project")
image_paths = ["image1.jpg", "image2.png", "chart.webp"]
image_embeddings = rag.embed_images(image_paths)
print(f"Generated {len(image_embeddings)} image embeddings")Base64 encoding large image files is time-consuming. Use batch processing and caching to optimize performance.
Video Embedding Implementation
Extracting Video Frames and Embedding
import cv2
from typing import List
class VideoEmbeddingProcessor:
"""Video processing and embedding generation"""
def __init__(self, rag_system: MultimodalRAGSystem):
self.rag = rag_system
def extract_frames(self, video_path, fps=1):
"""Extract frames from video"""
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)
# Calculate frame extraction interval
frame_interval = int(video_fps / fps)
while True:
ret, frame = cap.read()
if not ret:
break
if frame_count % frame_interval == 0:
# Save frame as JPEG (temporary)
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):
"""Embed entire video"""
frames = self.extract_frames(video_path, fps=fps)
# Vectorize frames
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
}
# Usage example
rag = MultimodalRAGSystem(project_id="my-project")
processor = VideoEmbeddingProcessor(rag)
video_embeddings = processor.embed_video(
"presentation.mp4",
fps=1 # Extract frames every second
)
print(f"Generated {video_embeddings['total_frames']} frame embeddings")Storing in Vector Database
Vertex AI Vector Search Integration
from google.cloud import aiplatform
from typing import List, Dict
class VectorSearchStorage:
"""Vertex AI Vector Search storage"""
def __init__(self, project_id, region="us-central1"):
self.project_id = project_id
self.region = region
def create_index(self, index_name, embedding_dimension=1408):
"""Create vector index"""
config = {
"indexDisplayName": index_name,
"dimensions": embedding_dimension,
"matchingLowLevelConfig": {
"linearSearchLowLevelConfig": {}
}
}
return config
# Lightweight implementation with ChromaDB
from chromadb import Client
from chromadb.config import Settings
class ChromaVectorStorage:
"""ChromaDB vector storage (development)"""
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):
"""Add documents"""
ids = []
embeddings = []
metadatas = []
docs = []
for doc in documents:
ids.append(doc["id"])
embeddings.append(doc["embedding"])
metadatas.append(doc["metadata"])
docs.append(doc.get("document", ""))
self.collection.add(
ids=ids,
embeddings=embeddings,
metadatas=metadatas,
documents=docs
)
def search(self, query_embedding, n_results=5):
"""Vector search"""
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results
)
return resultsUnified Search Pipeline
Hybrid Search Implementation
from typing import List, Dict
class HybridSearchPipeline:
"""Hybrid text + semantic search pipeline"""
def __init__(self, rag_system, storage):
self.rag = rag_system
self.storage = storage
def search(self, query, query_type="text", n_results=5):
"""
Unified search
query_type: 'text', 'image', or '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}")
# Vector search
results = self.storage.search(query_embedding, n_results=n_results)
# Score results
scored_results = []
for idx, (doc_id, metadata) in enumerate(
zip(results["ids"][0], results["metadatas"][0])
):
distance = results["distances"][0][idx]
# Convert cosine distance to similarity
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 ""
})
# Sort by score descending
scored_results.sort(key=lambda x: x["similarity_score"], reverse=True)
return scored_results
def multimodal_search(self, queries_dict, weights=None):
"""Unified search across multiple modalities"""
if weights is None:
weights = {k: 1.0 for k in queries_dict.keys()}
results_list = []
# Search for each query
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)
# Aggregate scores by 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
# Sort by aggregated score
final_results = sorted(
aggregated.values(),
key=lambda x: x["total_score"],
reverse=True
)
return final_results[:5]
# Usage example
rag = MultimodalRAGSystem(project_id="my-project")
storage = ChromaVectorStorage()
search_pipeline = HybridSearchPipeline(rag, storage)
# Text search
text_results = search_pipeline.search(
query="Python programming",
query_type="text"
)
# Multimodal search
multimodal_results = search_pipeline.multimodal_search({
'text': 'data analysis',
'image': 'chart.png'
}, weights={'text': 1.0, 'image': 0.8})In multimodal search, it's important to normalize and aggregate scores from different media types.
Answer Generation with Gemini
RAG Integration
from anthropic import Anthropic
class MultimodalRAGWithGeneration:
"""Multimodal RAG with generation"""
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):
"""Generate answer based on search results"""
# Execute search
search_results = self.search.search(
query=query,
query_type=query_type,
n_results=context_limit
)
# Build context
context_text = self._build_context(search_results)
# Generate with Gemini
prompt = f"""Answer the user's question based on these search results.
Search Results:
{context_text}
Question: {query}
Answer:"""
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):
"""Convert search results to context text"""
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"[Source {i}: {source} (Score: {score:.2f})]\n{document}\n"
)
return "\n".join(context_parts)
# Usage example
rag_gen = MultimodalRAGWithGeneration(search_pipeline)
result = rag_gen.generate_answer(
query="What are best practices for data analysis?",
query_type="text"
)
print("Answer:", result["answer"])
print("\nSources:")
for source in result["sources"]:
print(f" - {source['metadata']['source']}")Performance Optimization
Embedding Caching
import hashlib
from datetime import datetime, timedelta
class EmbeddingCache:
"""Embedding result caching"""
def __init__(self, ttl_hours=24):
self.cache = {}
self.ttl = timedelta(hours=ttl_hours)
def _get_key(self, content, content_type):
"""Generate hash key for content"""
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):
"""Get embedding from cache"""
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):
"""Store embedding in cache"""
key = self._get_key(content, content_type)
self.cache[key] = (embedding, datetime.now())
# Usage example
cache = EmbeddingCache(ttl_hours=24)
# Try to get from cache
cached = cache.get("query_text", "text")
if cached is not None:
embedding = cached
else:
# Generate and cache
embedding = rag.embed_text(["query_text"])[0]
cache.set("query_text", "text", embedding)Production Best Practices
Periodic Embedding Updates
import schedule
import time
from pathlib import Path
class DocumentIndexer:
"""Periodic document index updates"""
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):
"""Index documents in directory"""
documents = []
# Text files
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
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}"
})
# Store in vector database
self.storage.add_documents(documents)
def schedule_updates(self, interval_hours=24):
"""Schedule periodic updates"""
schedule.every(interval_hours).hours.do(self.index_documents)
while True:
schedule.run_pending()
time.sleep(60)
# Usage example
indexer = DocumentIndexer(rag, storage, "./documents")
# Run periodic updates in background
indexer.schedule_updates(interval_hours=24)Looking back
Building multimodal RAG systems enables:
- Unified Search: Search text, images, and video through same interface
- Semantic Understanding: Meaningful search across media types
- Automated Answer Generation: Natural answers based on search results
- Scalability: Production support via Vertex AI Vector Search
When implementing, incorporate caching strategies, periodic updates, and monitoring for stable production operations.