Module 6: Multimodal RAG

4. Hybrid Retrieval

Next →

Description

You have an index with text chunks and image descriptions. Now you need to search it. But searching a multimodal index isn't the same as searching a text-only index. When the user asks "how do the microservices connect?", you want to retrieve both the paragraphs explaining the connection and the architecture diagram that illustrates it. Hybrid retrieval combines text and image search, merges the results, and assigns them a unified ranking.

Hybrid retrieval solves a concrete problem: if you search by text only, the relevant diagrams fall to the bottom of the ranking. If you search by image only, the explanatory paragraphs are left out. Merging the results gives you the best of both worlds.

Why it matters: The quality of RAG answers depends directly on the quality of retrieval. Retrieval that only searches text loses visual context. Retrieval that doesn't re-rank returns disordered results. Hybrid retrieval with fusion is the difference between a system that "kind of finds something" and one that consistently retrieves the most relevant chunks.

Connection with the module: It uses the index built in capsule 03. The documents processed in capsule 05 are searched with these functions. LangChain (capsule 06) abstracts part of this, but understanding manual retrieval gives you fine-grained control.

Text Search

The foundation: a text query against the full index

import chromadb
from chromadb.utils import embedding_functions
from dotenv import load_dotenv
import os

load_dotenv()

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)


def search_by_text(
    collection,
    query: str,
    n: int = 10,
    content_type: str = None
) -> list[dict]:
    kwargs = {
        "query_texts": [query],
        "n_results": n,
    }
    if content_type:
        kwargs["where"] = {"type": content_type}

    results = collection.query(**kwargs)

    matches = []
    for i in range(len(results["documents"][0])):
        matches.append({
            "id": results["ids"][0][i],
            "content": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i],
            "score": 1 - results["distances"][0][i] / 2,
        })

    return matches

The score field

ChromaDB returns distance (cosine distance), not similarity. The conversion is:

cosine similarity = 1 - (cosine distance / 2)

Distance 0.0 → similarity 1.0  (perfect match)
Distance 1.0 → similarity 0.5  (orthogonal)
Distance 2.0 → similarity 0.0  (opposite)

Image Search

Query with a reference image

The user provides an image and wants to find similar content in the index.

from openai import OpenAI
import base64
from pathlib import Path

client = OpenAI()


def describe_image(image_path: str) -> str:
    path = Path(image_path)
    ext = path.suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp"}
    mime = mime_map.get(ext, "image/png")

    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode("utf-8")

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe this image in 1-2 sentences for semantic search."},
                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
            ]
        }],
        max_tokens=100
    )
    return response.choices[0].message.content


def search_by_image(
    collection,
    image_path: str,
    n: int = 10
) -> list[dict]:
    description = describe_image(image_path)
    results = search_by_text(collection, description, n=n)
    for r in results:
        r["query_type"] = "image"
        r["image_description"] = description
    return results

Search with CLIP (direct image)

If you have a CLIP collection, you can search directly with the image embedding without going through a textual description.

from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import torch

clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
clip_model.eval()


def search_by_image_clip(
    clip_collection,
    image_path: str,
    n: int = 10
) -> list[dict]:
    pil_image = Image.open(image_path).convert("RGB")
    inputs = clip_processor(images=pil_image, return_tensors="pt")

    with torch.no_grad():
        features = clip_model.get_image_features(**inputs)
    normalized = features / features.norm(dim=-1, keepdim=True)
    query_embedding = normalized[0].numpy().tolist()

    results = clip_collection.query(
        query_embeddings=[query_embedding],
        n_results=n
    )

    matches = []
    for i in range(len(results["documents"][0])):
        matches.append({
            "id": results["ids"][0][i],
            "content": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i],
            "score": 1 - results["distances"][0][i] / 2,
            "query_type": "clip_image",
        })
    return matches

Result Fusion

The problem

You have two lists of results: one from text search and one from image search. Each list has its own scores. You need to combine them into a single ordered list.

Text search:
  1. chunk_05 (score: 0.92) — paragraph about microservices
  2. chunk_12 (score: 0.87) — paragraph about API Gateway
  3. img_03   (score: 0.83) — diagram description

