Retrieval Optimization

---

Optimization Techniques Overview

---

Hybrid Search (Vector + Keyword)

Reciprocal Rank Fusion (RRF)

from dataclasses import dataclass
from typing import Callable

@dataclass
class SearchResult:
    id: str
    text: str
    score: float
    source: str  # "vector" or "keyword"

def reciprocal_rank_fusion(
    vector_results: list[SearchResult],
    keyword_results: list[SearchResult],
    k: int = 60,
    vector_weight: float = 0.5
) -> list[SearchResult]:
    """
    Combine vector and keyword results using RRF.
    k is a constant that reduces the impact of high rankings (typically 60).
    """
    scores: dict[str, float] = {}
    docs: dict[str, SearchResult] = {}

    # Score vector results
    for rank, result in enumerate(vector_results, 1):
        rrf_score = vector_weight * (1 / (k + rank))
        scores[result.id] = scores.get(result.id, 0) + rrf_score
        docs[result.id] = result

    # Score keyword results
    keyword_weight = 1 - vector_weight
    for rank, result in enumerate(keyword_results, 1):
        rrf_score = keyword_weight * (1 / (k + rank))
        scores[result.id] = scores.get(result.id, 0) + rrf_score
        if result.id not in docs:
            docs[result.id] = result

    # Sort by combined score
    sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)

    return [
        SearchResult(
            id=doc_id,
            text=docs[doc_id].text,
            score=scores[doc_id],
            source="hybrid"
        )
        for doc_id in sorted_ids
    ]

# Usage
hybrid_results = reciprocal_rank_fusion(
    vector_results=vector_search(query_embedding, top_k=20),
    keyword_results=bm25_search(query_text, top_k=20),
    vector_weight=0.6  # Favor semantic similarity
)

BM25 + Vector with Weaviate

from weaviate.classes.query import HybridFusion

collection = client.collections.get("Documents")

# Hybrid search with configurable fusion
results = collection.query.hybrid(
    query="how to configure authentication",
    alpha=0.5,  # 0 = pure BM25, 1 = pure vector
    fusion_type=HybridFusion.RELATIVE_SCORE,  # or RANKED
    limit=10,
    return_metadata=["score", "explain_score"]
)

# Iterate results
for obj in results.objects:
    print(f"Score: {obj.metadata.score}")
    print(f"Explanation: {obj.metadata.explain_score}")
    print(f"Text: {obj.properties['content'][:200]}")

Pinecone Sparse-Dense

from pinecone_text.sparse import BM25Encoder

# Train BM25 encoder on your corpus
bm25 = BM25Encoder()
bm25.fit(corpus_documents)

# Encode query for hybrid search
sparse_vector = bm25.encode_queries(query_text)
dense_vector = get_embedding(query_text)

# Search with both vectors
results = index.query(
    vector=dense_vector,
    sparse_vector=sparse_vector,
    top_k=10,
    include_metadata=True
)

---

Reranking

Cohere Rerank

co = cohere.Client(api_key="your-api-key")

def rerank_results(
    query: str,
    documents: list[str],
    top_n: int = 5,
    model: str = "rerank-english-v3.0"
) -> list[dict]:
    """Rerank documents using Cohere."""
    response = co.rerank(
        query=query,
        documents=documents,
        top_n=top_n,
        model=model,
        return_documents=True
    )

    return [
        {
            "text": result.document.text,
            "relevance_score": result.relevance_score,
            "original_index": result.index
        }
        for result in response.results
    ]

# Pipeline: retrieve more, rerank fewer
initial_results = vector_search(query_embedding, top_k=50)
documents = [r.text for r in initial_results]

reranked = rerank_results(
    query="how to configure OAuth2 authentication",
    documents=documents,
    top_n=5
)

# Use top 5 reranked docs for LLM context
context = "\
\
".join([r["text"] for r in reranked])

Cross-Encoder Reranking (Open Source)

from sentence_transformers import CrossEncoder

