xbe.at

LLMs and generative AI / Vector databases and embeddings

Understanding Vector Databases Fundamentals and Applications

14 min read, 13 sections

Contents
  1. Vector Space Fundamentals
  2. Vector Embedding Generation
  3. Similarity Search Implementation
  4. Dimensionality Reduction for Vector Storage
  5. Vector Quantization for Efficient Storage
  6. Approximate Nearest Neighbor Search Implementation
  7. Real-time Vector Database Updates
  8. HNSW Index Implementation
  9. Database Sharding and Distribution
  10. Vector Database Caching System
  11. Batch Vector Processing
  12. Vector Database Monitoring System
  13. Additional Resources

Vector Space Fundamentals

Vector databases operate on the principle of vector spaces where data points are represented as high-dimensional vectors. Understanding the mathematical foundation of vector spaces is crucial for implementing vector database operations efficiently using numerical computations and distance metrics.

import numpy as np

# Creating vector representations
class VectorSpace:
    def __init__(self, dimensions):
        self.dimensions = dimensions
        
    def create_vector(self, data):
        # Convert input data to vector representation
        vector = np.array(data)
        if len(vector) != self.dimensions:
            raise ValueError(f"Vector must have {self.dimensions} dimensions")
        return vector
    
    def euclidean_distance(self, vector1, vector2):
        return np.sqrt(np.sum((vector1 - vector2) ** 2))

# Example usage
vector_space = VectorSpace(3)
v1 = vector_space.create_vector([1, 2, 3])
v2 = vector_space.create_vector([4, 5, 6])
distance = vector_space.euclidean_distance(v1, v2)
print(f"Distance between vectors: {distance}")

Vector Embedding Generation

Vector embeddings transform raw data into dense numerical representations suitable for vector databases. This process involves using pre-trained models or custom embedding networks to generate fixed-length vectors that capture semantic meaning.

from sentence_transformers import SentenceTransformer
import numpy as np

class TextEmbedding:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)
        
    def generate_embedding(self, text):
        # Convert text to vector embedding
        embedding = self.model.encode(text)
        return embedding
    
    def batch_generate(self, texts):
        # Generate embeddings for multiple texts
        embeddings = self.model.encode(texts)
        return embeddings

# Example usage
embedder = TextEmbedding()
text = "Vector databases are efficient for similarity search"
embedding = embedder.generate_embedding(text)
print(f"Embedding shape: {embedding.shape}")
print(f"First 5 dimensions: {embedding[:5]}")

Similarity Search Implementation

The core functionality of vector databases relies on efficient similarity search algorithms. This implementation demonstrates the basic principles of nearest neighbor search using cosine similarity as the distance metric.

import numpy as np
from typing import List, Tuple

class VectorIndex:
    def __init__(self):
        self.vectors = []
        self.metadata = []
        
    def add_vector(self, vector: np.ndarray, metadata: dict):
        self.vectors.append(vector)
        self.metadata.append(metadata)
        
    def cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float:
        return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
    
    def search(self, query_vector: np.ndarray, k: int = 5) -> List[Tuple[int, float]]:
        similarities = [
            (i, self.cosine_similarity(query_vector, vec))
            for i, vec in enumerate(self.vectors)
        ]
        return sorted(similarities, key=lambda x: x[1], reverse=True)[:k]

# Example usage
index = VectorIndex()
# Add sample vectors
for i in range(5):
    vec = np.random.rand(128)
    index.add_vector(vec, {"id": i})

# Perform search
query = np.random.rand(128)
results = index.search(query, k=3)
print("Top 3 similar vectors:", results)

Dimensionality Reduction for Vector Storage

High-dimensional vectors can be computationally expensive to store and process. This implementation shows how to use Principal Component Analysis (PCA) to reduce vector dimensions while preserving important information.

from sklearn.decomposition import PCA
import numpy as np

class DimensionalityReducer:
    def __init__(self, target_dimensions: int):
        self.pca = PCA(n_components=target_dimensions)
        
    def fit_transform(self, vectors: np.ndarray) -> np.ndarray:
        # Reduce dimensionality of vectors
        reduced_vectors = self.pca.fit_transform(vectors)
        explained_variance = np.sum(self.pca.explained_variance_ratio_)
        print(f"Explained variance ratio: {explained_variance:.4f}")
        return reduced_vectors
    
    def transform(self, vectors: np.ndarray) -> np.ndarray:
        return self.pca.transform(vectors)

# Example usage
# Generate sample high-dimensional vectors
vectors = np.random.rand(1000, 512)

