Cápsula 03: Query expansion — cuando una query no es suficiente

Descripción de la cápsula

En la cápsula 02 viste que las queries del usuario son frecuentemente ambiguas, incompletas o mal formuladas. La técnica más directa para resolver el problema de ambigüedad es query expansion: en vez de buscar con la query original (única, posiblemente mal interpretada), generas múltiples versiones que cubren las interpretaciones probables, buscas con cada una, y combinas los rankings.

El insight pedagógico clave: si el usuario escribe "fastapi auth", no sabes si quiere OAuth2, JWT, API keys, basic auth o sesiones. Pero puedes generar 5 queries que cubran las 5 interpretaciones, buscar con cada una, y devolverle al LLM los mejores resultados de las 5 búsquedas combinadas. La probabilidad de capturar la respuesta correcta sube de ~50% (con la query original ambigua) a ~85% (con las 5 expandidas).

Esta cápsula te enseña cómo generar expansions de calidad con un LLM (no es trivial — un prompt mal diseñado produce expansions ruidosas), cómo combinar resultados con Reciprocal Rank Fusion, y cómo decidir cuándo expandir vs cuándo usar la query directa.

Al finalizar esta cápsula serás capaz de:

• ✅ Generar expansiones de queries con un LLM usando prompts calibrados
• ✅ Implementar Reciprocal Rank Fusion (RRF) para combinar múltiples rankings
• ✅ Decidir cuándo expandir (queries ambiguas) vs cuándo no (queries específicas)
• ✅ Calcular el costo extra (latencia + LLM calls + retrieval calls) y justificarlo
• ✅ Anticipar la trampa más cara: expansiones que cambian la intención del usuario
• ✅ Implementar expansion con cache para queries repetidas

Tiempo estimado: 30-35 minutos

El insight: una query, múltiples interpretaciones, múltiples búsquedas

Piensa en cómo googleas cuando no encuentras algo. La primera query no funciona, así que pruebas variantes:

Query 1: "fastapi auth"             → no encontró lo que querías
Query 2: "fastapi oauth2"           → encontró pero no específico
Query 3: "fastapi password bearer"  → encontró exactamente lo que querías

Lo que tú haces manualmente con 3 queries iterativas, query expansion lo hace automáticamente con 5 queries en paralelo y combina los resultados:

       ┌─────────────────────────────────────────────────┐
       │ User query: "fastapi auth"                       │
       └────────────────────┬────────────────────────────┘
                            │ LLM expansion
                            ▼
       ┌─────────────────────────────────────────────────┐
       │ Query 1: "How to implement OAuth2 in FastAPI?"  │
       │ Query 2: "FastAPI JWT authentication"            │
       │ Query 3: "FastAPI API key auth"                  │
       │ Query 4: "FastAPI session-based auth"            │
       │ Query 5: "FastAPI security best practices"       │
       └────────────────────┬────────────────────────────┘
                            │ Cada query → search
                            ▼
       ┌──────────────────────────────────────────────────┐
       │ 5 result sets (top-10 cada uno = 50 docs)        │
       └────────────────────┬─────────────────────────────┘
                            │ Reciprocal Rank Fusion
                            ▼
       ┌─────────────────────────────────────────────────┐
       │ Top-5 final: docs que aparecen alto en          │
       │ MÚLTIPLES queries (señal fuerte de relevancia)  │
       └─────────────────────────────────────────────────┘

Por qué funciona: un documento que aparece en posición #2 con la query "FastAPI OAuth2" y en posición #1 con "FastAPI security best practices" tiene señal de relevancia más fuerte que uno que aparece solo en una de las búsquedas. RRF captura esa intersección de rankings.

Generar expansions de calidad con un LLM

El prompt es la parte sensible. Un prompt mal diseñado genera expansions:

• Demasiado parecidas (no agregan diversidad de interpretación)
• Demasiado distintas (cambian la intención del usuario)
• Genéricas (no aprovechan vocabulario específico del dominio)

# query_expansion.py
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List
import os


client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


class ExpandedQueries(BaseModel):
    queries: List[str] = Field(
        description="Diverse alternative queries that cover different interpretations of the original",
        min_items=3,
        max_items=8,
    )
    reasoning: str = Field(
        description="One-sentence explanation of how the queries cover different interpretations"
    )


