A dynamic routing system that analyzes user query intent to automatically blend or switch between semantic dense vector search and traditional keyword sparse search (BM25).
Overview
RAG-001: Adaptive Hybrid RAG Layer
A production-grade, dynamic retrieval-augmented generation (RAG) gateway microservice designed to serve low-latency search requests. The gateway dynamically route queries between a dense semantic vector database (Qdrant) and a sparse keyword index (Elasticsearch) using intent-based classification weight routing.
System Topology & Architecture
- FastAPI Gateways: Manages concurrent calls to backend stores with strict 45ms database timeouts.
- Intent-Based Routing: Automatically determines the query intent $\alpha \in [0.0, 1.0]$. Technical exact tokens (UUIDs, files paths, programming signatures) route towards Elasticsearch ($\alpha=0.15$), while conceptual, open-ended questions route towards Qdrant ($\alpha=0.85$).
- Weighted RRF (Reciprocal Rank Fusion): Normalizes and blends disparate BM25 and cosine similarity scores.
- Cohere Rerank v3: Integrates cross-attention re-scoring with candidate pruning.
- Two-Tiered Semantic Cache: Exact query caching and high-similarity cache checks ($>0.96$ cosine threshold) served via Redis in <5ms.
- LLMLingua Proxy: Compresses return payloads by up to 60% by filtering low-surprise tokens.
Getting Started
1. Prerequisites
Ensure you have the following installed: * Docker & Docker Compose * Python 3.9 or higher
2. Environment Setup
Copy .env.example to .env and adjust settings:
cp .env.example .env
Default configuration values support running the pipeline locally with mock components if Cohere/LLMLingua dependencies are not configured.
3. Spin Up Infrastructure
Start the Elasticsearch, Qdrant, and Redis services using Docker Compose:
docker compose up -d
Verify that services are running:
* Elasticsearch: http://localhost:9200
* Qdrant: http://localhost:6333 (Dashboard: http://localhost:6333/dashboard)
* Redis: http://localhost:6379
4. Install Dependencies
Install the Python libraries listed in requirements.txt:
pip install -r requirements.txt
5. Setup Schemas & Indices
Run the database setup script to configure indices, analyzers, and mapping schemas:
python -m app.setup_indices
Running the Application
1. Start FastAPI Server
Launch the gateway server using Uvicorn:
uvicorn app.gateway:app --reload --port 8000
API Documentation is available at http://localhost:8000/docs.
2. Run the Verification Showcase Client
In a separate terminal, execute the showcase client to ingest mock documents, trigger keyword vs. semantic routing rules, and verify the semantic cache:
python showcase_client.py
Running Automated Tests
Run the Pytest suite to verify the ingestion sync, rate limiter, semantic cache, and the 45ms timeout SLA:
pytest -v
System Architecture Blueprint
Blueprint: RAG-001 — Adaptive Hybrid RAG Layer
This engineering specification details the architecture, math, data structures, and implementation logic for the Adaptive Hybrid RAG Layer (RAG-001). It serves as a self-contained guide for building, tuning, and hosting this high-performance retrieval microservice.
- System Specifications & Performance SLA
Target Latency:
Ingestion pipeline: $< 50\text{ms}$ per document batch.
Retrieval routing & parallel query execution: $< 45\text{ms}$.
Total round-trip latency (including reranking): $< 120\text{ms}$.
Target Accuracy: $> 92\%$ retrieval precision on key-value queries and $> 88\%$ on semantic questions.
Infrastructure Design: Stateless FastAPI container instances, scalable horizontally behind a Layer 7 proxy (e.g., Nginx, Envoy).
- Ingest Architecture & Dual-Index Configuration
Every document chunk processed by the system must be committed atomically to both the Dense Vector Database (Qdrant or Milvus) and the Sparse Index Engine (Elasticsearch). This guarantees index parity.
[ Incoming Chunk ]
│
┌────────┴────────┐
▼ ▼
[ Embedding API ] [ Raw Extraction ]
│ │
▼ ▼
[ Qdrant/Milvus ] [ Elasticsearch ]
2.1 Qdrant Vector Collection Configuration
For dense semantic retrieval, optimize the collections with pre-allocated payloads and HNSW index tweaks to manage memory usage.
{ "name": "enterprise_kb", "vectors": { "size": 1536, "distance": "Cosine", "hnsw_config": { "m": 16, "ef_construct": 128, "full_scan_threshold": 10000, "max_on_disk_vector_size_kb": 512 } }, "optimizers_config": { "default_segment_number": 2, "indexing_threshold_kb": 20480 } }
2.2 Elasticsearch Index Mapping Configuration
To optimize Elasticsearch for sparse indexing (BM25) and exact matches on custom strings (e.g., error codes, UUIDs), define explicit mappings and tokenizers.
{ "settings": { "analysis": { "analyzer": { "hybrid_code_analyzer": { "type": "custom", "tokenizer": "whitespace", "filter": ["lowercase", "word_delimiter_graph", "unique"] } } }, "index": { "similarity": { "default": { "type": "BM25", "b": 0.75, "k1": 1.2 } } } }, "mappings": { "properties": { "chunk_id": { "type": "keyword" }, "parent_doc_id": { "type": "keyword" }, "tenant_id": { "type": "keyword" }, "text_content": { "type": "text", "analyzer": "hybrid_code_analyzer" } } } }
- Mathematical Retrieval Formulation
3.1 Sparse Retrieval: BM25 Algorithm
The traditional sparse relevance score is computed over the term match space using the BM25 formulation:
$$score(D, Q) = \sum_{i=1}^{n} \text{IDF}(q_i) \cdot \frac{f(q_i, D) \cdot (k_1 + 1)}{f(q_i, D) + k_1 \cdot \left(1 - b + b \cdot \frac{\vert{}D\vert{}}{\text{avgdl}}\right)}$$
Where:
$f(q_i, D)$ is the frequency of query term $q_i$ in document $D$.
$\vert{}D\vert{}$ is the length of document $D$ in words.
$\text{avgdl}$ is the average document length in the collection.
$k_1$ is the term frequency saturation parameter ($1.2$).
$b$ is the document length penalty scaling parameter ($0.75$).
3.2 Dynamic Intent Routing Score ($\alpha$)
The balance between sparse and dense retrieval is controlled dynamically via a routing parameter $\alpha \in [0.0, 1.0]$. The value of $\alpha$ is derived from query characteristics:
$$\alpha = \text{logistic}\left(w_0 + w_1 \cdot \text{length}(Q) + w_2 \cdot I_{\text{semantic}} - w_3 \cdot I_{\text{token_match}}\right)$$
Where:
$I_{\text{token_match}} \in {0, 1}$ represents regex matches indicating strict technical terms (such as UUIDs, IPs, error codes, file extensions, or system variable structures).
$I_{\text{semantic}} \in {0, 1}$ indicates the presence of conceptual trigger words (e.g., "how", "why", "difference", "explain").
The final hybrid rank score is calculated using Reciprocal Rank Fusion (RRF) parameterized by $\alpha$:
$$RRF_Score(d) = \alpha \cdot \left( \frac{1}{k + r_{\text{dense}}(d)} \right) + (1 - \alpha) \cdot \left( \frac{1}{k + r_{\text{sparse}}(d)} \right)$$
Where:
$r_{\text{dense}}(d)$ is the index rank of document $d$ within the dense vector results.
$r_{\text{sparse}}(d)$ is the index rank of document $d$ within the sparse keyword results.
$k$ is a constant smoothing factor, typically set to $60$ to minimize the influence of outlier low-ranking candidates.
- Concurrent Implementation Pipeline
The FastAPI architecture leverages async/await execution paradigms to query both systems in parallel, ensuring that the total retrieval latency is determined by the slowest of the two engines, rather than their sum.
import asyncio import logging from typing import List, Dict, Any, Tuple from fastapi import FastAPI, HTTPException from pydantic import BaseModel
app = FastAPI(title="RAG-001: Adaptive Hybrid Gateway") logger = logging.getLogger("hybrid_retrieval")
class QueryPayload(BaseModel): query: str tenant_id: str top_k: int = 15
class DocumentChunk(BaseModel): chunk_id: str text: str score: float
async def search_dense(query: str, tenant_id: str, limit: int) -> List[Dict[str, Any]]: # Simulated Async Qdrant Client Call await asyncio.sleep(0.015) # Representing ~15ms round trip return [{"id": f"chunk_{i}", "text": "Semantic payload...", "rank": i + 1} for i in range(limit)]
async def search_sparse(query: str, tenant_id: str, limit: int) -> List[Dict[str, Any]]: # Simulated Async Elasticsearch Client Call await asyncio.sleep(0.022) # Representing ~22ms round trip return [{"id": f"chunk_{i}", "text": "Keyword payload...", "rank": i + 1} for i in range(limit)]
def calculate_intent_alpha(query: str) -> float: # Check for exact token matches (e.g., alphanumeric strings, UUIDs, code syntax symbols) import re token_patterns = [ r"\b[A-Za-z0-9_-]{8,12}\b", # Potential hash/IDs r"[a-zA-Z_][a-zA-Z0-9_].[a-zA-Z_][a-zA-Z0-9_]", # Dot notation methods r"0x[0-9a-fA-F]+" # Hex pointers ] if any(re.search(pattern, query) for pattern in token_patterns): return 0.15 # Heavily favor keyword sparse matches
# Query length-based evaluation
words = query.strip().split()
if len(words) < 3:
return 0.35 # Short queries point to exact matching keyword terms
return 0.85 # Conversational or descriptive queries use dense semantic mapping
def fuse_results(dense: List[Dict[str, Any]], sparse: List[Dict[str, Any]], alpha: float, k: int = 60) -> List[Tuple[str, float]]: rrf_map: Dict[str, float] = {}
# Process dense candidates
for item in dense:
doc_id = item["id"]
rank = item["rank"]
rrf_map[doc_id] = rrf_map.get(doc_id, 0.0) + alpha * (1.0 / (k + rank))
# Process sparse candidates
for item in sparse:
doc_id = item["id"]
rank = item["rank"]
rrf_map[doc_id] = rrf_map.get(doc_id, 0.0) + (1.0 - alpha) * (1.0 / (k + rank))
# Sort descending by fused RRF score
return sorted(rrf_map.items(), key=lambda x: x[1], reverse=True)
@app.post("/v1/retrieve/hybrid") async def retrieve_hybrid(payload: QueryPayload): alpha = calculate_intent_alpha(payload.query)
# Run sparse and dense queries concurrently to honor the SLA ceiling
try:
dense_task = search_dense(payload.query, payload.tenant_id, payload.top_k)
sparse_task = search_sparse(payload.query, payload.tenant_id, payload.top_k)
dense_results, sparse_results = await asyncio.wait_for(
asyncio.gather(dense_task, sparse_task),
timeout=0.080 # 80ms retrieval timeout barrier
)
except asyncio.TimeoutError:
logger.error("Database connection timeout during concurrent execution.")
raise HTTPException(status_code=504, detail="Primary index execution timed out.")
# Fuse scoring
fused_scores = fuse_results(dense_results, sparse_results, alpha)
return {
"query": payload.query,
"alpha_routing_factor": alpha,
"results": [{"chunk_id": doc_id, "rrf_score": score} for doc_id, score in fused_scores[:payload.top_k]]
}
- Cohere Cross-Encoder Reranking Execution
The initial retrieval step retrieves a wide net of candidate chunks (typically $40$ to $50$ entries). A cross-encoder neural model is then used to rerank these candidates. Unlike bi-encoder models, the cross-encoder processes both the query and the retrieved text in a single inference pass, yielding a highly accurate semantic match score.
Operational Sequence:
Ingestion: Collect top 40 RRF chunks from the parallel gateway.
API Transport: Dispatch payload to Cohere via their dynamic v3/rerank engine.
Cross-Attention Processing: Evaluates semantic alignments across all variables.
Sieving Logic: Retain only the high-signal chunks ($S_{\text{rerank}} \ge 0.40$).
- Verification and Validation Procedures
To verify your routing and retrieval logic, implement automated testing scenarios:
import pytest
def test_intent_classification(): # Technical Queries should yield low alpha (favoring sparse search) assert calculate_intent_alpha("ERR_CONNECTION_RESET") < 0.3 assert calculate_intent_alpha("0x7ffd5f") < 0.3
# Concept Queries should yield high alpha (favoring semantic search)
assert calculate_intent_alpha("What is the underlying mechanism of multi-tenancy context isolation?") > 0.75
Required Technical Skills
Advanced Engineering Skills Inventory: Adaptive Hybrid RAG Layer (RAG-001)
This matrix maps out the advanced full-stack systems engineering competencies, distributed systems patterns, data structures, and mathematical concepts required to build, secure, scale, and optimize high-throughput hybrid retrieval systems.
- Dual-Index Vector & Keyword Architecture
Building a performant hybrid search architecture requires deeply configuring and aligning two completely different index formats to operate in lockstep.
[ Document Chunk Ingestion ]
│
┌───────────────────────┴───────────────────────┐
▼ ▼
[ Dense Index Optimization ] [ Sparse Index Tuning ] • Qdrant / Milvus (HNSW) • Elasticsearch / BM25 • Cosine Similarity / Dot Product • Custom Tokenizers & Analyzers • In-Memory Vector Quantization • Sharding & Dynamic Mapping
Dense Semantic Indexing (Qdrant / Milvus):
Configuring Hierarchical Navigable Small World (HNSW) graph layers: tuning hyper-parameters such as $m$ (maximum connection links per node) and $ef_construct$ (search path optimization during indexing).
Managing memory profiles using Vector Quantization (Scalar Quantization SQ or Product Quantization PQ) to compress vector sizes with minimal recall loss.
Understanding the mathematical execution of distance metrics: executing Dot Product calculations on unit-normalized vectors to achieve faster indexing speeds compared to standard Cosine Similarity.
Sparse Keyword Indexing (Elasticsearch):
Tuning the Okapi BM25 ranking algorithm parameters: optimizing $k_1$ (term frequency saturation parameter) and $b$ (document length penalty weight) to calibrate matching precision on short technical search queries.
Designing custom tokenization and analysis filters (e.g., using Edge N-Grams, standard lowercasing, and word-delimiter graphs) to extract exact alpha-numeric codes, system error logs, and file directories.
Configuring index aliases, routing shards, and index lifecycles to maintain fast query throughput under dense writing pressure.
- Algorithmic Routing, Score Normalization, & Reranking
Integrating completely different database engines requires normalizing mathematical scoring models and assessing query intent dynamically.
Dynamic Intent Classification ($\alpha$):
Formulating logistic routing weights ($\alpha \in [0.0, 1.0]$) by processing queries through pattern-matching regex rules and linguistic classifiers to balance dense vector retrieval weights against sparse keyword indexing.
Reciprocal Rank Fusion (RRF) Implementation:
Writing customized RRF algorithms to merge disparate scoring profiles (e.g., raw BM25 scores which range from $0$ to $\infty$ versus cosine metrics which scale from $0.0$ to $1.0$).
Understanding and configuring the smoothing constant $k$ within the rank-decay formula:
$$RRF_Score(d) = \sum_{m \in M} \frac{1}{k + r_m(d)}$$
Cross-Encoder Reranking Architecture:
Integrating cross-attention neural models (e.g., Cohere Rerank v3 or local BGE-Reranker models) that compute query-to-document attention profiles dynamically, bypassing the context isolation of standard Bi-Encoder embedding vectors.
Constructing dynamic pruning thresholds to eliminate redundant chunks from the pipeline before passing them upstream to the generation LLM.
- High-Throughput Asynchronous Concurrency
To hit SLAs of $<150\text{ms}$ total round-trip retrieval latency, you must write concurrent, non-blocking asynchronous Python services.
[ API Request Gateway ]
│
[ asyncio.gather / Timeout ]
│
┌─────────────────────┴─────────────────────┐
▼ ▼
[ Dense Database Query ] [ Sparse Database Query ]
FastAPI Async/Await Execution Patterns:
Writing fully non-blocking asynchronous endpoints using Python's asyncio loop, ensuring database queries are triggered concurrently via asyncio.gather or asyncio.as_completed.
Implementing strict network request timeouts (asyncio.wait_for) to gracefully fall back on single-engine results if one database cluster slows down.
Thread-Safe Connection Pools:
Managing thread-safe connection pools for third-party databases, preventing TCP port exhaustion under high concurrent connection loads.
- Multi-Tenant SaaS Isolation Patterns
Maintaining cryptographic or logical boundaries between customer databases is an absolute requirement in enterprise environments.
Logical (Soft) Partition Models:
Implementing low-cost, high-density metadata filters on shared databases. Managing payload filter rules in Qdrant and metadata query term arrays in Elasticsearch to block index cross-talk or query leakage.
Physical (Hard) Partition Models:
Orchestrating dynamic client pool routers that map incoming JWT payloads at the API Gateway level to route queries directly to isolated physical indexes (tenant_docs_{tenant_id}).
Handling dynamic creation, indexing schema generation, and structural deprecation of collections on separate tenant registrations.
- Distributed Gateways, Caching, & Token Economics
SaaS deployments must shield upstream LLMs and vector indices from traffic bursts while optimizing input token margins.
[ User Retrieval Request ]
│
[ Redis Token Bucket ] ──► (Saves CPU, blocks abuse)
│
[ Redis Semantic Cache ] ──► (Returns context <5ms)
│
[ LLMLingua Compression ] ──► (Saves 60% on Token Cost)
Sliding Window Rate Limiting (Redis):
Developing distributed rate limiters using Redis sorted sets (ZSET) and multi-command pipelines to manage transaction histories within shifting sliding time blocks.
Semantic Cache Optimizations:
Constructing a local cache layer using Redis Vector Search. Indexing previous queries and cache contexts to bypass expensive database search loops and third-party APIs for repetitive or highly similar user queries.
Context Token Compression (LLMLingua):
Interfacing with context compressors (like LLMLingua) to evaluate linguistic perplexity scores across raw returned texts.
Calibrating optimal token compression margins (typically $50\%$ to $60\%$ reduction) to maximize context density while mitigating LLM "Lost in the Middle" issues.
- Next.js Full-Stack Stream Processing
Handling deep, multi-stage retrieval pipelines requires building rich, real-time feedback loops using streaming infrastructures.
Server-Sent Events (SSE) Pipelines:
Constructing backend event emitters in FastAPI that continuously push structural metadata (such as execution latencies, alpha metrics, and tokens saved) along with chunks to the client over an active HTTP stream.
Asynchronous Browser Stream Consumption:
Developing custom React hooks in Next.js using native browser ReadableStream APIs and TextDecoder buffers to parse SSE strings cleanly.
Managing incremental React state updates to guarantee smooth, low-latency client-side UI rendering.
🚀 Quick Installation & Setup Guide
To run this project locally, follow these simple steps designed for both developers and users with no prior programming experience.
- Download Python: Make sure you have Python 3.10+ installed on your computer. Download it from the official Python Website. During installation, make sure to check the box "Add Python to PATH".
- Extract / Setup Code: Open the terminal (Command Prompt on Windows) and navigate to the project directory, or download/extract the project files.
- Install Dependencies (pip): In Command Prompt, run the following command to install all necessary packages automatically:
pip install -r requirements.txt - Setup API Keys / Environment: Open the
.envfile in a text editor (like Notepad) and paste your API keys:GEMINI_API_KEY=your_gemini_api_key_here - Launch the Application: Double-click the
run_project.batorrun_pito.batfile, or run the following command:
(orpython app.pyapp_gui.py)