Multimodal RAG Engineering [2026]: Cross-Modal Retrieval

Score(q, d) = (1/M) * Σᵢ max_j(qᵢ · dⱼ)

final_score = α * normalize(dense_score) + (1 - α) * normalize(late_interaction_score)

import torch
import numpy as np
from PIL import Image
from pathlib import Path
from dataclasses import dataclass
from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, Distance, PointStruct,
    Filter, FieldCondition, MatchValue
)
from transformers import AutoModel, AutoProcessor, AutoTokenizer
from colpali_engine.models import ColPali, ColPaliProcessor


@dataclass
class RetrievalResult:
    content_id: str
    modality: str  # "text" | "image" | "document"
    score: float
    metadata: dict


class MultimodalRAGPipeline:
    """Production hybrid retrieval: SigLIP dense + ColPali re-ranking."""

    def __init__(self, qdrant_url: str = "localhost", collection: str = "multimodal_kb"):
        self.collection = collection
        self.client = QdrantClient(host=qdrant_url, port=6333)

        # Stage 1: Dense embedding (SigLIP-SO400M)
        self.siglip = AutoModel.from_pretrained(
            "google/siglip-so400m-patch14-384"
        ).eval().cuda()
        self.siglip_processor = AutoProcessor.from_pretrained(
            "google/siglip-so400m-patch14-384"
        )

        # Stage 2: Late-interaction re-ranker (ColPali v1.3)
        self.colpali = ColPali.from_pretrained(
            "vidore/colpali-v1.3",
            torch_dtype=torch.bfloat16
        ).eval().cuda()
        self.colpali_processor = ColPaliProcessor.from_pretrained(
            "vidore/colpali-v1.3"
        )

        self._ensure_collection()

    def _ensure_collection(self):
        """Create Qdrant collection with SigLIP vector config."""
        if not self.client.collection_exists(self.collection):
            self.client.create_collection(
                collection_name=self.collection,
                vectors_config=VectorParams(
                    size=1152,  # SigLIP-SO400M dimension
                    distance=Distance.COSINE
                )
            )

    def embed_image(self, image: Image.Image) -> np.ndarray:
        """Generate dense SigLIP embedding for an image."""
        inputs = self.siglip_processor(images=image, return_tensors="pt").to("cuda")
        with torch.no_grad():
            embedding = self.siglip.get_image_features(**inputs)
        return embedding.cpu().numpy().flatten()

    def embed_text(self, text: str) -> np.ndarray:
        """Generate dense SigLIP embedding for text."""
        inputs = self.siglip_processor(text=[text], return_tensors="pt", padding=True).to("cuda")
        with torch.no_grad():
            embedding = self.siglip.get_text_features(**inputs)
        return embedding.cpu().numpy().flatten()

    def colpali_embed_page(self, image: Image.Image) -> torch.Tensor:
        """Generate multi-vector ColPali embedding for a document page."""
        inputs = self.colpali_processor(images=[image], return_tensors="pt").to("cuda")
        with torch.no_grad():
            embeddings = self.colpali(**inputs)
        return embeddings  # Shape: [1, N_patches, 128]

    def colpali_embed_query(self, query: str) -> torch.Tensor:
        """Generate multi-vector ColPali embedding for a query."""
        inputs = self.colpali_processor(text=[query], return_tensors="pt").to("cuda")
        with torch.no_grad():
            embeddings = self.colpali(**inputs)
        return embeddings  # Shape: [1, M_tokens, 128]

    # --- Ingestion ---

    def ingest_document_page(self, page_image: Image.Image, doc_id: str, page_num: int, metadata: dict = None):
        """Ingest a document page: store dense vector + raw image for re-ranking."""
        dense_vector = self.embed_image(page_image)

        point = PointStruct(
            id=hash(f"{doc_id}_p{page_num}") % (2**63),
            vector=dense_vector.tolist(),
            payload={
                "doc_id": doc_id,
                "page_num": page_num,
                "modality": "document",
                "image_path": f"pages/{doc_id}/{page_num}.png",
                **(metadata or {})
            }
        )
        self.client.upsert(collection_name=self.collection, points=[point])

        # Save page image for ColPali re-ranking
        save_path = Path(f"pages/{doc_id}")
        save_path.mkdir(parents=True, exist_ok=True)
        page_image.save(save_path / f"{page_num}.png")

    def ingest_text_chunk(self, text: str, chunk_id: str, metadata: dict = None):
        """Ingest a text chunk with dense embedding."""
        dense_vector = self.embed_text(text)

        point = PointStruct(
            id=hash(chunk_id) % (2**63),
            vector=dense_vector.tolist(),
            payload={
                "chunk_id": chunk_id,
                "text": text,
                "modality": "text",
                **(metadata or {})
            }
        )
        self.client.upsert(collection_name=self.collection, points=[point])

    # --- Retrieval ---

    def retrieve(self, query: str, top_k: int = 10, rerank: bool = True) -> list[RetrievalResult]:
        """Two-stage retrieval: dense recall + ColPali re-ranking."""
        # Stage 1: Dense retrieval (top-100)
        query_vector = self.embed_text(query)
        candidates = self.client.search(
            collection_name=self.collection,
            query_vector=query_vector.tolist(),
            limit=100
        )

        if not rerank:
            return self._format_results(candidates[:top_k])

        # Stage 2: ColPali re-ranking for document pages
        query_embeds = self.colpali_embed_query(query)
        reranked = []

        for candidate in candidates:
            payload = candidate.payload
            if payload["modality"] == "document":
                # Load page image and compute late-interaction score
                page_img = Image.open(payload["image_path"])
                doc_embeds = self.colpali_embed_page(page_img)
                late_score = self._maxsim_score(query_embeds, doc_embeds)

                # Fuse scores: α=0.35 favors late-interaction
                fused = 0.35 * candidate.score + 0.65 * late_score
                reranked.append((candidate, fused))
            else:
                # Text chunks keep their dense score (no late-interaction)
                reranked.append((candidate, candidate.score))

        # Sort by fused score and return top-K
        reranked.sort(key=lambda x: x[1], reverse=True)
        return self._format_results_with_scores(reranked[:top_k])

    def _maxsim_score(self, query_embeds: torch.Tensor, doc_embeds: torch.Tensor) -> float:
        """Compute MaxSim late-interaction score."""
        # query_embeds: [1, M, 128], doc_embeds: [1, N, 128]
        sim_matrix = torch.einsum("bmd,bnd->bmn", query_embeds, doc_embeds)
        max_sim_per_query_token = sim_matrix.max(dim=-1).values  # [1, M]
        score = max_sim_per_query_token.mean().item()
        return score

    def _format_results(self, candidates) -> list[RetrievalResult]:
        return [
            RetrievalResult(
                content_id=c.payload.get("chunk_id") or c.payload.get("doc_id"),
                modality=c.payload["modality"],
                score=c.score,
                metadata=c.payload
            )
            for c in candidates
        ]

    def _format_results_with_scores(self, reranked) -> list[RetrievalResult]:
        return [
            RetrievalResult(
                content_id=c.payload.get("chunk_id") or c.payload.get("doc_id"),
                modality=c.payload["modality"],
                score=score,
                metadata=c.payload
            )
            for c, score in reranked
        ]