# Reduce dimensions
reducer = DimensionalityReducer(target_dimensions=128)
reduced_vectors = reducer.fit_transform(vectors)
print(f"Original shape: {vectors.shape}")
print(f"Reduced shape: {reduced_vectors.shape}")

Vector Quantization for Efficient Storage

Vector quantization reduces storage requirements by clustering similar vectors and representing them with centroids. This technique is crucial for scaling vector databases to handle millions of vectors while maintaining query performance.

import numpy as np
from sklearn.cluster import KMeans

class VectorQuantizer:
    def __init__(self, n_centroids: int):
        self.n_centroids = n_centroids
        self.kmeans = KMeans(n_clusters=n_centroids)
        
    def train(self, vectors: np.ndarray):
        # Train quantizer on input vectors
        self.kmeans.fit(vectors)
        
    def quantize(self, vector: np.ndarray) -> np.ndarray:
        # Find nearest centroid
        centroid_idx = self.kmeans.predict(vector.reshape(1, -1))[0]
        return self.kmeans.cluster_centers_[centroid_idx]
    
    def compression_ratio(self, original_vectors: np.ndarray) -> float:
        original_size = original_vectors.nbytes
        quantized_size = self.kmeans.cluster_centers_.nbytes
        return original_size / quantized_size

# Example usage
vectors = np.random.rand(1000, 128)
quantizer = VectorQuantizer(n_centroids=100)
quantizer.train(vectors)

# Quantize a single vector
sample_vector = np.random.rand(128)
quantized_vector = quantizer.quantize(sample_vector)
compression = quantizer.compression_ratio(vectors)
print(f"Compression ratio: {compression:.2f}x")

Approximate Nearest Neighbor Search Implementation

Exact nearest neighbor search becomes impractical for large datasets. This implementation demonstrates Locality-Sensitive Hashing (LSH) for approximate nearest neighbor search, significantly improving query performance.

import numpy as np
from typing import List, Tuple
from datasketch import MinHash, MinHashLSH

class LSHIndex:
    def __init__(self, num_perm: int = 128, threshold: float = 0.5):
        self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
        self.vectors = {}
        
    def _vector_to_minhash(self, vector: np.ndarray) -> MinHash:
        # Convert vector to MinHash representation
        m = MinHash(num_perm=128)
        # Quantize vector components to create hash input
        binary_vector = (vector > np.mean(vector)).astype(int)
        for idx, val in enumerate(binary_vector):
            if val == 1:
                m.update(str(idx).encode('utf-8'))
        return m
    
    def insert(self, key: str, vector: np.ndarray):
        minhash = self._vector_to_minhash(vector)
        self.lsh.insert(key, minhash)
        self.vectors[key] = vector
        
    def query(self, vector: np.ndarray) -> List[str]:
        query_minhash = self._vector_to_minhash(vector)
        return self.lsh.query(query_minhash)

# Example usage
index = LSHIndex()

# Insert vectors
for i in range(100):
    key = f"vec_{i}"
    vector = np.random.rand(128)
    index.insert(key, vector)

# Query similar vectors
query_vector = np.random.rand(128)
results = index.query(query_vector)
print(f"Similar vectors found: {len(results)}")
print(f"Sample results: {results[:5]}")

Real-time Vector Database Updates

Implementing efficient real-time updates in vector databases requires careful management of index structures. This implementation shows how to handle dynamic vector insertions and deletions while maintaining search performance.

import numpy as np
from collections import defaultdict
import threading

class DynamicVectorIndex:
    def __init__(self, dims: int, max_buffer_size: int = 1000):
        self.dims = dims
        self.max_buffer_size = max_buffer_size
        self.main_index = {}
        self.buffer_index = {}
        self.lock = threading.Lock()
        
    def insert(self, key: str, vector: np.ndarray):
        if vector.shape[0] != self.dims:
            raise ValueError(f"Vector must have {self.dims} dimensions")
            
        with self.lock:
            self.buffer_index[key] = vector
            if len(self.buffer_index) >= self.max_buffer_size:
                self._merge_buffer()
                
    def _merge_buffer(self):
        # Merge buffer into main index
        self.main_index.update(self.buffer_index)
        self.buffer_index.clear()
        
    def search(self, query_vector: np.ndarray, k: int = 5) -> List[Tuple[str, float]]:
        results = []
        
        # Search both indices
        with self.lock:
            all_vectors = {**self.main_index, **self.buffer_index}
            
            for key, vector in all_vectors.items():
                similarity = np.dot(query_vector, vector) / (
                    np.linalg.norm(query_vector) * np.linalg.norm(vector)
                )
                results.append((key, similarity))
                
        return sorted(results, key=lambda x: x[1], reverse=True)[:k]

