Postmortem: Our RAG Pipeline Returned Wrong Answers After Elasticsearch 8.15 and LlamaIndex 0.11 Had Index Corruption

import os
import logging
import hashlib
import json
from elasticsearch import Elasticsearch, ConnectionError, RequestError
from llamaindex.core import VectorStoreIndex, SimpleDirectoryReader
from llamaindex.vector_stores.elasticsearch import ElasticsearchVectorStore
from llamaindex.embeddings.openai import OpenAIEmbedding

# Configure logging to capture sync events and errors
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Bug: This sync logic does not validate ES segment checksums before overwriting local cache
# Leading to silent index corruption when ES 8.15 returns mismatched quantized vectors
def sync_llamaindex_to_elasticsearch(
    doc_dir: str,
    es_host: str = "https://localhost:9200",
    es_index: str = "rag_pipeline_v1",
    embed_model_name: str = "text-embedding-3-small"
) -> VectorStoreIndex:
    """
    Syncs local documents to Elasticsearch vector store using LlamaIndex.
    Contains the buggy logic that caused corruption in ES 8.15 + LlamaIndex 0.11.
    """
    try:
        # Initialize Elasticsearch client with error handling for connection issues
        es_client = Elasticsearch(
            es_host,
            basic_auth=("elastic", os.getenv("ES_PASSWORD")),
            verify_certs=False,  # Dev only, do not use in prod
            request_timeout=30
        )
        # Check if ES is available
        if not es_client.ping():
            raise ConnectionError("Elasticsearch cluster is not reachable")
        logger.info(f"Connected to Elasticsearch at {es_host}")

        # Initialize LlamaIndex vector store with ES backend
        vector_store = ElasticsearchVectorStore(
            es_client=es_client,
            index_name=es_index,
            embed_dim=1536  # Matches text-embedding-3-small dimensions
        )

        # Load documents from local directory
        logger.info(f"Loading documents from {doc_dir}")
        documents = SimpleDirectoryReader(doc_dir).load_data()
        if not documents:
            raise ValueError(f"No documents found in {doc_dir}")

        # Initialize embedding model
        embed_model = OpenAIEmbedding(model=embed_model_name)
        logger.info(f"Using embedding model: {embed_model_name}")

        # Create VectorStoreIndex with buggy sync logic:
        # No checksum validation of ES-stored vectors before building index
        # ES 8.15 returns quantized vectors with mismatched checksums here
        logger.info("Building VectorStoreIndex (buggy sync logic)")
        index = VectorStoreIndex.from_documents(
            documents,
            vector_store=vector_store,
            embed_model=embed_model,
            show_progress=True
        )

        # Persist index to disk (overwrites local cache without validation)
        index.storage_context.persist("./storage")
        logger.info(f"Index synced to ES index {es_index} and local storage")

        return index

    except ConnectionError as ce:
        logger.error(f"Elasticsearch connection failed: {ce}")
        raise
    except RequestError as re:
        logger.error(f"Elasticsearch request error: {re}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error during sync: {e}")
        raise

if __name__ == "__main__":
    # Example usage of buggy sync logic
    try:
        index = sync_llamaindex_to_elasticsearch(
            doc_dir="./data/prod_docs",
            es_host="https://es-prod.internal:9200",
            es_index="rag_prod_v2"
        )
        logger.info("Sync completed successfully (corruption risk present)")
    except Exception as e:
        logger.error(f"Sync failed: {e}")
        exit(1)

import os
import logging
import hashlib
import json
from elasticsearch import Elasticsearch, RequestError
from llamaindex.core import VectorStoreIndex, StorageContext
from llamaindex.vector_stores.elasticsearch import ElasticsearchVectorStore
from llamaindex.embeddings.openai import OpenAIEmbedding

# Configure logging for corruption detection
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def compute_vector_checksum(vector: list) -> str:
    """Compute SHA-256 checksum of a vector payload for validation."""
    try:
        # Convert vector to sorted JSON string to ensure consistent hashing
        vector_str = json.dumps(vector, sort_keys=True)
        return hashlib.sha256(vector_str.encode("utf-8")).hexdigest()
    except Exception as e:
        logger.error(f"Failed to compute checksum for vector: {e}")
        return ""

