Implementing Dynamic Chunking and Sparse Vector Search in Qdrant for Multi-Tenant RAG — code snippets

snippet-1.py

import re
from typing import List, Dict, Any

class MarkdownStructuralChunker:
    def __init__(self, max_chunk_size: int = 1000):
        self.max_chunk_size = max_chunk_size
        self.header_regex = re.compile(r'^(#{1,6})\s+(.+)$', re.MULTILINE)

    def chunk_document(self, text: str) -> List[Dict[str, Any]]:
        lines = text.split('\n')
        chunks = []
        current_headers = {1: "", 2: "", 3: "", 4: "", 5: "", 6: ""}
        current_chunk_buffer = []
        current_size = 0

        for line in lines:
            header_match = self.header_regex.match(line)
            if header_match:
                # Flush the current buffer if it has content
                if current_chunk_buffer:
                    chunks.append(self._create_chunk(current_chunk_buffer, current_headers))
                    current_chunk_buffer = []
                    current_size = 0
                
                # Update current headers context
                level = len(header_match.group(1))
                title = header_match.group(2).strip()
                current_headers[level] = title
                # Clear deeper header levels
                for l in range(level + 1, 7):
                    current_headers[l] = ""
            
            line_len = len(line)
            if current_size + line_len > self.max_chunk_size and current_chunk_buffer:
                chunks.append(self._create_chunk(current_chunk_buffer, current_headers))
                current_chunk_buffer = [line]
                current_size = line_len
            else:
                current_chunk_buffer.append(line)
                current_size += line_len + 1 # Include newline character

        if current_chunk_buffer:
            chunks.append(self._create_chunk(current_chunk_buffer, current_headers))

        return chunks

    def _create_chunk(self, lines: List[str], headers: Dict[int, str]) -> Dict[str, Any]:
        header_path = [headers[i] for i in sorted(headers.keys()) if headers[i]]
        header_context = " > ".join(header_path)
        content = "\n".join(lines).strip()
        
        # Inject structural context directly into the text for better embedding alignment
        enriched_content = f"Context: {header_context}\n\nContent:\n{content}" if header_context else content
        
        return {
            "text": enriched_content,
            "metadata": {
                "header_path": header_path,
                "raw_length": len(content)
            }
        }

snippet-2.py

from typing import List, Dict, Any, Tuple
from fastembed import TextEmbedding, SparseTextEmbedding

class HybridEmbedder:
    def __init__(self):
        # Initialize standard lightweight production-grade models
        self.dense_model = TextEmbedding(model_name="BAAI/bge-small-en-v1.5")
        self.sparse_model = SparseTextEmbedding(model_name="prithivida/Splade_PP_en_v1")

    def embed_texts(self, texts: List[str]) -> Tuple[List[List[float]], List[Dict[int, float]]]:
        # Generate dense embeddings
        dense_embeddings = list(self.dense_model.embed(texts))
        
        # Generate sparse embeddings
        sparse_embeddings_raw = list(self.sparse_model.embed(texts))
        
        processed_sparse = []
        for sparse_vec in sparse_embeddings_raw:
            # Map sparse vector lists to a dictionary of {index: value} for Qdrant compatibility
            sparse_dict = {
                int(idx): float(val) 
                for idx, val in zip(sparse_vec.indices, sparse_vec.values)
            }
            processed_sparse.append(sparse_dict)
            
        return [list(d) for d in dense_embeddings], processed_sparse

snippet-3.py

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, 
    VectorParams, 
    SparseVectorParams, 
    SparseIndexParams, 
    PayloadSchemaType,
    HnswConfigDiff
)

def configure_qdrant_collection(client: QdrantClient, collection_name: str):
    # Ensure collection is configured with both dense and sparse configurations
    client.recreate_collection(
        collection_name=collection_name,
        vectors_config={
            "dense-vector": VectorParams(
                size=384,  # Dimensionality of BAAI/bge-small-en-v1.5
                distance=Distance.COSINE,
                hnsw_config=HnswConfigDiff(
                    m=16,
                    ef_construction=100,
                    on_disk=True  # Production tip: keep dense index on disk to save RAM
                )
            )
        },
        sparse_vectors_config={
            "sparse-vector": SparseVectorParams(
                index=SparseIndexParams(
                    on_disk=True  # Production tip: keep sparse index on disk to save RAM
                )
            )
        }
    )

    # CRITICAL: Create keyword index for hard tenant isolation
    client.create_payload_index(
        collection_name=collection_name,
        field_name="tenant_id",
        field_schema=PayloadSchemaType.KEYWORD
    )