# Example usage
index = DynamicVectorIndex(dims=128)

# Simulate real-time updates
for i in range(100):
    key = f"doc_{i}"
    vector = np.random.rand(128)
    index.insert(key, vector)

# Search
query = np.random.rand(128)
results = index.search(query, k=5)
print("Top 5 results:", results)

HNSW Index Implementation

Hierarchical Navigable Small World (HNSW) is a state-of-the-art algorithm for approximate nearest neighbor search. This implementation demonstrates the core concepts of building and searching a hierarchical graph structure.

import numpy as np
from typing import Dict, List, Set
import heapq

class HNSWNode:
    def __init__(self, id: int, vector: np.ndarray):
        self.id = id
        self.vector = vector
        self.neighbors: Dict[int, Set[int]] = {}  # layer -> neighbor_ids
        
class HNSWIndex:
    def __init__(self, dim: int, M: int = 16, ef_construction: int = 200):
        self.dim = dim
        self.M = M  # max neighbors per layer
        self.ef_construction = ef_construction
        self.nodes: Dict[int, HNSWNode] = {}
        self.max_layer = 0
        self.entry_point = None
        
    def _get_random_level(self) -> int:
        level = 0
        while np.random.random() < 0.5 and level < 32:
            level += 1
        return level
        
    def insert(self, id: int, vector: np.ndarray):
        level = self._get_random_level()
        node = HNSWNode(id, vector)
        
        # Initialize layers
        for l in range(level + 1):
            node.neighbors[l] = set()
            
        self.nodes[id] = node
        
        if self.entry_point is None:
            self.entry_point = id
            self.max_layer = level
            return
            
        # Insert into layers
        curr = self.entry_point
        for l in range(min(self.max_layer, level), -1, -1):
            curr = self._search_layer(vector, curr, l, 1)[0]
            
        if level > self.max_layer:
            self.max_layer = level
            self.entry_point = id
            
    def _search_layer(self, query: np.ndarray, entry_id: int, 
                     layer: int, ef: int) -> List[int]:
        visited = set([entry_id])
        candidates = [(1.0 - self._distance(query, self.nodes[entry_id].vector), entry_id)]
        heapq.heapify(candidates)
        
        return [candidate[1] for candidate in sorted(candidates, reverse=True)[:ef]]
    
    def _distance(self, v1: np.ndarray, v2: np.ndarray) -> float:
        return np.sqrt(np.sum((v1 - v2) ** 2))
    
    def search(self, query: np.ndarray, k: int = 5) -> List[int]:
        curr = self.entry_point
        for l in range(self.max_layer, 0, -1):
            curr = self._search_layer(query, curr, l, 1)[0]
        
        return self._search_layer(query, curr, 0, k)

# Example usage
index = HNSWIndex(dim=128)

# Insert vectors
for i in range(100):
    vector = np.random.rand(128)
    index.insert(i, vector)

# Search
query = np.random.rand(128)
results = index.search(query, k=5)
print(f"Query results: {results}")

Database Sharding and Distribution

Implementing sharding strategies for distributed vector databases enables horizontal scaling. This implementation shows how to partition and distribute vectors across multiple nodes while maintaining query accuracy.

import numpy as np
from typing import List, Dict, Tuple
from concurrent.futures import ThreadPoolExecutor
import hashlib

class VectorShard:
    def __init__(self, shard_id: int):
        self.shard_id = shard_id
        self.vectors: Dict[str, np.ndarray] = {}
        
    def insert(self, key: str, vector: np.ndarray):
        self.vectors[key] = vector
        
    def search(self, query: np.ndarray, k: int) -> List[Tuple[str, float]]:
        results = []
        for key, vector in self.vectors.items():
            similarity = np.dot(query, vector) / (
                np.linalg.norm(query) * np.linalg.norm(vector)
            )
            results.append((key, similarity))
        return sorted(results, key=lambda x: x[1], reverse=True)[:k]