def detect_index_corruption(
    es_host: str,
    es_index: str,
    local_storage_dir: str = "./storage"
) -> dict:
    """
    Detects index corruption by comparing ES-stored vector checksums
    with locally cached LlamaIndex checksums.
    Returns a report of mismatches and corruption rate.
    """
    report = {
        "total_vectors": 0,
        "corrupted_vectors": 0,
        "mismatch_details": [],
        "corruption_rate": 0.0
    }

    try:
        # Initialize ES client
        es_client = Elasticsearch(
            es_host,
            basic_auth=("elastic", os.getenv("ES_PASSWORD")),
            verify_certs=False,
            request_timeout=30
        )
        if not es_client.ping():
            raise ConnectionError("Elasticsearch cluster is not reachable")

        # Initialize LlamaIndex vector store and storage context
        vector_store = ElasticsearchVectorStore(
            es_client=es_client,
            index_name=es_index,
            embed_dim=1536
        )
        storage_context = StorageContext.from_defaults(
            vector_store=vector_store,
            persist_dir=local_storage_dir
        )

        # Load local index to get cached vectors
        logger.info(f"Loading local index from {local_storage_dir}")
        local_index = VectorStoreIndex.load_from_storage(storage_context)
        local_vectors = local_index.vector_store._get_all_vectors()  # Internal method, prod use stable API

        # Fetch all vectors from Elasticsearch (paginated to handle large indices)
        logger.info(f"Fetching vectors from ES index {es_index}")
        es_vectors = []
        scroll_id = None
        while True:
            if scroll_id:
                res = es_client.scroll(scroll_id=scroll_id, scroll="1m")
            else:
                res = es_client.search(
                    index=es_index,
                    scroll="1m",
                    body={"query": {"match_all": {}}, "size": 1000}
                )
            scroll_id = res["_scroll_id"]
            hits = res["hits"]["hits"]
            if not hits:
                break
            es_vectors.extend([hit["_source"] for hit in hits])
            if len(hits) < 1000:
                break
        es_client.clear_scroll(scroll_id=scroll_id)

        report["total_vectors"] = len(es_vectors)
        logger.info(f"Total vectors to validate: {report['total_vectors']}")

        # Compare checksums
        for es_vec in es_vectors:
            vec_id = es_vec.get("doc_id", "unknown")
            es_embedding = es_vec.get("embedding", [])
            es_checksum = es_vec.get("checksum", "")

            # Get local vector checksum
            local_embedding = local_vectors.get(vec_id, {}).get("embedding", [])
            if not local_embedding:
                logger.warning(f"No local vector found for ID {vec_id}")
                continue

            local_checksum = compute_vector_checksum(local_embedding)
            computed_es_checksum = compute_vector_checksum(es_embedding)

            # Check for mismatches
            if es_checksum != local_checksum or computed_es_checksum != local_checksum:
                report["corrupted_vectors"] += 1
                report["mismatch_details"].append({
                    "vector_id": vec_id,
                    "es_stored_checksum": es_checksum,
                    "es_computed_checksum": computed_es_checksum,
                    "local_checksum": local_checksum
                })
                logger.warning(f"Corruption detected for vector {vec_id}")

        # Calculate corruption rate
        if report["total_vectors"] > 0:
            report["corruption_rate"] = (report["corrupted_vectors"] / report["total_vectors"]) * 100

        logger.info(f"Corruption detection complete. Rate: {report['corruption_rate']:.2f}%")
        return report

    except ConnectionError as ce:
        logger.error(f"ES connection failed: {ce}")
        raise
    except RequestError as re:
        logger.error(f"ES request error: {re}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error during corruption detection: {e}")
        raise

if __name__ == "__main__":
    try:
        report = detect_index_corruption(
            es_host="https://es-prod.internal:9200",
            es_index="rag_prod_v2",
            local_storage_dir="./storage"
        )
        print(json.dumps(report, indent=2))
    except Exception as e:
        logger.error(f"Corruption detection failed: {e}")
        exit(1)

import os
import logging
import hashlib
import json
from elasticsearch import Elasticsearch, ConnectionError, RequestError
from llamaindex.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext
from llamaindex.vector_stores.elasticsearch import ElasticsearchVectorStore
from llamaindex.embeddings.openai import OpenAIEmbedding

# Configure logging for fixed sync logic
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def compute_vector_checksum(vector: list) -> str:
    """Compute SHA-256 checksum of a vector payload for validation."""
    try:
        vector_str = json.dumps(vector, sort_keys=True)
        return hashlib.sha256(vector_str.encode("utf-8")).hexdigest()
    except Exception as e:
        logger.error(f"Failed to compute checksum: {e}")
        return ""

