Building RAG from Scratch: Article 4

Embeddings and Vector Databases

Series: Building RAG from Scratch
Article: 4 of 12
Level: Intermediate
Prerequisites: Articles 1-3, Linear algebra basics (helpful but not required)
Reading Time: 40-45 minutes
Coding Time: 60-90 minutes

A comprehensive overview of the semantic search pipeline, featuring popular embedding models and vector database options.

Introduction

You've prepared your data into clean, well-chunked documents. Now comes the magic that makes RAG work: transforming text into searchable vectors.

In this article, we'll demystify one of the most fascinating aspects of modern AI: embeddings. We'll explore how machines understand meaning, why "similar meanings produce similar vectors," and how to build systems that can search millions of documents in milliseconds.

By the end, you'll understand:

What embeddings are and the mathematics behind them
How embedding models work
Comparing embedding models and choosing the right one
Vector databases and their architecture
Similarity metrics and search algorithms
Building a complete indexing pipeline
Optimizing for speed and quality

Let's dive into the world of semantic search!

Part 1: Understanding Embeddings

1.1 What Are Embeddings?

Simple definition: Embeddings are numerical representations of text that capture semantic meaning.

More technical: Embeddings are dense vectors (arrays of floating-point numbers) in a high-dimensional space where the geometric relationships between vectors correspond to semantic relationships between the text they represent.

Let's unpack this with examples:

# Text strings (what we have)
text1 = "The cat sat on the mat"
text2 = "A feline rested on the rug"
text3 = "Python programming tutorial"

# Embeddings (what the model creates)
embedding1 = [0.23, -0.45, 0.67, 0.12, -0.89, ...]  # 384 numbers
embedding2 = [0.25, -0.43, 0.65, 0.14, -0.87, ...]  # Very similar!
embedding3 = [-0.78, 0.91, -0.34, 0.56, 0.22, ...]  # Very different!

Key insight: Sentences with similar meanings have similar embeddings (vectors close together in space), even if they use different words!

1.2 Why Can't We Just Use Keywords?

Let's see why traditional keyword search fails:

# Traditional keyword search
query = "how to fix Python errors"
doc1 = "Debugging Python code and resolving issues"  # ✗ No matching keywords!
doc2 = "how to fix Python errors in your program"    # ✓ Exact match

# But doc1 is MORE relevant than doc2!
# Keyword search would miss doc1 entirely.

# Semantic search with embeddings
# Both would be retrieved because their MEANING is similar

Problems with keyword search:

❌ Misses synonyms ("car" vs "automobile")
❌ Ignores context ("bank" = money or river?)
❌ No understanding of semantics
❌ Requires exact word matches

Advantages of embedding-based search:

✅ Understands meaning, not just words
✅ Finds conceptually similar content
✅ Handles paraphrases
✅ Context-aware

1.3 The Mathematics Behind Embeddings

Don't worry - you don't need to be a mathematician! But understanding the basics helps.

What's a vector?

# A vector is just a list of numbers
vector_2d = [3, 4]           # 2 dimensions
vector_3d = [1, 2, 3]        # 3 dimensions
embedding = [0.1, 0.2, ...]  # 384 or 1536 dimensions

Vector operations:

import numpy as np

# Two embeddings
v1 = np.array([1, 2, 3])
v2 = np.array([4, 5, 6])

# Dot product (how similar are they?)
similarity = np.dot(v1, v2)  # = 1*4 + 2*5 + 3*6 = 32

# Magnitude (length of vector)
magnitude = np.linalg.norm(v1)  # = √(1² + 2² + 3²) = 3.74

# Cosine similarity (standard for embeddings)
cosine_sim = np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
# Range: -1 (opposite) to 1 (identical)

Visual intuition (simplified to 2D):

     y
     |
     |    • v2 (Python tutorial)
     |   /
     |  /
     | /
     |/_________ x
    /|
   / |
  /  |
 •   |  v1 (Python guide)
     |
     
v1 and v2 are close together = similar meaning

1.4 How Embedding Models Are Trained

Embedding models learn through a process called contrastive learning:

Training process (simplified):

Positive pairs: Take similar sentences
- "The cat sleeps" and "A feline rests"
- Train model so their embeddings are close
Negative pairs: Take dissimilar sentences
- "The cat sleeps" and "Database architecture"
- Train model so their embeddings are far apart
Repeat millions of times with diverse text
- Model learns what "similar meaning" means
- Captures semantic relationships in the vector space

The result: A model that can embed ANY text into a meaningful vector space!

Part 2: Embedding Models - A Deep Dive

2.1 Popular Embedding Models

Let's explore the most common models and when to use each.

OpenAI Embeddings

Models:

text-embedding-3-small - 1536 dimensions
text-embedding-3-large - 3072 dimensions
text-embedding-ada-002 - 1536 dimensions (older)

Pros:

✅ Excellent quality
✅ Easy API integration
✅ Regular updates
✅ Good documentation

