Vector Databases

Vector databases are specialized databases optimized for storing and searching embeddings. They are like a library with a magical catalog that finds similar books based on their content!

Popular Vector Databases

1┌─────────────────────────────────────────────────────────┐
2│              Vector Databases Landscape                  │
3├─────────────────────────────────────────────────────────┤
4│                                                          │
5│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
6│  │   Pinecone   │  │    Qdrant    │  │   Chroma     │  │
7│  │   (Cloud)    │  │  (Self-host) │  │   (Local)    │  │
8│  └──────────────┘  └──────────────┘  └──────────────┘  │
9│                                                          │
10│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
11│  │   Weaviate   │  │    Milvus    │  │    FAISS     │  │
12│  │  (Semantic)  │  │  (Enterprise)│  │  (In-memory) │  │
13│  └──────────────┘  └──────────────┘  └──────────────┘  │
14│                                                          │
15└─────────────────────────────────────────────────────────┘

Chroma - Local Vector Database

1import chromadb
2from chromadb.utils import embedding_functions
3
4# Initialize client
5client = chromadb.Client()  # In-memory
6# or: client = chromadb.PersistentClient(path="./chroma_db")  # Persistent
7
8# Configure embedding function
9openai_ef = embedding_functions.OpenAIEmbeddingFunction(
10    api_key="your-key",
11    model_name="text-embedding-3-small"
12)
13
14# Create collection
15collection = client.create_collection(
16    name="python_safari",
17    embedding_function=openai_ef,
18    metadata={"description": "Python Safari course documentation"}
19)
20
21# Add documents
22collection.add(
23    documents=[
24        "Python is a high-level programming language",
25        "RAG combines retrieval with text generation",
26        "Vector databases store embeddings"
27    ],
28    metadatas=[
29        {"topic": "python", "level": "beginner"},
30        {"topic": "ai", "level": "advanced"},
31        {"topic": "databases", "level": "intermediate"}
32    ],
33    ids=["doc1", "doc2", "doc3"]
34)
35
36# Search
37results = collection.query(
38    query_texts=["How to start learning programming?"],
39    n_results=2,
40    where={"level": "beginner"}  # Filter by metadata
41)
42
43print(results)

Qdrant - Production Vector Search

1from qdrant_client import QdrantClient
2from qdrant_client.models import Distance, VectorParams, PointStruct
3import numpy as np
4
5# Connect to Qdrant
6client = QdrantClient(host="localhost", port=6333)
7# or: client = QdrantClient(":memory:")  # In-memory
8
9# Create collection
10client.create_collection(
11    collection_name="documents",
12    vectors_config=VectorParams(
13        size=1536,  # Embedding size
14        distance=Distance.COSINE
15    )
16)
17
18# Add points
19def add_documents(texts: list[str], embeddings: list[list[float]]):
20    """Adds documents to Qdrant."""
21    points = [
22        PointStruct(
23            id=i,
24            vector=embedding,
25            payload={"text": text}
26        )
27        for i, (text, embedding) in enumerate(zip(texts, embeddings))
28    ]
29
30    client.upsert(
31        collection_name="documents",
32        points=points
33    )
34
35# Search
36def search(query_embedding: list[float], limit: int = 5):
37    """Searches for similar documents."""
38    results = client.search(
39        collection_name="documents",
40        query_vector=query_embedding,
41        limit=limit
42    )
43
44    return [
45        {
46            "text": hit.payload["text"],
47            "score": hit.score
48        }
49        for hit in results
50    ]
51
52# Filtering
53from qdrant_client.models import Filter, FieldCondition, MatchValue
54
55filtered_results = client.search(
56    collection_name="documents",
57    query_vector=query_embedding,
58    query_filter=Filter(
59        must=[
60            FieldCondition(
61                key="category",
62                match=MatchValue(value="python")
63            )
64        ]
65    ),
66    limit=5
67)

FAISS - Fast In-Memory Search