# --- Usage Example ---

if __name__ == "__main__":
    pipeline = MultimodalRAGPipeline(qdrant_url="localhost")

    # Ingest a PDF: convert pages to images, then ingest
    from pdf2image import convert_from_path
    pages = convert_from_path("financial_report_2026.pdf", dpi=200)
    for i, page in enumerate(pages):
        pipeline.ingest_document_page(page, doc_id="fin_report_2026", page_num=i)

    # Ingest text chunks
    pipeline.ingest_text_chunk(
        "Revenue grew 34% YoY reaching $2.1B in Q4 2025.",
        chunk_id="fin_report_2026_summary_01"
    )

    # Query: cross-modal retrieval
    results = pipeline.retrieve("What was the revenue trend in Q4?")
    for r in results[:5]:
        print(f"[{r.modality}] score={r.score:.4f} - {r.content_id}")

    # Output:
    # [document] score=0.8934 - fin_report_2026 (page 12 - revenue chart)
    # [text] score=0.8721 - fin_report_2026_summary_01
    # [document] score=0.8156 - fin_report_2026 (page 3 - exec summary)

import { QdrantClient } from "@qdrant/js-client-rest";
import { pipeline, env } from "@xenova/transformers";
import sharp from "sharp";
import * as fs from "fs/promises";

