Essential Techniques for Production Vector Search Systems Part 2 - Binary Quantization

pip install qdrant-client fastembed numpy

# Add to your ~/.bashrc or ~/.zshrc
export QDRANT_URL="https://your-cluster-url.cloud.qdrant.io:6333"
export QDRANT_API_KEY="your-api-key-here"

# .env file
QDRANT_URL=https://your-cluster-url.cloud.qdrant.io:6333
QDRANT_API_KEY=your-api-key-here

from qdrant_client import QdrantClient
from dotenv import load_dotenv
import os

# Load environment variables from .env file
load_dotenv()

# Initialize client
client = QdrantClient(
    url=os.getenv("QDRANT_URL"),
    api_key=os.getenv("QDRANT_API_KEY"),
)

# Test connection
try:
    collections = client.get_collections()
    print(f" Connected successfully!")
    print(f"   Current collections: {len(collections.collections)}")
except Exception as e:
    print(f" Connection failed: {e}")
    print("   Check your .env file has QDRANT_URL and QDRANT_API_KEY")

python verify-connection.py
Connected successfully!
   Current collections: 2

"""Generic Binary Quantization Implementation for Qdrant"""

from qdrant_client import QdrantClient
from dotenv import load_dotenv
import os
from typing import List, Dict, Any, Optional
import numpy as np
import time

# Cache the embedding model globally
_embedding_model = None

def get_embedding_model():
    """Get or create the embedding model (cached)."""
    global _embedding_model
    if _embedding_model is None:
        try:
            from sentence_transformers import SentenceTransformer
            _embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        except ImportError:
            raise ImportError(
                "sentence-transformers not installed. Install it with: pip install sentence-transformers"
            )
    return _embedding_model


def get_qdrant_client() -> QdrantClient:
    """Initialize and return Qdrant client."""
    load_dotenv()
    return QdrantClient(
        url=os.getenv("QDRANT_URL"),
        api_key=os.getenv("QDRANT_API_KEY"),
    )


def float_to_binary(vector: List[float], threshold: Optional[float] = None) -> List[int]:
    """Convert a float vector to a binary vector."""
    vector_array = np.array(vector)
    if threshold is None:
        threshold = np.median(vector_array)
    return (vector_array >= threshold).astype(int).tolist()


def hamming_distance(binary_vec1: List[int], binary_vec2: List[int]) -> float:
    """Calculate Hamming similarity between two binary vectors."""
    if len(binary_vec1) != len(binary_vec2):
        raise ValueError("Vectors must have the same length")
    vec1_array = np.array(binary_vec1, dtype=np.uint8)
    vec2_array = np.array(binary_vec2, dtype=np.uint8)
    differences = np.sum(vec1_array != vec2_array)
    return 1.0 - (differences / len(binary_vec1))


def binary_quantize_search(
    collection_name: str,
    query: str,
    client: Optional[QdrantClient] = None,
    limit: int = 10,
    threshold: Optional[float] = None
) -> List[Dict[str, Any]]:
    """Perform vector search using binary quantized vectors."""
    if client is None:
        client = get_qdrant_client()
    
    model = get_embedding_model()
    query_vector = model.encode(query).tolist()
    query_binary = float_to_binary(query_vector, threshold)
    query_binary_array = np.array(query_binary, dtype=np.uint8)
    
    try:
        candidate_limit = limit * 5
        search_response = client.query_points(
            collection_name=collection_name,
            query=query_vector,
            limit=candidate_limit,
            with_payload=True,
            with_vectors=True
        )
        
        if not search_response.points:
            return []
        
        results = []
        threshold_val = threshold if threshold is not None else np.median(query_vector)
        
        for point in search_response.points:
            if point.vector is None:
                continue
            
            vector = list(point.vector.values())[0] if isinstance(point.vector, dict) else point.vector
            vector_array = np.array(vector)
            stored_binary_array = (vector_array >= threshold_val).astype(np.uint8)
            differences = np.sum(query_binary_array != stored_binary_array)
            similarity = 1.0 - (differences / len(query_binary))
            
            results.append({
                "id": point.id,
                "score": similarity,
                "payload": point.payload,
                "binary_similarity": similarity
            })
        
        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:limit]
        
    except Exception as e:
        print(f"Binary quantized search error: {e}")
        return []


def full_precision_search(
    collection_name: str,
    query: str,
    client: Optional[QdrantClient] = None,
    limit: int = 10
) -> List[Dict[str, Any]]:
    """Perform standard full precision vector search."""
    if client is None:
        client = get_qdrant_client()
    
    model = get_embedding_model()
    query_vector = model.encode(query).tolist()
    
    try:
        search_response = client.query_points(
            collection_name=collection_name,
            query=query_vector,
            limit=limit,
            with_payload=True
        )
        
        return [
            {
                "id": r.id,
                "score": r.score,
                "payload": r.payload,
                "cosine_similarity": r.score
            }
            for r in search_response.points
        ]
    except Exception as e:
        print(f"Full precision search error: {e}")
        return []