class DistributedVectorDB:
    def __init__(self, num_shards: int):
        self.num_shards = num_shards
        self.shards = [VectorShard(i) for i in range(num_shards)]
        self.executor = ThreadPoolExecutor(max_workers=num_shards)
        
    def _get_shard_id(self, key: str) -> int:
        # Consistent hashing for shard selection
        hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
        return hash_val % self.num_shards
        
    def insert(self, key: str, vector: np.ndarray):
        shard_id = self._get_shard_id(key)
        self.shards[shard_id].insert(key, vector)
        
    def search(self, query: np.ndarray, k: int) -> List[Tuple[str, float]]:
        # Search all shards in parallel
        future_results = [
            self.executor.submit(shard.search, query, k)
            for shard in self.shards
        ]
        
        # Merge results
        all_results = []
        for future in future_results:
            all_results.extend(future.result())
            
        return sorted(all_results, key=lambda x: x[1], reverse=True)[:k]

# Example usage
db = DistributedVectorDB(num_shards=4)

# Insert vectors
for i in range(1000):
    key = f"doc_{i}"
    vector = np.random.rand(128)
    db.insert(key, vector)

# Search across shards
query = np.random.rand(128)
results = db.search(query, k=5)
print(f"Top 5 results across shards: {results}")

Vector Database Caching System

This implementation demonstrates an intelligent caching system for vector databases that stores frequently accessed vectors and query results to reduce computation overhead and improve response times.

import numpy as np
from collections import OrderedDict
from typing import List, Tuple, Optional
import time

class VectorCache:
    def __init__(self, capacity: int = 1000, ttl: int = 3600):
        self.capacity = capacity
        self.ttl = ttl  # Time-to-live in seconds
        self.vector_cache = OrderedDict()
        self.query_cache = OrderedDict()
        
    def _cleanup_expired(self, cache: OrderedDict):
        current_time = time.time()
        expired_keys = [
            k for k, (_, timestamp) in cache.items()
            if current_time - timestamp > self.ttl
        ]
        for k in expired_keys:
            cache.pop(k)
            
    def get_vector(self, key: str) -> Optional[np.ndarray]:
        self._cleanup_expired(self.vector_cache)
        if key in self.vector_cache:
            vector, timestamp = self.vector_cache[key]
            self.vector_cache.move_to_end(key)
            return vector
        return None
        
    def cache_vector(self, key: str, vector: np.ndarray):
        self._cleanup_expired(self.vector_cache)
        self.vector_cache[key] = (vector, time.time())
        if len(self.vector_cache) > self.capacity:
            self.vector_cache.popitem(last=False)
            
    def get_query_result(self, query_hash: str) -> Optional[List[Tuple[str, float]]]:
        self._cleanup_expired(self.query_cache)
        if query_hash in self.query_cache:
            results, timestamp = self.query_cache[query_hash]
            self.query_cache.move_to_end(query_hash)
            return results
        return None
        
    def cache_query_result(self, query_hash: str, results: List[Tuple[str, float]]):
        self._cleanup_expired(self.query_cache)
        self.query_cache[query_hash] = (results, time.time())
        if len(self.query_cache) > self.capacity:
            self.query_cache.popitem(last=False)

class CachedVectorDB:
    def __init__(self):
        self.vectors = {}
        self.cache = VectorCache()
        
    def insert(self, key: str, vector: np.ndarray):
        self.vectors[key] = vector
        self.cache.cache_vector(key, vector)
        
    def _compute_query_hash(self, query: np.ndarray) -> str:
        return hashlib.md5(query.tobytes()).hexdigest()
        
    def search(self, query: np.ndarray, k: int = 5) -> List[Tuple[str, float]]:
        query_hash = self._compute_query_hash(query)
        
        # Check cache first
        cached_results = self.cache.get_query_result(query_hash)
        if cached_results is not None:
            return cached_results
            
        # Compute search results
        results = []
        for key, vector in self.vectors.items():
            cached_vector = self.cache.get_vector(key)
            if cached_vector is None:
                cached_vector = vector
                self.cache.cache_vector(key, vector)
                
            similarity = np.dot(query, cached_vector) / (
                np.linalg.norm(query) * np.linalg.norm(cached_vector)
            )
            results.append((key, similarity))
            
        results = sorted(results, key=lambda x: x[1], reverse=True)[:k]
        self.cache.cache_query_result(query_hash, results)
        return results

# Example usage
db = CachedVectorDB()

# Insert vectors
for i in range(1000):
    key = f"doc_{i}"
    vector = np.random.rand(128)
    db.insert(key, vector)

# Perform searches
query = np.random.rand(128)
results1 = db.search(query, k=5)
print("First search results:", results1)

# Second search should be faster due to caching
results2 = db.search(query, k=5)
print("Second search results (cached):", results2)

Batch Vector Processing