// Configure transformers.js for server-side
env.cacheDir = "./model-cache";

interface RetrievalResult {
  contentId: string;
  modality: "text" | "image" | "document";
  score: number;
  metadata: Record<string, unknown>;
}

interface CandidateResult {
  id: string | number;
  score: number;
  payload: Record<string, unknown>;
}

class MultimodalRAGService {
  private client: QdrantClient;
  private collection: string;
  private embedder: Awaited<ReturnType<typeof pipeline>> | null = null;

  constructor(qdrantUrl = "http://localhost:6333", collection = "multimodal_kb") {
    this.client = new QdrantClient({ url: qdrantUrl });
    this.collection = collection;
  }

  async initialize(): Promise<void> {
    // Load CLIP model via transformers.js (runs on CPU/WASM in Node)
    this.embedder = await pipeline(
      "feature-extraction",
      "Xenova/clip-vit-base-patch32"
    );

    // Ensure collection exists
    const collections = await this.client.getCollections();
    const exists = collections.collections.some(c => c.name === this.collection);
    if (!exists) {
      await this.client.createCollection(this.collection, {
        vectors: { size: 512, distance: "Cosine" }
      });
    }
  }

  async embedText(text: string): Promise<number[]> {
    if (!this.embedder) throw new Error("Call initialize() first");
    const output = await this.embedder(text, { pooling: "mean", normalize: true });
    return Array.from(output.data as Float32Array).slice(0, 512);
  }

  async embedImage(imagePath: string): Promise<number[]> {
    if (!this.embedder) throw new Error("Call initialize() first");
    // Preprocess image to 224x224 RGB
    const buffer = await sharp(imagePath)
      .resize(224, 224)
      .toFormat("png")
      .toBuffer();

    const output = await this.embedder(buffer, { pooling: "mean", normalize: true });
    return Array.from(output.data as Float32Array).slice(0, 512);
  }

  // --- Ingestion ---

  async ingestTextChunk(text: string, chunkId: string, metadata: Record<string, unknown> = {}): Promise<void> {
    const vector = await this.embedText(text);

    await this.client.upsert(this.collection, {
      points: [{
        id: this.hashId(chunkId),
        vector,
        payload: { chunkId, text, modality: "text", ...metadata }
      }]
    });
  }

  async ingestImage(imagePath: string, imageId: string, metadata: Record<string, unknown> = {}): Promise<void> {
    const vector = await this.embedImage(imagePath);

    await this.client.upsert(this.collection, {
      points: [{
        id: this.hashId(imageId),
        vector,
        payload: { imageId, imagePath, modality: "image", ...metadata }
      }]
    });
  }

  // --- Retrieval ---

  async retrieve(query: string, topK = 10): Promise<RetrievalResult[]> {
    const queryVector = await this.embedText(query);

    const results = await this.client.search(this.collection, {
      vector: queryVector,
      limit: topK,
      with_payload: true
    });

    return results.map((r: CandidateResult) => ({
      contentId: (r.payload.chunkId || r.payload.imageId) as string,
      modality: r.payload.modality as "text" | "image" | "document",
      score: r.score,
      metadata: r.payload
    }));
  }