SYSTEM_PROMPT = """You are an expert at expanding search queries for retrieval systems.

Given a short or ambiguous user query, generate 5 alternative queries that:

1. Cover the most likely DIFFERENT interpretations of the original
2. Use natural language (full questions, not keywords)
3. Preserve the user's INTENT — don't add unrelated topics
4. Use specific terminology when relevant (e.g., if the topic is FastAPI, use FastAPI-specific terms)
5. Are mutually distinct (each one explores a different angle)

Examples:

Input: "fastapi auth"
Output queries:
1. How to implement OAuth2 authentication in FastAPI?
2. FastAPI JWT token authentication tutorial
3. How to use API keys for authentication in FastAPI?
4. FastAPI session-based authentication with cookies
5. Best practices for securing FastAPI endpoints

Input: "ml deployment"
Output queries:
1. How to deploy machine learning models to production?
2. ML model deployment with Docker containers
3. Serverless deployment for machine learning models (AWS Lambda, Cloud Run)
4. Real-time vs batch ML model serving
5. CI/CD pipelines for ML model deployment"""


def expand_query(query: str, num_expansions: int = 5) -> ExpandedQueries:
    """Generate diverse query expansions preserving original intent."""
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": f'User query: "{query}"\n\nGenerate {num_expansions} alternative queries.',
            },
        ],
        response_format=ExpandedQueries,
        temperature=0.5,  # algo de creatividad para diversidad
    )
    return response.choices[0].message.parsed


# Probar
expanded = expand_query("fastapi auth", num_expansions=5)
print(f"Reasoning: {expanded.reasoning}\n")
for i, q in enumerate(expanded.queries, 1):
    print(f"{i}. {q}")

Output esperado:

Reasoning: The original query 'fastapi auth' is ambiguous and could refer to several authentication methods. The expansions cover the main implementation patterns.

1. How to implement OAuth2 authentication in FastAPI applications?
2. FastAPI JWT token authentication implementation guide
3. How to use API key authentication in FastAPI?
4. Implementing session-based authentication in FastAPI with cookies
5. FastAPI security best practices and common authentication patterns

Por qué structured outputs

Sin structured outputs, los LLMs tienden a:

• Devolver formatos inconsistentes (a veces numerados, a veces bullets, a veces texto libre)
• Agregar comentarios o explicaciones en el medio que rompen el parsing
• Devolver más o menos queries que las pedidas

response_format=ExpandedQueries (con Pydantic) garantiza una lista limpia parseable.

Por qué temperature=0.5

• temperature=0 produce expansions casi idénticas a la query original (poca diversidad).
• temperature=1.0 produce expansions creativas pero a veces fuera de tema.
• 0.5 es sweet spot: diversas pero respetan la intención.

Reciprocal Rank Fusion (RRF): la fusión correcta

Después de buscar con las 5 queries, tienes 5 result sets. ¿Cómo combinar 5 rankings en uno solo?

Opción mala 1 — sumar scores: los scores de cosine de distintas queries no son comparables. Sumarlos da rankings sin sentido.

Opción mala 2 — usar solo el top-1 de cada uno: desperdicia los demás resultados. Si un documento aparece como #2 en 4 queries distintas, eso es señal fuerte que se ignora.

Opción correcta — Reciprocal Rank Fusion (RRF): combina rankings ignorando los scores absolutos. Solo importa la posición de cada documento en cada ranking.

Fórmula

RRF_score(doc) = Σ_i  1 / (k + rank_i(doc))

donde:
  k = constante (típicamente 60)
  rank_i(doc) = posición del doc en el ranking i (1, 2, 3, ...)

Si el doc no aparece en un ranking, contribuye 0.

Intuición:

• Doc en posición #1 contribuye 1/(60+1) = 0.0164
• Doc en posición #5 contribuye 1/(60+5) = 0.0154
• Doc en posición #20 contribuye 1/(60+20) = 0.0125

Las contribuciones son no lineales — diferencia entre top-5 y top-20 importa, pero diferencia entre #1 y #2 es pequeña. La función premia documentos que aparecen consistentemente alto en múltiples rankings.

Implementación

# rrf.py
from typing import List
from collections import defaultdict