This implementation showcases efficient batch processing for vector operations, essential for handling large-scale vector insertions and queries in production environments while optimizing memory usage and computational resources.

import numpy as np
from typing import List, Dict, Tuple
import concurrent.futures
from dataclasses import dataclass

@dataclass
class BatchResult:
    vectors_processed: int
    time_taken: float
    memory_used: int

class BatchVectorProcessor:
    def __init__(self, batch_size: int = 1000, num_workers: int = 4):
        self.batch_size = batch_size
        self.num_workers = num_workers
        self.vector_store = {}
        
    def _process_batch(self, batch_vectors: Dict[str, np.ndarray]) -> BatchResult:
        start_time = time.time()
        memory_before = self._get_memory_usage()
        
        # Normalize vectors in batch
        for key, vector in batch_vectors.items():
            normalized_vector = vector / np.linalg.norm(vector)
            self.vector_store[key] = normalized_vector
            
        memory_after = self._get_memory_usage()
        return BatchResult(
            vectors_processed=len(batch_vectors),
            time_taken=time.time() - start_time,
            memory_used=memory_after - memory_before
        )
        
    def _get_memory_usage(self) -> int:
        import psutil
        return psutil.Process().memory_info().rss
        
    def batch_insert(self, vectors: Dict[str, np.ndarray]) -> List[BatchResult]:
        results = []
        batches = self._create_batches(vectors, self.batch_size)
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.num_workers) as executor:
            future_to_batch = {
                executor.submit(self._process_batch, batch): i 
                for i, batch in enumerate(batches)
            }
            
            for future in concurrent.futures.as_completed(future_to_batch):
                batch_idx = future_to_batch[future]
                results.append(future.result())
                
        return results
    
    def _create_batches(self, vectors: Dict[str, np.ndarray], 
                       batch_size: int) -> List[Dict[str, np.ndarray]]:
        batches = []
        current_batch = {}
        
        for key, vector in vectors.items():
            current_batch[key] = vector
            if len(current_batch) >= batch_size:
                batches.append(current_batch)
                current_batch = {}
                
        if current_batch:
            batches.append(current_batch)
            
        return batches

    def batch_search(self, queries: List[np.ndarray], k: int = 5) -> List[List[Tuple[str, float]]]:
        def process_query(query):
            results = []
            query_norm = query / np.linalg.norm(query)
            
            for key, vector in self.vector_store.items():
                similarity = np.dot(query_norm, vector)
                results.append((key, similarity))
                
            return sorted(results, key=lambda x: x[1], reverse=True)[:k]
            
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.num_workers) as executor:
            return list(executor.map(process_query, queries))

# Example usage
processor = BatchVectorProcessor()

# Create sample vectors
vectors_to_insert = {
    f"doc_{i}": np.random.rand(128)
    for i in range(5000)
}

# Batch insert
results = processor.batch_insert(vectors_to_insert)
print(f"Processed {len(results)} batches")
for i, result in enumerate(results):
    print(f"Batch {i}: processed {result.vectors_processed} vectors "
          f"in {result.time_taken:.2f} seconds")

# Batch search
queries = [np.random.rand(128) for _ in range(10)]
search_results = processor.batch_search(queries, k=5)
print(f"\nProcessed {len(search_results)} queries")
print(f"Sample results for first query: {search_results[0][:2]}")

Vector Database Monitoring System

This implementation provides a comprehensive monitoring system for vector databases, tracking key performance metrics, resource utilization, and query patterns to ensure optimal performance and reliability in production environments.

import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional
import time
import threading
from collections import deque

@dataclass
class QueryMetrics:
    latency: float
    vectors_scanned: int
    memory_used: float
    timestamp: float