Cons:

❌ Paid (though cheap: ~$0.02 / 1M tokens)
❌ Requires internet connection
❌ Data leaves your infrastructure

Implementation:

# embeddings/openai_embedder.py
from openai import OpenAI
import os
from typing import List

class OpenAIEmbedder:
    """
    Wrapper for OpenAI embedding models
    """
    
    def __init__(
        self,
        api_key: str = None,
        model: str = "text-embedding-3-small"
    ):
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
        self.model = model
        self.client = OpenAI(api_key=self.api_key)
        
        # Dimension info
        self.dimensions = {
            "text-embedding-3-small": 1536,
            "text-embedding-3-large": 3072,
            "text-embedding-ada-002": 1536
        }
    
    def embed_text(self, text: str) -> List[float]:
        """
        Embed a single text
        
        Args:
            text: Text to embed
            
        Returns:
            List of floats (embedding vector)
        """
        response = self.client.embeddings.create(
            model=self.model,
            input=text
        )
        return response.data[0].embedding
    
    def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
        """
        Embed multiple texts efficiently
        
        Args:
            texts: List of texts to embed
            batch_size: How many to embed at once (max 2048 for OpenAI)
            
        Returns:
            List of embedding vectors
        """
        all_embeddings = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            
            response = self.client.embeddings.create(
                model=self.model,
                input=batch
            )
            
            batch_embeddings = [item.embedding for item in response.data]
            all_embeddings.extend(batch_embeddings)
            
            print(f"Embedded {i + len(batch)}/{len(texts)} texts")
        
        return all_embeddings
    
    def get_dimension(self) -> int:
        """Get the dimension of embeddings for this model"""
        return self.dimensions[self.model]

# Example usage
if __name__ == "__main__":
    embedder = OpenAIEmbedder()
    
    # Single embedding
    text = "How do embeddings work in RAG systems?"
    embedding = embedder.embed_text(text)
    
    print(f"Text: {text}")
    print(f"Embedding dimension: {len(embedding)}")
    print(f"First 5 values: {embedding[:5]}")
    
    # Batch embedding
    texts = [
        "RAG combines retrieval with generation",
        "Vector databases store embeddings",
        "Python is a programming language"
    ]
    embeddings = embedder.embed_batch(texts)
    print(f"\nEmbedded {len(embeddings)} texts")

Sentence Transformers (Open Source)

Popular models:

all-MiniLM-L6-v2 - 384 dimensions (fast, good)
all-mpnet-base-v2 - 768 dimensions (better quality)
bge-large-en-v1.5 - 1024 dimensions (SOTA open source)

Pros:

✅ Completely free
✅ Run locally (no API calls)
✅ No data leaves your infrastructure
✅ Fast inference
✅ Many models to choose from

Cons:

❌ Requires local compute (CPU/GPU)
❌ Need to manage model files
❌ Slightly lower quality than OpenAI

Implementation:

# embeddings/sentence_transformer_embedder.py
from sentence_transformers import SentenceTransformer
from typing import List
import numpy as np

class SentenceTransformerEmbedder:
    """
    Wrapper for Sentence Transformer models
    """
    
    def __init__(
        self,
        model_name: str = "all-MiniLM-L6-v2",
        device: str = None
    ):
        """
        Initialize embedder
        
        Args:
            model_name: Name of sentence-transformers model
            device: 'cuda' for GPU, 'cpu' for CPU, None for auto
        """
        self.model_name = model_name
        self.model = SentenceTransformer(model_name, device=device)
        self.dimension = self.model.get_sentence_embedding_dimension()
        
        print(f"Loaded {model_name}")
        print(f"Embedding dimension: {self.dimension}")
    
    def embed_text(self, text: str) -> List[float]:
        """Embed a single text"""
        embedding = self.model.encode(text, convert_to_numpy=True)
        return embedding.tolist()
    
    def embed_batch(
        self,
        texts: List[str],
        batch_size: int = 32,
        show_progress: bool = True
    ) -> List[List[float]]:
        """
        Embed multiple texts
        
        Args:
            texts: List of texts
            batch_size: Batch size for encoding
            show_progress: Show progress bar
            
        Returns:
            List of embeddings
        """
        embeddings = self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=show_progress,
            convert_to_numpy=True
        )
        return embeddings.tolist()
    
    def get_dimension(self) -> int:
        """Get embedding dimension"""
        return self.dimension

# Example usage
if __name__ == "__main__":
    # Fast, lightweight model (good for development)
    embedder = SentenceTransformerEmbedder("all-MiniLM-L6-v2")
    
    # Or higher quality model
    # embedder = SentenceTransformerEmbedder("all-mpnet-base-v2")
    
    texts = [
        "How do vector databases work?",
        "What are vector databases?",
        "Python programming tutorial"
    ]
    
    embeddings = embedder.embed_batch(texts)
    
    # Check similarity between first two (should be high)
    from sklearn.metrics.pairwise import cosine_similarity
    
    sim = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
    print(f"\nSimilarity between query 1 and 2: {sim:.3f}")
    
    sim = cosine_similarity([embeddings[0]], [embeddings[2]])[0][0]
    print(f"Similarity between query 1 and 3: {sim:.3f}")

