Adaptive Retrieval: Dynamically Adjusting Chunk Count and Search Strategy Based on Query Complexity at Runtime

Introduction

Picture this: a user asks your RAG system two questions in rapid succession. The first is "What's the capital of France?" The second is "Compare the macroeconomic factors that drove the 2008 financial crisis with the conditions preceding the 1929 crash, and explain how the regulatory responses differed." If your retrieval pipeline treats both queries identically — fetching the same number of chunks using the same vector similarity search — you've already made a serious architectural mistake. The first query needs almost nothing from your knowledge base. The second needs a careful, multi-angle retrieval strategy pulling from potentially dozens of source documents, balanced across time periods and topic domains.

Most production RAG implementations I've seen use a fixed top_k — typically somewhere between 3 and 10 — and a single retrieval strategy. This is the equivalent of using the same size wrench for every bolt in an engine. It works often enough that teams don't notice how badly it fails on the queries that actually matter. Simple factual queries get bloated with irrelevant context, inflating token costs and injecting noise that confuses the generation step. Complex analytical queries get starved of the context they need, producing shallow, incomplete, or confidently wrong answers.

By the end of this lesson, you'll be able to design and implement a runtime adaptive retrieval system that classifies incoming queries along multiple complexity dimensions, then dynamically routes each query to an appropriate retrieval strategy with a calibrated chunk count. This isn't about making things complicated for the sake of it — it's about matching retrieval behavior to the actual information needs of each query.

What you'll learn:

• How to model query complexity as a multi-dimensional feature space and classify queries at runtime
• Why different retrieval strategies (dense vector search, sparse BM25, hybrid, multi-hop, decomposed) are appropriate for different query types
• How to implement a dynamic retrieval orchestrator that selects strategy and top_k without adding prohibitive latency
• How to design fallback and escalation logic when initial retrieval confidence is low
• How to measure whether adaptive retrieval is actually improving your system, and what metrics to trust

---

Prerequisites

This is an expert-level lesson. You should be comfortable with:

• Building and querying vector stores (Pinecone, Weaviate, Chroma, pgvector — the concepts transfer)
• The mechanics of BM25 and dense embedding retrieval
• LangChain or LlamaIndex at an intermediate level (we'll write framework-agnostic code but reference these)
• Python async patterns and basic classifier design
• Familiarity with LLM API usage (OpenAI, Anthropic, or equivalent)

If RAG is relatively new to you, start with the foundational RAG lessons in this learning path first.

---

The Problem with Fixed Retrieval Parameters

Before we build the solution, let's establish exactly why fixed retrieval fails — not just intuitively, but mechanically.

When you set top_k=5 globally, you're implicitly claiming that five chunks is the right amount of context for every query your system will ever receive. Think about what "right amount" actually means here. It means:

1. Coverage: Enough context to contain the information needed to answer the question
2. Precision: Not so much irrelevant context that it degrades generation quality
3. Cost efficiency: Not burning tokens on content that adds nothing

For a simple lookup query — "What year was the Clean Air Act passed?" — top_k=1 is probably sufficient. Fetching five chunks means four chunks of noise. For an LLM, noise in context is not neutral. Multiple studies on RAG systems have shown that adding irrelevant retrieved documents to context actively degrades answer quality, a phenomenon sometimes called "context pollution." The model attends to that noise.

For a complex comparative query — "Analyze how Tesla's approach to battery technology differs from traditional automotive manufacturers and explain the supply chain implications" — top_k=5 is likely insufficient. You'd want chunks covering Tesla's battery strategy, traditional OEM battery approaches, supply chain structures for both, raw material sourcing, and possibly some financial context. You're realistically looking at top_k=15 to top_k=25 with a smart deduplication pass.

The failure modes are asymmetric and non-obvious:

# This is what your fixed-k system does internally, every time
def naive_retrieve(query: str, vector_store, k: int = 5) -> list[Document]:
    # No consideration of query complexity
    # No consideration of expected document diversity needed
    # No consideration of whether dense search is even right for this query
    return vector_store.similarity_search(query, k=k)

This function is doing something subtle and harmful: it's treating the query as a black box and applying a uniform information retrieval strategy to it. The query isn't a black box — it's a structured signal with a lot of recoverable information about what kind of retrieval it needs.

---

Modeling Query Complexity

To build adaptive retrieval, we first need to define what "complexity" actually means in a way that maps to retrieval behavior. I want to move away from a vague notion of "simple vs. complex" toward a concrete feature space.

Here are the dimensions that matter:

1. Scope Breadth — How many distinct topics or entities does the query span? "What is photosynthesis?" has a scope breadth of 1. "Compare photosynthesis in C3 and C4 plants under drought conditions" has a scope breadth of roughly 4-5 (C3 plants, C4 plants, drought conditions, efficiency comparison, mechanism differences).

2. Temporal Span — Does the query require information from a single point in time, or does it require reasoning across time periods? This is critical for document-heavy knowledge bases.

3. Reasoning Depth — Is the query asking for recall ("who founded X?") or reasoning ("why did X succeed while Y failed despite similar conditions?"). Reasoning queries often need more context to support the inference chain.

4. Entity Density — How many named entities appear in the query? High entity density often signals that multiple retrieval passes targeting each entity will outperform a single broad search.

5. Query Length and Syntactic Complexity — A strong signal, though not deterministic. Longer queries with subordinate clauses tend to require more retrieval context.

6. Ambiguity — Does the query have a clear referent, or could it mean several things? Ambiguous queries often need exploratory retrieval before a focused pass.

These dimensions aren't just theoretical. Each one has a practical mapping to retrieval behavior:

from dataclasses import dataclass
from enum import Enum

class RetrievalStrategy(Enum):
    SINGLE_VECTOR = "single_vector"       # Simple semantic search, low k
    HYBRID = "hybrid"                     # Dense + sparse (BM25), moderate k
    MULTI_QUERY = "multi_query"           # Multiple reformulated queries, moderate-high k
    DECOMPOSED = "decomposed"             # Break query into sub-queries, high k
    MULTI_HOP = "multi_hop"              # Sequential retrieval with intermediate reasoning

@dataclass
class QueryComplexityProfile:
    scope_breadth: float        # 0.0 to 1.0
    temporal_span: float        # 0.0 to 1.0
    reasoning_depth: float      # 0.0 to 1.0
    entity_density: float       # 0.0 to 1.0
    ambiguity_score: float      # 0.0 to 1.0
    recommended_strategy: RetrievalStrategy
    recommended_k: int
    confidence: float           # How confident we are in this classification

The key insight here is that we're not trying to classify queries into buckets like "simple" and "hard." We're building a profile that maps directly to retrieval parameters. A query can be low-breadth but high-reasoning-depth, which would push you toward a deep single-topic retrieval rather than a scattered multi-topic one.

---

Building the Query Complexity Classifier

Now let's build the actual classifier. There are two viable approaches here, and they have different trade-offs.

Approach A: Heuristic + lightweight model classifier

This uses a combination of measurable query features (token count, named entity count, presence of comparison/causal keywords) fed into a small classifier. It's fast (sub-millisecond) and interpretable, but misses nuance.

Approach B: LLM-based classification

Use a fast, cheap LLM call (GPT-4o-mini or similar) to assess query complexity. More accurate, but adds 100-300ms latency. Appropriate if you're not serving real-time interactive queries.

For production systems, I recommend a hybrid: heuristic pre-screening with LLM fallback for ambiguous cases.

Let's build both:

import re
import spacy
from typing import Optional

# Load a lightweight NLP model for entity detection
# python -m spacy download en_core_web_sm
nlp = spacy.load("en_core_web_sm")

COMPARISON_SIGNALS = [
    "compare", "contrast", "difference between", "versus", "vs",
    "how does .* differ", "similarities", "unlike", "whereas",
    "on the other hand", "relative to"
]

TEMPORAL_SIGNALS = [
    "history", "evolution", "over time", "since", "before", "after",
    "trend", "changed", "originally", "currently", "used to", "now"
]

CAUSAL_REASONING_SIGNALS = [
    "why", "because", "cause", "reason", "explain", "how did",
    "what led to", "impact of", "effect of", "consequence",
    "result in", "due to"
]

MULTI_ENTITY_THRESHOLD = 3  # NER count above this = high entity density

def compute_heuristic_complexity(query: str) -> QueryComplexityProfile:
    query_lower = query.lower()
    tokens = query.split()
    doc = nlp(query)
    
    # Scope breadth: normalized by query length, boosted by comparison signals
    comparison_hits = sum(
        1 for pattern in COMPARISON_SIGNALS 
        if re.search(pattern, query_lower)
    )
    scope_breadth = min(1.0, (len(tokens) / 50.0) + (comparison_hits * 0.2))
    
    # Temporal span
    temporal_hits = sum(
        1 for signal in TEMPORAL_SIGNALS 
        if signal in query_lower
    )
    temporal_span = min(1.0, temporal_hits * 0.3)
    
    # Reasoning depth
    causal_hits = sum(
        1 for signal in CAUSAL_REASONING_SIGNALS 
        if signal in query_lower
    )
    reasoning_depth = min(1.0, causal_hits * 0.25 + (len(tokens) / 80.0))
    
    # Entity density
    entities = [ent for ent in doc.ents]
    entity_density = min(1.0, len(entities) / MULTI_ENTITY_THRESHOLD)
    
    # Ambiguity: short queries with pronouns or vague references
    vague_terms = ["it", "this", "that", "they", "the thing", "the concept"]
    ambiguity_score = min(
        1.0,
        sum(1 for term in vague_terms if term in query_lower) * 0.3 +
        (0.4 if len(tokens) < 5 else 0.0)
    )
    
    # Compute composite score
    composite = (
        scope_breadth * 0.25 +
        temporal_span * 0.15 +
        reasoning_depth * 0.30 +
        entity_density * 0.20 +
        ambiguity_score * 0.10
    )
    
    # Map composite score to strategy and k
    strategy, k = _map_composite_to_retrieval(composite, scope_breadth, entity_density)
    
    return QueryComplexityProfile(
        scope_breadth=scope_breadth,
        temporal_span=temporal_span,
        reasoning_depth=reasoning_depth,
        entity_density=entity_density,
        ambiguity_score=ambiguity_score,
        recommended_strategy=strategy,
        recommended_k=k,
        confidence=0.65  # Heuristic confidence is inherently limited
    )

def _map_composite_to_retrieval(
    composite: float, 
    scope_breadth: float,
    entity_density: float
) -> tuple[RetrievalStrategy, int]:
    
    if composite < 0.2:
        return RetrievalStrategy.SINGLE_VECTOR, 3
    elif composite < 0.4:
        return RetrievalStrategy.HYBRID, 6
    elif composite < 0.6:
        if entity_density > 0.6:
            return RetrievalStrategy.MULTI_QUERY, 10
        return RetrievalStrategy.HYBRID, 10
    elif composite < 0.75:
        return RetrievalStrategy.DECOMPOSED, 15
    else:
        return RetrievalStrategy.DECOMPOSED, 20

Now the LLM-based classifier, which handles the fuzzy cases where heuristics fall short:

from openai import AsyncOpenAI
import json

client = AsyncOpenAI()

COMPLEXITY_CLASSIFICATION_PROMPT = """You are a query complexity analyzer for a RAG system. 
Analyze the following query and return a JSON object with these exact fields:

- scope_breadth: float 0-1 (how many distinct topics/entities are involved)
- temporal_span: float 0-1 (does this require info across time periods?)
- reasoning_depth: float 0-1 (recall=0, complex causal reasoning=1)
- entity_density: float 0-1 (how many named entities, normalized)
- ambiguity_score: float 0-1 (how unclear or ambiguous is the query?)
- recommended_k: int between 2 and 25
- recommended_strategy: one of ["single_vector", "hybrid", "multi_query", "decomposed", "multi_hop"]
- reasoning: string explaining your classification (max 50 words)

Return only valid JSON. No markdown, no explanation outside the JSON.

Query: {query}"""

async def classify_with_llm(query: str) -> QueryComplexityProfile:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "user", 
                "content": COMPLEXITY_CLASSIFICATION_PROMPT.format(query=query)
            }
        ],
        temperature=0.1,  # Low temperature for consistent classification
        response_format={"type": "json_object"}
    )
    
    result = json.loads(response.choices[0].message.content)
    
    return QueryComplexityProfile(
        scope_breadth=result["scope_breadth"],
        temporal_span=result["temporal_span"],
        reasoning_depth=result["reasoning_depth"],
        entity_density=result["entity_density"],
        ambiguity_score=result["ambiguity_score"],
        recommended_strategy=RetrievalStrategy(result["recommended_strategy"]),
        recommended_k=result["recommended_k"],
        confidence=0.85  # LLM classification is more reliable
    )