  async hybridRetrieve(
    query: string,
    topK = 10,
    modalityWeights: Record<string, number> = { text: 1.0, image: 0.9, document: 0.95 }
  ): Promise<RetrievalResult[]> {
    const queryVector = await this.embedText(query);

    // Overretrieve for re-ranking headroom
    const candidates = await this.client.search(this.collection, {
      vector: queryVector,
      limit: topK * 5,
      with_payload: true
    });

    // Apply modality-aware score calibration
    const calibrated = candidates.map((r: CandidateResult) => {
      const modality = r.payload.modality as string;
      const weight = modalityWeights[modality] ?? 1.0;
      return {
        ...r,
        calibratedScore: r.score * weight
      };
    });

    // Sort by calibrated score
    calibrated.sort((a, b) => b.calibratedScore - a.calibratedScore);

    return calibrated.slice(0, topK).map(r => ({
      contentId: (r.payload.chunkId || r.payload.imageId) as string,
      modality: r.payload.modality as "text" | "image" | "document",
      score: r.calibratedScore,
      metadata: r.payload
    }));
  }

  private hashId(str: string): number {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
      const char = str.charCodeAt(i);
      hash = ((hash << 5) - hash) + char;
      hash = hash & hash; // Convert to 32-bit integer
    }
    return Math.abs(hash);
  }
}

// --- Usage ---

async function main() {
  const rag = new MultimodalRAGService();
  await rag.initialize();

  // Ingest text
  await rag.ingestTextChunk(
    "The transformer architecture uses self-attention mechanisms to process sequences in parallel.",
    "ml_textbook_ch3_p1",
    { source: "ml_textbook", chapter: 3 }
  );

  // Ingest image
  await rag.ingestImage(
    "./diagrams/transformer_architecture.png",
    "transformer_arch_diagram",
    { source: "ml_textbook", chapter: 3, type: "architecture_diagram" }
  );

  // Cross-modal query: text query finds relevant image
  const results = await rag.hybridRetrieve("How does the transformer process input sequences?");
  console.log("Results:", JSON.stringify(results, null, 2));

  // Expected output:
  // [
  //   { contentId: "transformer_arch_diagram", modality: "image", score: 0.847, ... },
  //   { contentId: "ml_textbook_ch3_p1", modality: "text", score: 0.823, ... }
  // ]
}

main().catch(console.error);

def maxsim_rerank(query_tokens: torch.Tensor, doc_tokens: torch.Tensor) -> float:
    """
    query_tokens: [M, 128] - M query token embeddings
    doc_tokens: [N, 128] - N document patch embeddings
    """
    # Compute all-pairs similarity
    sim_matrix = query_tokens @ doc_tokens.T  # [M, N]
    # For each query token, take max similarity across all doc tokens
    max_sims = sim_matrix.max(dim=1).values  # [M]
    # Average across query tokens
    return max_sims.mean().item()

def calibrated_fusion(
    candidates: list[dict],
    alpha: float = 0.6  # Weight for late-interaction score
) -> list[dict]:
    """Fuse dense and late-interaction scores with per-modality calibration."""

    # Group by modality
    by_modality = {}
    for c in candidates:
        mod = c["modality"]
        by_modality.setdefault(mod, []).append(c)

    # Normalize scores within each modality to [0, 1]
    for mod, items in by_modality.items():
        dense_scores = [c["dense_score"] for c in items]
        min_d, max_d = min(dense_scores), max(dense_scores)
        range_d = max_d - min_d or 1.0

        for c in items:
            c["norm_dense"] = (c["dense_score"] - min_d) / range_d
            if "late_score" in c:
                c["final_score"] = (1 - alpha) * c["norm_dense"] + alpha * c["late_score"]
            else:
                c["final_score"] = c["norm_dense"]

    # Flatten and sort
    all_items = [c for items in by_modality.values() for c in items]
    all_items.sort(key=lambda x: x["final_score"], reverse=True)
    return all_items