class Reranker:
    """Rerank using cross-encoder model."""

    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model_name)

    def rerank(
        self,
        query: str,
        documents: list[str],
        top_k: int = 5
    ) -> list[tuple[str, float]]:
        """Rerank documents by relevance to query."""
        # Create query-document pairs
        pairs = [[query, doc] for doc in documents]

        # Get relevance scores
        scores = self.model.predict(pairs)

        # Sort by score
        doc_scores = list(zip(documents, scores))
        doc_scores.sort(key=lambda x: x[1], reverse=True)

        return doc_scores[:top_k]

# Usage
reranker = Reranker()
top_docs = reranker.rerank(
    query="OAuth2 setup guide",
    documents=retrieved_documents,
    top_k=5
)

ColBERT-Style Late Interaction

from colbert import Searcher
from colbert.infra import Run, RunConfig

# Setup ColBERT index (one-time)
with Run().context(RunConfig(nranks=1)):
    searcher = Searcher(index="path/to/colbert_index")

# Search with late interaction scoring
results = searcher.search(
    query="how to configure authentication",
    k=10
)

# Results include token-level matching scores
for passage_id, rank, score in zip(*results):
    print(f"Rank {rank}: Doc {passage_id}, Score: {score}")

---

Query Expansion

LLM-Based Query Expansion

from openai import OpenAI

client = OpenAI()

def expand_query(query: str, num_expansions: int = 3) -> list[str]:
    """Generate query variations using LLM."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"""Generate {num_expansions} alternative search queries
                that would help find relevant documents for the user's question.
                Include:
                - Synonym variations
                - More specific versions
                - More general versions
                Return as JSON array of strings."""
            },
            {
                "role": "user",
                "content": query
            }
        ],
        response_format={"type": "json_object"}
    )

    result = json.loads(response.choices[0].message.content)
    return [query] + result.get("queries", [])

# Usage
original_query = "how to fix memory leak"
expanded_queries = expand_query(original_query)
# ["how to fix memory leak", "debug memory issues", "memory leak detection",
#  "troubleshoot high memory usage"]

# Search with all queries and merge results
all_results = []
for q in expanded_queries:
    results = vector_search(get_embedding(q), top_k=10)
    all_results.extend(results)

# Deduplicate and rank by frequency
deduped = deduplicate_by_id(all_results)

Query Rewriting

def rewrite_query_for_retrieval(
    conversational_query: str,
    chat_history: list[dict]
) -> str:
    """Rewrite conversational query to standalone search query."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """Rewrite the user's question as a standalone search query.
                Include relevant context from chat history.
                Output only the rewritten query, nothing else."""
            },
            {
                "role": "user",
                "content": f"""Chat history:
{format_chat_history(chat_history)}

User's question: {conversational_query}

