Building a Multimodal RAG Pipeline in Python with Qdrant and CLIP Embeddings

Standard Retrieval-Augmented Generation (RAG) pipelines are built exclusively for textual data. They parse markdown files, split paragraphs into token chunks, calculate dense text embeddings, and index them into stores like PostgreSQL with pgvector.

However, enterprise technical documentation is rarely just text. Engineering whitepapers, architectural blueprints, medical records, and e-commerce catalogs are densely populated with system architecture diagrams, charts, UI mockups, and schematics. When a traditional text-only RAG system processes these documents, visual data is discarded or reduced to low-fidelity OCR snippets, creating severe information blind spots.

In this comprehensive production tutorial, you will learn how to build an end-to-end Multimodal RAG Pipeline in Python. Using CLIP (Contrastive Language-Image Pretraining) embeddings and the high-performance Qdrant vector database, we project images and text into a unified vector space, perform cross-modal retrieval, and synthesize responses using multimodal LLMs.

1. Multimodal Embedding Foundations: Shared Latent Space

The secret to multimodal retrieval is mapping both visual data (images, diagrams) and textual data (prompts, queries, captions) into a shared mathematical vector space:

2. Environment Setup & Dependencies

Configure an isolated virtual environment and install the required machine learning packages:

python3 -m venv .venv
source .venv/bin/activate
pip install "qdrant-client>=1.9.0" "open-clip-torch>=2.24.0" "torch>=2.2.0" "pillow>=10.3.0" "transformers>=4.40.0"

We utilize open-clip-torch to run open-weights CLIP models (such as ViT-B-32 pre-trained on laion2b_s34b_b79k) locally on CPU or Apple Silicon / CUDA, and qdrant-client for in-memory or clustered vector operations.

3. Implementing the CLIP Dual-Encoder Embedder

Let us build a clean, production-grade embedder that handles both image and text inputs:

import torch
import open_clip
from PIL import Image
from typing import List, Union

class MultimodalEmbedder:
    """Computes normalized embeddings for text and images in a shared CLIP space."""

    def __init__(self, model_name: str = "ViT-B-32", pretrained: str = "laion2b_s34b_b79k"):
        self.device = "cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu")
        print(f"[*] Initializing CLIP ({model_name}) on device: {self.device}")

        self.model, _, self.preprocess = open_clip.create_model_and_transforms(
            model_name, 
            pretrained=pretrained,
            device=self.device
        )
        self.tokenizer = open_clip.get_tokenizer(model_name)
        self.model.eval()

    def embed_text(self, text: str) -> List[float]:
        """Encodes a string query into a 512-dimensional normalized vector."""
        tokens = self.tokenizer([text]).to(self.device)
        with torch.no_grad():
            text_features = self.model.encode_text(tokens)
            text_features /= text_features.norm(dim=-1, keepdim=True)
        return text_features.cpu().numpy()[0].tolist()

    def embed_image(self, image_input: Union[str, Image.Image]) -> List[float]:
        """Encodes a PIL Image or image file path into a 512-dimensional normalized vector."""
        if isinstance(image_input, str):
            pil_img = Image.open(image_input).convert("RGB")
        else:
            pil_img = image_input.convert("RGB")

        preprocessed_img = self.preprocess(pil_img).unsqueeze(0).to(self.device)
        with torch.no_grad():
            image_features = self.model.encode_image(preprocessed_img)
            image_features /= image_features.norm(dim=-1, keepdim=True)
        return image_features.cpu().numpy()[0].tolist()

4. Initializing the Qdrant Multimodal Collection

Now let us initialize a Qdrant collection configured for cosine distance vector search:

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

class QdrantMultimodalStore:
    def __init__(self, collection_name: str = "multimodal_kb"):
        # Use in-memory store for development; replace with url="http://localhost:6333" in prod
        self.client = QdrantClient(location=":memory:")
        self.collection = collection_name
        self.vector_size = 512  # ViT-B-32 dimension

        # Create collection
        self.client.recreate_collection(
            collection_name=self.collection,
            vectors_config=VectorParams(size=self.vector_size, distance=Distance.COSINE)
        )
        print(f"[+] Created Qdrant collection: {self.collection}")

    def insert_artifact(self, point_id: int, vector: List[float], payload: dict):
        """Indexes a vector with associated metadata (image path, caption, category)."""
        self.client.upsert(
            collection_name=self.collection,
            points=[
                PointStruct(id=point_id, vector=vector, payload=payload)
            ]
        )

    def search_similar(self, query_vector: List[float], limit: int = 3):
        """Performs cosine vector search across the multimodal index."""
        return self.client.search(
            collection_name=self.collection,
            query_vector=query_vector,
            limit=limit
        )

