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 (2026)

Model Dimensions Max Tokens Best For

voyage-3-large 1024 32000 Claude apps (Anthropic recommended)

voyage-3 1024 32000 Claude apps, cost-effective

voyage-code-3 1024 32000 Code search

voyage-finance-2 1024 32000 Financial documents

voyage-law-2 1024 32000 Legal documents

text-embedding-3-large 3072 8191 OpenAI apps, high accuracy

text-embedding-3-small 1536 8191 OpenAI apps, cost-effective

bge-large-en-v1.5 1024 512 Open source, local deployment

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: Voyage AI Embeddings (Recommended for Claude)

from langchain_voyageai import VoyageAIEmbeddings from typing import List import os

Initialize Voyage AI embeddings (recommended by Anthropic for Claude)

embeddings = VoyageAIEmbeddings( model="voyage-3-large", voyage_api_key=os.environ.get("VOYAGE_API_KEY") )

def get_embeddings(texts: List[str]) -> List[List[float]]: """Get embeddings from Voyage AI.""" return embeddings.embed_documents(texts)

def get_query_embedding(query: str) -> List[float]: """Get single query embedding.""" return embeddings.embed_query(query)

Specialized models for domains

code_embeddings = VoyageAIEmbeddings(model="voyage-code-3") finance_embeddings = VoyageAIEmbeddings(model="voyage-finance-2") legal_embeddings = VoyageAIEmbeddings(model="voyage-law-2")

Template 2: 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 with optional dimension reduction.""" # 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:
        # Matryoshka dimensionality reduction
        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 Matryoshka embeddings

def get_reduced_embedding(text: str, dimensions: int = 512) -> List[float]: """Get embedding with reduced dimensions (Matryoshka).""" return get_embedding( text, model="text-embedding-3-small", dimensions=dimensions )

Template 3: 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)
    self.model_name = model_name

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 appropriate prefix for retrieval models."""
    # BGE and similar models benefit from query prefix
    if "bge" in self.model_name.lower():
        query = f"Represent this sentence for searching relevant passages: {query}"
    return self.embed([query])[0]

def embed_documents(self, documents: List[str]) -> np.ndarray:
    """Embed documents for indexing."""
    return self.embed(documents)

E5 model with instructions

class E5Embedder: def init(self, model_name: str = "intfloat/multilingual-e5-large"): self.model = SentenceTransformer(model_name)

def embed_query(self, query: str) -> np.ndarray:
    """E5 requires 'query:' prefix for queries."""
    return self.model.encode(f"query: {query}")

def embed_document(self, document: str) -> np.ndarray:
    """E5 requires 'passage:' prefix for documents."""
    return self.model.encode(f"passage: {document}")

Template 4: 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 = 0

    current_chunk.append(sentence)
    current_size += sentence_size

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

return chunks

def chunk_by_semantic_sections( text: str, headers_pattern: str = r'^#{1,3}\s+.+$' ) -> List[Tuple[str, str]]: """Chunk markdown by headers, preserving hierarchy.""" lines = text.split('\n') chunks = [] current_header = "" current_content = []

for line in lines:
    if re.match(headers_pattern, line, re.MULTILINE):
        if current_content:
            chunks.append((current_header, '\n'.join(current_content)))
        current_header = line
        current_content = []
    else:
        current_content.append(line)

if current_content:
    chunks.append((current_header, '\n'.join(current_content)))

return chunks

def recursive_character_splitter( text: str, chunk_size: int = 1000, chunk_overlap: int = 200, separators: List[str] = None ) -> List[str]: """LangChain-style recursive splitter.""" separators = separators or ["\n\n", "\n", ". ", " ", ""]

def split_text(text: str, separators: List[str]) -> List[str]:
    if not text:
        return []

    separator = separators[0]
    remaining_separators = separators[1:]

    if separator == "":
        # Character-level split
        return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size - chunk_overlap)]

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

    for split in splits:
        split_length = len(split) + len(separator)

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

            # Recursively split if still too large
            if len(chunk_text) > chunk_size and remaining_separators:
                chunks.extend(split_text(chunk_text, remaining_separators))
            else:
                chunks.append(chunk_text)

            # Start new chunk with overlap
            overlap_splits = []
            overlap_length = 0
            for s in reversed(current_chunk):
                if overlap_length + len(s) <= chunk_overlap:
                    overlap_splits.insert(0, s)
                    overlap_length += len(s)
                else:
                    break
            current_chunk = overlap_splits
            current_length = overlap_length

        current_chunk.append(split)
        current_length += split_length

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

    return chunks

return split_text(text, separators)

Template 5: Domain-Specific Embedding Pipeline

import re from typing import List, Optional from dataclasses import dataclass

@dataclass class EmbeddedDocument: id: str document_id: str chunk_index: int text: str embedding: List[float] metadata: dict

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

def __init__(
    self,
    embedding_model: str = "voyage-3-large",
    chunk_size: int = 512,
    chunk_overlap: int = 50,
    preprocessing_fn=None
):
    self.embeddings = VoyageAIEmbeddings(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 (customize for your domain)
    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: Optional[List[str]] = None
) -> List[EmbeddedDocument]:
    """Process documents for vector storage."""
    processed = []

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

        # Preprocess
        cleaned = self.preprocess(content)

        # Chunk
        chunks = chunk_by_tokens(
            cleaned,
            self.chunk_size,
            self.chunk_overlap
        )

        # Create embeddings
        embeddings = await self.embeddings.aembed_documents(chunks)

        # Create records
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            metadata = {"document_id": doc_id, "chunk_index": i}

            # Add specified metadata fields
            if metadata_fields:
                for field in metadata_fields:
                    if field in doc:
                        metadata[field] = doc[field]

            processed.append(EmbeddedDocument(
                id=f"{doc_id}_chunk_{i}",
                document_id=doc_id,
                chunk_index=i,
                text=chunk,
                embedding=embedding,
                metadata=metadata
            ))

    return processed