---

Implementing the Retrieval Strategies

Now that we can classify queries, we need the actual retrieval strategies to select from. This is where the real engineering lives.

Strategy 1: Single Vector Search (Simple Queries)

For low-complexity queries, a straightforward semantic search with a small k is best. The key constraint is that you should still enforce a minimum relevance threshold — don't return chunks just because they're the "closest" if they're not actually relevant.

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

async def single_vector_retrieve(
    query: str,
    vector_store: Chroma,
    k: int,
    relevance_threshold: float = 0.75
) -> list[Document]:
    
    results = vector_store.similarity_search_with_relevance_scores(query, k=k)
    
    # Filter by relevance threshold — this is non-negotiable
    filtered = [
        doc for doc, score in results 
        if score >= relevance_threshold
    ]
    
    if not filtered:
        # Don't return empty — escalate k slightly and lower threshold
        results = vector_store.similarity_search_with_relevance_scores(
            query, k=k * 2
        )
        filtered = [
            doc for doc, score in results 
            if score >= relevance_threshold * 0.85
        ]
    
    return filtered

Strategy 2: Hybrid Search (Moderate Complexity)

Hybrid search combines dense vector similarity with sparse BM25 retrieval, then uses Reciprocal Rank Fusion (RRF) to merge the result lists. This is dramatically better for queries involving specific terminology, product names, codes, or technical jargon where semantic search can miss exact matches.

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, vector_store, documents: list[Document]):
        self.vector_store = vector_store
        self.documents = documents
        
        # Build BM25 index from document content
        tokenized_corpus = [doc.page_content.lower().split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_corpus)
    
    def _reciprocal_rank_fusion(
        self, 
        rankings: list[list[int]], 
        k: int = 60
    ) -> list[tuple[int, float]]:
        """
        RRF score = sum(1 / (k + rank)) across all ranking lists.
        k=60 is the standard constant from the original RRF paper.
        """
        scores = {}
        for ranking in rankings:
            for rank, doc_idx in enumerate(ranking, start=1):
                scores[doc_idx] = scores.get(doc_idx, 0) + (1 / (k + rank))
        
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)
    
    async def retrieve(self, query: str, k: int) -> list[Document]:
        # Dense retrieval: get top 2k candidates for reranking
        dense_results = self.vector_store.similarity_search(query, k=k * 2)
        
        # Map dense results back to document indices
        # This assumes a stable document order — in production, use doc IDs
        dense_ranking = []
        for dense_doc in dense_results:
            for idx, doc in enumerate(self.documents):
                if doc.page_content == dense_doc.page_content:
                    dense_ranking.append(idx)
                    break
        
        # Sparse retrieval: BM25 scores
        tokenized_query = query.lower().split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        sparse_ranking = np.argsort(bm25_scores)[::-1][:k * 2].tolist()
        
        # Fuse rankings
        fused = self._reciprocal_rank_fusion([dense_ranking, sparse_ranking])
        
        # Return top k documents
        top_indices = [idx for idx, _ in fused[:k]]
        return [self.documents[idx] for idx in top_indices]

