Comprehensive Guide to Vector Search Implementation

import os
from langchain.document_loaders import DirectoryLoader, TextLoader, PDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.vectorstores import Chroma

# Document loading - adapt to your data sources
loader = DirectoryLoader('./data', glob="**/*.pdf", loader_cls=PDFLoader)
documents = loader.load()

# Document chunking - parameters highly impact retrieval quality
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# Metadata enrichment
for i, chunk in enumerate(chunks):
    # Add source tracking and other useful metadata
    chunk.metadata["chunk_id"] = i
    chunk.metadata["source_type"] = "pdf"
    # Extract or compute additional metadata as needed
    
# Embedding generation
# Option 1: OpenAI embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Option 2: Open source embeddings
# embeddings = HuggingFaceEmbeddings(
#     model_name="BAAI/bge-large-en-v1.5",
#     model_kwargs={'device': 'cuda'},
#     encode_kwargs={'normalize_embeddings': True}
# )

# Store in vector database
vectordb = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
  )

# Example: Integrating with Qdrant as a production vector store

import qdrant_client
from qdrant_client.http import models as rest
from qdrant_client.http.models import Distance, VectorParams, OptimizersConfigDiff
from langchain_qdrant import QdrantVectorStore

# Initialize client
client = qdrant_client.QdrantClient(
    url="https://your-qdrant-instance.com",
    api_key="your-api-key"  # For cloud deployments
)

# Create collection with optimized configuration
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,  # Dimension of your embeddings
        distance=Distance.COSINE,  # COSINE, DOT, EUCLID
    ),
    optimizers_config=OptimizersConfigDiff(
        indexing_threshold=20000,  # Points to build index
        memmap_threshold=100000,   # When to use memory mapping
    ),
    hnsw_config=rest.HnswConfigDiff(
        m=16,              # Number of bidirectional links created for each new element
        ef_construct=128,  # Size of the dynamic list for the nearest neighbors
        full_scan_threshold=10000,  # When to use brute force vs hnsw
    )
)

# Configure payload indexes for efficient filtering
client.create_payload_index(
    collection_name="documents",
    field_name="metadata.source_type",
    field_schema=rest.PayloadSchemaType.KEYWORD,
)

client.create_payload_index(
    collection_name="documents",
    field_name="metadata.date",
    field_schema=rest.PayloadSchemaType.DATETIME,
)

# Batch upload points (more efficient than individual uploads)
from langchain_openai import OpenAIEmbeddings
embedding_model = OpenAIEmbeddings()

vectors = []
metadatas = []
ids = []
texts = []

# Prepare batches of 100 documents
for i, doc in enumerate(chunks):
    ids.append(str(i))
    texts.append(doc.page_content)
    metadatas.append(doc.metadata)
    
    # Process in batches of 100
    if len(texts) >= 100 or i == len(chunks) - 1:
        # Generate embeddings in batch
        batch_embeddings = embedding_model.embed_documents(texts)
        
        # Create point objects
        points = [
            rest.PointStruct(
                id=id,
                vector=embedding.tolist(),
                payload=metadata
            )
            for id, embedding, metadata in zip(ids, batch_embeddings, metadatas)
        ]
        
        # Upload batch
        client.upsert(
            collection_name="documents",
            points=points
        )
        
        # Clear batch
        vectors = []
        metadatas = []
        ids = []
    texts = []

from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import EmbeddingsFilter

# Initialize retriever with vector store
embeddings = OpenAIEmbeddings()
vector_store = QdrantVectorStore(
    client=client,
    collection_name="documents",
    embeddings=embeddings,
)

# Basic retriever
basic_retriever = vector_store.as_retriever(
    search_type="similarity",  # "similarity", "mmr", or "similarity_score_threshold"
    search_kwargs={
        "k": 10,  # Number of results to retrieve
        "score_threshold": 0.7,  # Only for similarity_score_threshold
        # "fetch_k": 50,  # For MMR, fetch this many candidates
        # "lambda_mult": 0.5,  # For MMR, diversity vs relevance control
    }
)

