Lesson 16.17: Keyword Search, Hybrid Retrieval & Reciprocal Rank Fusion

                          User Query: "Price of BYT-KB-1042?"
                                       │
                     ┌─────────────────┴─────────────────┐
                     ▼                                   ▼
          Dense Embedding Model                  Sparse BM25 Encoder
         (text-embedding-3-small)                  (FastEmbed BM25)
                     │                                   │
                     ▼                                   ▼
             Dense Query Vector                  Sparse Query Vector
                     │                                   │
                     ▼                                   ▼
         Qdrant HNSW Graph Search              Qdrant Sparse Index Search
        (Tenant Filter Applied)                 (Tenant Filter Applied)
                     │                                   │
                     ▼                                   ▼
            Top-20 Dense Hits                   Top-20 BM25 Hits
         [Doc B, Doc A, Doc D...]            [Doc C, Doc B, Doc A...]
                     │                                   │
                     └─────────────────┬─────────────────┘
                                       │
                                       ▼
                       Reciprocal Rank Fusion (RRF)
                     Combined Score = Σ 1 / (60 + Rank)
                                       │
                                       ▼
                    Final Authorized Top-K Chunks for LLM
        

import os
from dotenv import load_dotenv
from qdrant_client import QdrantClient, models

load_dotenv()
COLLECTION_NAME = "knowledge_hybrid"
VECTOR_SIZE = 1536

qdrant = QdrantClient(url=os.getenv("QDRANT_URL", "http://localhost:6333"))

def ensure_hybrid_collection() -> None:
    if qdrant.collection_exists(COLLECTION_NAME):
        return

    qdrant.create_collection(
        collection_name=COLLECTION_NAME,
        # 1. Dense Semantic Vector Configuration
        vectors_config={
            "dense": models.VectorParams(
                size=VECTOR_SIZE,
                distance=models.Distance.COSINE,
            ),
        },
        # 2. Sparse BM25 Vector Configuration
        sparse_vectors_config={
            "bm25": models.SparseVectorParams(
                modifier=models.Modifier.IDF,  # Mandatory for FastEmbed BM25
            ),
        },
    )
    
    # Critical payload index for multi-tenant isolation
    qdrant.create_payload_index(
        collection_name=COLLECTION_NAME,
        field_name="tenant_id",
        field_schema=models.PayloadSchemaType.KEYWORD,
        is_tenant=True
    )

ensure_hybrid_collection()

import os
from uuid import NAMESPACE_URL, uuid5
from openai import OpenAI
from fastembed import SparseTextEmbedding
from qdrant_client import models

openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
bm25_model = SparseTextEmbedding(model_name="Qdrant/bm25")

def embed_dense(text: str) -> list[float]:
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def embed_sparse(text: str):
    return next(iter(bm25_model.passage_embed([text])))

def index_chunk(chunk: dict) -> None:
    text = chunk["text"].strip()
    if not text:
        return

    # Generate both representations for the same chunk
    dense_vec = embed_dense(text)
    sparse_vec = embed_sparse(text)

    point_id = str(uuid5(NAMESPACE_URL, chunk["chunk_id"]))

    point = models.PointStruct(
        id=point_id,
        vector={
            "dense": dense_vec,
            "bm25": models.SparseVector(
                indices=sparse_vec.indices.tolist(),
                values=sparse_vec.values.tolist(),
            ),
        },
        payload={
            **chunk["metadata"],
            "chunk_id": chunk["chunk_id"],
            "text": text,
        },
    )

    qdrant.upsert(
        collection_name="knowledge_hybrid",
        points=[point],
        wait=True
    )

from typing import List, Dict, Any
from qdrant_client import models

def hybrid_search(
    query: str,
    tenant_id: str,
    limit: int = 5
) -> List[Dict[str, Any]]:
    # 1. Formulate dual query representations
    dense_query_vec = embed_dense(query)
    sparse_query = next(iter(bm25_model.query_embed([query])))

    # 2. Enforce strict tenant boundary
    tenant_filter = models.Filter(
        must=[
            models.FieldCondition(
                key="tenant_id",
                match=models.MatchValue(value=tenant_id),
            ),
        ],
    )

    # 3. Concurrent Prefetch and RRF Fusion
    response = qdrant.query_points(
        collection_name="knowledge_hybrid",
        prefetch=[
            # Branch 1: Dense Semantic Candidate Retrieval
            models.Prefetch(
                query=dense_query_vec,
                using="dense",
                filter=tenant_filter,
                limit=20,  # Gather candidate pool
            ),
            # Branch 2: Sparse BM25 Candidate Retrieval
            models.Prefetch(
                query=models.SparseVector(
                    indices=sparse_query.indices.tolist(),
                    values=sparse_query.values.tolist(),
                ),
                using="bm25",
                filter=tenant_filter,
                limit=20,
            ),
        ],
        # In-Engine Rank Fusion
        query=models.FusionQuery(
            fusion=models.Fusion.RRF,
        ),
        query_filter=tenant_filter,
        limit=limit,
        with_payload=True,
    )

    results = []
    for point in response.points:
        payload = point.payload or {}
        results.append({
            "chunk_id": payload.get("chunk_id"),
            "text": payload.get("text"),
            "source": payload.get("source"),
            "score": point.score,  # Unified RRF ranking score
        })

    return results

from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel, Field

app = FastAPI()

class HybridSearchRequest(BaseModel):
    query: str = Field(..., min_length=2, max_length=1000)
    limit: int = Field(default=5, ge=1, le=20)

async def get_current_tenant(authorization: str = Header(...)) -> str:
    # Decode cryptographic token to extract trusted tenant ID
    token = authorization.replace("Bearer ", "")
    payload = decode_jwt(token)
    return payload["tenant_id"]

@app.post("/api/v1/knowledge/hybrid-search")
async def execute_hybrid_search(
    req: HybridSearchRequest,
    tenant_id: str = Depends(get_current_tenant)
):
    try:
        results = hybrid_search(
            query=req.query,
            tenant_id=tenant_id,
            limit=req.limit
        )
        return {"query": req.query, "results": results}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

import pytest

def test_hybrid_search_sku_retrieval():
    """Ensure exact SKU inquiries retrieve the exact matching product chunk."""
    results = hybrid_search(query="Price of BYT-KB-1042?", tenant_id="tenant_123", limit=5)
    assert len(results) > 0
    assert "BYT-KB-1042" in results[0]["text"]

def test_hybrid_search_tenant_isolation():
    """Tenant A's caller must never retrieve Tenant B's confidential points."""
    results = hybrid_search(query="leave policy", tenant_id="tenant_alpha", limit=5)
    for r in results:
        assert r.get("tenant_id", "tenant_alpha") == "tenant_alpha"

def test_empty_query_raises_validation_error():
    """Empty or whitespace-only queries must trigger a validation error."""
    with pytest.raises(ValueError):
        hybrid_search(query="   ", tenant_id="tenant_123")