2.2 Comparing Embedding Models

Let's test different models on the same data:

# embeddings/model_comparison.py
from typing import List, Dict
import time
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def compare_embedding_models(test_queries: List[str], documents: List[str]):
    """
    Compare different embedding models
    
    Args:
        test_queries: List of test queries
        documents: List of documents to search
    """
    models_to_test = [
        # (name, embedder_class, init_params)
        ("OpenAI Small", OpenAIEmbedder, {"model": "text-embedding-3-small"}),
        ("MiniLM", SentenceTransformerEmbedder, {"model_name": "all-MiniLM-L6-v2"}),
        ("MPNet", SentenceTransformerEmbedder, {"model_name": "all-mpnet-base-v2"}),
    ]
    
    results = {}
    
    for model_name, embedder_class, params in models_to_test:
        print(f"\nTesting {model_name}...")
        print("="*60)
        
        # Initialize
        embedder = embedder_class(**params)
        
        # Measure embedding time
        start = time.time()
        doc_embeddings = embedder.embed_batch(documents)
        embed_time = time.time() - start
        
        # Test retrieval quality
        for query in test_queries:
            query_embedding = embedder.embed_text(query)
            
            # Calculate similarities
            similarities = cosine_similarity(
                [query_embedding],
                doc_embeddings
            )[0]
            
            # Get top 3
            top_indices = np.argsort(similarities)[::-1][:3]
            
            print(f"\nQuery: {query}")
            for idx in top_indices:
                print(f"  {similarities[idx]:.3f} - {documents[idx][:60]}...")
        
        results[model_name] = {
            'dimension': embedder.get_dimension(),
            'embed_time': embed_time,
            'avg_time_per_doc': embed_time / len(documents)
        }
        
        print(f"\nPerformance:")
        print(f"  Dimension: {results[model_name]['dimension']}")
        print(f"  Total time: {embed_time:.2f}s")
        print(f"  Per document: {results[model_name]['avg_time_per_doc']*1000:.2f}ms")
    
    return results

# Example usage
if __name__ == "__main__":
    test_queries = [
        "How do I implement RAG?",
        "What are vector databases?"
    ]
    
    documents = [
        "RAG systems combine retrieval with generation for better answers",
        "Vector databases store embeddings and enable fast similarity search",
        "Python is a popular programming language for AI",
        "Implementing RAG requires chunking, embedding, and retrieval steps"
    ]
    
    results = compare_embedding_models(test_queries, documents)

2.3 Model Selection Guide

Factor	OpenAI	Sentence Transformers	Recommendation
Quality	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	OpenAI slight edge
Cost	Paid (~$0.02/1M)	Free	ST for budget
Speed	Network latency	Local (fast)	ST for speed
Privacy	Data leaves infra	Fully private	ST for privacy
Setup	API key only	Model download	OpenAI easier
Offline	❌ Needs internet	✅ Works offline	ST for offline

Recommendation:

Development/Testing: Use Sentence Transformers (all-MiniLM-L6-v2)
Production (quality matters): OpenAI or bge-large-en-v1.5
Production (privacy matters): Sentence Transformers
Large scale (cost matters): Sentence Transformers on your own GPUs

Part 3: Vector Databases

3.1 Why We Need Vector Databases

The problem:

# Naive approach - BAD for scale
embeddings = []  # List of 1 million vectors
query_embedding = model.embed("your query")

# Calculate similarity with EVERY vector
for embedding in embeddings:
    similarity = cosine_similarity(query_embedding, embedding)

# This is O(n) - slow for large datasets!
# 1 million vectors × 1536 dimensions = 1.5 billion calculations per query

Vector databases solve this with:

✅ Specialized data structures (indices)
✅ Approximate nearest neighbor (ANN) search
✅ O(log n) instead of O(n) complexity
✅ Can handle billions of vectors
✅ Millisecond query times

3.2 How Vector Databases Work

Key concepts:

Index structures:
- HNSW (Hierarchical Navigable Small World)
- IVF (Inverted File Index)
- Product Quantization
Approximate search:
- Trade slight accuracy for massive speed
- 95% accuracy, 100x faster
Metadata filtering:
- Filter by attributes before vector search
- E.g., "Only search PDFs from 2024"

3.3 ChromaDB - Getting Started

ChromaDB is perfect for development and small-medium projects.

Installation:

pip install chromadb

Basic usage:

# vector_stores/chroma_store.py
import chromadb
from chromadb.config import Settings
from typing import List, Dict
import uuid