Strategy 3: Multi-Query Retrieval (High Entity Density)

When a query involves multiple distinct entities or perspectives, a single query vector is a compromise that may not center on any of them well. Multi-query retrieval generates several reformulations of the original query and retrieves for each, then deduplicates.

from typing import AsyncIterator

async def generate_query_variants(
    query: str, 
    n_variants: int = 3
) -> list[str]:
    """Generate semantically diverse reformulations of a query."""
    
    prompt = f"""Generate {n_variants} different ways to search for information 
needed to answer this question. Each variant should emphasize a different 
aspect or use different terminology. Return as a JSON array of strings.

Original query: {query}

Return only a JSON array, no explanation."""
    
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.4,
        response_format={"type": "json_object"}
    )
    
    # Parse carefully — LLMs sometimes wrap in an object
    content = json.loads(response.choices[0].message.content)
    if isinstance(content, list):
        variants = content
    else:
        # Try to extract array from object
        variants = next(iter(content.values()), [query])
    
    return [query] + variants  # Always include original

async def multi_query_retrieve(
    query: str,
    vector_store,
    k: int,
    n_variants: int = 3
) -> list[Document]:
    
    variants = await generate_query_variants(query, n_variants)
    
    # Retrieve for each variant
    all_results = []
    seen_content_hashes = set()
    
    k_per_variant = max(3, k // len(variants))
    
    for variant in variants:
        results = vector_store.similarity_search(variant, k=k_per_variant)
        for doc in results:
            # Deduplicate by content hash
            content_hash = hash(doc.page_content[:200])  # First 200 chars as fingerprint
            if content_hash not in seen_content_hashes:
                seen_content_hashes.add(content_hash)
                all_results.append(doc)
    
    return all_results[:k]  # Return at most k unique documents

Strategy 4: Decomposed Retrieval (Complex Analytical Queries)

This is the heavyweight strategy, designed for queries that are essentially multiple sub-questions bundled together. The idea is to decompose the query into atomic sub-questions, retrieve for each, then reassemble.

@dataclass
class SubQuery:
    text: str
    retrieval_weight: float  # How important is this sub-question?
    expected_doc_count: int

async def decompose_query(query: str) -> list[SubQuery]:
    """Break a complex query into atomic sub-questions with retrieval weights."""
    
    prompt = f"""Decompose this complex query into atomic sub-questions that can 
each be answered with a focused document search. For each sub-question, estimate 
how many document chunks you'd need (1-5) and its importance (0.0-1.0).

Return a JSON object with a "sub_queries" array. Each element should have:
- "text": the sub-question
- "retrieval_weight": importance 0.0-1.0
- "expected_doc_count": int 1-5

Query: {query}"""
    
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
        response_format={"type": "json_object"}
    )
    
    result = json.loads(response.choices[0].message.content)
    
    return [
        SubQuery(
            text=sq["text"],
            retrieval_weight=sq["retrieval_weight"],
            expected_doc_count=sq["expected_doc_count"]
        )
        for sq in result.get("sub_queries", [{"text": query, "retrieval_weight": 1.0, "expected_doc_count": 5}])
    ]

