| 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 (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
Resources