class ChromaVectorStore:
    """
    Wrapper for ChromaDB vector database
    """
    
    def __init__(
        self,
        collection_name: str = "rag_collection",
        persist_directory: str = "./chroma_db"
    ):
        """
        Initialize ChromaDB
        
        Args:
            collection_name: Name of the collection
            persist_directory: Where to store the database
        """
        self.client = chromadb.Client(Settings(
            persist_directory=persist_directory,
            anonymized_telemetry=False
        ))
        
        # Get or create collection
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"description": "RAG document collection"}
        )
        
        print(f"Initialized ChromaDB collection: {collection_name}")
        print(f"Current count: {self.collection.count()} documents")
    
    def add_documents(
        self,
        documents: List[str],
        embeddings: List[List[float]],
        metadatas: List[Dict] = None,
        ids: List[str] = None
    ):
        """
        Add documents to the collection
        
        Args:
            documents: List of text documents
            embeddings: List of embedding vectors
            metadatas: Optional metadata for each document
            ids: Optional IDs (generated if not provided)
        """
        # Generate IDs if not provided
        if ids is None:
            ids = [str(uuid.uuid4()) for _ in documents]
        
        # Add to collection
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            metadatas=metadatas,
            ids=ids
        )
        
        print(f"Added {len(documents)} documents")
        print(f"Total documents: {self.collection.count()}")
    
    def search(
        self,
        query_embedding: List[float],
        n_results: int = 5,
        where: Dict = None
    ) -> Dict:
        """
        Search for similar documents
        
        Args:
            query_embedding: Query vector
            n_results: Number of results to return
            where: Metadata filter (e.g., {"source": "manual.pdf"})
            
        Returns:
            Dictionary with results
        """
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            where=where
        )
        
        return {
            'documents': results['documents'][0],
            'distances': results['distances'][0],
            'metadatas': results['metadatas'][0],
            'ids': results['ids'][0]
        }
    
    def delete_collection(self):
        """Delete the entire collection"""
        self.client.delete_collection(self.collection.name)
        print(f"Deleted collection: {self.collection.name}")
    
    def get_stats(self) -> Dict:
        """Get collection statistics"""
        return {
            'count': self.collection.count(),
            'name': self.collection.name
        }

# Example usage
if __name__ == "__main__":
    from embeddings.sentence_transformer_embedder import SentenceTransformerEmbedder
    
    # Initialize embedder
    embedder = SentenceTransformerEmbedder()
    
    # Initialize vector store
    store = ChromaVectorStore(collection_name="demo")
    
    # Sample documents
    documents = [
        "RAG systems combine retrieval with generation",
        "Vector databases enable fast similarity search",
        "ChromaDB is a vector database for AI applications",
        "Embeddings represent text as numerical vectors"
    ]
    
    # Create embeddings
    embeddings = embedder.embed_batch(documents)
    
    # Add to store
    metadatas = [
        {"source": "doc1.txt", "category": "rag"},
        {"source": "doc2.txt", "category": "databases"},
        {"source": "doc3.txt", "category": "databases"},
        {"source": "doc4.txt", "category": "embeddings"}
    ]
    
    store.add_documents(
        documents=documents,
        embeddings=embeddings,
        metadatas=metadatas
    )
    
    # Search
    query = "How do RAG systems work?"
    query_embedding = embedder.embed_text(query)
    
    results = store.search(query_embedding, n_results=3)
    
    print(f"\nQuery: {query}")
    print("\nTop results:")
    for i, (doc, distance, meta) in enumerate(zip(
        results['documents'],
        results['distances'],
        results['metadatas']
    )):
        print(f"\n{i+1}. Distance: {distance:.3f}")
        print(f"   Document: {doc}")
        print(f"   Metadata: {meta}")

3.4 Advanced Vector Stores

Pinecone (Production-Ready, Managed)

# vector_stores/pinecone_store.py
import pinecone
from typing import List, Dict

class PineconeVectorStore:
    """
    Wrapper for Pinecone vector database
    """
    
    def __init__(
        self,
        api_key: str,
        environment: str,
        index_name: str,
        dimension: int = 1536
    ):
        """
        Initialize Pinecone
        
        Args:
            api_key: Pinecone API key
            environment: Pinecone environment (e.g., 'us-west1-gcp')
            index_name: Name of the index
            dimension: Embedding dimension
        """
        pinecone.init(api_key=api_key, environment=environment)
        
        # Create index if it doesn't exist
        if index_name not in pinecone.list_indexes():
            pinecone.create_index(
                name=index_name,
                dimension=dimension,
                metric='cosine'
            )
        
        self.index = pinecone.Index(index_name)
        print(f"Connected to Pinecone index: {index_name}")
    
    def upsert_documents(
        self,
        ids: List[str],
        embeddings: List[List[float]],
        metadatas: List[Dict]
    ):
        """
        Add or update documents
        
        Args:
            ids: Document IDs
            embeddings: Embedding vectors
            metadatas: Metadata dicts
        """
        vectors = [
            (id, embedding, metadata)
            for id, embedding, metadata in zip(ids, embeddings, metadatas)
        ]
        
        # Upsert in batches
        batch_size = 100
        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i + batch_size]
            self.index.upsert(vectors=batch)
        
        print(f"Upserted {len(vectors)} vectors")
    
    def search(
        self,
        query_embedding: List[float],
        top_k: int = 5,
        filter: Dict = None
    ) -> List[Dict]:
        """
        Search for similar vectors
        
        Args:
            query_embedding: Query vector
            top_k: Number of results
            filter: Metadata filter
            
        Returns:
            List of results
        """
        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            include_metadata=True,
            filter=filter
        )
        
        return results['matches']