async def decomposed_retrieve(
    query: str,
    vector_store,
    total_k: int
) -> list[Document]:
    
    sub_queries = await decompose_query(query)
    
    # Allocate k proportionally to sub-query weights
    total_weight = sum(sq.retrieval_weight for sq in sub_queries)
    
    all_docs = []
    seen_hashes = set()
    
    for sub_query in sub_queries:
        # Weight-proportional k, bounded by expected_doc_count
        allocated_k = min(
            sub_query.expected_doc_count,
            max(2, int((sub_query.retrieval_weight / total_weight) * total_k))
        )
        
        results = vector_store.similarity_search(sub_query.text, k=allocated_k)
        
        for doc in results:
            content_hash = hash(doc.page_content[:200])
            if content_hash not in seen_hashes:
                seen_hashes.add(content_hash)
                # Tag the document with its source sub-query for context assembly
                doc.metadata["source_sub_query"] = sub_query.text
                doc.metadata["sub_query_weight"] = sub_query.retrieval_weight
                all_docs.append(doc)
    
    # Sort by sub-query weight to put most important context first
    all_docs.sort(key=lambda d: d.metadata.get("sub_query_weight", 0), reverse=True)
    
    return all_docs[:total_k]

---

The Adaptive Retrieval Orchestrator

Now we wire everything together into an orchestrator that takes a raw query and returns appropriately retrieved documents, with no manual configuration per query.

import asyncio
from typing import Optional

class AdaptiveRetrievalOrchestrator:
    def __init__(
        self,
        vector_store,
        documents: list[Document],
        use_llm_classifier: bool = True,
        llm_classifier_threshold: float = 0.70,
        max_retrieval_latency_ms: Optional[float] = None
    ):
        self.vector_store = vector_store
        self.documents = documents
        self.use_llm_classifier = use_llm_classifier
        self.llm_classifier_threshold = llm_classifier_threshold
        self.max_retrieval_latency_ms = max_retrieval_latency_ms
        
        # Initialize hybrid retriever (needs the document corpus)
        self.hybrid_retriever = HybridRetriever(vector_store, documents)
        
        # Telemetry — track what strategies are being used
        self._strategy_counts = {s.value: 0 for s in RetrievalStrategy}
        self._total_queries = 0
    
    async def retrieve(self, query: str) -> tuple[list[Document], QueryComplexityProfile]:
        """
        Main entry point. Returns retrieved documents and the complexity profile
        used (useful for logging and debugging).
        """
        self._total_queries += 1
        
        # Step 1: Classify query complexity
        profile = await self._classify_query(query)
        
        # Step 2: Route to appropriate strategy
        docs = await self._execute_strategy(query, profile)
        
        # Step 3: Post-retrieval validation and escalation
        docs = await self._validate_and_escalate(query, docs, profile)
        
        # Track telemetry
        self._strategy_counts[profile.recommended_strategy.value] += 1
        
        return docs, profile
    
    async def _classify_query(self, query: str) -> QueryComplexityProfile:
        # Always run fast heuristic first
        heuristic_profile = compute_heuristic_complexity(query)
        
        if not self.use_llm_classifier:
            return heuristic_profile
        
        # Only invoke LLM if heuristic confidence is below threshold
        if heuristic_profile.confidence >= self.llm_classifier_threshold:
            return heuristic_profile
        
        # Check latency budget — if we have a tight constraint, skip LLM
        if self.max_retrieval_latency_ms and self.max_retrieval_latency_ms < 500:
            return heuristic_profile
        
        try:
            # Run LLM classification with a timeout
            llm_profile = await asyncio.wait_for(
                classify_with_llm(query),
                timeout=3.0  # 3 second timeout on classification
            )
            return llm_profile
        except (asyncio.TimeoutError, Exception) as e:
            # Classification failure is non-fatal — fall back to heuristic
            print(f"LLM classification failed, using heuristic: {e}")
            return heuristic_profile
    
    async def _execute_strategy(
        self, 
        query: str, 
        profile: QueryComplexityProfile
    ) -> list[Document]:
        
        strategy = profile.recommended_strategy
        k = profile.recommended_k
        
        if strategy == RetrievalStrategy.SINGLE_VECTOR:
            return await single_vector_retrieve(query, self.vector_store, k)
        
        elif strategy == RetrievalStrategy.HYBRID:
            return await self.hybrid_retriever.retrieve(query, k)
        
        elif strategy == RetrievalStrategy.MULTI_QUERY:
            return await multi_query_retrieve(query, self.vector_store, k)
        
        elif strategy == RetrievalStrategy.DECOMPOSED:
            return await decomposed_retrieve(query, self.vector_store, k)
        
        elif strategy == RetrievalStrategy.MULTI_HOP:
            # Multi-hop is a special case — covered below
            return await self._multi_hop_retrieve(query, k)
        
        else:
            # Safe default
            return await single_vector_retrieve(query, self.vector_store, k)
    
    async def _validate_and_escalate(
        self,
        query: str,
        docs: list[Document],
        profile: QueryComplexityProfile
    ) -> list[Document]:
        """
        Check whether retrieval results are adequate.
        If not, escalate to a more powerful strategy.
        """
        
        if len(docs) == 0:
            # Nothing retrieved — try hybrid with minimum k
            return await self.hybrid_retriever.retrieve(query, k=5)
        
        if len(docs) < profile.recommended_k * 0.4:
            # Got significantly fewer than expected — try with lower threshold
            # or escalate strategy
            current_strategy = profile.recommended_strategy
            strategies = list(RetrievalStrategy)
            current_idx = strategies.index(current_strategy)
            
            if current_idx < len(strategies) - 2:
                next_strategy = strategies[current_idx + 1]
                # Create an escalated profile
                escalated_profile = QueryComplexityProfile(
                    **{**profile.__dict__,
                       "recommended_strategy": next_strategy,
                       "recommended_k": profile.recommended_k + 3}
                )
                return await self._execute_strategy(query, escalated_profile)
        
        return docs
    
    async def _multi_hop_retrieve(self, query: str, k: int) -> list[Document]:
        """
        Multi-hop retrieval: retrieve initial documents, extract bridging concepts,
        then retrieve again using those concepts.
        """
        # First hop: standard dense retrieval
        initial_docs = await single_vector_retrieve(
            query, self.vector_store, k=k // 2
        )
        
        if not initial_docs:
            return []
        
        # Extract bridging concepts from initial results
        initial_context = "\n\n".join(doc.page_content for doc in initial_docs[:3])
        
        bridging_prompt = f"""Given this initial context and the original query, 
what follow-up search queries would help gather additional needed information?
Return a JSON array of 2-3 search strings.

Query: {query}
Initial context (first 1000 chars): {initial_context[:1000]}"""
        
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": bridging_prompt}],
            temperature=0.3,
            response_format={"type": "json_object"}
        )
        
        content = json.loads(response.choices[0].message.content)
        follow_up_queries = content if isinstance(content, list) else next(iter(content.values()), [])
        
        # Second hop: retrieve for each bridging query
        all_docs = list(initial_docs)
        seen_hashes = {hash(doc.page_content[:200]) for doc in initial_docs}
        
        for follow_up in follow_up_queries[:3]:
            hop_docs = await single_vector_retrieve(
                follow_up, self.vector_store, k=3
            )
            for doc in hop_docs:
                content_hash = hash(doc.page_content[:200])
                if content_hash not in seen_hashes:
                    seen_hashes.add(content_hash)
                    doc.metadata["hop"] = "second"
                    all_docs.append(doc)
        
        return all_docs[:k]
    
    def get_strategy_distribution(self) -> dict:
        """Useful for monitoring — shows what strategies are being used."""
        if self._total_queries == 0:
            return {}
        return {
            strategy: count / self._total_queries 
            for strategy, count in self._strategy_counts.items()
            if count > 0
        }

