embedding-strategies

name	embedding-strategies
description	Select and optimize embedding models for semantic search and RAG applications. Use when choosing embedding models, implementing chunking strategies, or optimizing embedding quality for specific domains.

Embedding Strategies

Guide to selecting and optimizing embedding models for vector search applications.

When to Use This Skill

• Choosing embedding models for RAG
• Optimizing chunking strategies
• Fine-tuning embeddings for domains
• Comparing embedding model performance
• Reducing embedding dimensions
• Handling multilingual content

Core Concepts

1. Embedding Model Comparison

Model	Dimensions	Max Tokens	Best For
text-embedding-3-large	3072	8191	High accuracy
text-embedding-3-small	1536	8191	Cost-effective
voyage-2	1024	4000	Code, legal
bge-large-en-v1.5	1024	512	Open source
all-MiniLM-L6-v2	384	256	Fast, lightweight
multilingual-e5-large	1024	512	Multi-language

2. Embedding Pipeline

Document → Chunking → Preprocessing → Embedding Model → Vector
                ↓
        [Overlap, Size]  [Clean, Normalize]  [API/Local]

Templates

Template 1: OpenAI Embeddings

from openai import OpenAI
from typing import List
import numpy as np

client = OpenAI()

def get_embeddings(
    texts: List[str],
    model: str = "text-embedding-3-small",
    dimensions: int = None
) -> List[List[float]]:
    """Get embeddings from OpenAI."""
    # Handle batching for large lists
    batch_size = 100
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]

        kwargs = {"input": batch, "model": model}
        if dimensions:
            kwargs["dimensions"] = dimensions

        response = client.embeddings.create(**kwargs)
        embeddings = [item.embedding for item in response.data]
        all_embeddings.extend(embeddings)

    return all_embeddings


def get_embedding(text: str, **kwargs) -> List[float]:
    """Get single embedding."""
    return get_embeddings([text], **kwargs)[0]


# Dimension reduction with OpenAI
def get_reduced_embedding(text: , dimensions:  = ) -> []:
    
     get_embedding(
        text,
        model=,
        dimensions=dimensions
    )

Template 2: Local Embeddings with Sentence Transformers

from sentence_transformers import SentenceTransformer
from typing import List, Optional
import numpy as np

class LocalEmbedder:
    """Local embedding with sentence-transformers."""

    def __init__(
        self,
        model_name: str = "BAAI/bge-large-en-v1.5",
        device: str = "cuda"
    ):
        self.model = SentenceTransformer(model_name, device=device)

    def embed(
        self,
        texts: List[str],
        normalize: bool = True,
        show_progress: bool = False
    ) -> np.ndarray:
        """Embed texts with optional normalization."""
        embeddings = self.model.encode(
            texts,
            normalize_embeddings=normalize,
            show_progress_bar=show_progress,
            convert_to_numpy=True
        )
        return embeddings

    def embed_query(self, query: str) -> np.ndarray:
        """Embed a query with BGE-style prefix."""
        # BGE models benefit from query prefix
        if "bge" in self.model.get_sentence_embedding_dimension():
            query = f"Represent this sentence for searching relevant passages: "
         .embed([query])[]

     () -> np.ndarray:
        
         .embed(documents)



 :
     ():
        .model = SentenceTransformer(model_name)

     () -> np.ndarray:
         .model.encode()

     () -> np.ndarray:
         .model.encode()

Template 3: Chunking Strategies

from typing import List, Tuple
import re

def chunk_by_tokens(
    text: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
    tokenizer=None
) -> List[str]:
    """Chunk text by token count."""
    import tiktoken
    tokenizer = tokenizer or tiktoken.get_encoding("cl100k_base")

    tokens = tokenizer.encode(text)
    chunks = []

    start = 0
    while start < len(tokens):
        end = start + chunk_size
        chunk_tokens = tokens[start:end]
        chunk_text = tokenizer.decode(chunk_tokens)
        chunks.append(chunk_text)
        start = end - chunk_overlap

    return chunks


def chunk_by_sentences(
    text: str,
    max_chunk_size: int = 1000,
    min_chunk_size: int = 100
) -> List[str]:
    """Chunk text by sentences, respecting size limits."""
    import nltk
    sentences = nltk.sent_tokenize(text)

    chunks = []
    current_chunk = []
    current_size = 0

    for sentence in sentences:
        sentence_size = len(sentence)

        if current_size + sentence_size > max_chunk_size and current_chunk:
            chunks.append(" ".join(current_chunk))
            current_chunk = []
            current_size = 

        current_chunk.append(sentence)
        current_size += sentence_size

     current_chunk:
        chunks.append(.join(current_chunk))

     chunks


 () -> [[, ]]:
    
    lines = text.split()
    chunks = []
    current_header = 
    current_content = []

     line  lines:
         re.(headers_pattern, line, re.MULTILINE):
             current_content:
                chunks.append((current_header, .join(current_content)))
            current_header = line
            current_content = []
        :
            current_content.append(line)

     current_content:
        chunks.append((current_header, .join(current_content)))

     chunks


 () -> []:
    
    separators = separators  [, , , , ]

     () -> []:
          text:
             []

        separator = separators[]
        remaining_separators = separators[:]

         separator == :
            
             [text[i:i+chunk_size]  i  (, (text), chunk_size - chunk_overlap)]

        splits = text.split(separator)
        chunks = []
        current_chunk = []
        current_length = 

         split  splits:
            split_length = (split) + (separator)

             current_length + split_length > chunk_size  current_chunk:
                chunk_text = separator.join(current_chunk)

                
                 (chunk_text) > chunk_size  remaining_separators:
                    chunks.extend(split_text(chunk_text, remaining_separators))
                :
                    chunks.append(chunk_text)

                
                overlap_splits = []
                overlap_length = 
                 s  (current_chunk):
                     overlap_length + (s) <= chunk_overlap:
                        overlap_splits.insert(, s)
                        overlap_length += (s)
                    :
                        
                current_chunk = overlap_splits
                current_length = overlap_length

            current_chunk.append(split)
            current_length += split_length

         current_chunk:
            chunks.append(separator.join(current_chunk))

         chunks

     split_text(text, separators)

