Módulo 6: RAG Multimodal

4. Retrieval Híbrido

Siguiente →

Descripción

Tienes un índice con chunks de texto y descripciones de imágenes. Ahora necesitas buscar en él. Pero buscar en un índice multimodal no es lo mismo que buscar en un índice de solo texto. Cuando el usuario pregunta "¿cómo se conectan los microservicios?", quieres recuperar tanto los párrafos que explican la conexión como el diagrama de arquitectura que la ilustra. El retrieval híbrido combina búsqueda por texto y por imagen, fusiona los resultados, y les asigna un ranking unificado.

El retrieval híbrido resuelve un problema concreto: si buscas solo por texto, los diagramas relevantes quedan abajo del ranking. Si buscas solo por imagen, los párrafos explicativos quedan fuera. La fusión de resultados te da lo mejor de ambos mundos.

Por qué importa: La calidad de las respuestas en RAG depende directamente de la calidad del retrieval. Un retrieval que solo busca en texto pierde contexto visual. Un retrieval que no hace re-ranking devuelve resultados desordenados. El retrieval híbrido con fusión es la diferencia entre un sistema que "más o menos encuentra algo" y uno que consistentemente recupera los chunks más relevantes.

Conexión con el módulo: Usa el índice construido en la cápsula 03. Los documentos procesados en la cápsula 05 se buscan con estas funciones. LangChain (cápsula 06) abstrae parte de esto, pero entender el retrieval manual te da control fino.

Búsqueda por Texto

La base: query de texto contra el índice completo

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

El campo score

ChromaDB retorna distance (distancia coseno), no similitud. La conversión es:

similitud coseno = 1 - (distancia coseno / 2)

Distancia 0.0 → similitud 1.0  (match perfecto)
Distancia 1.0 → similitud 0.5  (ortogonal)
Distancia 2.0 → similitud 0.0  (opuesto)

Búsqueda por Imagen

Query con una imagen de referencia

El usuario proporciona una imagen y quiere encontrar contenido similar en el índice.

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 esta imagen en 1-2 oraciones para búsqueda semántica."},
                {"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

Búsqueda con CLIP (imagen directa)

Si tienes una collection CLIP, puedes buscar directamente con el embedding de la imagen sin pasar por descripción textual.

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

Fusión de Resultados

El problema

Tienes dos listas de resultados: una de búsqueda por texto y otra de búsqueda por imagen. Cada lista tiene sus propios scores. Necesitas combinarlas en una sola lista ordenada.

Búsqueda por texto:
  1. chunk_05 (score: 0.92) — párrafo sobre microservicios
  2. chunk_12 (score: 0.87) — párrafo sobre API Gateway
  3. img_03   (score: 0.83) — descripción de diagrama

Búsqueda por imagen:
  1. img_03   (score: 0.91) — descripción de diagrama
  2. img_07   (score: 0.78) — otro diagrama
  3. chunk_05 (score: 0.72) — párrafo sobre microservicios

Fusión → ¿Qué va primero?

Estrategia 1: Score promedio

La más simple: si un documento aparece en ambas listas, promedia los 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]

Estrategia 2: Reciprocal Rank Fusion (RRF)

RRF es una técnica probada en information retrieval. No usa scores directamente — usa la posición (rank) de cada documento en cada lista. Los documentos que aparecen en altas posiciones en ambas listas obtienen scores RRF altos.

Fórmula: score_rrf(d) = Σ 1/(k + rank_i(d)) donde k es una constante (típicamente 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

Por qué RRF funciona bien

Ejemplo con k=60:

Búsqueda texto → chunk_05 rank 0, img_03 rank 2
Búsqueda imagen → img_03 rank 0, chunk_05 rank 2

Score RRF de chunk_05:
  1/(60+1) + 1/(60+3) = 0.01639 + 0.01587 = 0.03226

Score RRF de img_03:
  1/(60+3) + 1/(60+1) = 0.01587 + 0.01639 = 0.03226

Ambos obtienen el mismo score porque están en posiciones simétricas.
El documento que aparece arriba en ambas listas gana.

Estrategia 3: Ponderación por tipo de query

Si sabes que la pregunta del usuario es más textual o más visual, puedes ponderar.

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]

Clasificar queries para auto-ponderar

def classify_query_intent(query: str) -> dict:
    visual_keywords = {
        "diagrama", "imagen", "foto", "gráfico", "tabla",
        "captura", "screenshot", "figura", "visual", "muestra",
        "ilustración", "esquema", "mapa", "chart", "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]

Pipeline Completo de Retrieval Híbrido

Reunir todo en una función

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"Estrategia no soportada: {strategy}")


results = hybrid_retrieve(
    collection,
    query="cómo se conectan los microservicios",
    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 con LLM

Por qué re-rankear

Los embeddings son buenos para encontrar candidatos, pero no siempre ordenan perfectamente. Un LLM puede re-evaluar los top candidatos y reordenarlos con mejor criterio.

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}] (tipo: {content_type}) {preview}"

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Pregunta del usuario: {query}\n\n"
                f"Candidatos recuperados:{candidates_text}\n\n"
                "Ordena los candidatos por relevancia para responder la pregunta. "
                "Devuelve SOLO los índices ordenados de más a menos relevante, "
                "separados por comas. Ejemplo: 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]

