Architecting a Real-Time Hybrid Search System: Combining Dense Vectors and BM25 Indexes in Elasticsearch — code snippets

snippet-1.json

{
  "settings": {
    "index": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "30s",
      "codec": "best_compression"
    }
  },
  "mappings": {
    "properties": {
      "id": { "type": "keyword" },
      "sku": { "type": "keyword" },
      "title": {
        "type": "text",
        "analyzer": "english",
        "fields": {
          "keyword": { "type": "keyword", "ignore_above": 256 }
        }
      },
      "description": {
        "type": "text",
        "analyzer": "english"
      },
      "title_vector": {
        "type": "dense_vector",
        "dims": 768,
        "index": true,
        "similarity": "cosine",
        "index_options": {
          "type": "hnsw",
          "m": 16,
          "ef_construction": 100,
          "confidence_interval": 0.99,
          "quantization": {
            "type": "int8"
          }
        }
      },
      "category": { "type": "keyword" },
      "price": { "type": "double" },
      "in_stock": { "type": "boolean" },
      "created_at": { "type": "date" }
    }
  }
}

snippet-2.py

import os
import time
import logging
from typing import List, Dict, Any
from elasticsearch import Elasticsearch, helpers
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

ES_HOSTS = os.getenv("ES_HOSTS", "http://localhost:9200").split(",")
TEI_ENDPOINT = os.getenv("TEI_ENDPOINT", "http://localhost:8080/embed")
es_client = Elasticsearch(ES_HOSTS)

def get_embeddings(texts: List[str]) -> List[List[float]]:
    """Fetches embeddings from HuggingFace Text Embeddings Inference (TEI) service in batches."""
    start_time = time.time()
    try:
        response = requests.post(
            TEI_ENDPOINT,
            json={"inputs": texts},
            headers={"Content-Type": "application/json"},
            timeout=5.0
        )
        response.raise_for_status()
        embeddings = response.json()
        logger.debug(f"Embedded batch of {len(texts)} in {time.time() - start_time:.4f}s")
        return embeddings
    except Exception as e:
        logger.error(f"Failed to generate embeddings: {e}")
        raise

def bulk_index_documents(index_name: str, documents: List[Dict[str, Any]]):
    """Batches document embedding and indexes them into Elasticsearch."""
    if not documents:
        return

    titles = [doc["title"] for doc in documents]
    try:
        embeddings = get_embeddings(titles)
    except Exception as e:
        logger.error(f"Aborting batch write due to embedding generation failure: {e}")
        return

    actions = []
    for doc, embedding in zip(documents, embeddings):
        actions.append({
            "_index": index_name,
            "_id": doc["id"],
            "_source": {
                "id": doc["id"],
                "sku": doc["sku"],
                "title": doc["title"],
                "description": doc["description"],
                "title_vector": embedding,
                "category": doc["category"],
                "price": doc["price"],
                "in_stock": doc["in_stock"],
                "created_at": doc["created_at"]
            }
        })

    try:
        success, failed = helpers.bulk(es_client, actions, raise_on_error=False, chunk_size=500)
        logger.info(f"Successfully indexed {success} documents. Failed: {len(failed) if isinstance(failed, list) else failed}")
        if failed:
            logger.warning(f"Failed operations sample: {failed[:3]}")
    except Exception as e:
        logger.error(f"Elasticsearch bulk index write failure: {e}")

snippet-3.py

import os
import requests
from typing import Dict, Any
from elasticsearch import Elasticsearch

ES_HOST = os.getenv("ES_HOSTS", "http://localhost:9200")
TEI_ENDPOINT = os.getenv("TEI_ENDPOINT", "http://localhost:8080/embed")
es = Elasticsearch([ES_HOST])

def generate_query_vector(query_text: str) -> list[float]:
    response = requests.post(
        TEI_ENDPOINT,
        json={"inputs": [query_text]},
        headers={"Content-Type": "application/json"},
        timeout=2.0
    )
    response.raise_for_status()
    return response.json()[0]

def execute_hybrid_search_rrf(index_name: str, query_text: str, category_filter: str = None) -> Dict[str, Any]:
    try:
        query_vector = generate_query_vector(query_text)
    except Exception as e:
        logger.error(f"Inference failure, falling back to pure BM25: {e}")
        # Production fallback logic should be placed here
        raise

    search_payload = {
        "retrieval": {
            "rrf": {
                "window_size": 100,
                "rank_constant": 60
            }
        },
        "sub_searches": [
            {
                "query": {
                    "bool": {
                        "must": [
                            {
                                "multi_match": {
                                    "query": query_text,
                                    "fields": ["title^3", "description"],
                                    "type": "best_fields"
                                }
                            }
                        ],
                        "filter": [
                            {"term": {"category": category_filter}}
                        ] if category_filter else []
                    }
                }
            },
            {
                "query": {
                    "knn": {
                        "field": "title_vector",
                        "query_vector": query_vector,
                        "k": 50,
                        "num_candidates": 100,
                        "filter": [
                            {"term": {"category": category_filter}}
                        ] if category_filter else []
                    }
                }
            }
        ],
        "size": 20
    }
    
    response = es.search(index=index_name, body=search_payload)
    return response["hits"]

snippet-4.json

{
  "query": {
    "bool": {
      "must": [
        {
          "multi_match": {
            "query": "high-performance running shoes",
            "fields": ["title^2", "description"],
            "boost": 0.3
          }
        }
      ],
      "filter": [
        { "term": { "in_stock": true } }
      ]
    }
  },
  "knn": {
    "field": "title_vector",
    "query_vector": [0.015, -0.023, 0.084], 
    "k": 50,
    "num_candidates": 100,
    "boost": 0.7,
    "filter": [
      { "term": { "in_stock": true } }
    ]
  },
  "size": 10
}

snippet-5.txt

# Tuning index settings for high-throughput initial vector ingestion
PUT /products/_settings
Content-Type: application/json

{
  "index": {
    "refresh_interval": "-1",
    "number_of_replicas": 0,
    "translog": {
      "durability": "async",
      "sync_interval": "10s"
    }
  }
}

# Re-enabling and optimizing settings after bulk indexing complete
PUT /products/_settings
Content-Type: application/json

{
  "index": {
    "refresh_interval": "30s",
    "number_of_replicas": 1
  }
}

# Forcing segment merge to collapse HNSW graphs into a single segment for faster query times
POST /products/_forcemerge?max_num_segments=1

snippet-6.py

import hashlib
import json
import redis
from typing import List

redis_client = redis.Redis(host="localhost", port=6379, db=0)
CACHE_TTL_SECONDS = 86400  # 24 hours

def get_query_embedding_cached(query_text: str, tei_endpoint: str) -> List[float]:
    normalized_query = query_text.strip().lower()
    query_hash = hashlib.sha256(normalized_query.encode("utf-8")).hexdigest()
    cache_key = f"emb:{query_hash}"

    # Try to fetch cached embedding
    cached_data = redis_client.get(cache_key)
    if cached_data:
        return json.loads(cached_data)

    # Cache miss - hit embedding server
    import requests
    response = requests.post(
        tei_endpoint,
        json={"inputs": [query_text]},
        headers={"Content-Type": "application/json"},
        timeout=2.0
    )
    response.raise_for_status()
    embedding = response.json()[0]

    # Save to Redis
    redis_client.setex(cache_key, CACHE_TTL_SECONDS, json.dumps(embedding))
    return embedding