snippet-4.py

import time
import uuid
from typing import List, Dict, Any
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
from qdrant_client.http.exceptions import UnexpectedResponse

def ingest_tenant_chunks(
    client: QdrantClient, 
    collection_name: str, 
    tenant_id: str, 
    chunks: List[Dict[str, Any]], 
    embedder: HybridEmbedder,
    batch_size: int = 64
):
    texts = [chunk["text"] for chunk in chunks]
    dense_vecs, sparse_vecs = embedder.embed_texts(texts)
    
    points = []
    for i, chunk in enumerate(chunks):
        point_id = str(uuid.uuid4())
        
        # Format sparse vector as Qdrant expects it
        sparse_vector_data = {
            "indices": list(sparse_vecs[i].keys()),
            "values": list(sparse_vecs[i].values())
        }

        points.append(
            PointStruct(
                id=point_id,
                vector={
                    "dense-vector": dense_vecs[i],
                    "sparse-vector": sparse_vector_data
                },
                payload={
                    "tenant_id": tenant_id,
                    "text": chunk["text"],
                    "metadata": chunk["metadata"]
                }
            )
        )

    # Batch upload with retry logic
    for i in range(0, len(points), batch_size):
        batch = points[i : i + batch_size]
        retries = 3
        while retries > 0:
            try:
                client.upsert(
                    collection_name=collection_name,
                    points=batch,
                    wait=True
                )
                break
            except UnexpectedResponse as e:
                retries -= 1
                if retries == 0:
                    raise RuntimeError(f"Failed to upsert batch to Qdrant after retries: {e}")
                time.sleep(2 ** (3 - retries)) # Exponential backoff

snippet-5.py

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

def query_tenant_hybrid(
    client: QdrantClient,
    collection_name: str,
    tenant_id: str,
    query_text: str,
    embedder: HybridEmbedder,
    limit: int = 5
) -> List[Dict[str, Any]]:
    # Generate query embeddings
    dense_queries, sparse_queries = embedder.embed_texts([query_text])
    dense_query = dense_queries[0]
    sparse_query = sparse_queries[0]
    
    # Format sparse query vector for API
    sparse_vector_query = models.SparseVector(
        indices=list(sparse_query.keys()),
        values=list(sparse_query.values())
    )

    # Core filter to enforce hard tenant isolation
    tenant_filter = models.Filter(
        must=[
            models.FieldCondition(
                key="tenant_id",
                match=models.MatchValue(value=tenant_id)
            )
        ]
    )

    # Execute Hybrid query using Reciprocal Rank Fusion (RRF)
    search_result = client.query_points(
        collection_name=collection_name,
        prefetch=[
            models.Prefetch(
                query=dense_query,
                using="dense-vector",
                filter=tenant_filter,
                limit=limit * 3
            ),
            models.Prefetch(
                query=sparse_vector_query,
                using="sparse-vector",
                filter=tenant_filter,
                limit=limit * 3
            )
        ],
        query=models.FusionQuery(
            fusion=models.Fusion.RRF
        ),
        limit=limit
    )

    results = []
    for point in search_result.points:
        results.append({
            "id": point.id,
            "score": point.score,
            "text": point.payload.get("text"),
            "metadata": point.payload.get("metadata")
        })
        
    return results

snippet-6.yaml

# config.yaml (Qdrant Production Tunings)
storage:
  performance:
    max_search_threads: 0 # Automatically scale to CPU cores
  mmap_threshold_kb: 20000 # Memory-map vector segments larger than 20MB to disk

optimizers:
  indexing_threshold_kb: 20000 # Minimum segment size to build HNSW index
  default_vacuum_min_vector_number: 1000

service:
  grpc_port: 6334
  enable_cors: false

snippet-7.py

from qdrant_client import QdrantClient
from qdrant_client import models

def hard_delete_tenant_data(client: QdrantClient, collection_name: str, tenant_id: str):
    # Perform a payload-filtered delete to ensure isolation
    client.delete(
        collection_name=collection_name,
        points_selector=models.FilterSelector(
            filter=models.Filter(
                must=[
                    models.FieldCondition(
                        key="tenant_id",
                        match=models.MatchValue(value=tenant_id)
                    )
                ]
            )
        )
    )