Image search:
  1. img_03   (score: 0.91) — diagram description
  2. img_07   (score: 0.78) — another diagram
  3. chunk_05 (score: 0.72) — paragraph about microservices

Fusion → Which comes first?

Strategy 1: Average score

The simplest: if a document appears in both lists, average the scores.

def merge_by_average(
    text_results: list[dict],
    image_results: list[dict],
    top_k: int = 5
) -> list[dict]:
    scores = {}
    details = {}

    for r in text_results:
        doc_id = r["id"]
        scores[doc_id] = scores.get(doc_id, [])
        scores[doc_id].append(r["score"])
        details[doc_id] = r

    for r in image_results:
        doc_id = r["id"]
        scores[doc_id] = scores.get(doc_id, [])
        scores[doc_id].append(r["score"])
        if doc_id not in details:
            details[doc_id] = r

    merged = []
    for doc_id, score_list in scores.items():
        avg_score = sum(score_list) / len(score_list)
        entry = details[doc_id].copy()
        entry["merged_score"] = avg_score
        entry["appeared_in"] = len(score_list)
        merged.append(entry)

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

Strategy 2: Reciprocal Rank Fusion (RRF)

RRF is a proven technique in information retrieval. It doesn't use scores directly — it uses the position (rank) of each document in each list. Documents that appear in high positions in both lists get high RRF scores.

Formula: rrf_score(d) = Σ 1/(k + rank_i(d)) where k is a constant (typically 60).

def reciprocal_rank_fusion(
    *result_lists: list[dict],
    k: int = 60,
    top_n: int = 5
) -> list[dict]:
    rrf_scores = {}
    doc_details = {}

    for results in result_lists:
        for rank, result in enumerate(results):
            doc_id = result["id"]
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0)
            rrf_scores[doc_id] += 1.0 / (k + rank + 1)

            if doc_id not in doc_details:
                doc_details[doc_id] = result

    ranked = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)

    output = []
    for doc_id, rrf_score in ranked[:top_n]:
        entry = doc_details[doc_id].copy()
        entry["rrf_score"] = round(rrf_score, 6)
        output.append(entry)

    return output

Why RRF works well

Example with k=60:

Text search → chunk_05 rank 0, img_03 rank 2
Image search → img_03 rank 0, chunk_05 rank 2

RRF score of chunk_05:
  1/(60+1) + 1/(60+3) = 0.01639 + 0.01587 = 0.03226

RRF score of img_03:
  1/(60+3) + 1/(60+1) = 0.01587 + 0.01639 = 0.03226

Both get the same score because they're in symmetric positions.
The document that appears at the top in both lists wins.

Strategy 3: Weighting by query type

If you know the user's question is more textual or more visual, you can weight accordingly.

def weighted_hybrid_search(
    collection,
    text_query: str = None,
    image_path: str = None,
    text_weight: float = 0.6,
    image_weight: float = 0.4,
    n: int = 10,
    top_k: int = 5
) -> list[dict]:
    scores = {}
    details = {}

    if text_query:
        text_results = search_by_text(collection, text_query, n=n)
        for r in text_results:
            doc_id = r["id"]
            scores[doc_id] = scores.get(doc_id, 0.0)
            scores[doc_id] += text_weight * r["score"]
            details[doc_id] = r

    if image_path:
        image_results = search_by_image(collection, image_path, n=n)
        for r in image_results:
            doc_id = r["id"]
            scores[doc_id] = scores.get(doc_id, 0.0)
            scores[doc_id] += image_weight * r["score"]
            if doc_id not in details:
                details[doc_id] = r

    merged = []
    for doc_id, weighted_score in scores.items():
        entry = details[doc_id].copy()
        entry["weighted_score"] = round(weighted_score, 4)
        merged.append(entry)

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

Classify queries to auto-weight

def classify_query_intent(query: str) -> dict:
    visual_keywords = {
        "diagram", "image", "photo", "chart", "table",
        "capture", "screenshot", "figure", "visual", "shows",
        "illustration", "schematic", "map", "graph", "plot"
    }
    query_words = set(query.lower().split())
    visual_matches = query_words & visual_keywords
    has_visual_intent = len(visual_matches) > 0

    if has_visual_intent:
        return {"text_weight": 0.3, "image_weight": 0.7, "reason": "visual_keywords_detected"}
    return {"text_weight": 0.7, "image_weight": 0.3, "reason": "default_text_priority"}


