Module 3: Embedding Models Compared

Mini-Project: Benchmark Framework for Embedding Models

Next →

Project overview

You'll build a complete benchmarking framework that compares 3+ embedding models across multiple dimensions: performance (MTEB-style), latency, throughput, and cost. The framework generates automatic reports and recommends the best model for your use case.

By completing this project, you'll have a reusable tool to evaluate and select embedding models objectively.

Project objectives

Features:

  1. ✅ Compare 3+ models (OpenAI, SBERT, BGE)
  2. ✅ Evaluate performance (retrieval accuracy)
  3. ✅ Measure latency (ms/query)
  4. ✅ Measure throughput (QPS)
  5. ✅ Calculate costs (monthly)
  6. ✅ Generate a comparative report
  7. ✅ Recommend the optimal model

Project structure

embeddings-benchmark/
├── src/
│   ├── __init__.py
│   ├── models.py          # Model wrapper
│   ├── evaluator.py       # Performance evaluation
│   ├── latency_bench.py   # Latency/throughput
│   ├── cost_calculator.py # Cost calculation
│   └── reporter.py        # Report generation
├── data/
│   └── eval_dataset.json  # Evaluation dataset
├── results/
│   └── benchmark_report.md  # Generated report
├── requirements.txt
├── main.py
└── README.md

Initial setup

requirements.txt

openai==1.54.0
sentence-transformers==2.3.1
python-dotenv==1.0.0
numpy==1.26.4
tabulate==0.9.0

.env

OPENAI_API_KEY=your-api-key-here

Install dependencies

pip install -r requirements.txt

Implementation

Step 1: models.py (Model wrapper)

"""
Unified wrapper for different embedding models
"""
from openai import OpenAI
from sentence_transformers import SentenceTransformer
import os
from dotenv import load_dotenv
from typing import List
import numpy as np

load_dotenv()

class EmbeddingModel:
    """Base class for models"""
    
    def __init__(self, name: str):
        self.name = name
    
    def encode(self, texts: List[str]) -> np.ndarray:
        """Generate embeddings"""
        raise NotImplementedError

class OpenAIModel(EmbeddingModel):
    """Wrapper for OpenAI embeddings"""
    
    def __init__(self, model_name: str):
        super().__init__(f"OpenAI-{model_name}")
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model_name = model_name
    
    def encode(self, texts: List[str]) -> np.ndarray:
        """Generate embeddings"""
        if isinstance(texts, str):
            texts = [texts]
        
        response = self.client.embeddings.create(
            model=self.model_name,
            input=texts
        )
        
        embeddings = [item.embedding for item in response.data]
        return np.array(embeddings)

class SentenceTransformerModel(EmbeddingModel):
    """Wrapper for Sentence-Transformers"""
    
    def __init__(self, model_name: str):
        super().__init__(f"SBERT-{model_name}")
        self.model = SentenceTransformer(model_name)
    
    def encode(self, texts: List[str]) -> np.ndarray:
        """Generate embeddings"""
        return self.model.encode(texts)

# Factory
def get_model(model_type: str, model_name: str) -> EmbeddingModel:
    """Create a model based on type"""
    if model_type == "openai":
        return OpenAIModel(model_name)
    elif model_type == "sbert":
        return SentenceTransformerModel(model_name)
    else:
        raise ValueError(f"Unknown model type: {model_type}")

Step 2: evaluator.py (Performance evaluation)

"""
Performance evaluation (retrieval accuracy)
"""
import numpy as np
from typing import List, Dict
from src.models import EmbeddingModel

