embedding-strategies

埋め込み戦略

ベクトル検索アプリケーション向けの埋め込みモデルの選定と最適化に関するガイドです。

このスキルを使用しない場合

• タスクが埋め込み戦略に関係しない場合
• この範囲外の異なる分野やツールが必要な場合

手順

• 目標、制約、必要な入力を明確にします。
• 関連するベストプラクティスを適用し、結果を検証します。
• 実行可能な手順と検証方法を提示します。
• 詳細な例が必要な場合は、resources/implementation-playbook.md を開きます。

このスキルを使用する場合

• RAG 用の埋め込みモデルを選ぶ場合
• チャンク分割戦略を最適化する場合
• 特定の分野に合わせて埋め込みをファインチューニングする場合
• 埋め込みモデルの性能を比較する場合
• 埋め込みの次元数を削減する場合
• 多言語コンテンツを扱う場合

主な概念

モデル一覧は過去の候補リストであり、最新のランキングではありません。使用前に、モデルカードで正確なトークン制限、次元数、ライセンス、リビジョン、クエリ／ドキュメントのプレフィックスを確認してください。トークン数は、そのモデルのトークナイザーで数えてください。機密文書の転送やプロバイダーの費用について承認を得てください。元のテキストとソース内のオフセットを保持してください。記号を取り除くと、コード、通貨記号、数式、識別子が損なわれることがあります。

1. 埋め込みモデルの比較

モデル	次元数	最大トークン数	最適な用途
text-embedding-3-large	3072	8191	高精度
text-embedding-3-small	1536	8191	コスト効率
voyage-2	1024	4000	コード、法律
bge-large-en-v1.5	1024	512	オープンソース
all-MiniLM-L6-v2	384	256	高速、軽量
multilingual-e5-large	1024	512	多言語

2. 埋め込みパイプライン

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

テンプレート

テンプレート 1：OpenAI の埋め込み

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: str, dimensions: int = 512) -> List[float]:
    """Get embedding with reduced dimensions (Matryoshka)."""
    return get_embedding(
        text,
        model="text-embedding-3-small",
        dimensions=dimensions
    )

テンプレート 2：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 = "cpu",
        query_prefix: str = ""
    ):
        self.model = SentenceTransformer(model_name, device=device)
        self.query_prefix = query_prefix  # Choose from the selected model card.

    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 the explicitly configured model prefix."""
        return self.embed([self.query_prefix + 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:
        return self.model.encode(f"query: {query}")

    def embed_document(self, document: str) -> np.ndarray:
        return self.model.encode(f"passage: {document}")

テンプレート 3：チャンク分割戦略

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."""
    if not isinstance(chunk_size, int) or not isinstance(chunk_overlap, int):
        raise ValueError("chunk size and overlap must be integers")
    if chunk_size <= 0 or not 0 <= chunk_overlap < chunk_size:
        raise ValueError("require chunk_size > 0 and 0 <= overlap < chunk_size")
    if tokenizer is None:
        import tiktoken
        tokenizer = 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)
        if end >= len(tokens):
            break
        start = end - chunk_overlap

    return chunks


def chunk_by_sentences(
    text: str,
    max_chunk_size: int = 1000,
    min_chunk_size: int = 100
) -> List[str]:
    """Group sentences by approximate character length, not tokenizer length.

    min_chunk_size is not enforced; oversized individual sentences require a
    second token-bounded splitting pass before embedding. NLTK resources must
    already be installed; no implicit network download is supplied here.
    """
    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]:
    """Bounded character chunks, preferring a separator inside each window."""
    if chunk_size <= 0 or not 0 <= chunk_overlap < chunk_size:
        raise ValueError("require chunk_size > 0 and 0 <= overlap < chunk_size")
    separators = separators or ["\n\n", "\n", ". ", " "]
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        if end < len(text):
            for separator in separators:
                if not separator:
                    continue
                boundary = text.rfind(separator, start + chunk_overlap + 1, end)
                if boundary >= 0:
                    end = boundary + len(separator)
                    break
        chunks.append(text[start:end])
        if end == len(text):
            break
        start = end - chunk_overlap
    return chunks

テンプレート 4：特定分野向け埋め込みパイプライン

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)
        return text.strip()

    def process_documents(
        self,
        documents: List[dict],
        id_field: str = "id",
        content_field: str = "content",
        metadata_fields: List[str] = None
    ) -> List[dict]:
        """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 = get_embeddings(chunks, self.embedding_model)

            # Create records
            for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
                record = {
                    "id": f"{doc_id}_chunk_{i}",
                    "document_id": doc_id,
                    "chunk_index": i,
                    "text": chunk,
                    "embedding": embedding
                }

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

                processed.append(record)

        return processed


# Code-specific pipeline
class CodeEmbeddingPipeline:
    """Specialized pipeline for code embeddings."""

    def __init__(self, embed_fn):
        self.embed_fn = embed_fn  # Supply a reviewed provider-specific embedding adapter.

    def chunk_code(self, code: str, language: str) -> List[dict]:
        """Chunk code by functions/classes."""
        import tree_sitter

        # Parse with tree-sitter
        # Extract functions, classes, methods
        # Return chunks with context
        raise NotImplementedError("Supply the installed parser and language grammar")

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

テンプレート 5：埋め込み品質の評価

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

    if k <= 0 or not queries or len(queries) != len(relevant_docs) or len(queries) != len(retrieved_docs):
        raise ValueError("Require positive k and equal nonempty query/relevance/result lists")
    if any(len(ids) != len(set(ids)) for ids in retrieved_docs):
        raise ValueError("Deduplicate retrieved IDs before ranking metrics")
    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
        lengths1 = np.linalg.norm(embeddings1, axis=1, keepdims=True)
        lengths2 = np.linalg.norm(embeddings2, axis=1, keepdims=True)
        if np.any(lengths1 == 0) or np.any(lengths2 == 0):
            raise ValueError("Cosine similarity is undefined for zero vectors")
        norm1 = embeddings1 / lengths1
        norm2 = embeddings2 / lengths2
        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
    raise ValueError(f"Unsupported metric: {metric}")

ベストプラクティス

推奨事項

• 用途に合ったモデルを選ぶ - コードか、文章か、多言語か
• 慎重にチャンク分割する - 意味上の境界を保つ
• 埋め込みを正規化する - コサイン類似度を使う場合
• リクエストをバッチ処理する - 1 件ずつ処理するより効率的
• 埋め込みをキャッシュする - 再計算を避ける

避けるべきこと

• トークン制限を無視しない - 切り捨てると情報が失われます
• 異なる埋め込みモデルを混在させない - ベクトル空間に互換性がありません
• 前処理を省略しない - 入力が悪ければ出力も悪くなります
• 過剰にチャンク分割しない - 文脈が失われます

リソース

• OpenAI の埋め込み
• Sentence Transformers
• MTEB ベンチマーク

制限事項

• タスクが上記の範囲に明確に当てはまる場合にのみ、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認してください。