def compare_search_methods(
    collection_name: str,
    query: str,
    client: Optional[QdrantClient] = None,
    limit: int = 10,
    threshold: Optional[float] = None
) -> Dict[str, Any]:
    """Compare full precision search vs binary quantized search."""
    if client is None:
        client = get_qdrant_client()
    
    print(f"\nComparing Search Methods for: '{query}'")
    print("=" * 80)
    
    # Full precision search
    print("\n1. Full Precision Search (Cosine Similarity)")
    print("-" * 80)
    start_time = time.time()
    full_results = full_precision_search(
        collection_name=collection_name,
        query=query,
        client=client,
        limit=limit
    )
    full_time = time.time() - start_time
    
    # Binary quantized search
    print("\n2. Binary Quantized Search (Hamming Similarity)")
    print("-" * 80)
    start_time = time.time()
    binary_results = binary_quantize_search(
        collection_name=collection_name,
        query=query,
        client=client,
        limit=limit,
        threshold=threshold
    )
    binary_time = time.time() - start_time
    
    # Compare results
    full_ids = {r["id"] for r in full_results}
    binary_ids = {r["id"] for r in binary_results}
    
    overlap = len(full_ids & binary_ids)
    overlap_ratio = overlap / limit if limit > 0 else 0.0
    
    comparison = {
        "query": query,
        "full_precision": {
            "results": full_results,
            "time_ms": full_time * 1000,
            "count": len(full_results)
        },
        "binary_quantized": {
            "results": binary_results,
            "time_ms": binary_time * 1000,
            "count": len(binary_results)
        },
        "overlap": overlap,
        "overlap_ratio": overlap_ratio
    }
    
    # Display comparison summary
    print("\n" + "=" * 80)
    print("COMPARISON SUMMARY")
    print("=" * 80)
    print(f"Full Precision Search:")
    print(f"  Time: {full_time*1000:.2f} ms")
    print(f"  Results: {len(full_results)}")
    if full_results:
        print(f"  Top Score: {full_results[0]['score']:.4f}")
    
    print(f"\nBinary Quantized Search:")
    print(f"  Time: {binary_time*1000:.2f} ms")
    print(f"  Results: {len(binary_results)}")
    if binary_results:
        print(f"  Top Score: {binary_results[0]['score']:.4f}")
    
    print(f"\nOverlap:")
    print(f"  Common Results: {overlap} / {limit} ({overlap_ratio*100:.1f}%)")
    
    speedup = full_time / binary_time if binary_time > 0 else 0
    if speedup > 1:
        print(f"  Binary search is {speedup:.2f}x faster")
    elif speedup > 0:
        print(f"  Full precision search is {1/speedup:.2f}x faster")
    
    print("\n" + "=" * 80)
    
    return comparison


def display_binary_results(results: List[Dict[str, Any]], query: str, show_fields: Optional[List[str]] = None):
    """Display binary quantized search results."""
    if show_fields is None:
        show_fields = ['name', 'title', 'description', 'text']
    
    print(f"\nBinary Quantized Search Results for: '{query}'")
    print("=" * 80)
    
    if not results:
        print("No results found.")
        return
    
    print(f"Found {len(results)} results using binary quantization (Hamming similarity)\n")
    
    for i, result in enumerate(results, 1):
        payload = result["payload"]
        
        # Try to find a display name from common fields
        display_name = "Result"
        for field in ['name', 'title', 'part_name', 'id', 'part_id']:
            if field in payload:
                display_name = str(payload[field])
                break
        
        print(f"\n{i}. {display_name}")
        
        # Display other fields
        for field in show_fields:
            if field in payload and field not in ['name', 'title']:
                value = payload[field]
                if isinstance(value, str):
                    print(f"   {field.capitalize()}: {value[:100]}{'...' if len(value) > 100 else ''}")
        
        print(f"   Binary Similarity Score: {result['score']:.4f} (Hamming)")
        print(f"   {result['score']*100:.1f}% bit similarity")
        
        print("-" * 80)


if __name__ == "__main__":
    """
    Example usage - customize for your collection.
    """
    # Example 1: Binary quantized search
    print("=" * 80)
    print("EXAMPLE 1: Binary Quantized Search")
    print("=" * 80)
    print("This demonstrates search using binary quantized vectors.\n")
    
    # Replace with your collection name and query
    collection_name = os.getenv("QDRANT_COLLECTION", "your_collection")
    query1 = "example query"
    
    try:
        client = get_qdrant_client()
        results1 = binary_quantize_search(
            collection_name=collection_name,
            query=query1,
            client=client,
            limit=5
        )
        display_binary_results(results1, query1, show_fields=['name', 'description'])
    except Exception as e:
        print(f"Error: {e}")
        print("\nTo use this script:")
        print("1. Set QDRANT_URL and QDRANT_API_KEY in your .env file")
        print("2. Set QDRANT_COLLECTION environment variable or update collection_name in code")
        print("3. Ensure your collection has vectors stored")
        print("4. Adjust show_fields to match your collection's payload structure")