| 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. Use when this capability is needed. |
| metadata | {"author":"ericgrill"} |
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
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)
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."""
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]
() -> []:
get_embedding(
text,
model=,
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."""
if "bge" in self.model_name.lower():
query =
.embed([query])[]
() -> np.ndarray:
.embed(documents)
:
():
.model = SentenceTransformer(model_name)
() -> np.ndarray:
.model.encode()
() -> np.ndarray:
.model.encode()
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 =
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 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."""
text = re.sub(r'\s+', ' ', text)
text = re.sub(, , text)
text.strip()
() -> [EmbeddedDocument]:
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 = .embeddings.aembed_documents(chunks)
i, (chunk, embedding) ((chunks, embeddings)):
metadata = {: doc_id, : i}
metadata_fields:
field metadata_fields:
field doc:
metadata[field] = doc[field]
processed.append(EmbeddedDocument(
=,
document_id=doc_id,
chunk_index=i,
text=chunk,
embedding=embedding,
metadata=metadata
))
processed
:
():
.embeddings = VoyageAIEmbeddings(model=)
() -> []:
:
tree_sitter_languages
parser = tree_sitter_languages.get_parser(language)
tree = parser.parse((code, ))
chunks = []
._extract_nodes(tree.root_node, code, chunks)
chunks
ImportError:
[{: code, : }]
():
node. [, , ]:
text = source_code[node.start_byte:node.end_byte]
chunks.append({
: text,
: node.,
: ._get_name(node),
: node.start_point[],
: node.end_point[]
})
child node.children:
._extract_nodes(child, source_code, chunks)
() -> :
child node.children:
child. == child. == :
child.text.decode()
() -> []:
context:
combined =
:
combined = chunk
.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]],
retrieved_docs: List[List[str]],
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 / (relevant) relevant
() -> :
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
:
ValueError()
() -> [, [, ]]:
results = {}
model_name, embed_fn models.items():
doc_embeddings = np.array(embed_fn(texts))
retrieved_per_query = []
query queries:
query_embedding = np.array(embed_fn([query])[])
similarities = compute_embedding_similarity(
query_embedding.reshape(, -),
doc_embeddings,
metric=
)[]
top_k_indices = np.argsort(similarities)[::-][:k]
retrieved_per_query.append([(i) i top_k_indices])
relevant_docs = [[(i) i indices] indices relevant_indices]
results[model_name] = evaluate_retrieval_quality(
queries, relevant_docs, retrieved_per_query, k
)
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
Converted and distributed by TomeVault — claim your Tome and manage your conversions.