---

Latency Budgets and Cascading Complexity

Here's where many implementations fall apart in production: adaptive retrieval adds overhead, and that overhead is not uniform. A DECOMPOSED strategy with LLM-based decomposition adds 2-4 LLM API calls before you've even done a single vector lookup. For an interactive application, this can push total response time from 1.5 seconds to 4+ seconds.

The solution isn't to abandon complexity — it's to budget your latency explicitly.

@dataclass
class LatencyBudget:
    total_ms: float
    classification_ms: float   # Budget for query classification
    retrieval_ms: float         # Budget for actual retrieval
    post_processing_ms: float  # Budget for reranking, dedup etc.

LATENCY_PRESETS = {
    "interactive": LatencyBudget(
        total_ms=800,
        classification_ms=50,   # Heuristic only
        retrieval_ms=600,
        post_processing_ms=150
    ),
    "near_realtime": LatencyBudget(
        total_ms=2000,
        classification_ms=300,  # Can afford LLM classification
        retrieval_ms=1200,
        post_processing_ms=500
    ),
    "batch": LatencyBudget(
        total_ms=10000,
        classification_ms=1000,
        retrieval_ms=7000,
        post_processing_ms=2000
    )
}

When you're inside the interactive budget, the system should automatically cap itself to strategies that fit. The max_retrieval_latency_ms parameter on the orchestrator controls this — but you should also implement strategy-level fallbacks:

STRATEGY_TYPICAL_LATENCY_MS = {
    RetrievalStrategy.SINGLE_VECTOR: 50,
    RetrievalStrategy.HYBRID: 150,
    RetrievalStrategy.MULTI_QUERY: 600,    # Includes LLM call for variants
    RetrievalStrategy.DECOMPOSED: 1200,   # Multiple LLM calls + multiple searches
    RetrievalStrategy.MULTI_HOP: 800,
}

