Módulo 3: Modelos de Embeddings Comparison

Mini-Proyecto: Benchmark Framework para Modelos de Embeddings

Siguiente →

Descripción del proyecto

Construirás un framework completo de benchmarking que compara 3+ modelos de embeddings en múltiples dimensiones: performance (MTEB-style), latencia, throughput, y costo. El framework genera reportes automáticos y recomienda el mejor modelo según tu caso de uso.

Al completar este proyecto, tendrás una herramienta reutilizable para evaluar y seleccionar modelos de embeddings objetivamente.

Objetivos del proyecto

Funcionalidades:

  1. ✅ Comparar 3+ modelos (OpenAI, SBERT, BGE)
  2. ✅ Evaluar performance (retrieval accuracy)
  3. ✅ Medir latencia (ms/query)
  4. ✅ Medir throughput (QPS)
  5. ✅ Calcular costos (mensual)
  6. ✅ Generar reporte comparativo
  7. ✅ Recomendar modelo óptimo

Estructura del proyecto

embeddings-benchmark/
├── src/
│   ├── __init__.py
│   ├── models.py          # Wrapper de modelos
│   ├── evaluator.py       # Performance evaluation
│   ├── latency_bench.py   # Latencia/throughput
│   ├── cost_calculator.py # Cálculo de costos
│   └── reporter.py        # Generación de reportes
├── data/
│   └── eval_dataset.json  # Dataset de evaluación
├── results/
│   └── benchmark_report.md  # Reporte generado
├── requirements.txt
├── main.py
└── README.md

Setup inicial

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=tu-api-key-aqui

Instalar dependencias

pip install -r requirements.txt

Implementación

Paso 1: models.py (Wrapper de modelos)

"""
Wrapper unificado para diferentes modelos de embeddings
"""
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:
    """Clase base para modelos"""
    
    def __init__(self, name: str):
        self.name = name
    
    def encode(self, texts: List[str]) -> np.ndarray:
        """Generar embeddings"""
        raise NotImplementedError

class OpenAIModel(EmbeddingModel):
    """Wrapper para 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:
        """Generar 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 para 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:
        """Generar embeddings"""
        return self.model.encode(texts)

# Factory
def get_model(model_type: str, model_name: str) -> EmbeddingModel:
    """Crear modelo según tipo"""
    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}")

Paso 2: evaluator.py (Performance evaluation)

"""
Evaluación de performance (retrieval accuracy)
"""
import numpy as np
from typing import List, Dict
from src.models import EmbeddingModel

class PerformanceEvaluator:
    """Evaluador de performance"""
    
    def __init__(self, eval_dataset: List[Dict]):
        """
        Args:
            eval_dataset: Lista de dicts con keys:
                - query: str
                - doc_relevant: str
                - doc_irrelevant: str
        """
        self.eval_dataset = eval_dataset
    
    def evaluate(self, model: EmbeddingModel) -> Dict:
        """
        Evaluar modelo
        
        Returns:
            Dict con métricas
        """
        correct = 0
        total = len(self.eval_dataset)
        
        for item in self.eval_dataset:
            # Generar 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 si relevante > irrelevante
            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))

Paso 3: latency_bench.py (Latencia/Throughput)

"""
Benchmark de latencia y throughput
"""
import time
import numpy as np
from typing import Dict
from src.models import EmbeddingModel

class LatencyBenchmark:
    """Benchmark de latencia"""
    
    def __init__(self, n_runs: int = 20):
        self.n_runs = n_runs
    
    def measure_latency(self, model: EmbeddingModel, text: str) -> Dict:
        """
        Medir latencia
        
        Returns:
            Dict con latencia promedio/min/max
        """
        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:
        """
        Medir throughput
        
        Returns:
            Dict con QPS (queries per second)
        """
        texts = [f"Texto {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
        }

Paso 4: cost_calculator.py (Cálculo de costos)

"""
Calculadora de costos
"""
from typing import Dict

class CostCalculator:
    """Calculadora de costos"""
    
    # Pricing (actualizar según fecha)
    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 requerida
            "monthly_infra": 15  # Storage + bandwidth
        }
    }
    
    def calculate_cost(
        self,
        model_name: str,
        queries_per_month: int,
        tokens_per_query: int = 50
    ) -> Dict:
        """
        Calcular costo mensual
        
        Returns:
            Dict con breakdown de costos
        """
        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"]
                }
            }