class PerformanceEvaluator:
    """Performance evaluator"""
    
    def __init__(self, eval_dataset: List[Dict]):
        """
        Args:
            eval_dataset: List of dicts with keys:
                - query: str
                - doc_relevant: str
                - doc_irrelevant: str
        """
        self.eval_dataset = eval_dataset
    
    def evaluate(self, model: EmbeddingModel) -> Dict:
        """
        Evaluate a model
        
        Returns:
            Dict with metrics
        """
        correct = 0
        total = len(self.eval_dataset)
        
        for item in self.eval_dataset:
            # Generate embeddings
            query_emb = model.encode([item['query']])[0]
            rel_emb = model.encode([item['doc_relevant']])[0]
            irrel_emb = model.encode([item['doc_irrelevant']])[0]
            
            # Cosine similarity
            sim_rel = self._cosine_similarity(query_emb, rel_emb)
            sim_irrel = self._cosine_similarity(query_emb, irrel_emb)
            
            # Check if relevant > irrelevant
            if sim_rel > sim_irrel:
                correct += 1
        
        accuracy = correct / total
        
        return {
            'accuracy': accuracy,
            'correct': correct,
            'total': total
        }
    
    @staticmethod
    def _cosine_similarity(a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

Step 3: latency_bench.py (Latency/Throughput)

"""
Latency and throughput benchmark
"""
import time
import numpy as np
from typing import Dict
from src.models import EmbeddingModel

class LatencyBenchmark:
    """Latency benchmark"""
    
    def __init__(self, n_runs: int = 20):
        self.n_runs = n_runs
    
    def measure_latency(self, model: EmbeddingModel, text: str) -> Dict:
        """
        Measure latency
        
        Returns:
            Dict with average/min/max latency
        """
        latencies = []
        
        # Warmup
        _ = model.encode([text])
        
        for _ in range(self.n_runs):
            start = time.time()
            _ = model.encode([text])
            latency = (time.time() - start) * 1000  # ms
            latencies.append(latency)
        
        return {
            'avg_ms': np.mean(latencies),
            'min_ms': np.min(latencies),
            'max_ms': np.max(latencies),
            'std_ms': np.std(latencies)
        }
    
    def measure_throughput(self, model: EmbeddingModel, n_texts: int = 500) -> Dict:
        """
        Measure throughput
        
        Returns:
            Dict with QPS (queries per second)
        """
        texts = [f"Text {i}" for i in range(n_texts)]
        
        start = time.time()
        _ = model.encode(texts)
        elapsed = time.time() - start
        
        throughput = n_texts / elapsed
        
        return {
            'qps': throughput,
            'total_time_s': elapsed,
            'n_texts': n_texts
        }

Step 4: cost_calculator.py (Cost calculation)

"""
Cost calculator
"""
from typing import Dict

class CostCalculator:
    """Cost calculator"""
    
    # Pricing (update by date)
    PRICING = {
        "OpenAI-text-embedding-3-small": {
            "type": "api",
            "cost_per_1m_tokens": 0.020
        },
        "OpenAI-text-embedding-3-large": {
            "type": "api",
            "cost_per_1m_tokens": 0.130
        },
        "SBERT-all-MiniLM-L6-v2": {
            "type": "self-hosted",
            "monthly_gpu": 0,  # CPU
            "monthly_infra": 0
        },
        "SBERT-all-mpnet-base-v2": {
            "type": "self-hosted",
            "monthly_gpu": 0,  # CPU
            "monthly_infra": 0
        },
        "SBERT-BAAI/bge-large-en-v1.5": {
            "type": "self-hosted",
            "monthly_gpu": 200,  # GPU required
            "monthly_infra": 15  # Storage + bandwidth
        }
    }
    
    def calculate_cost(
        self,
        model_name: str,
        queries_per_month: int,
        tokens_per_query: int = 50
    ) -> Dict:
        """
        Calculate the monthly cost
        
        Returns:
            Dict with a cost breakdown
        """
        pricing = self.PRICING.get(model_name, None)
        
        if not pricing:
            return {"error": "Pricing not found"}
        
        if pricing["type"] == "api":
            # API cost
            total_tokens = queries_per_month * tokens_per_query
            cost = (total_tokens / 1_000_000) * pricing["cost_per_1m_tokens"]
            
            return {
                "type": "API",
                "cost_per_query": cost / queries_per_month if queries_per_month > 0 else 0,
                "cost_monthly": cost,
                "breakdown": {
                    "api": cost
                }
            }
        
        else:  # self-hosted
            cost = pricing["monthly_gpu"] + pricing["monthly_infra"]
            
            return {
                "type": "Self-hosted",
                "cost_per_query": cost / queries_per_month if queries_per_month > 0 else 0,
                "cost_monthly": cost,
                "breakdown": {
                    "gpu": pricing["monthly_gpu"],
                    "infra": pricing["monthly_infra"]
                }
            }

Step 5: reporter.py (Report generation)

"""
Report generation
"""
from typing import List, Dict
from tabulate import tabulate

class BenchmarkReporter:
    """Report generator"""
    
    def generate_report(self, results: List[Dict], output_file: str = "results/benchmark_report.md"):
        """
        Generate a report in Markdown
        
        Args:
            results: List of dicts with results per model
            output_file: Path of the output file
        """
        with open(output_file, 'w') as f:
            f.write("# Embeddings Benchmark Report\n\n")
            
            # Performance table
            f.write("## Performance (Retrieval Accuracy)\n\n")
            perf_table = [
                [r['model'], f"{r['performance']['accuracy']:.2%}"]
                for r in results
            ]
            f.write(tabulate(perf_table, headers=["Model", "Accuracy"], tablefmt="github"))
            f.write("\n\n")
            
            # Latency table
            f.write("## Latency\n\n")
            latency_table = [
                [r['model'], f"{r['latency']['avg_ms']:.1f}ms"]
                for r in results
            ]
            f.write(tabulate(latency_table, headers=["Model", "Avg Latency"], tablefmt="github"))
            f.write("\n\n")
            
            # Throughput table
            f.write("## Throughput\n\n")
            throughput_table = [
                [r['model'], f"{r['throughput']['qps']:.0f} QPS"]
                for r in results
            ]
            f.write(tabulate(throughput_table, headers=["Model", "Throughput"], tablefmt="github"))
            f.write("\n\n")
            
            # Cost table
            f.write("## Cost (100K queries/month)\n\n")
            cost_table = [
                [r['model'], r['cost']['type'], f"${r['cost']['cost_monthly']:.2f}"]
                for r in results
            ]
            f.write(tabulate(cost_table, headers=["Model", "Type", "Monthly Cost"], tablefmt="github"))
            f.write("\n\n")
            
            # Recommendation
            f.write("## Recommendation\n\n")
            f.write(self._generate_recommendation(results))
        
        print(f"✅ Report generated: {output_file}")
    
    def _generate_recommendation(self, results: List[Dict]) -> str:
        """Generate a recommendation based on results"""
        # Sort by accuracy
        sorted_by_acc = sorted(results, key=lambda x: x['performance']['accuracy'], reverse=True)
        best_acc = sorted_by_acc[0]
        
        # Sort by latency
        sorted_by_lat = sorted(results, key=lambda x: x['latency']['avg_ms'])
        best_lat = sorted_by_lat[0]
        
        # Sort by cost
        sorted_by_cost = sorted(results, key=lambda x: x['cost']['cost_monthly'])
        best_cost = sorted_by_cost[0]
        
        rec = f"**Best Performance:** {best_acc['model']} ({best_acc['performance']['accuracy']:.2%})\n\n"
        rec += f"**Lowest Latency:** {best_lat['model']} ({best_lat['latency']['avg_ms']:.1f}ms)\n\n"
        rec += f"**Lowest Cost:** {best_cost['model']} (${best_cost['cost']['cost_monthly']:.2f}/month)\n\n"
        
        return rec

Step 6: data/eval_dataset.json (Evaluation dataset)

[
  {
    "query": "How to install Python?",
    "doc_relevant": "Download Python from python.org and run installer",
    "doc_irrelevant": "JavaScript is a web programming language"
  },
  {
    "query": "Python list comprehension",
    "doc_relevant": "[x for x in range(10)] creates a list of numbers",
    "doc_irrelevant": "Arrays in Java are fixed size"
  },
  {
    "query": "Django web framework",
    "doc_relevant": "Django is a Python framework for building web apps",
    "doc_irrelevant": "React is a JavaScript library for UIs"
  },
  {
    "query": "NumPy arrays",
    "doc_relevant": "NumPy provides efficient array operations in Python",
    "doc_irrelevant": "MATLAB is used for numerical computing"
  },
  {
    "query": "Virtual environment Python",
    "doc_relevant": "Use venv or virtualenv to create isolated Python environments",
    "doc_irrelevant": "Docker containers provide application isolation"
  },
  {
    "query": "Pandas DataFrame",
    "doc_relevant": "Pandas DataFrame is a 2D data structure for data analysis",
    "doc_irrelevant": "Excel spreadsheets store tabular data"
  },
  {
    "query": "FastAPI tutorial",
    "doc_relevant": "FastAPI is a modern Python web framework for APIs",
    "doc_irrelevant": "Express.js is a Node.js web framework"
  },
  {
    "query": "Python decorators",
    "doc_relevant": "@decorator syntax modifies function behavior in Python",
    "doc_irrelevant": "Annotations in Java provide metadata"
  },
  {
    "query": "Async await Python",
    "doc_relevant": "async/await enables asynchronous programming in Python",
    "doc_irrelevant": "Callbacks handle asynchronous code in JavaScript"
  },
  {
    "query": "Python type hints",
    "doc_relevant": "Type hints specify variable types in Python 3.5+",
    "doc_irrelevant": "TypeScript adds static typing to JavaScript"
  }
]

Step 7: main.py (Main script)

"""
Main benchmark script
"""
import json
from src.models import get_model
from src.evaluator import PerformanceEvaluator
from src.latency_bench import LatencyBenchmark
from src.cost_calculator import CostCalculator
from src.reporter import BenchmarkReporter

def main():
    """Run the complete benchmark"""
    print("=== Embeddings Benchmark Framework ===\n")
    
    # Configuration
    models_to_test = [
        ("openai", "text-embedding-3-small"),
        ("sbert", "all-MiniLM-L6-v2"),
        ("sbert", "all-mpnet-base-v2")
    ]
    
    queries_per_month = 100_000
    
    # Load the dataset
    with open("data/eval_dataset.json", 'r') as f:
        eval_dataset = json.load(f)
    
    # Initialize evaluators
    perf_evaluator = PerformanceEvaluator(eval_dataset)
    latency_bench = LatencyBenchmark(n_runs=10)
    cost_calc = CostCalculator()
    
    # Results
    results = []
    
    for model_type, model_name in models_to_test:
        print(f"Testing {model_type}/{model_name}...")
        
        model = get_model(model_type, model_name)
        
        # Performance
        print("  - Evaluating performance...")
        perf = perf_evaluator.evaluate(model)
        
        # Latency
        print("  - Measuring latency...")
        latency = latency_bench.measure_latency(model, "Python is popular")
        
        # Throughput
        print("  - Measuring throughput...")
        throughput = latency_bench.measure_throughput(model, n_texts=100)
        
        # Cost
        print("  - Calculating cost...")
        cost = cost_calc.calculate_cost(model.name, queries_per_month)
        
        results.append({
            'model': model.name,
            'performance': perf,
            'latency': latency,
            'throughput': throughput,
            'cost': cost
        })
        
        print(f"  ✅ Done\n")
    
    # Generate the report
    print("Generating report...")
    reporter = BenchmarkReporter()
    reporter.generate_report(results)
    
    print("\n✅ Benchmark completed!")

if __name__ == "__main__":
    main()

Execution

python main.py

Expected output:

=== Embeddings Benchmark Framework ===

Testing openai/text-embedding-3-small...
  - Evaluating performance...
  - Measuring latency...
  - Measuring throughput...
  - Calculating cost...
  ✅ Done

Testing sbert/all-MiniLM-L6-v2...
  - Evaluating performance...
  - Measuring latency...
  - Measuring throughput...
  - Calculating cost...
  ✅ Done

Testing sbert/all-mpnet-base-v2...
  - Evaluating performance...
  - Measuring latency...
  - Measuring throughput...
  - Calculating cost...
  ✅ Done

Generating report...
✅ Report generated: results/benchmark_report.md

✅ Benchmark completed!

Generated report (benchmark_report.md)

# Embeddings Benchmark Report

## Performance (Retrieval Accuracy)

| Model                        | Accuracy |
|------------------------------|----------|
| OpenAI-text-embedding-3-small| 100.00%  |
| SBERT-all-mpnet-base-v2      | 100.00%  |
| SBERT-all-MiniLM-L6-v2       | 100.00%  |

## Latency

| Model                        | Avg Latency |
|------------------------------|-------------|
| SBERT-all-MiniLM-L6-v2       | 4.8ms       |
| SBERT-all-mpnet-base-v2      | 14.2ms      |
| OpenAI-text-embedding-3-small| 87.3ms      |

## Throughput

| Model                        | Throughput  |
|------------------------------|-------------|
| SBERT-all-MiniLM-L6-v2       | 215 QPS     |
| SBERT-all-mpnet-base-v2      | 68 QPS      |
| OpenAI-text-embedding-3-small| 11 QPS      |

## Cost (100K queries/month)

| Model                        | Type        | Monthly Cost |
|------------------------------|-------------|--------------|
| SBERT-all-MiniLM-L6-v2       | Self-hosted | $0.00        |
| SBERT-all-mpnet-base-v2      | Self-hosted | $0.00        |
| OpenAI-text-embedding-3-small| API         | $0.10        |

## Recommendation

**Best Performance:** OpenAI-text-embedding-3-small (100.00%)

**Lowest Latency:** SBERT-all-MiniLM-L6-v2 (4.8ms)

**Lowest Cost:** SBERT-all-MiniLM-L6-v2 ($0.00/month)

Optional extensions

1. Add more models:

models_to_test = [
    ("openai", "text-embedding-3-small"),
    ("openai", "text-embedding-3-large"),
    ("sbert", "all-MiniLM-L6-v2"),
    ("sbert", "all-mpnet-base-v2"),
    ("sbert", "BAAI/bge-large-en-v1.5"),  # Add BGE
]

2. Add MTEB scores:

# In models.py, add:
MTEB_SCORES = {
    "OpenAI-text-embedding-3-small": 62.3,
    "SBERT-all-MiniLM-L6-v2": 56.3,
    "SBERT-all-mpnet-base-v2": 57.8
}

3. Visualization (plots):

import matplotlib.pyplot as plt

def plot_comparison(results):
    """Generate comparison charts"""
    models = [r['model'] for r in results]
    accuracies = [r['performance']['accuracy'] for r in results]
    
    plt.bar(models, accuracies)
    plt.ylabel('Accuracy')
    plt.title('Model Comparison')
    plt.savefig('results/comparison.png')

Project validation

Checklist:

  • Framework compares 3+ models ✅
  • Evaluates performance (accuracy) ✅
  • Measures latency and throughput ✅
  • Calculates costs ✅
  • Generates a Markdown report ✅
  • Recommends the optimal model ✅

Module 3 summary

What you learned in the module:

Landscape:

  • ✅ OpenAI models (3-small vs 3-large)
  • ✅ Open-source (SBERT, BGE, Instructor, E5)

Evaluation:

  • ✅ MTEB benchmark (58 datasets, 8 tasks)
  • ✅ Latency/throughput (API ~90ms, local ~5ms)
  • ✅ Cost analysis (break-even ~500M queries/month)

Specialization:

  • ✅ Domain-specific (legal, medical, code)
  • ✅ Multilingual (mBERT, XLM-R, BGE-M3)

Project:

  • ✅ Complete benchmark framework (~600 lines)

Module 3 conclusion

What you implemented:

  • ✅ Benchmark framework: Systematic model comparison
  • ✅ Multiple dimensions: Performance, latency, cost
  • ✅ Automatic reports: Markdown with tables
  • ✅ Recommendation: Based on objective data

Patterns applied:

  1. Strategy pattern (different models, same interface)
  2. Factory pattern (get_model)
  3. Single Responsibility (each module one function)

Next module

Module 4: Chunking and Embedding Evaluation

You'll learn:

  • Chunking strategies (fixed, semantic, recursive)
  • Overlap and optimal chunk size
  • Evaluating embedding quality
  • Retrieval metrics (nDCG, MRR, Recall@K)
  • Project: A RAG system with intelligent chunking

From model comparison to RAG implementation.

Module 3 completed ✅ Benchmark framework: choosing the right model with data

← PreviousDomain-Specific and Multilingual EmbeddingsNext →Introduction to Module 4: Evaluation and Chunking Strategies