def cap_strategy_to_budget(
    profile: QueryComplexityProfile,
    budget: LatencyBudget
) -> QueryComplexityProfile:
    """Downgrade strategy if it won't fit in the latency budget."""
    
    strategy = profile.recommended_strategy
    
    while (
        STRATEGY_TYPICAL_LATENCY_MS.get(strategy, 0) > budget.retrieval_ms
        and strategy != RetrievalStrategy.SINGLE_VECTOR
    ):
        # Walk down the strategy list
        strategies = list(RetrievalStrategy)
        current_idx = strategies.index(strategy)
        strategy = strategies[max(0, current_idx - 1)]
    
    return QueryComplexityProfile(
        **{**profile.__dict__, "recommended_strategy": strategy}
    )

---

Measuring Whether Adaptive Retrieval is Actually Working

Building something sophisticated doesn't mean it's better. You need to measure this rigorously, and the metrics that matter are often counterintuitive.

Metric 1: Retrieval Recall @ k

For a labeled evaluation set, measure the fraction of queries where the ground-truth answer chunks are in the retrieved set. Compare fixed-k baseline vs. adaptive retrieval.

def retrieval_recall_at_k(
    retrieved_docs: list[Document],
    ground_truth_doc_ids: list[str]
) -> float:
    """Fraction of ground truth docs that were retrieved."""
    retrieved_ids = {doc.metadata.get("doc_id") for doc in retrieved_docs}
    hits = sum(1 for gt_id in ground_truth_doc_ids if gt_id in retrieved_ids)
    return hits / len(ground_truth_doc_ids) if ground_truth_doc_ids else 0.0

Metric 2: Context Precision

Of the chunks retrieved, what fraction were actually used by the LLM? This is harder to measure directly, but you can approximate it by checking attribution in the generated answer.

Metric 3: Token Efficiency

Simple but important: how many tokens are you sending to the generator on average, and how does that correlate with answer quality?

def token_efficiency_score(
    retrieved_docs: list[Document],
    answer_quality_score: float,  # 0.0-1.0 from your eval framework
    cost_per_token: float = 0.000015  # GPT-4o pricing, approximate
) -> float:
    """Quality per unit of token cost — higher is better."""
    total_tokens = sum(
        len(doc.page_content.split()) * 1.3  # Rough token estimate
        for doc in retrieved_docs
    )
    cost = total_tokens * cost_per_token
    return answer_quality_score / cost if cost > 0 else 0.0

What Good Adaptive Retrieval Looks Like in Practice

When you instrument a well-tuned adaptive system, you should see something like:

• Simple factual queries: avg 2-4 chunks, avg 200-400 tokens in context, answer quality equivalent to fixed-k=10
• Moderate queries: avg 6-8 chunks, hybrid strategy used ~60% of the time
• Complex analytical queries: avg 14-20 chunks, decomposed or multi-query, meaningfully better answer quality vs. fixed-k=5 baseline

If your adaptive system is retrieving more chunks on average than your fixed baseline, something's wrong with your complexity classifier — you're treating too many queries as complex.

---

Hands-On Exercise

Here's a realistic exercise to cement everything covered in this lesson. You'll build a complete adaptive retrieval system for a knowledge base of financial regulatory documents (an excellent test case because queries range from "what does SEC stand for?" to "analyze the jurisdictional differences between GDPR and CCPA data retention requirements for financial institutions").

Setup: Download a set of financial regulation PDFs (SEC filings guidance, GDPR text, Dodd-Frank summaries — all publicly available). Chunk them into 512-token overlapping chunks with 50-token overlap. Index them in Chroma or your preferred vector store.

Task 1: Implement compute_heuristic_complexity and test it against these five queries:

1. "What is the SEC?"
2. "What are the penalties for insider trading under the Securities Exchange Act?"
3. "Compare SEC enforcement mechanisms with FCA enforcement under UK financial services law, focusing on civil vs. criminal penalties"
4. "How has the definition of a material adverse effect evolved in M&A agreements between 2008 and 2023?"
5. "What are reporting requirements for broker-dealers?"

For each query, print the QueryComplexityProfile and verify that the classification makes intuitive sense. Query 1 should get SINGLE_VECTOR with k=3. Query 3 and 4 should get DECOMPOSED or MULTI_QUERY with k>=12.

Task 2: Implement the full AdaptiveRetrievalOrchestrator and run all five queries through it. For each query, print:

• Strategy used
• k used
• Number of unique documents retrieved
• First 100 characters of each retrieved chunk

Task 3: Create a fixed-k baseline (k=5, single vector) and compare retrieval recall for queries 3 and 4, assuming you've manually identified which chunks are ground-truth relevant. You should see the adaptive system outperform on complex queries and at least match on simple ones.

Task 4 (Advanced): Add a latency budget of 800ms ("interactive" preset) and observe how the system degrades gracefully for complex queries — it should cap at HYBRID or MULTI_QUERY instead of DECOMPOSED.

---

Common Mistakes & Troubleshooting

Mistake 1: Using adaptive k without a relevance threshold