def reciprocal_rank_fusion(
    rankings: List[List[str]],
    k: int = 60,
) -> List[tuple[str, float]]:
    """
    Combine multiple rankings using Reciprocal Rank Fusion.

    Args:
        rankings: List of rankings, each ranking is a list of doc_ids ordered by relevance.
        k: RRF constant (default 60, value used in original paper).

    Returns:
        List of (doc_id, rrf_score) tuples, sorted by score descending.
    """
    rrf_scores = defaultdict(float)

    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            rrf_scores[doc_id] += 1.0 / (k + rank)

    sorted_results = sorted(rrf_scores.items(), key=lambda x: -x[1])
    return sorted_results


# Ejemplo de uso
ranking_1 = ["doc_a", "doc_b", "doc_c", "doc_d", "doc_e"]      # query 1
ranking_2 = ["doc_b", "doc_a", "doc_f", "doc_c", "doc_g"]      # query 2
ranking_3 = ["doc_c", "doc_a", "doc_b", "doc_h", "doc_d"]      # query 3

fused = reciprocal_rank_fusion([ranking_1, ranking_2, ranking_3])
print("Documento : Score RRF")
for doc_id, score in fused[:5]:
    print(f"  {doc_id}: {score:.4f}")

Output:

Documento : Score RRF
  doc_a: 0.0492    ← aparece en posición 1, 2, 2 (top 3 en todos)
  doc_b: 0.0476    ← aparece en posición 2, 1, 3 (top 3 en todos)
  doc_c: 0.0467    ← aparece en posición 3, 4, 1
  doc_d: 0.0312    ← aparece en posición 4, -, 5
  doc_e: 0.0164    ← aparece solo en una

doc_a gana porque está consistentemente en top-3 de las 3 queries. doc_e queda último porque solo aparece en una.

Pipeline completo end-to-end

# expansion_pipeline.py
import chromadb
from chromadb.utils import embedding_functions
import os

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small",
)
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_collection("docs", embedding_function=openai_ef)


def search_with_expansion(query: str, top_k: int = 5, num_expansions: int = 5):
    """
    Pipeline completo: expand query → search each → fuse with RRF → top-K final.
    """
    # 1. Expandir
    expansion_result = expand_query(query, num_expansions=num_expansions)
    queries_to_search = [query] + expansion_result.queries  # incluir la original también

    # 2. Buscar con cada query (en paralelo si es posible)
    all_rankings = []
    for q in queries_to_search:
        results = collection.query(query_texts=[q], n_results=10)
        all_rankings.append(results['ids'][0])

    # 3. Fusion con RRF
    fused = reciprocal_rank_fusion(all_rankings)
    top_ids = [doc_id for doc_id, _ in fused[:top_k]]

    # 4. Recuperar documentos finales
    final_docs = collection.get(ids=top_ids)
    return final_docs


# Probar
query = "fastapi auth"
results = search_with_expansion(query, top_k=5)

print(f"Query: {query}\n")
print(f"Top 5 después de expansion + RRF:")
for i, (doc, doc_id) in enumerate(zip(results['documents'], results['ids']), 1):
    print(f"\n#{i} [{doc_id}]")
    print(f"   {doc[:120]}...")

Optimización: queries en paralelo

Si tienes 5 queries que buscar, hacerlo secuencial agrega 5x la latencia. Paraleliza:

from concurrent.futures import ThreadPoolExecutor


def parallel_retrieve(queries: list[str], top_k_per_query: int = 10):
    """Ejecuta múltiples retrievals en paralelo."""
    def search_one(q):
        return collection.query(query_texts=[q], n_results=top_k_per_query)

    with ThreadPoolExecutor(max_workers=5) as executor:
        results_list = list(executor.map(search_one, queries))

    return [r['ids'][0] for r in results_list]

Con 5 workers paralelos, las 5 queries de retrieval ejecutan en el tiempo de 1.

Cuándo expandir y cuándo no

Query expansion no es gratis: agrega ~600-800ms (1 LLM call + 5 retrieval calls + RRF) y ~$0.001 por query (LLM call). No siempre vale.

Cuándo SÍ expandir

Cuándo NO expandir

Skip dinámico

Lo ideal es decidir por query si vale expandir:

def should_expand(query: str) -> bool:
    """Heurística simple para decidir si expandir."""
    word_count = len(query.split())

    # Queries muy cortas son ambiguas → expandir
    if word_count <= 3:
        return True

    # Queries con identificadores exactos → no expandir
    has_identifier = any(c in query for c in ['_', '`', '(', ')', '/']) or \
                     any(word.isupper() for word in query.split() if len(word) > 2)
    if has_identifier:
        return False

    # Queries largas y bien formuladas → no expandir
    if word_count > 12 and "?" in query:
        return False

    # Default: expandir
    return True


def smart_search(query: str, top_k: int = 5):
    if should_expand(query):
        return search_with_expansion(query, top_k=top_k)
    else:
        results = collection.query(query_texts=[query], n_results=top_k)
        return results

Beneficio: ~30-50% de queries no necesitan expansion. Ahorra latencia y costo en esos casos sin perder calidad.

Trampas y errores comunes

Trampa 1: prompt sin ejemplos lleva a expansions débiles

El error:

Generate 5 search queries related to: "fastapi auth"

Síntoma: el LLM genera variaciones triviales ("fastapi authentication", "fastapi auth tutorial") que no agregan diversidad.

Cómo prevenir: prompt con ejemplos concretos (few-shot). Los LLMs aprenden mucho mejor de ejemplos que de instrucciones abstractas.

Trampa 2: temperature=0 genera expansions idénticas

El error: temperature=0 por consistencia.

Síntoma: las 5 "expansions" son casi idénticas. RRF no aprovecha diversidad.

Cómo prevenir: temperature=0.4-0.6 para diversidad sin desviarse de la intención.

Trampa 3: ignorar la query original

El error:

queries_to_search = expanded_queries  # solo las 5 generadas

Síntoma: si la query original ya era buena, pierdes su ranking en la fusión.

Cómo prevenir: siempre incluir la query original además de las expandidas. La query original es una "expansion" más, con peso igual:

queries_to_search = [query] + expanded_queries

Trampa 4: número de expansions demasiado alto

El error: generar 10-20 expansions "para estar seguros".

Síntoma:

• Latencia se dispara (10 retrievals + LLM call más larga).
• Costo sube linealmente.
• Calidad mejora marginalmente — 5 expansions cubren las interpretaciones probables; las 10-20 agregan ruido.

Cómo prevenir: mantener entre 4 y 6 expansions. Sweet spot empírico.

Trampa 5: expansions que cambian la intención

El error: prompt sin guardrails. El LLM expande "why is FastAPI slow" a queries como "how to make FastAPI faster".

Síntoma: la expansion ahora busca sobre "cómo optimizar" cuando el usuario quería entender "por qué a veces es lento". El sistema responde lo opuesto.

Cómo prevenir: instrucción explícita en el prompt: "Preserve the user's INTENT — don't add unrelated topics or change the question's stance."

Trampa 6: cachear sin considerar variantes triviales

El error: cachear expansion para "fastapi auth" y "FastAPI Auth" como queries distintas.

Síntoma: cache hit rate bajo (~10%) porque cada variante de capitalization/whitespace genera nueva expansion.

Cómo prevenir: normalizar la query antes de cachear:

import hashlib

def cache_key(query: str) -> str:
    normalized = query.lower().strip()
    normalized = " ".join(normalized.split())  # collapse whitespace
    return hashlib.md5(normalized.encode()).hexdigest()

Hit rate típico con normalización: 40-60%.

Ejercicio aplicado

Escenario: eres AI Engineer en una empresa de soporte técnico para herramientas DevOps. Datos del log:

• 10,000 queries reales del último mes
• 51% son queries cortas/ambiguas (1-4 tokens) — "k8s deploy", "docker timeout", "jenkins fail"
• 30% son queries con identificadores exactos — "kubectl get pods not working", "ERR_NETWORK_TIMEOUT_504"
• 19% son queries bien formuladas

Sistema actual: cosine + cross-encoder rerank. Métricas:

• Precision@5: 84%
• Recall@5: 68% (este es el problema — sistema no encuentra muchos docs relevantes)

Tu trabajo:

1. Decide si query expansion ayudaría aquí. Justifica con los datos.
2. Diseña la implementación incluyendo skip dinámico.
3. Estima impacto esperado y costo extra mensual (5K queries/día).

