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:
top_k without adding prohibitive latencyThis is an expert-level lesson. You should be comfortable with:
If RAG is relatively new to you, start with the foundational RAG lessons in this learning path first.
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:
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.
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.
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
)
Warning: Don't run the LLM classifier on every single query in a high-throughput system. A 200ms classification call that precedes every retrieval will noticeably degrade perceived performance. Use the heuristic classifier as your primary path and invoke LLM classification only when the heuristic confidence is below 0.7 or when the query triggers specific ambiguity signals.
Now that we can classify queries, we need the actual retrieval strategies to select from. This is where the real engineering lives.
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
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]
Tip: The RRF constant
k=60is not a retrievalk— it's the fusion constant from the RRF formula. Don't confuse these. The original paper by Cormack et al. (2009) showed that k=60 is robust across many retrieval tasks. You generally don't need to tune this.
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
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]
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
}
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}
)
Warning: The latency estimates in
STRATEGY_TYPICAL_LATENCY_MSwill vary significantly based on your LLM API latency, your vector store's query performance, and your chunk corpus size. Instrument these empirically in your environment before hardcoding them into budget logic.
Building something sophisticated doesn't mean it's better. You need to measure this rigorously, and the metrics that matter are often counterintuitive.
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
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.
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
When you instrument a well-tuned adaptive system, you should see something like:
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.
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:
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:
k usedTask 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.
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
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.
Learning Path: RAG & AI Agents