# Advanced retriever with metadata filtering
def retrieve_with_filters(query, filters=None):
    # Create filter dict for Qdrant if filters provided
    filter_dict = None
    if filters:
        filter_dict = {
            "must": [
                {
                    "key": f"metadata.{key}",
                    "match": {"value": value}
                }
                for key, value in filters.items()
            ]
        }
    
    # Convert query to embedding
    query_embedding = embeddings.embed_query(query)
    
    # Search with filters
    results = client.search(
        collection_name="documents",
        query_vector=query_embedding,
        limit=10,
        filter=filter_dict,
        with_payload=True,
        with_vectors=False  # Save bandwidth by not returning vectors
    )
    
    # Process and return results
    documents = []
    for result in results:
        doc = Document(
            page_content=result.payload.get("text", ""),
            metadata=result.payload.get("metadata", {})
        )
        doc.metadata["score"] = result.score  # Add similarity score
        documents.append(doc)
    
    return documents

# Implement hybrid search (combine vector search with keyword search)
def hybrid_search(query, filters=None, keyword_weight=0.2):
    # Vector search
    vector_results = retrieve_with_filters(query, filters)
    
    # Keyword search using your preferred text search engine
    # This is a simplified example - replace with actual keyword search implementation
    keyword_results = keyword_search_function(query, filters)
    
    # Score normalization and fusion
    combined_results = {}
    
    # Process vector results (assuming scores are 0-1, higher is better)
    for doc in vector_results:
        doc_id = doc.metadata.get("chunk_id")
        combined_results[doc_id] = {
            "doc": doc,
            "vector_score": doc.metadata.get("score", 0),
            "keyword_score": 0
        }
    
    # Process keyword results and combine scores
    for doc in keyword_results:
        doc_id = doc.metadata.get("chunk_id")
        keyword_score = doc.metadata.get("keyword_score", 0)
        
        if doc_id in combined_results:
            combined_results[doc_id]["keyword_score"] = keyword_score
        else:
            combined_results[doc_id] = {
                "doc": doc,
                "vector_score": 0,
                "keyword_score": keyword_score
            }
    
    # Calculate hybrid score and sort
    results_with_scores = []
    for doc_id, scores in combined_results.items():
        hybrid_score = (1 - keyword_weight) * scores["vector_score"] + keyword_weight * scores["keyword_score"]
        scores["doc"].metadata["hybrid_score"] = hybrid_score
        results_with_scores.append((scores["doc"], hybrid_score))
    
    # Sort by hybrid score and return documents
    results_with_scores.sort(key=lambda x: x[1], reverse=True)
    return [doc for doc, score in results_with_scores[:10]]

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# Initialize LLM
llm = ChatOpenAI(model="gpt-4-turbo")

# Define RAG prompt template
prompt_template = """
You are an assistant with access to a knowledge base. 
Answer the user's question based ONLY on the following context:

{context}

If the answer is not contained within the context, say "I don't have enough information to answer this question" and suggest a follow-up question.

Question: {question}

Answer:
"""
prompt = ChatPromptTemplate.from_template(prompt_template)

# Function to format context documents
def format_docs(docs):
    return "\n\n".join([f"Document {i+1}:\n" + doc.page_content for i, doc in enumerate(docs)])