# Note: Requires Pinecone account and API key

FAISS (Facebook AI Similarity Search)

# vector_stores/faiss_store.py
import faiss
import numpy as np
from typing import List, Dict
import pickle

class FAISSVectorStore:
    """
    Wrapper for FAISS vector database
    """
    
    def __init__(self, dimension: int):
        """
        Initialize FAISS index
        
        Args:
            dimension: Embedding dimension
        """
        self.dimension = dimension
        
        # Create index (using L2 distance)
        self.index = faiss.IndexFlatL2(dimension)
        
        # For IDs and metadata
        self.ids = []
        self.metadatas = []
        
        print(f"Initialized FAISS index with dimension {dimension}")
    
    def add_documents(
        self,
        embeddings: List[List[float]],
        ids: List[str],
        metadatas: List[Dict]
    ):
        """Add documents to index"""
        embeddings_array = np.array(embeddings).astype('float32')
        
        self.index.add(embeddings_array)
        self.ids.extend(ids)
        self.metadatas.extend(metadatas)
        
        print(f"Added {len(embeddings)} vectors")
        print(f"Total vectors: {self.index.ntotal}")
    
    def search(
        self,
        query_embedding: List[float],
        k: int = 5
    ) -> List[Dict]:
        """Search for similar vectors"""
        query_array = np.array([query_embedding]).astype('float32')
        
        distances, indices = self.index.search(query_array, k)
        
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            results.append({
                'id': self.ids[idx],
                'distance': float(dist),
                'metadata': self.metadatas[idx]
            })
        
        return results
    
    def save(self, filepath: str):
        """Save index to disk"""
        faiss.write_index(self.index, f"{filepath}.index")
        
        with open(f"{filepath}.meta", 'wb') as f:
            pickle.dump({
                'ids': self.ids,
                'metadatas': self.metadatas,
                'dimension': self.dimension
            }, f)
        
        print(f"Saved index to {filepath}")
    
    def load(self, filepath: str):
        """Load index from disk"""
        self.index = faiss.read_index(f"{filepath}.index")
        
        with open(f"{filepath}.meta", 'rb') as f:
            data = pickle.load(f)
            self.ids = data['ids']
            self.metadatas = data['metadatas']
            self.dimension = data['dimension']
        
        print(f"Loaded index from {filepath}")
        print(f"Total vectors: {self.index.ntotal}")

# Example usage
if __name__ == "__main__":
    # Create index
    store = FAISSVectorStore(dimension=384)
    
    # Add vectors
    embeddings = [[0.1, 0.2, ...], ...]  # Your embeddings here
    ids = ["doc1", "doc2", ...]
    metadatas = [{"source": "..."}, ...]
    
    store.add_documents(embeddings, ids, metadatas)
    
    # Search
    query_embedding = [0.15, 0.25, ...]
    results = store.search(query_embedding, k=3)
    
    # Save for later
    store.save("my_faiss_index")

Part 4: Similarity Metrics

4.1 Cosine Similarity (Most Common)

Formula:

cosine_similarity(A, B) = (A · B) / (||A|| × ||B||)

Range: -1 (opposite) to 1 (identical)

Implementation:

# similarity/metrics.py
import numpy as np
from typing import List

def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
    """
    Calculate cosine similarity between two vectors
    
    Args:
        vec1: First vector
        vec2: Second vector
        
    Returns:
        Similarity score between -1 and 1
    """
    v1 = np.array(vec1)
    v2 = np.array(vec2)
    
    dot_product = np.dot(v1, v2)
    norm_product = np.linalg.norm(v1) * np.linalg.norm(v2)
    
    return dot_product / norm_product

def euclidean_distance(vec1: List[float], vec2: List[float]) -> float:
    """
    Calculate Euclidean distance (L2 distance)
    
    Smaller = more similar
    
    Args:
        vec1: First vector
        vec2: Second vector
        
    Returns:
        Distance (0 = identical)
    """
    v1 = np.array(vec1)
    v2 = np.array(vec2)
    
    return np.linalg.norm(v1 - v2)

def dot_product_similarity(vec1: List[float], vec2: List[float]) -> float:
    """
    Calculate dot product similarity
    
    Faster but assumes normalized vectors
    """
    v1 = np.array(vec1)
    v2 = np.array(vec2)
    
    return np.dot(v1, v2)