def auto_weighted_search(
    collection,
    query: str,
    n: int = 10,
    top_k: int = 5
) -> list[dict]:
    weights = classify_query_intent(query)

    text_results = search_by_text(collection, query, n=n)

    scores = {}
    details = {}
    for r in text_results:
        doc_id = r["id"]
        is_image = r["metadata"].get("type") == "image"
        weight = weights["image_weight"] if is_image else weights["text_weight"]
        scores[doc_id] = r["score"] * weight
        details[doc_id] = r

    merged = []
    for doc_id, weighted_score in scores.items():
        entry = details[doc_id].copy()
        entry["weighted_score"] = round(weighted_score, 4)
        entry["weight_reason"] = weights["reason"]
        merged.append(entry)

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

Complete Hybrid Retrieval Pipeline

Bring it all together in one function

def hybrid_retrieve(
    collection,
    query: str,
    image_path: str = None,
    strategy: str = "rrf",
    top_k: int = 5,
    text_weight: float = 0.6
) -> list[dict]:
    text_results = search_by_text(collection, query, n=top_k * 3)

    image_results = []
    if image_path:
        image_results = search_by_image(collection, image_path, n=top_k * 3)

    if strategy == "rrf":
        if image_results:
            return reciprocal_rank_fusion(text_results, image_results, top_n=top_k)
        return reciprocal_rank_fusion(text_results, top_n=top_k)

    elif strategy == "weighted":
        return weighted_hybrid_search(
            collection, text_query=query, image_path=image_path,
            text_weight=text_weight, image_weight=1 - text_weight,
            n=top_k * 3, top_k=top_k
        )

    elif strategy == "average":
        if image_results:
            return merge_by_average(text_results, image_results, top_k=top_k)
        return text_results[:top_k]

    else:
        raise ValueError(f"Unsupported strategy: {strategy}")


results = hybrid_retrieve(
    collection,
    query="how do the microservices connect",
    strategy="rrf",
    top_k=5
)

for i, r in enumerate(results):
    content_type = r["metadata"].get("type", "?")
    score_key = "rrf_score" if "rrf_score" in r else "score"
    print(f"  {i+1}. [{content_type}] ({r.get(score_key, 0):.4f}) {r['content'][:60]}...")

Re-ranking with an LLM

Why re-rank

Embeddings are good at finding candidates, but they don't always order them perfectly. An LLM can re-evaluate the top candidates and reorder them with better judgment.

def rerank_with_llm(
    query: str,
    candidates: list[dict],
    top_k: int = 5
) -> list[dict]:
    if not candidates:
        return []

    candidates_text = ""
    for i, c in enumerate(candidates):
        content_type = c["metadata"].get("type", "text")
        preview = c["content"][:200]
        candidates_text += f"\n[{i}] (type: {content_type}) {preview}"

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"User question: {query}\n\n"
                f"Retrieved candidates:{candidates_text}\n\n"
                "Order the candidates by relevance for answering the question. "
                "Return ONLY the indices ordered from most to least relevant, "
                "separated by commas. Example: 2,0,4,1,3"
            )
        }],
        max_tokens=50,
        temperature=0
    )

    try:
        indices_str = response.choices[0].message.content.strip()
        indices = [int(x.strip()) for x in indices_str.split(",")]
        reranked = [candidates[i] for i in indices if i < len(candidates)]
        return reranked[:top_k]
    except (ValueError, IndexError):
        return candidates[:top_k]

Complete pipeline: retrieve → re-rank

def retrieve_and_rerank(
    collection,
    query: str,
    image_path: str = None,
    retrieve_k: int = 15,
    final_k: int = 5
) -> list[dict]:
    candidates = hybrid_retrieve(
        collection, query,
        image_path=image_path,
        strategy="rrf",
        top_k=retrieve_k
    )

    reranked = rerank_with_llm(query, candidates, top_k=final_k)

    for i, r in enumerate(reranked):
        r["final_rank"] = i + 1

    return reranked

Generate an Answer with Retrieved Context

From retrieval to answer