class VectorDBMonitor:
    def __init__(self, metrics_window: int = 3600):
        self.metrics_window = metrics_window  # Store metrics for 1 hour
        self.query_metrics = deque(maxlen=1000)
        self.index_size = 0
        self.total_memory = 0
        self.queries_per_second = 0
        self._lock = threading.Lock()
        self._start_monitoring()
        
    def _start_monitoring(self):
        self.monitor_thread = threading.Thread(target=self._monitor_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
        
    def _monitor_loop(self):
        while True:
            self._update_metrics()
            time.sleep(1)
            
    def _update_metrics(self):
        current_time = time.time()
        with self._lock:
            # Clean old metrics
            while (self.query_metrics and 
                   current_time - self.query_metrics[0].timestamp > self.metrics_window):
                self.query_metrics.popleft()
            
            # Calculate queries per second
            recent_queries = sum(1 for m in self.query_metrics 
                               if current_time - m.timestamp <= 60)
            self.queries_per_second = recent_queries / 60
            
    def record_query(self, metrics: QueryMetrics):
        with self._lock:
            self.query_metrics.append(metrics)
            
    def get_performance_stats(self) -> Dict[str, float]:
        with self._lock:
            if not self.query_metrics:
                return {
                    "avg_latency": 0,
                    "qps": 0,
                    "p95_latency": 0,
                    "memory_utilization": 0
                }
                
            latencies = [m.latency for m in self.query_metrics]
            return {
                "avg_latency": np.mean(latencies),
                "qps": self.queries_per_second,
                "p95_latency": np.percentile(latencies, 95),
                "memory_utilization": self.total_memory / (1024 * 1024 * 1024)  # GB
            }

class MonitoredVectorDB:
    def __init__(self):
        self.vectors = {}
        self.monitor = VectorDBMonitor()
        
    def insert(self, key: str, vector: np.ndarray):
        start_time = time.time()
        memory_before = self._get_memory_usage()
        
        self.vectors[key] = vector
        
        memory_after = self._get_memory_usage()
        self.monitor.total_memory = memory_after
        self.monitor.index_size = len(self.vectors)
        
    def search(self, query: np.ndarray, k: int = 5) -> List[Tuple[str, float]]:
        start_time = time.time()
        memory_before = self._get_memory_usage()
        
        results = []
        for key, vector in self.vectors.items():
            similarity = np.dot(query, vector) / (
                np.linalg.norm(query) * np.linalg.norm(vector)
            )
            results.append((key, similarity))
            
        results = sorted(results, key=lambda x: x[1], reverse=True)[:k]
        
        # Record metrics
        memory_after = self._get_memory_usage()
        self.monitor.record_query(QueryMetrics(
            latency=time.time() - start_time,
            vectors_scanned=len(self.vectors),
            memory_used=memory_after - memory_before,
            timestamp=time.time()
        ))
        
        return results
        
    def _get_memory_usage(self) -> float:
        import psutil
        return psutil.Process().memory_info().rss

# Example usage
db = MonitoredVectorDB()

# Insert vectors
for i in range(1000):
    vector = np.random.rand(128)
    db.insert(f"doc_{i}", vector)

# Perform searches
for _ in range(100):
    query = np.random.rand(128)
    results = db.search(query, k=5)
    time.sleep(0.1)  # Simulate real-world query pattern

# Get performance stats
stats = db.monitor.get_performance_stats()
print("Performance Statistics:")
print(f"Average Latency: {stats['avg_latency']*1000:.2f}ms")
print(f"Queries per Second: {stats['qps']:.2f}")
print(f"95th Percentile Latency: {stats['p95_latency']*1000:.2f}ms")
print(f"Memory Utilization: {stats['memory_utilization']:.2f}GB")

Additional Resources

  • Vector Database Research Papers:
  • Recommended Resources for Further Learning:
    • Google Scholar search for "Vector Database Optimization"
    • Pinecone Documentation and Technical Blog
    • Weaviate Vector Database Documentation
    • FAISS Performance Optimization Guide
    • Milvus Architecture Design Docs
  • Open Source Implementations:
    • FAISS (Facebook AI Similarity Search)
    • Annoy (Spotify)
    • Hnswlib
    • ScaNN (Google Research)

More in Vector databases and embeddings

  1. Vector Databases and Qd-trees for Building RAG Systems8 min

    Vector databases are specialized systems designed to store and efficiently query high-dimensional vector data. They are crucial for many machine learning and...

  2. Techniques for Compressing Large Language Models in Python7 min

    Model compression is crucial for deploying large language models (LLMs) in resource-constrained environments. It aims to reduce model size and computational...

  3. Vector Databases Powering AI and Machine Learning4 min

    Vector databases are specialized systems designed to store and efficiently query high-dimensional vector data. They are becoming increasingly important in...

  4. Running Open-Source LLMs Locally with Ollama10 min

    The first step involves configuring the Python environment to interact with Ollama's API endpoints. This allows seamless communication between Python...

  5. Running Open-Source Large Language Models Locally with Python5 min

    Running large language models (LLMs) locally can be a cost-effective and privacy-conscious approach for researchers, developers, and enthusiasts. This...

  6. Quantization for Efficient Large Language Models10 min

    Quantization is a technique that reduces the precision of model weights and activations, allowing for more efficient storage and computation of Large...