Template 4: Domain-Specific Embedding Pipeline

class DomainEmbeddingPipeline:
    """Pipeline for domain-specific embeddings."""

    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        chunk_size: int = 512,
        chunk_overlap: int = 50,
        preprocessing_fn=None
    ):
        self.embedding_model = embedding_model
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.preprocess = preprocessing_fn or self._default_preprocess

    def _default_preprocess(self, text: str) -> str:
        """Default preprocessing."""
        # Remove excessive whitespace
        text = re.sub(r'\s+', ' ', text)
        # Remove special characters
        text = re.sub(r'[^\w\s.,!?-]', '', text)
        return text.strip()

    async def process_documents(
        self,
        documents: List[dict],
        id_field: str = "id",
        content_field: str = "content",
        metadata_fields: List[str] = None
    ) -> []:
        
        processed = []

         doc  documents:
            content = doc[content_field]
            doc_id = doc[id_field]

            
            cleaned = .preprocess(content)

            
            chunks = chunk_by_tokens(
                cleaned,
                .chunk_size,
                .chunk_overlap
            )

            
            embeddings = get_embeddings(chunks, .embedding_model)

            
             i, (chunk, embedding)  ((chunks, embeddings)):
                record = {
                    : ,
                    : doc_id,
                    : i,
                    : chunk,
                    : embedding
                }

                
                 metadata_fields:
                     field  metadata_fields:
                         field  doc:
                            record[field] = doc[field]

                processed.append(record)

         processed



 :
    

     ():
        .model = model

     () -> []:
        
         tree_sitter

        
        
        
        

     () -> []:
        
        combined = 
         get_embedding(combined, model=.model)

Template 5: Embedding Quality Evaluation

import numpy as np
from typing import List, Tuple

def evaluate_retrieval_quality(
    queries: List[str],
    relevant_docs: List[List[str]],  # List of relevant doc IDs per query
    retrieved_docs: List[List[str]],  # List of retrieved doc IDs per query
    k: int = 10
) -> dict:
    """Evaluate embedding quality for retrieval."""

    def precision_at_k(relevant: set, retrieved: List[str], k: int) -> float:
        retrieved_k = retrieved[:k]
        relevant_retrieved = len(set(retrieved_k) & relevant)
        return relevant_retrieved / k

    def recall_at_k(relevant: set, retrieved: List[str], k: int) -> float:
        retrieved_k = retrieved[:k]
        relevant_retrieved = len(set(retrieved_k) & relevant)
        return relevant_retrieved / len(relevant) if relevant else 0

    def () -> :
         i, doc  (retrieved):
             doc  relevant:
                  / (i + )
         

     () -> :
        dcg = (
             / np.log2(i + )  doc  relevant  
             i, doc  (retrieved[:k])
        )
        ideal_dcg = ( / np.log2(i + )  i  (((relevant), k)))
         dcg / ideal_dcg  ideal_dcg >   

    metrics = {
        : [],
        : [],
        : [],
        : []
    }

     relevant, retrieved  (relevant_docs, retrieved_docs):
        relevant_set = (relevant)
        metrics[].append(precision_at_k(relevant_set, retrieved, k))
        metrics[].append(recall_at_k(relevant_set, retrieved, k))
        metrics[].append(mrr(relevant_set, retrieved))
        metrics[].append(ndcg_at_k(relevant_set, retrieved, k))

     {name: np.mean(values)  name, values  metrics.items()}


 () -> np.ndarray:
    
     metric == :
        
        norm1 = embeddings1 / np.linalg.norm(embeddings1, axis=, keepdims=)
        norm2 = embeddings2 / np.linalg.norm(embeddings2, axis=, keepdims=)
         norm1 @ norm2.T
     metric == :
         scipy.spatial.distance  cdist
         -cdist(embeddings1, embeddings2, metric=)
     metric == :
         embeddings1 @ embeddings2.T

Best Practices

Do's

• Match model to use case - Code vs prose vs multilingual
• Chunk thoughtfully - Preserve semantic boundaries
• Normalize embeddings - For cosine similarity
• Batch requests - More efficient than one-by-one
• Cache embeddings - Avoid recomputing

Don'ts

• Don't ignore token limits - Truncation loses info
• Don't mix embedding models - Incompatible spaces
• Don't skip preprocessing - Garbage in, garbage out
• Don't over-chunk - Lose context

Resources

• OpenAI Embeddings
• Sentence Transformers
• MTEB Benchmark