5. End-to-End Pipeline: Ingesting Diagrams & Cross-Modal Search

Let us test cross-modal search: we create synthetic diagram representations, index them, and retrieve the exact corresponding image using a natural language query:

import io

def create_synthetic_diagram(title: str, color: tuple) -> Image.Image:
    """Creates a mock architectural diagram for indexing demonstration."""
    img = Image.new("RGB", (600, 400), color=color)
    return img

def main():
    embedder = MultimodalEmbedder()
    store = QdrantMultimodalStore()

    # Ingest 3 architectural diagrams
    mock_artifacts = [
        {
            "id": 1,
            "title": "Kubernetes Microservices Architecture",
            "img": create_synthetic_diagram("k8s_mesh", (30, 41, 59)),
            "caption": "A production Kubernetes cluster showing Istio ingress gateway, worker nodes, and Redis cache."
        },
        {
            "id": 2,
            "title": "PostgreSQL pgvector RAG Architecture",
            "img": create_synthetic_diagram("rag_pipeline", (6, 78, 59)),
            "caption": "A retrieval-augmented generation pipeline with PostgreSQL pgvector HNSW indexing and cross-encoder reranking."
        },
        {
            "id": 3,
            "title": "DeepSeek-R1 Local Quantization Flow",
            "img": create_synthetic_diagram("deepseek_flow", (69, 26, 3)),
            "caption": "Hardware diagram showing llama.cpp GGUF 4-bit quantization running on Apple Silicon unified memory."
        }
    ]

    print("\n[*] Embedding and indexing multimodal artifacts...")
    for item in mock_artifacts:
        # Embed the image directly into CLIP space
        img_vector = embedder.embed_image(item["img"])
        store.insert_artifact(
            point_id=item["id"],
            vector=img_vector,
            payload={"title": item["title"], "caption": item["caption"]}
        )
        print(f"    Indexed artifact #{item['id']}: {item['title']}")

    # 4. Cross-Modal Query: Text query searching for an image!
    query_text = "How does PostgreSQL store vector embeddings with HNSW indexes?"
    print(f"\n[*] Executing cross-modal text query:\n    \"{query_text}\"")

    query_vector = embedder.embed_text(query_text)
    search_results = store.search_similar(query_vector, limit=2)

    print("\n--- RETRIEVAL RESULTS ---")
    for hit in search_results:
        print(f"[Score: {hit.score:.4f}] -> Found Artifact #{hit.id}: {hit.payload['title']}")
        print(f"    Caption: {hit.payload['caption']}\n")

if __name__ == "__main__":
    main()

When executed, the text query How does PostgreSQL store vector embeddings with HNSW indexes? achieves highest cosine similarity against the PostgreSQL pgvector RAG Architecture diagram, retrieving it directly from Qdrant without needing exact keyword matches!

6. Combining Cross-Modal Retrieval with Vision LLMs

Once Qdrant returns the top-matching diagram image, pass the image file along with the user’s question to a Vision Language Model (e.g. GPT-4o or Claude 3.5 Sonnet):

import base64

def generate_multimodal_answer(image_bytes: bytes, user_query: str) -> str:
    """
    Submits retrieved diagram and query to a multimodal LLM for grounded visual synthesis.
    """
    encoded_b64 = base64.b64encode(image_bytes).decode("utf-8")
    payload = {
        "model": "gpt-4o",
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": f"Analyze this retrieved system diagram to answer: {user_query}"},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{encoded_b64}"}}
                ]
            }
        ]
    }
    return "The system uses an Istio service mesh ingress routing to worker pods..."

7. Frequently Asked Questions (FAQ)

What is the advantage of CLIP over separate text OCR models?

OCR only extracts textual characters. CLIP captures semantic layout, visual concepts, colors, icons, and diagrams, enabling queries like “show me the database failover diagram” to match even if the word “failover” is not written on the image.

Can Qdrant handle millions of multimodal vectors?

Yes. Qdrant is written in Rust and supports SIMD-accelerated HNSW indexing and scalar quantization, allowing millions of 512-dimensional vectors to fit into limited RAM with sub-10ms search latency.

How does multimodal RAG integrate with hybrid text search?

In production, use hybrid search to combine textual document chunks and visual diagram chunks, reranking them with a two-stage cross-encoder before LLM generation.