Code-specific pipeline

class CodeEmbeddingPipeline: """Specialized pipeline for code embeddings."""

def __init__(self):
    # Use Voyage's code-specific model
    self.embeddings = VoyageAIEmbeddings(model="voyage-code-3")

def chunk_code(self, code: str, language: str) -> List[dict]:
    """Chunk code by functions/classes using tree-sitter."""
    try:
        import tree_sitter_languages
        parser = tree_sitter_languages.get_parser(language)
        tree = parser.parse(bytes(code, "utf8"))

        chunks = []
        # Extract function and class definitions
        self._extract_nodes(tree.root_node, code, chunks)
        return chunks
    except ImportError:
        # Fallback to simple chunking
        return [{"text": code, "type": "module"}]

def _extract_nodes(self, node, source_code: str, chunks: list):
    """Recursively extract function/class definitions."""
    if node.type in ['function_definition', 'class_definition', 'method_definition']:
        text = source_code[node.start_byte:node.end_byte]
        chunks.append({
            "text": text,
            "type": node.type,
            "name": self._get_name(node),
            "start_line": node.start_point[0],
            "end_line": node.end_point[0]
        })
    for child in node.children:
        self._extract_nodes(child, source_code, chunks)

def _get_name(self, node) -> str:
    """Extract name from function/class node."""
    for child in node.children:
        if child.type == 'identifier' or child.type == 'name':
            return child.text.decode('utf8')
    return "unknown"

async def embed_with_context(
    self,
    chunk: str,
    context: str = ""
) -> List[float]:
    """Embed code with surrounding context."""
    if context:
        combined = f"Context: {context}\n\nCode:\n{chunk}"
    else:
        combined = chunk
    return await self.embeddings.aembed_query(combined)

Template 6: Embedding Quality Evaluation

import numpy as np from typing import List, Dict

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[str, float]: """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 if k > 0 else 0

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 mrr(relevant: set, retrieved: List[str]) -> float:
    for i, doc in enumerate(retrieved):
        if doc in relevant:
            return 1 / (i + 1)
    return 0

def ndcg_at_k(relevant: set, retrieved: List[str], k: int) -> float:
    dcg = sum(
        1 / np.log2(i + 2) if doc in relevant else 0
        for i, doc in enumerate(retrieved[:k])
    )
    ideal_dcg = sum(1 / np.log2(i + 2) for i in range(min(len(relevant), k)))
    return dcg / ideal_dcg if ideal_dcg > 0 else 0

metrics = {
    f"precision@{k}": [],
    f"recall@{k}": [],
    "mrr": [],
    f"ndcg@{k}": []
}

for relevant, retrieved in zip(relevant_docs, retrieved_docs):
    relevant_set = set(relevant)
    metrics[f"precision@{k}"].append(precision_at_k(relevant_set, retrieved, k))
    metrics[f"recall@{k}"].append(recall_at_k(relevant_set, retrieved, k))
    metrics["mrr"].append(mrr(relevant_set, retrieved))
    metrics[f"ndcg@{k}"].append(ndcg_at_k(relevant_set, retrieved, k))

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

def compute_embedding_similarity( embeddings1: np.ndarray, embeddings2: np.ndarray, metric: str = "cosine" ) -> np.ndarray: """Compute similarity matrix between embedding sets.""" if metric == "cosine": # Normalize and compute dot product norm1 = embeddings1 / np.linalg.norm(embeddings1, axis=1, keepdims=True) norm2 = embeddings2 / np.linalg.norm(embeddings2, axis=1, keepdims=True) return norm1 @ norm2.T elif metric == "euclidean": from scipy.spatial.distance import cdist return -cdist(embeddings1, embeddings2, metric='euclidean') elif metric == "dot": return embeddings1 @ embeddings2.T else: raise ValueError(f"Unknown metric: {metric}")

def compare_embedding_models( texts: List[str], models: Dict[str, callable], queries: List[str], relevant_indices: List[List[int]], k: int = 5 ) -> Dict[str, Dict[str, float]]: """Compare multiple embedding models on retrieval quality.""" results = {}

for model_name, embed_fn in models.items():
    # Embed all texts
    doc_embeddings = np.array(embed_fn(texts))

    retrieved_per_query = []
    for query in queries:
        query_embedding = np.array(embed_fn([query])[0])
        # Compute similarities
        similarities = compute_embedding_similarity(
            query_embedding.reshape(1, -1),
            doc_embeddings,
            metric="cosine"
        )[0]
        # Get top-k indices
        top_k_indices = np.argsort(similarities)[::-1][:k]
        retrieved_per_query.append([str(i) for i in top_k_indices])

    # Convert relevant indices to string IDs
    relevant_docs = [[str(i) for i in indices] for indices in relevant_indices]

    results[model_name] = evaluate_retrieval_quality(
        queries, relevant_docs, retrieved_per_query, k
    )

return results

Best Practices

Do's

• Match model to use case: Code vs prose vs multilingual

• Chunk thoughtfully: Preserve semantic boundaries

• Normalize embeddings: For cosine similarity search

• Batch requests: More efficient than one-by-one

• Cache embeddings: Avoid recomputing for static content

• Use Voyage AI for Claude apps: Recommended by Anthropic

Don'ts

• Don't ignore token limits: Truncation loses information

• Don't mix embedding models: Incompatible vector spaces

• Don't skip preprocessing: Garbage in, garbage out

• Don't over-chunk: Lose important context

• Don't forget metadata: Essential for filtering and debugging