# Create RAG pipeline
rag_chain = (
    {"context": basic_retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Process a user query
user_query = "How do I implement vector search with PostgreSQL?"
response = rag_chain.invoke(user_query)
print(response)

# Enhanced RAG with re-ranking
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# Document extractor to get only relevant parts of retrieved documents
compressor = LLMChainExtractor.from_llm(llm)

# Create a compressing retriever
compression_retriever = ContextualCompressionRetriever(
    base_retriever=basic_retriever,
    base_compressor=compressor,
)

# Enhanced RAG chain with compressed retrieval
enhanced_rag_chain = (
    {"context": compression_retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Process the same query with enhanced retrieval
enhanced_response = enhanced_rag_chain.invoke(user_query)
print(enhanced_response)

from langchain.evaluation import EvaluatorChain
from langchain.evaluation.criteria import LLMCriteriaEvaluator
from langchain.smith import RunEvalConfig, run_on_dataset
import pandas as pd
import logging
from prometheus_client import Counter, Histogram, start_http_server

# Set up basic logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("vector_search.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("vector_search")

# Prometheus metrics (for production monitoring)
QUERY_COUNTER = Counter('vector_search_queries_total', 'Total number of vector search queries')
RETRIEVAL_LATENCY = Histogram('vector_search_latency_seconds', 'Vector search latency in seconds')
RESULTS_COUNT = Histogram('vector_search_results_count', 'Number of results returned per query')
RELEVANCE_SCORE = Histogram('vector_search_relevance_score', 'Relevance scores of retrieved documents')

# Start Prometheus HTTP server on port 8000
start_http_server(8000)

# Wrapper function with instrumentation
def instrumented_vector_search(query, filters=None):
    QUERY_COUNTER.inc()
    
    # Record latency
    start_time = time.time()
    
    try:
        # Perform the search
        results = retrieve_with_filters(query, filters)
        
        # Record metrics
        search_time = time.time() - start_time
        RETRIEVAL_LATENCY.observe(search_time)
        RESULTS_COUNT.observe(len(results))
        
        # Log relevance scores if available
        if results and hasattr(results[0], 'metadata') and 'score' in results[0].metadata:
            for result in results:
                RELEVANCE_SCORE.observe(result.metadata['score'])
        
        # Log search info
        logger.info(f"Query: '{query}' | Results: {len(results)} | Time: {search_time:.3f}s")
        
        return results
    
    except Exception as e:
        logger.error(f"Search error for query '{query}': {str(e)}")
        raise

# Evaluation framework for retrieval quality
def evaluate_retrieval_quality(test_queries, ground_truth):
    """
    Evaluate retrieval quality using a test set
    
    Args:
        test_queries: List of test queries
        ground_truth: Dictionary mapping queries to relevant document IDs
    
    Returns:
        Dictionary with evaluation metrics
    """
    results = {}
    
    for query in test_queries:
        retrieved_docs = retrieve_with_filters(query)
        retrieved_ids = [doc.metadata.get("chunk_id") for doc in retrieved_docs]
        
        # Calculate metrics
        relevant_ids = ground_truth.get(query, [])
        
        # Precision at k
        precision = len(set(retrieved_ids) & set(relevant_ids)) / len(retrieved_ids) if retrieved_ids else 0
        
        # Recall
        recall = len(set(retrieved_ids) & set(relevant_ids)) / len(relevant_ids) if relevant_ids else 0
        
        # F1 score
        f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        # MRR (Mean Reciprocal Rank)
        mrr = 0
        for i, doc_id in enumerate(retrieved_ids):
            if doc_id in relevant_ids:
                mrr = 1 / (i + 1)
                break
        
        results[query] = {
            "precision": precision,
            "recall": recall,
            "f1": f1,
            "mrr": mrr
        }
    
    # Calculate averages
    avg_metrics = {
        "avg_precision": sum(r["precision"] for r in results.values()) / len(results),
        "avg_recall": sum(r["recall"] for r in results.values()) / len(results),
        "avg_f1": sum(r["f1"] for r in results.values()) / len(results),
        "avg_mrr": sum(r["mrr"] for r in results.values()) / len(results),
    }
    
    return {"per_query": results, "average": avg_metrics}

# LLM-based evaluation for RAG responses
evaluator = LLMCriteriaEvaluator(
    llm=llm,
    criteria={
        "relevance": "Is the response relevant to the query?",
        "accuracy": "Is the response accurate based on the retrieved context?",
        "completeness": "Does the response completely address all aspects of the query?",
        "hallucination": "Does the response contain information not supported by the retrieved context?"
    }
)