def generate_answer(
    query: str,
    retrieved_chunks: list[dict],
    model: str = "gpt-4o"
) -> dict:
    context_parts = []
    sources = []

    for i, chunk in enumerate(retrieved_chunks):
        content_type = chunk["metadata"].get("type", "text")
        source = chunk["metadata"].get("source", "unknown")
        page = chunk["metadata"].get("page", "?")

        prefix = "[IMAGE]" if content_type == "image" else "[TEXT]"
        context_parts.append(f"{prefix} (Source: {source}, p.{page})\n{chunk['content']}")
        sources.append({"source": source, "page": page, "type": content_type})

    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are an assistant that answers questions based ONLY on the provided context. "
                    "If the context includes image descriptions marked with [IMAGE], "
                    "incorporate them into your answer, mentioning what they show. "
                    "If you can't find the answer in the context, say you don't have enough information. "
                    "Cite the sources at the end."
                )
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {query}"
            }
        ],
        max_tokens=500,
        temperature=0
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": sources,
        "chunks_used": len(retrieved_chunks),
    }

Complete pipeline: query → retrieve → rerank → answer

def rag_query(
    collection,
    query: str,
    image_path: str = None,
    strategy: str = "rrf",
    use_rerank: bool = True,
    top_k: int = 5
) -> dict:
    if use_rerank:
        chunks = retrieve_and_rerank(
            collection, query,
            image_path=image_path,
            retrieve_k=top_k * 3,
            final_k=top_k
        )
    else:
        chunks = hybrid_retrieve(
            collection, query,
            image_path=image_path,
            strategy=strategy,
            top_k=top_k
        )

    answer = generate_answer(query, chunks)
    return answer


result = rag_query(collection, "How do the microservices connect?")
print(f"Answer: {result['answer']}")
print(f"\nSources:")
for s in result["sources"]:
    print(f"  - {s['source']} p.{s['page']} ({s['type']})")

Retrieval Evaluation

Basic metrics

def evaluate_retrieval(
    collection,
    test_queries: list[dict],
    strategy: str = "rrf",
    top_k: int = 5
) -> dict:
    """
    test_queries: [{"query": "...", "expected_ids": ["id1", "id2"]}, ...]
    """
    total_precision = 0.0
    total_recall = 0.0
    total_mrr = 0.0

    for tq in test_queries:
        results = hybrid_retrieve(
            collection, tq["query"],
            strategy=strategy, top_k=top_k
        )
        retrieved_ids = [r["id"] for r in results]
        expected = set(tq["expected_ids"])

        hits = [1 if rid in expected else 0 for rid in retrieved_ids]
        precision = sum(hits) / len(hits) if hits else 0
        recall = sum(hits) / len(expected) if expected else 0

        mrr = 0.0
        for i, hit in enumerate(hits):
            if hit:
                mrr = 1.0 / (i + 1)
                break

        total_precision += precision
        total_recall += recall
        total_mrr += mrr

    n = len(test_queries)
    return {
        "avg_precision": round(total_precision / n, 4) if n else 0,
        "avg_recall": round(total_recall / n, 4) if n else 0,
        "mrr": round(total_mrr / n, 4) if n else 0,
        "queries_evaluated": n,
    }

Troubleshooting

Image results always fall to the bottom of the ranking

Image descriptions tend to be shorter and more generic than text chunks. Detailed text embeddings tend to have higher similarity with queries.

Solution 1: Use RRF instead of direct scores — RRF normalizes by position, not by score.
Solution 2: Enrich image descriptions with more context (page, section).
Solution 3: Use weighting that boosts image results.

LLM re-ranking is slow

Re-ranking adds an extra LLM call per query.

Solution: Only re-rank the top 10-15 candidates, not the full index.
For production, consider a dedicated re-ranking model (Cohere Rerank, cross-encoders).

Duplicate results

If you index the same document twice, it appears duplicated in the results.

def deduplicate_results(results: list[dict]) -> list[dict]:
    seen_content = set()
    unique = []
    for r in results:
        content_hash = hash(r["content"][:100])
        if content_hash not in seen_content:
            seen_content.add(content_hash)
            unique.append(r)
    return unique

Ambiguous queries return irrelevant results

Solution: Rewrite the query before searching:

def expand_query(query: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Rewrite this question for semantic search. "
                f"Add synonyms and related terms. "
                f"Return only the expanded query.\n\nQuestion: {query}"
            )
        }],
        max_tokens=100,
        temperature=0
    )
    return response.choices[0].message.content

Exercises

Exercise 1: RRF with multiple lists

Implement RRF that combines 3+ result lists (for example: text, image, and metadata).

See solution 
def multi_source_rrf(
    collection,
    query: str,
    top_k: int = 5,
    k: int = 60
) -> list[dict]:
    all_results = search_by_text(collection, query, n=top_k * 3)
    text_only = search_by_text(collection, query, n=top_k * 3, content_type="text")
    image_only = search_by_text(collection, query, n=top_k * 3, content_type="image")

    return reciprocal_rank_fusion(
        all_results,
        text_only,
        image_only,
        k=k,
        top_n=top_k
    )


results = multi_source_rrf(collection, "microservices", top_k=5)
for r in results:
    print(f"  RRF={r['rrf_score']:.6f} [{r['metadata'].get('type')}] {r['content'][:50]}...")

Exercise 2: Weight text vs image dynamically

Implement a system that detects whether the question is more "visual" or more "textual" and adjusts the weights automatically.

See solution 
def dynamic_weighted_search(
    collection,
    query: str,
    top_k: int = 5
) -> list[dict]:
    weights = classify_query_intent(query)
    print(f"  Detected weights: text={weights['text_weight']}, image={weights['image_weight']}")

    results = search_by_text(collection, query, n=top_k * 3)

    for r in results:
        is_image = r["metadata"].get("type") == "image"
        w = weights["image_weight"] if is_image else weights["text_weight"]
        r["dynamic_score"] = r["score"] * w

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


for q in ["architecture diagram", "returns policy", "show the price table"]:
    print(f"\nQuery: {q}")
    results = dynamic_weighted_search(collection, q)
    for r in results:
        print(f"  [{r['metadata'].get('type')}] {r['dynamic_score']:.4f} — {r['content'][:50]}...")

Exercise 3: Compare fusion strategies

Given the same query, run all three strategies (average, RRF, weighted) and compare the rankings.

See solution 
def compare_fusion_strategies(
    collection,
    query: str,
    top_k: int = 5
) -> dict:
    strategies = {}

    for strategy in ["average", "rrf", "weighted"]:
        results = hybrid_retrieve(
            collection, query,
            strategy=strategy,
            top_k=top_k
        )
        strategies[strategy] = [
            {"id": r["id"], "type": r["metadata"].get("type", "?")}
            for r in results
        ]

    print(f"Query: {query}\n")
    for name, results in strategies.items():
        print(f"  {name}:")
        for i, r in enumerate(results):
            print(f"    {i+1}. [{r['type']}] {r['id']}")

    avg_ids = [r["id"] for r in strategies["average"]]
    rrf_ids = [r["id"] for r in strategies["rrf"]]
    weighted_ids = [r["id"] for r in strategies["weighted"]]

    overlap_all = set(avg_ids) & set(rrf_ids) & set(weighted_ids)
    print(f"\n  Overlap among the 3: {len(overlap_all)}/{top_k}")

    return strategies

Summary

  • Text search is the foundation: query → embedding → search in the vector store.
  • Image search describes the image with Vision and searches with the resulting text (or uses CLIP directly).
  • Result fusion combines lists from different sources with techniques like score averaging, RRF, or weighting.
  • RRF is the most robust technique: it uses positions in the ranking, not raw scores.
  • Dynamic weighting adjusts the weights according to the query type (textual vs visual).
  • LLM re-ranking improves precision by reordering the top candidates with a language model.
  • Answer generation uses the retrieved context (text + images) to produce an informed answer.
  • Evaluate retrieval with precision, recall and MRR over test queries.

Additional Resources

  1. Reciprocal Rank Fusion (Cormack et al.) — Original RRF paper
  2. Pinecone: Hybrid Search — Hybrid search guide
  3. Cohere Rerank — Re-ranking as a service
  4. Cross-Encoders for Re-ranking — Re-ranking with Sentence Transformers
  5. RAG Evaluation Metrics — RAG evaluation framework
← Previous3. Multimodal IndexingNext →5. Documents with Images