# Example: Understanding the differences
if __name__ == "__main__":
    # Two similar embeddings
    v1 = [0.5, 0.8, 0.2]
    v2 = [0.6, 0.7, 0.3]
    
    # Very different embedding
    v3 = [-0.5, -0.8, -0.2]
    
    print("Similar vectors:")
    print(f"  Cosine similarity: {cosine_similarity(v1, v2):.3f}")
    print(f"  Euclidean distance: {euclidean_distance(v1, v2):.3f}")
    print(f"  Dot product: {dot_product_similarity(v1, v2):.3f}")
    
    print("\nDifferent vectors:")
    print(f"  Cosine similarity: {cosine_similarity(v1, v3):.3f}")
    print(f"  Euclidean distance: {euclidean_distance(v1, v3):.3f}")
    print(f"  Dot product: {dot_product_similarity(v1, v3):.3f}")

When to use each:

Metric	Use Case	Notes
Cosine	Most RAG systems	Direction matters, not magnitude
Euclidean	When magnitude matters	Actual distance in space
Dot Product	Normalized vectors	Fastest, similar to cosine

Part 5: Complete Indexing Pipeline

Let's build a production-ready pipeline that ties everything together.

# pipeline/indexing_pipeline.py
from typing import List, Dict
from langchain.schema import Document
import json
from pathlib import Path

class EmbeddingIndexPipeline:
    """
    Complete pipeline for embedding and indexing documents
    """
    
    def __init__(
        self,
        embedder,
        vector_store,
        batch_size: int = 100
    ):
        """
        Initialize pipeline
        
        Args:
            embedder: Embedding model (OpenAI or SentenceTransformer)
            vector_store: Vector database (Chroma, Pinecone, FAISS)
            batch_size: Batch size for embedding
        """
        self.embedder = embedder
        self.vector_store = vector_store
        self.batch_size = batch_size
    
    def index_documents(
        self,
        documents: List[Document],
        show_progress: bool = True
    ) -> Dict:
        """
        Embed and index documents
        
        Args:
            documents: List of Document objects (from Article 3)
            show_progress: Show progress updates
            
        Returns:
            Statistics about indexing
        """
        print("="*60)
        print("Embedding and Indexing Pipeline")
        print("="*60)
        print()
        
        # Step 1: Extract text and metadata
        print("Step 1: Preparing documents...")
        texts = [doc.page_content for doc in documents]
        metadatas = [doc.metadata for doc in documents]
        print(f"  {len(texts)} documents to process\n")
        
        # Step 2: Generate embeddings
        print("Step 2: Generating embeddings...")
        embeddings = self.embedder.embed_batch(
            texts,
            batch_size=self.batch_size
        )
        print(f"  Generated {len(embeddings)} embeddings")
        print(f"  Dimension: {len(embeddings[0])}\n")
        
        # Step 3: Add to vector store
        print("Step 3: Adding to vector database...")
        
        # Generate IDs
        ids = [f"doc_{i}" for i in range(len(documents))]
        
        self.vector_store.add_documents(
            documents=texts,
            embeddings=embeddings,
            metadatas=metadatas,
            ids=ids
        )
        
        print()
        print("="*60)
        print("Indexing Complete!")
        print("="*60)
        
        return {
            'total_documents': len(documents),
            'embedding_dimension': len(embeddings[0]),
            'vector_store': type(self.vector_store).__name__
        }
    
    def query(
        self,
        query_text: str,
        n_results: int = 5,
        return_distances: bool = True
    ) -> Dict:
        """
        Query the indexed documents
        
        Args:
            query_text: Query string
            n_results: Number of results to return
            return_distances: Include distance scores
            
        Returns:
            Search results
        """
        # Embed query
        query_embedding = self.embedder.embed_text(query_text)
        
        # Search
        results = self.vector_store.search(
            query_embedding=query_embedding,
            n_results=n_results
        )
        
        # Format results
        formatted_results = {
            'query': query_text,
            'results': []
        }
        
        for i, (doc, distance, metadata) in enumerate(zip(
            results['documents'],
            results['distances'],
            results['metadatas']
        )):
            result = {
                'rank': i + 1,
                'document': doc,
                'metadata': metadata
            }
            
            if return_distances:
                result['distance'] = distance
                result['similarity_score'] = 1 - (distance / 2)  # Convert to 0-1
            
            formatted_results['results'].append(result)
        
        return formatted_results
    
    def save_index(self, filepath: str):
        """Save the index to disk"""
        # Note: Implementation depends on vector store type
        print(f"Saving index to {filepath}...")
        # Vector store-specific save logic here