Raising k doesn't help if the extra documents are irrelevant. Always filter by a minimum similarity score. Without this, you're adding noise proportional to your k increase.

Mistake 2: Not deduplicating across multi-query or decomposed retrieval

When you run three query variants and retrieve 5 documents each, you often get the same high-relevance documents multiple times. Without deduplication, you're sending the same context three times, inflating token costs and causing the LLM to over-weight that content.

Mistake 3: Treating the LLM classifier as an oracle

The LLM complexity classifier is better than heuristics but still wrong often enough to need fallback logic. Always track classification confidence and validate empirically that the recommended strategies actually improve outcomes.

Mistake 4: Forgetting that decomposition costs latency even for simple queries

If your heuristic classifier is poorly calibrated, it might classify simple queries as complex and trigger expensive decomposition unnecessarily. Monitor _strategy_counts in your orchestrator regularly. If more than 30% of queries are hitting DECOMPOSED, your complexity scoring is probably too aggressive.

Mistake 5: Not tagging retrieved documents with their provenance

When you do multi-query or decomposed retrieval, tag each document with which sub-query or variant retrieved it (as shown in the decomposed_retrieve function with source_sub_query metadata). This is invaluable both for debugging and for building better context assembly prompts.

Mistake 6: Ignoring chunk boundary effects at high k

When k is large, you're more likely to retrieve adjacent chunks from the same document that cover slightly different ranges of the same content. Consider a post-processing step that merges adjacent chunks from the same source document:

def merge_adjacent_chunks(docs: list[Document]) -> list[Document]:
    """Merge chunks from the same document that are adjacent by index."""
    from itertools import groupby
    
    # Group by source document
    doc_groups = {}
    for doc in docs:
        source = doc.metadata.get("source", "unknown")
        chunk_idx = doc.metadata.get("chunk_index", -1)
        if source not in doc_groups:
            doc_groups[source] = []
        doc_groups[source].append((chunk_idx, doc))
    
    merged = []
    for source, chunk_list in doc_groups.items():
        chunk_list.sort(key=lambda x: x[0])
        
        # Find consecutive runs
        current_text = chunk_list[0][1].page_content
        current_metadata = chunk_list[0][1].metadata.copy()
        
        for i in range(1, len(chunk_list)):
            prev_idx, _ = chunk_list[i-1]
            curr_idx, curr_doc = chunk_list[i]
            
            if curr_idx == prev_idx + 1:
                # Adjacent — merge
                current_text += " " + curr_doc.page_content
            else:
                # Gap — emit current and start new
                merged.append(Document(page_content=current_text, metadata=current_metadata))
                current_text = curr_doc.page_content
                current_metadata = curr_doc.metadata.copy()
        
        merged.append(Document(page_content=current_text, metadata=current_metadata))
    
    return merged

---

Summary & Next Steps

You've now built a complete adaptive retrieval system from the ground up. Let's crystallize the key architectural principles:

The core mental model: Every query is a specification of information needs. Your job as the retrieval architect is to extract that specification at runtime and match it to the appropriate retrieval behavior — not to apply a universal policy.

The classification layer is the brain: Whether you use heuristics, an LLM, or a trained classifier, investing in good query complexity profiling pays compounding returns. Every downstream component — strategy selection, k calibration, context assembly — depends on this classification being correct.

Strategies exist on a cost-quality curve: SINGLE_VECTOR is cheap and good enough for simple queries. DECOMPOSED is expensive but necessary for complex ones. The adaptive system's value is in routing each query to the appropriate point on that curve, not always using the most powerful strategy available.

Latency budgets are first-class constraints: In production systems, you don't get to ignore latency. Build budget awareness into your orchestrator from day one, not as an afterthought.

Instrument everything: Adaptive systems are harder to debug than fixed systems because their behavior is dynamic. Log strategy selections, retrieval counts, and confidence scores for every query. Build dashboards. Run offline evaluations regularly with a labeled query set.

Where to go next:

• Reranking as a post-retrieval step: Cross-encoder rerankers (Cohere Rerank, BGE-reranker) can dramatically improve precision after retrieval. This pairs naturally with adaptive retrieval — use higher k with a reranker to compensate for recall pressure.

• Contextual compression: After retrieving documents, compress each one to only the portion relevant to the query before sending to the generator. LangChain's ContextualCompressionRetriever implements this.

• Query routing to specialized indexes: For large knowledge bases with distinct domains (e.g., legal, financial, technical documentation), route different query types to domain-specific indexes rather than a single monolithic one.

• Learned complexity classifiers: If you have enough query-feedback pairs, train a small logistic regression or gradient-boosted classifier on query features. This will outperform both heuristics and LLM classification at scale, and it'll be much faster.

• Agentic RAG: When retrieval complexity is very high and the information need isn't well-defined upfront, a ReAct-style agent that iteratively retrieves and reasons is more appropriate than any fixed pipeline — even an adaptive one.