1import faiss
2import numpy as np
3from dataclasses import dataclass
4
5@dataclass
6class FAISSIndex:
7    """Wrapper for FAISS."""
8
9    dimension: int
10    index: faiss.Index = None
11    documents: list[str] = None
12
13    def __post_init__(self):
14        # Different index types
15        # Flat - exact, slower
16        self.index = faiss.IndexFlatL2(self.dimension)
17
18        # IVF - faster, approximate
19        # quantizer = faiss.IndexFlatL2(self.dimension)
20        # self.index = faiss.IndexIVFFlat(quantizer, self.dimension, 100)
21
22        self.documents = []
23
24    def add(self, embeddings: np.ndarray, documents: list[str]):
25        """Adds vectors to the index."""
26        embeddings = np.array(embeddings).astype('float32')
27        self.index.add(embeddings)
28        self.documents.extend(documents)
29
30    def search(self, query_embedding: np.ndarray, k: int = 5) -> list[tuple[str, float]]:
31        """Searches for k nearest neighbors."""
32        query = np.array([query_embedding]).astype('float32')
33        distances, indices = self.index.search(query, k)
34
35        results = []
36        for idx, dist in zip(indices[0], distances[0]):
37            if idx < len(self.documents):
38                results.append((self.documents[idx], float(dist)))
39
40        return results
41
42# Usage example
43faiss_index = FAISSIndex(dimension=384)
44
45# Add documents
46embeddings = np.random.rand(100, 384).astype('float32')
47documents = [f"Document {i}" for i in range(100)]
48faiss_index.add(embeddings, documents)
49
50# Search
51query = np.random.rand(384).astype('float32')
52results = faiss_index.search(query, k=5)

Pinecone - Managed Vector Database

1from pinecone import Pinecone, ServerlessSpec
2
3# Initialize
4pc = Pinecone(api_key="your-key")
5
6# Create index
7pc.create_index(
8    name="python-safari",
9    dimension=1536,
10    metric="cosine",
11    spec=ServerlessSpec(
12        cloud="aws",
13        region="us-east-1"
14    )
15)
16
17# Connect to index
18index = pc.Index("python-safari")
19
20# Upsert (insert/update)
21index.upsert(
22    vectors=[
23        {
24            "id": "doc1",
25            "values": [0.1, 0.2, ...],  # 1536 values
26            "metadata": {
27                "text": "Python is a programming language",
28                "category": "programming",
29                "level": 1
30            }
31        }
32    ],
33    namespace="tutorials"
34)
35
36# Search
37results = index.query(
38    vector=[0.1, 0.2, ...],
39    top_k=10,
40    include_metadata=True,
41    namespace="tutorials",
42    filter={
43        "category": {"$eq": "programming"},
44        "level": {"$lte": 3}
45    }
46)
47
48# Statistics
49stats = index.describe_index_stats()
50print(f"Number of vectors: {stats['total_vector_count']}")

Hybrid Search - Combining Vectors with BM25

1from rank_bm25 import BM25Okapi
2import numpy as np
3
4class HybridSearch:
5    """Combines semantic search with keyword search."""
6
7    def __init__(self, documents: list[str], embeddings: np.ndarray):
8        self.documents = documents
9        self.embeddings = embeddings
10
11        # BM25 for keyword search
12        tokenized = [doc.lower().split() for doc in documents]
13        self.bm25 = BM25Okapi(tokenized)
14
15    def search(
16        self,
17        query: str,
18        query_embedding: np.ndarray,
19        alpha: float = 0.5,  # Semantic search weight
20        top_k: int = 5
21    ) -> list[tuple[str, float]]:
22        """Hybrid search with configurable weights."""
23
24        # Semantic search scores
25        semantic_scores = np.dot(self.embeddings, query_embedding)
26        semantic_scores /= np.linalg.norm(self.embeddings, axis=1)
27        semantic_scores /= np.linalg.norm(query_embedding)
28
29        # Normalize to [0, 1]
30        semantic_scores = (semantic_scores - semantic_scores.min()) / (semantic_scores.max() - semantic_scores.min())
31
32        # BM25 scores
33        bm25_scores = np.array(self.bm25.get_scores(query.lower().split()))
34        if bm25_scores.max() > 0:
35            bm25_scores = bm25_scores / bm25_scores.max()
36
37        # Hybrid score
38        hybrid_scores = alpha * semantic_scores + (1 - alpha) * bm25_scores
39
40        # Top K
41        top_indices = np.argsort(hybrid_scores)[::-1][:top_k]
42
43        return [(self.documents[i], hybrid_scores[i]) for i in top_indices]

Vector databases are the infrastructure of every RAG system. In the next lesson you will learn about LlamaIndex - a framework that simplifies building RAG applications!