async def vlm_rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
    """Use a VLM as a cross-encoder re-ranker (expensive, high precision)."""
    import openai

    scored = []
    for candidate in candidates[:20]:  # Limit to top-20 for cost
        if candidate["modality"] == "document":
            response = await openai.chat.completions.create(
                model="gpt-4o",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": f"Rate relevance 0-10: Query: '{query}'"},
                        {"type": "image_url", "image_url": {"url": candidate["image_url"]}}
                    ]
                }],
                max_tokens=10
            )
            score = int(response.choices[0].message.content.strip()) / 10
        else:
            score = candidate["dense_score"]  # Text candidates keep original score

        scored.append({**candidate, "vlm_score": score})

    scored.sort(key=lambda x: x["vlm_score"], reverse=True)
    return scored[:top_k]

class AlignmentDriftMonitor:
    """Monitor cross-modal alignment quality over time."""

    def __init__(self, window_size: int = 1000):
        self.window_size = window_size
        self.text_text_scores = []
        self.text_image_scores = []
        self.baseline_gap = None

    def record_retrieval(self, query: str, results: list[RetrievalResult]):
        """Record score distributions from each retrieval."""
        for r in results:
            if r.modality == "text":
                self.text_text_scores.append(r.score)
            elif r.modality in ("image", "document"):
                self.text_image_scores.append(r.score)

        # Keep rolling window
        self.text_text_scores = self.text_text_scores[-self.window_size:]
        self.text_image_scores = self.text_image_scores[-self.window_size:]

    def compute_modality_gap(self) -> float:
        """Compute the mean score difference between modalities."""
        if not self.text_text_scores or not self.text_image_scores:
            return 0.0
        return np.mean(self.text_text_scores) - np.mean(self.text_image_scores)

    def check_drift(self, threshold: float = 0.05) -> dict:
        """Check if alignment has drifted beyond threshold."""
        current_gap = self.compute_modality_gap()

        if self.baseline_gap is None:
            self.baseline_gap = current_gap
            return {"drifted": False, "gap": current_gap, "baseline": current_gap}

        drift = abs(current_gap - self.baseline_gap)
        return {
            "drifted": drift > threshold,
            "gap": current_gap,
            "baseline": self.baseline_gap,
            "drift_magnitude": drift
        }

from sentence_transformers import SentenceTransformer, InputExample, losses

model = SentenceTransformer("google/siglip-so400m-patch14-384")

# Create training pairs from your domain
train_examples = [
    InputExample(texts=["revenue chart Q4 2025", "path/to/q4_chart.png"], label=1.0),
    InputExample(texts=["org structure diagram", "path/to/unrelated_chart.png"], label=0.0),
]

train_loss = losses.CosineSimilarityLoss(model)
model.fit(train_objectives=[(train_dataloader, train_loss)], epochs=3)

def adaptive_fusion_weight(query: str) -> float:
    """Return alpha (late-interaction weight) based on query type."""
    visual_indicators = ["chart", "diagram", "figure", "table", "image", "show", "graph"]
    text_indicators = ["define", "explain", "what is", "describe", "summarize"]

    query_lower = query.lower()
    if any(ind in query_lower for ind in visual_indicators):
        return 0.8  # Trust late-interaction more for visual queries
    elif any(ind in query_lower for ind in text_indicators):
        return 0.3  # Trust dense embedding more for text queries
    return 0.6  # Default balanced

metrics = {
    "text_recall@5": compute_recall(text_results, text_ground_truth),
    "image_recall@5": compute_recall(image_results, image_ground_truth),
    "document_recall@5": compute_recall(doc_results, doc_ground_truth),
    "cross_modal_recall@5": compute_recall(all_results, mixed_ground_truth),
}