# Complete example
if __name__ == "__main__":
    from embeddings.sentence_transformer_embedder import SentenceTransformerEmbedder
    from vector_stores.chroma_store import ChromaVectorStore
    from langchain.schema import Document
    
    # Initialize components
    embedder = SentenceTransformerEmbedder()
    vector_store = ChromaVectorStore(collection_name="rag_demo")
    
    # Create pipeline
    pipeline = EmbeddingIndexPipeline(
        embedder=embedder,
        vector_store=vector_store
    )
    
    # Sample documents (in real app, these come from Article 3 pipeline)
    documents = [
        Document(
            page_content="RAG combines retrieval with generation to provide accurate, grounded responses.",
            metadata={"source": "doc1.txt", "category": "rag"}
        ),
        Document(
            page_content="Vector databases store embeddings and enable fast similarity search using ANN algorithms.",
            metadata={"source": "doc2.txt", "category": "databases"}
        ),
        Document(
            page_content="Embeddings are numerical representations of text that capture semantic meaning.",
            metadata={"source": "doc3.txt", "category": "embeddings"}
        ),
        Document(
            page_content="ChromaDB is an open-source vector database designed for AI applications.",
            metadata={"source": "doc4.txt", "category": "databases"}
        )
    ]
    
    # Index documents
    stats = pipeline.index_documents(documents)
    
    print("\nIndexing Statistics:")
    for key, value in stats.items():
        print(f"  {key}: {value}")
    
    # Query the index
    print("\n" + "="*60)
    print("Testing Queries")
    print("="*60)
    
    queries = [
        "How does RAG work?",
        "What are vector databases?",
        "Tell me about embeddings"
    ]
    
    for query in queries:
        print(f"\nQuery: {query}")
        print("-"*60)
        
        results = pipeline.query(query, n_results=2)
        
        for result in results['results']:
            print(f"\n{result['rank']}. Similarity: {result['similarity_score']:.3f}")
            print(f"   {result['document'][:100]}...")
            print(f"   Source: {result['metadata']['source']}")

Part 6: Performance Optimization

6.1 Embedding Optimization

# optimization/embedding_optimization.py
import time
from typing import List
import numpy as np

class EmbeddingOptimizer:
    """
    Techniques to optimize embedding performance
    """
    
    @staticmethod
    def batch_optimize(
        texts: List[str],
        embedder,
        optimal_batch_size: int = None
    ) -> List[List[float]]:
        """
        Find optimal batch size for your hardware
        
        Args:
            texts: Texts to embed
            embedder: Embedding model
            optimal_batch_size: If known, use this. Otherwise, auto-detect.
            
        Returns:
            Embeddings
        """
        if optimal_batch_size:
            return embedder.embed_batch(texts, batch_size=optimal_batch_size)
        
        # Auto-detect optimal batch size
        print("Auto-detecting optimal batch size...")
        test_sizes = [8, 16, 32, 64, 128]
        test_texts = texts[:100]  # Use subset for testing
        
        results = {}
        for size in test_sizes:
            start = time.time()
            _ = embedder.embed_batch(test_texts, batch_size=size)
            elapsed = time.time() - start
            
            per_text = elapsed / len(test_texts)
            results[size] = per_text
            print(f"  Batch size {size}: {per_text*1000:.2f}ms per text")
        
        # Find fastest
        optimal = min(results, key=results.get)
        print(f"\nOptimal batch size: {optimal}")
        
        # Use optimal size for full embedding
        return embedder.embed_batch(texts, batch_size=optimal)
    
    @staticmethod
    def cache_embeddings(
        texts: List[str],
        embedder,
        cache_file: str = "embedding_cache.npz"
    ) -> List[List[float]]:
        """
        Cache embeddings to avoid re-computing
        
        Args:
            texts: Texts to embed
            embedder: Embedding model
            cache_file: Where to cache
            
        Returns:
            Embeddings
        """
        from pathlib import Path
        import hashlib
        
        # Create hash of texts
        text_hash = hashlib.md5(''.join(texts).encode()).hexdigest()
        cache_path = Path(cache_file)
        
        # Check cache
        if cache_path.exists():
            data = np.load(cache_path, allow_pickle=True)
            if data['hash'] == text_hash:
                print("Loading from cache...")
                return data['embeddings'].tolist()
        
        # Compute embeddings
        print("Computing embeddings...")
        embeddings = embedder.embed_batch(texts)
        
        # Save to cache
        np.savez(
            cache_file,
            embeddings=np.array(embeddings),
            hash=text_hash
        )
        print(f"Cached to {cache_file}")
        
        return embeddings

6.2 Index Optimization

# optimization/index_optimization.py