Paso 5: reporter.py (Generación de reportes)

"""
Generación de reportes
"""
from typing import List, Dict
from tabulate import tabulate

class BenchmarkReporter:
    """Generador de reportes"""
    
    def generate_report(self, results: List[Dict], output_file: str = "results/benchmark_report.md"):
        """
        Generar reporte en Markdown
        
        Args:
            results: Lista de dicts con resultados por modelo
            output_file: Path del archivo de salida
        """
        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:
        """Generar recomendación basada en resultados"""
        # 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

Paso 6: data/eval_dataset.json (Dataset de evaluación)

[
  {
    "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"
  }
]

Paso 7: main.py (Script principal)

"""
Script principal del benchmark
"""
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():
    """Ejecutar benchmark completo"""
    print("=== Embeddings Benchmark Framework ===\n")
    
    # Configuración
    models_to_test = [
        ("openai", "text-embedding-3-small"),
        ("sbert", "all-MiniLM-L6-v2"),
        ("sbert", "all-mpnet-base-v2")
    ]
    
    queries_per_month = 100_000
    
    # Cargar dataset
    with open("data/eval_dataset.json", 'r') as f:
        eval_dataset = json.load(f)
    
    # Inicializar evaluadores
    perf_evaluator = PerformanceEvaluator(eval_dataset)
    latency_bench = LatencyBenchmark(n_runs=10)
    cost_calc = CostCalculator()
    
    # Resultados
    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)
        
        # Latencia
        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)
        
        # Costo
        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")
    
    # Generar reporte
    print("Generating report...")
    reporter = BenchmarkReporter()
    reporter.generate_report(results)
    
    print("\n✅ Benchmark completed!")

if __name__ == "__main__":
    main()

Ejecución

python main.py

Output esperado:

=== 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!

Reporte generado (benchmark_report.md)

# Embeddings Benchmark Report

## Performance (Retrieval Accuracy)

| Model                        | Accuracy |
|------------------------------|----------|
| OpenAI-text-embedding-3-small| 90.00%   |
| SBERT-all-mpnet-base-v2      | 85.00%   |
| SBERT-all-MiniLM-L6-v2       | 80.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         | $1.00        |

## Recommendation

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

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

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

Extensiones opcionales

1. Agregar más modelos:

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"),  # Agregar BGE
]

2. Agregar MTEB scores:

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

3. Visualización (plots):

import matplotlib.pyplot as plt

def plot_comparison(results):
    """Generar gráficos de comparación"""
    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')

Validación del proyecto

Checklist:

  • Framework compara 3+ modelos ✅
  • Evalúa performance (accuracy) ✅
  • Mide latencia y throughput ✅
  • Calcula costos ✅
  • Genera reporte Markdown ✅
  • Recomienda modelo óptimo ✅

Resumen del Módulo 3

Qué aprendiste en el módulo:

Landscape:

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

Evaluación:

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

Specialization:

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

Proyecto:

  • ✅ Benchmark framework completo (~600 líneas)

Conclusión del Módulo 3

Qué implementaste:

  • ✅ Benchmark framework: Comparación sistemática de modelos
  • ✅ Múltiples dimensiones: Performance, latencia, costo
  • ✅ Reportes automáticos: Markdown con tablas
  • ✅ Recomendación: Basada en datos objetivos

Patrones aplicados:

  1. Strategy pattern (diferentes modelos, misma interfaz)
  2. Factory pattern (get_model)
  3. Single Responsibility (cada módulo una función)

Siguiente módulo

Módulo 4: Chunking y Evaluación de Embeddings

Aprenderás:

  • Estrategias de chunking (fixed, semantic, recursive)
  • Overlap y tamaño óptimo de chunks
  • Evaluación de calidad de embeddings
  • Métricas de retrieval (nDCG, MRR, Recall@K)
  • Proyecto: Sistema RAG con chunking inteligente

De comparación de modelos a implementación de RAG.

Módulo 3 completado ✅ Benchmark framework: eligiendo el modelo correcto con datos

← AnteriorDomain-Specific y Multilingual EmbeddingsSiguiente →Introducción al Módulo 4: Evaluación y Chunking Strategies