Pipeline completo: 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

Generar Respuesta con Contexto Recuperado

De retrieval a respuesta

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", "desconocido")
        page = chunk["metadata"].get("page", "?")

        prefix = "[IMAGEN]" if content_type == "image" else "[TEXTO]"
        context_parts.append(f"{prefix} (Fuente: {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": (
                    "Eres un asistente que responde preguntas basándose SOLO en el contexto proporcionado. "
                    "Si el contexto incluye descripciones de imágenes marcadas con [IMAGEN], "
                    "incorpóralas en tu respuesta mencionando qué muestran. "
                    "Si no encuentras la respuesta en el contexto, di que no tienes información suficiente. "
                    "Cita las fuentes al final."
                )
            },
            {
                "role": "user",
                "content": f"Contexto:\n{context}\n\nPregunta: {query}"
            }
        ],
        max_tokens=500,
        temperature=0
    )

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

Pipeline completo: 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, "¿Cómo se conectan los microservicios?")
print(f"Respuesta: {result['answer']}")
print(f"\nFuentes:")
for s in result["sources"]:
    print(f"  - {s['source']} p.{s['page']} ({s['type']})")

Evaluación del Retrieval

Métricas básicas

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

Los resultados de imagen siempre quedan abajo del ranking

Las descripciones de imagen suelen ser más cortas y genéricas que los chunks de texto. Los embeddings de texto detallado tienden a tener mayor similitud con las queries.

Solución 1: Usa RRF en lugar de scores directos — RRF normaliza por posición, no por score.
Solución 2: Enriquece las descripciones de imagen con más contexto (página, sección).
Solución 3: Usa ponderación que booste resultados de imagen.

El re-ranking con LLM es lento

El re-ranking añade una llamada adicional al LLM por query.

Solución: Solo re-rankea los top 10-15 candidatos, no el índice completo.
Para producción, considera un modelo de re-ranking dedicado (Cohere Rerank, cross-encoders).

Resultados duplicados

Si indexas el mismo documento dos veces, aparece duplicado en resultados.

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

Queries ambiguas retornan resultados irrelevantes

Solución: Reescribir el query antes de buscar:

def expand_query(query: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Reescribe esta pregunta para búsqueda semántica. "
                f"Añade sinónimos y términos relacionados. "
                f"Solo devuelve la query expandida.\n\nPregunta: {query}"
            )
        }],
        max_tokens=100,
        temperature=0
    )
    return response.choices[0].message.content

Ejercicios

Ejercicio 1: RRF con múltiples listas

Implementa RRF que combine 3+ listas de resultados (por ejemplo: texto, imagen, y metadata).

Ver solución 
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, "microservicios", top_k=5)
for r in results:
    print(f"  RRF={r['rrf_score']:.6f} [{r['metadata'].get('type')}] {r['content'][:50]}...")

Ejercicio 2: Ponderar texto vs imagen dinámicamente

Implementa un sistema que detecte si la pregunta es más "visual" o más "textual" y ajuste los pesos automáticamente.

Ver solución 
def dynamic_weighted_search(
    collection,
    query: str,
    top_k: int = 5
) -> list[dict]:
    weights = classify_query_intent(query)
    print(f"  Pesos detectados: texto={weights['text_weight']}, imagen={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 ["diagrama de arquitectura", "política de devoluciones", "muestra la tabla de precios"]:
    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]}...")

Ejercicio 3: Comparar estrategias de fusión

Dada una misma query, ejecuta las tres estrategias (average, RRF, weighted) y compara los rankings.

Ver solución 
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 entre las 3: {len(overlap_all)}/{top_k}")

    return strategies

Resumen

  • Búsqueda por texto es la base: query → embedding → buscar en vector store.
  • Búsqueda por imagen describe la imagen con Vision y busca con el texto resultante (o usa CLIP directamente).
  • Fusión de resultados combina listas de diferentes fuentes con técnicas como promedio de scores, RRF, o ponderación.
  • RRF es la técnica más robusta: usa posiciones en el ranking, no scores crudos.
  • Ponderación dinámica ajusta los pesos según el tipo de query (textual vs visual).
  • Re-ranking con LLM mejora la precisión reordenando los top candidatos con un modelo de lenguaje.
  • La generación de respuesta usa el contexto recuperado (texto + imágenes) para producir una respuesta informada.
  • Evalúa el retrieval con precision, recall y MRR sobre queries de test.

Recursos Adicionales

  1. Reciprocal Rank Fusion (Cormack et al.) — Paper original de RRF
  2. Pinecone: Hybrid Search — Guía de búsqueda híbrida
  3. Cohere Rerank — Re-ranking como servicio
  4. Cross-Encoders for Re-ranking — Re-ranking con Sentence Transformers
  5. RAG Evaluation Metrics — Framework de evaluación de RAG
← Anterior3. Indexación MultimodalSiguiente →5. Documentos con Imágenes