def fixed_sync_llamaindex_to_elasticsearch(
    doc_dir: str,
    es_host: str = "https://localhost:9200",
    es_index: str = "rag_pipeline_v1",
    embed_model_name: str = "text-embedding-3-small",
    validate_checksums: bool = True
) -> VectorStoreIndex:
    """
    Fixed sync logic with checksum validation to prevent index corruption.
    Compatible with Elasticsearch 8.14.1 and LlamaIndex 0.10.43.
    """
    try:
        # Initialize ES client with production-grade settings
        es_client = Elasticsearch(
            es_host,
            basic_auth=("elastic", os.getenv("ES_PASSWORD")),
            verify_certs=True,  # Enforce cert validation in prod
            ca_certs="./es_ca.pem",
            request_timeout=30
        )
        if not es_client.ping():
            raise ConnectionError("Elasticsearch cluster is not reachable")
        logger.info(f"Connected to Elasticsearch at {es_host}")

        # Check ES version for compatibility
        es_version = es_client.info()["version"]["number"]
        if es_version.startswith("8.15"):
            logger.warning(f"Elasticsearch {es_version} has known corruption issues with LlamaIndex 0.11")
            if validate_checksums:
                logger.info("Enabling strict checksum validation for ES 8.15")

        # Initialize vector store and embedding model
        vector_store = ElasticsearchVectorStore(
            es_client=es_client,
            index_name=es_index,
            embed_dim=1536
        )
        embed_model = OpenAIEmbedding(model=embed_model_name)
        logger.info(f"Using embedding model: {embed_model_name}")

        # Load and process documents
        logger.info(f"Loading documents from {doc_dir}")
        documents = SimpleDirectoryReader(doc_dir).load_data()
        if not documents:
            raise ValueError(f"No documents found in {doc_dir}")

        # Create index with checksum validation
        logger.info("Building VectorStoreIndex with checksum validation")
        index = VectorStoreIndex.from_documents(
            documents,
            vector_store=vector_store,
            embed_model=embed_model,
            show_progress=True
        )

        # Validate checksums before persisting (if enabled)
        if validate_checksums:
            logger.info("Validating vector checksums before persistence")
            # Get all vectors from ES
            es_vectors = []
            scroll_id = None
            while True:
                if scroll_id:
                    res = es_client.scroll(scroll_id=scroll_id, scroll="1m")
                else:
                    res = es_client.search(
                        index=es_index,
                        scroll="1m",
                        body={"query": {"match_all": {}}, "size": 1000}
                    )
                scroll_id = res["_scroll_id"]
                hits = res["hits"]["hits"]
                if not hits:
                    break
                es_vectors.extend([hit["_source"] for hit in hits])
                if len(hits) < 1000:
                    break
            es_client.clear_scroll(scroll_id=scroll_id)

            # Compute and compare checksums
            corruption_found = False
            for vec in es_vectors:
                vec_id = vec.get("doc_id", "unknown")
                es_embedding = vec.get("embedding", [])
                stored_checksum = vec.get("checksum", "")
                computed_checksum = compute_vector_checksum(es_embedding)

                if stored_checksum != computed_checksum:
                    logger.error(f"Checksum mismatch for vector {vec_id}: stored {stored_checksum}, computed {computed_checksum}")
                    corruption_found = True

            if corruption_found:
                raise ValueError("Checksum validation failed. Aborting persistence to prevent corruption.")

        # Persist index to disk only if validation passes
        storage_context = StorageContext.from_defaults(
            vector_store=vector_store,
            persist_dir="./storage"
        )
        index.storage_context.persist("./storage")
        logger.info(f"Index synced successfully to ES {es_index} and local storage")

        return index

    except ConnectionError as ce:
        logger.error(f"ES connection failed: {ce}")
        raise
    except RequestError as re:
        logger.error(f"ES request error: {re}")
        raise
    except ValueError as ve:
        logger.error(f"Validation error: {ve}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

if __name__ == "__main__":
    try:
        index = fixed_sync_llamaindex_to_elasticsearch(
            doc_dir="./data/prod_docs",
            es_host="https://es-prod.internal:9200",
            es_index="rag_prod_v2_fixed"
        )
        logger.info("Fixed sync completed successfully with no corruption risk")
    except Exception as e:
        logger.error(f"Fixed sync failed: {e}")
        exit(1)

# pytest test to validate version compatibility
import pytest
from elasticsearch import Elasticsearch
from llamaindex import __version__ as llamaindex_version

def test_es_llamaindex_compatibility():
    es_client = Elasticsearch("https://localhost:9200")
    es_version = es_client.info()["version"]["number"]
    # Compatible pairs: ES 8.14.x + LlamaIndex 0.10.x, ES 8.15.x + LlamaIndex 0.11.x (with patches)
    if es_version.startswith("8.15") and not llamaindex_version.startswith("0.11"):
        pytest.fail(f"ES {es_version} requires LlamaIndex 0.11.x with patches")
    if es_version.startswith("8.14") and not llamaindex_version.startswith("0.10"):
        pytest.fail(f"ES {es_version} requires LlamaIndex 0.10.x")

# Function to compute vector checksum
import hashlib
import json

def compute_vector_checksum(vector: list) -> str:
    """Compute SHA-256 checksum of a vector embedding."""
    try:
        # Sort keys to ensure consistent JSON serialization
        vector_str = json.dumps(vector, sort_keys=True)
        return hashlib.sha256(vector_str.encode("utf-8")).hexdigest()
    except Exception as e:
        raise ValueError(f"Checksum computation failed: {e}")

# Check LLM response confidence from LlamaIndex metadata
from llamaindex.core.response import Response

def check_answer_confidence(response: Response, threshold: float = 0.7) -> bool:
    """Return True if answer confidence is above threshold."""
    confidence = response.metadata.get("confidence", 0.0)
    if confidence < threshold:
        logger.warning(f"Low confidence answer: {confidence} < {threshold}")
        return False
    return True