def smart_pipeline(query: str, top_k: int = 5):
    """Pipeline con expansion condicional."""

    # Heurística para decidir si expandir
    word_count = len(query.split())
    has_identifier = any(c in query for c in ['_', '`', '/']) or \
                     any(re.match(r'[A-Z][A-Z_]+\d*', w) for w in query.split())  # ej: ERR_TIMEOUT_504
    is_well_formed = word_count > 8 and "?" in query

    if has_identifier or is_well_formed:
        # Skip expansion: query directa + retrieval + rerank
        return search_with_rerank_only(query, top_k=top_k)
    elif word_count <= 5:
        # Query corta/ambigua: usar expansion
        return search_with_expansion_and_rerank(query, top_k=top_k)
    else:
        # Default: pipeline directo
        return search_with_rerank_only(query, top_k=top_k)


def search_with_expansion_and_rerank(query: str, top_k: int = 5):
    # 1. Expandir
    expansion = expand_query(query, num_expansions=5)
    queries_to_search = [query] + expansion.queries

    # 2. Retrieval paralelo
    all_rankings = parallel_retrieve(queries_to_search, top_k_per_query=15)

    # 3. RRF para combinar
    fused = reciprocal_rank_fusion(all_rankings)
    candidate_ids = [doc_id for doc_id, _ in fused[:30]]

    # 4. Recuperar documentos completos
    candidates = collection.get(ids=candidate_ids)

    # 5. Cross-encoder rerank sobre los 30 candidatos
    reranked = cross_encoder_rerank(query, candidates['documents'], top_k=top_k)
    return reranked

queries_per_day = 5000
expansion_rate = 0.51  # 51% expanden
expanded_queries_per_day = queries_per_day * expansion_rate

# Cada expansion: 1 LLM call (~500 tokens in, ~200 tokens out con gpt-4o-mini)
cost_per_expansion = (500 / 1_000_000 * 0.15) + (200 / 1_000_000 * 0.60)
# = $0.000195

monthly_cost = expanded_queries_per_day * 30 * cost_per_expansion
print(f"Costo mensual de expansion: ${monthly_cost:.2f}")

Resumen y siguiente paso

Lo que aprendiste:

• Query expansion genera múltiples versiones de una query ambigua y combina los resultados para mejorar recall.
• Mejora típica: +20-30% recall, especialmente en queries cortas o conceptuales.
• Costo: ~600-800ms latencia + ~$0.001 por query (LLM call para expansion).
• Generar expansions de calidad requiere prompt con ejemplos (few-shot), structured outputs, y temperature=0.4-0.6.
• Reciprocal Rank Fusion combina rankings ignorando scores absolutos. Es la técnica correcta vs sumar scores ingenuamente.
• Skip dinámico: no expandir queries con identificadores exactos o bien formuladas. Ahorra ~50% del costo extra.
• Incluir la query original en el set de búsqueda, no solo las expansiones.
• Trampa común: expansiones que cambian la intención del usuario. Prompt explícito previene.

Checkpoint: antes de avanzar, deberías poder:

• Implementar query expansion con structured outputs y RRF en menos de 100 líneas.
• Diseñar skip dinámico con heurísticas para detectar queries que NO necesitan expansion.
• Calcular costo extra mensual de query expansion para un volumen dado.

Siguiente cápsula: 04 — Query Rewriting.

Query expansion ataca queries ambiguas. Query rewriting ataca queries incompletas o mal formuladas. La idea: en vez de generar múltiples queries paralelas, transformar la query original en una mejor versión (con más contexto, mejor estructurada). Es la técnica complementaria — y juntas cubren los tres modos de falla que viste en M03/02.

Recursos

1. Reciprocal Rank Fusion Paper (Cormack et al., 2009) — Paper original
2. LangChain — Multi-Query Retriever — Implementación con LangChain
3. LlamaIndex — Query Transformation — Patrones alternativos
4. Pinecone — Query Optimization — Tutorial práctico
5. OpenAI — Structured Outputs — Para implementación robusta
6. Anthropic — Contextual Retrieval — Técnica complementaria

Tiempo estimado: 30-35 minutos Siguiente: 04-query-rewriting.md