class IndexOptimizer:
    """
    Optimize vector database performance
    """
    
    @staticmethod
    def analyze_index_size(vector_store) -> dict:
        """
        Analyze memory usage and suggest optimizations
        """
        stats = vector_store.get_stats()
        count = stats['count']
        
        # Estimate memory (rough)
        dimension = 384  # Typical
        bytes_per_float = 4
        vectors_size_mb = (count * dimension * bytes_per_float) / (1024 * 1024)
        
        print(f"Index Statistics:")
        print(f"  Documents: {count:,}")
        print(f"  Estimated size: {vectors_size_mb:.2f} MB")
        
        # Recommendations
        if vectors_size_mb > 1000:  # > 1GB
            print("\nRecommendation: Consider using quantization or PQ compression")
        
        return {
            'count': count,
            'size_mb': vectors_size_mb
        }
    
    @staticmethod
    def benchmark_search_speed(
        vector_store,
        embedder,
        test_queries: List[str],
        n_results: int = 5
    ):
        """
        Benchmark search performance
        """
        print("Benchmarking search speed...")
        
        times = []
        for query in test_queries:
            query_embedding = embedder.embed_text(query)
            
            start = time.time()
            _ = vector_store.search(query_embedding, n_results=n_results)
            elapsed = time.time() - start
            
            times.append(elapsed)
        
        avg_time = np.mean(times)
        p95_time = np.percentile(times, 95)
        
        print(f"\nSearch Performance:")
        print(f"  Average: {avg_time*1000:.2f}ms")
        print(f"  95th percentile: {p95_time*1000:.2f}ms")
        print(f"  Queries/second: {1/avg_time:.1f}")
        
        return {
            'avg_ms': avg_time * 1000,
            'p95_ms': p95_time * 1000
        }

Part 7: Best Practices

7.1 Production Checklist

Embedding:

✅ Choose model based on quality/cost/privacy needs
✅ Batch embeddings for efficiency
✅ Cache embeddings when possible
✅ Handle encoding errors gracefully
✅ Monitor embedding costs (if using API)

Vector Database:

✅ Choose database based on scale
✅ Use appropriate index type for your data size
✅ Implement proper error handling
✅ Back up indices regularly
✅ Monitor query latency

Search Quality:

✅ Test similarity thresholds
✅ Implement metadata filtering
✅ Add re-ranking if needed
✅ Log queries for analysis
✅ A/B test different embedding models

7.2 Common Pitfalls

Pitfall 1: Wrong Embedding Model

# Bad: Using a random model
embedder = SentenceTransformer('random-model-name')

# Good: Choose based on benchmarks and your needs
embedder = SentenceTransformer('all-mpnet-base-v2')  # Tested, proven

Pitfall 2: Not Normalizing Vectors

# Some models require normalization
from sklearn.preprocessing import normalize

embeddings = embedder.embed_batch(texts)
embeddings_normalized = normalize(embeddings, norm='l2')

Pitfall 3: Ignoring Metadata

# Bad: No metadata
store.add_documents(docs, embeddings)

# Good: Rich metadata for filtering
metadatas = [
    {"source": "doc.pdf", "date": "2024-01", "category": "tech"},
    # ...
]
store.add_documents(docs, embeddings, metadatas=metadatas)

Part 8: Practice Exercises

Exercise 1: Embedding Model Comparison

Test 3 different embedding models on your domain-specific data. Measure quality and speed.

Exercise 2: Build a Search Engine

Create a mini search engine for your documents using embeddings and a vector database.

Exercise 3: Optimize Performance

Take a large document collection (10,000+ docs) and optimize embedding and indexing time.

Exercise 4: Metadata Filtering

Implement advanced metadata filtering (date ranges, categories, tags) in your searches.

Key Takeaways

Core Concepts

✅ Embeddings convert text to numerical vectors
✅ Similar meanings → similar vectors
✅ Vector databases enable fast similarity search
✅ Cosine similarity is the standard metric
✅ Batch processing is crucial for performance

Embedding Models

OpenAI: Best quality, paid
Sentence Transformers: Free, local, good quality
BGE: SOTA open source

Vector Databases

ChromaDB: Development, local
Pinecone: Production, managed
FAISS: High performance, self-hosted
Weaviate: Advanced features, flexible

Performance Tips

Batch embeddings for efficiency
Cache when possible
Choose appropriate index type
Monitor and optimize continuously

What's Next?

In Article 5: Retrieval Techniques and Optimization, we'll explore:

Advanced retrieval strategies
Hybrid search (semantic + keyword)
Re-ranking techniques
Maximal Marginal Relevance (MMR)
Query optimization
Measuring retrieval quality

You now have searchable vectors. Next, we'll make that search even better!

Review Questions

What's the difference between embeddings and keywords?
Why do we need vector databases instead of regular databases?
What's cosine similarity and why is it used?
When would you choose OpenAI embeddings over Sentence Transformers?
What's the purpose of batch processing embeddings?
Name three vector database options and their use cases.
What's an approximate nearest neighbor (ANN) search?

If you can answer these, you've mastered embeddings and vector databases!

Next Article: Retrieval Techniques and Optimization
Coming Soon: Where we'll make search even better with advanced techniques!

This is Article 4 of the "Building RAG from Scratch" series. Your data is now embedded and searchable - time to optimize retrieval!

semantic

Related Topics

Building RAG from Scratch: Article 4