Rewritten search query:"""
            }
        ],
        max_tokens=100
    )

    return response.choices[0].message.content.strip()

# Example
history = [
    {"role": "user", "content": "Tell me about Python web frameworks"},
    {"role": "assistant", "content": "Popular Python web frameworks include Django, Flask, and FastAPI..."}
]
query = "Which one is best for APIs?"

rewritten = rewrite_query_for_retrieval(query, history)
# Output: "Best Python web framework for building REST APIs: Django vs Flask vs FastAPI"

---

HyDE (Hypothetical Document Embeddings)

def hyde_search(
    query: str,
    vector_store,
    embedding_model,
    top_k: int = 10
) -> list[SearchResult]:
    """
    Generate hypothetical answer, embed it, and search.
    Aligns query embedding space with document embedding space.
    """
    # Generate hypothetical document
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """Write a passage that would answer the user's question.
                Write as if you're an expert documentation author.
                Be specific and technical. About 100-200 words."""
            },
            {
                "role": "user",
                "content": query
            }
        ],
        max_tokens=300
    )

    hypothetical_doc = response.choices[0].message.content

    # Embed hypothetical document
    hyde_embedding = embedding_model.encode(hypothetical_doc)

    # Search with hypothetical doc embedding
    results = vector_store.search(
        vector=hyde_embedding,
        top_k=top_k
    )

    return results

# Usage
results = hyde_search(
    query="How do I handle rate limiting in my API?",
    vector_store=qdrant_client,
    embedding_model=sentence_transformer
)

Multi-HyDE (Multiple Perspectives)

def multi_hyde_search(
    query: str,
    vector_store,
    embedding_model,
    num_hypotheticals: int = 3,
    top_k: int = 10
) -> list[SearchResult]:
    """Generate multiple hypothetical docs for diverse retrieval."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"""Generate {num_hypotheticals} different passages
                that could answer the question from different angles:
                1. Technical deep-dive
                2. Beginner-friendly explanation
                3. Best practices summary

                Return as JSON with "passages" array."""
            },
            {
                "role": "user",
                "content": query
            }
        ],
        response_format={"type": "json_object"}
    )

    passages = json.loads(response.choices[0].message.content)["passages"]

    # Embed all hypotheticals
    all_results = []
    for passage in passages:
        embedding = embedding_model.encode(passage)
        results = vector_store.search(vector=embedding, top_k=top_k)
        all_results.extend(results)

    # Deduplicate and combine scores
    return deduplicate_and_merge(all_results)

---

Metadata Filtering

Multi-Tenant Filtering

class MultiTenantRetriever:
    """Retriever with mandatory tenant isolation."""

    def __init__(self, vector_store):
        self.vector_store = vector_store

    def search(
        self,
        query_embedding: list[float],
        tenant_id: str,
        top_k: int = 10,
        additional_filters: dict | None = None
    ) -> list[SearchResult]:
        """Search with mandatory tenant filter."""
        # Build filter - tenant is always required
        filters = {"tenant_id": {"$eq": tenant_id}}

        if additional_filters:
            filters = {"$and": [filters, additional_filters]}

        return self.vector_store.search(
            vector=query_embedding,
            filter=filters,
            top_k=top_k
        )

# Usage
retriever = MultiTenantRetriever(pinecone_index)
results = retriever.search(
    query_embedding=embedding,
    tenant_id="acme-corp",
    additional_filters={
        "doc_type": {"$in": ["manual", "faq"]},
        "published": {"$eq": True}
    }
)

Temporal Filtering

from datetime import datetime, timedelta

def search_recent_documents(
    query_embedding: list[float],
    vector_store,
    days_back: int = 30,
    top_k: int = 10
) -> list[SearchResult]:
    """Search documents updated within time window."""
    cutoff_date = datetime.utcnow() - timedelta(days=days_back)

    return vector_store.search(
        vector=query_embedding,
        filter={
            "updated_at": {"$gte": cutoff_date.isoformat()}
        },
        top_k=top_k
    )

def search_with_recency_boost(
    query_embedding: list[float],
    vector_store,
    recency_weight: float = 0.2,
    top_k: int = 10
) -> list[SearchResult]:
    """Boost recent documents in ranking."""
    # Get more results to apply post-filtering
    results = vector_store.search(
        vector=query_embedding,
        top_k=top_k * 3
    )

    now = datetime.utcnow()

    def compute_boosted_score(result):
        doc_date = datetime.fromisoformat(result.metadata["updated_at"])
        days_old = (now - doc_date).days
        recency_score = max(0, 1 - (days_old / 365))  # Decay over 1 year
        return result.score * (1 - recency_weight) + recency_score * recency_weight

    # Rerank with recency boost
    for result in results:
        result.boosted_score = compute_boosted_score(result)

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

---

Query Decomposition

def decompose_complex_query(query: str) -> list[str]:
    """Break complex query into sub-questions."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """Break this complex question into simpler sub-questions
                that can be answered independently. Each sub-question should be
                searchable. Return as JSON with "questions" array."""
            },
            {
                "role": "user",
                "content": query
            }
        ],
        response_format={"type": "json_object"}
    )

    result = json.loads(response.choices[0].message.content)
    return result.get("questions", [query])

def search_with_decomposition(
    complex_query: str,
    vector_store,
    embedding_model,
    top_k_per_subquery: int = 5
) -> dict:
    """Search for each sub-question and aggregate results."""
    sub_questions = decompose_complex_query(complex_query)

    a