埋め込み戦略

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

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

• RAG 向けの埋め込みモデル選択
• チャンキング戦略の最適化
• ドメイン固有の埋め込みの微調整
• 埋め込みモデルパフォーマンスの比較
• 埋め込み次元の削減
• 多言語コンテンツの処理

コア概念

1. 埋め込みモデル比較 (2026)

モデル	次元数	最大トークン	最適用途
voyage-3-large	1024	32000	Claude アプリ (Anthropic 推奨)
voyage-3	1024	32000	Claude アプリ、コスト効率的
voyage-code-3	1024	32000	コード検索
voyage-finance-2	1024	32000	金融ドキュメント
voyage-law-2	1024	32000	法律ドキュメント
text-embedding-3-large	3072	8191	OpenAI アプリ、高精度
text-embedding-3-small	1536	8191	OpenAI アプリ、コスト効率的
bge-large-en-v1.5	1024	512	オープンソース、ローカルデプロイ
all-MiniLM-L6-v2	384	256	高速、軽量
multilingual-e5-large	1024	512	多言語対応

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

ドキュメント → チャンキング → 前処理 → 埋め込みモデル → ベクトル
            ↓
   [オーバーラップ、サイズ]  [クリーン、正規化]  [API/ローカル]

テンプレート

テンプレート 1: Voyage AI 埋め込み (Claude 向け推奨)

from langchain_voyageai import VoyageAIEmbeddings
from typing import List
import os

# Voyage AI 埋め込みを初期化 (Anthropic による 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]]:
    """Voyage AI から埋め込みを取得。"""
    return embeddings.embed_documents(texts)

def get_query_embedding(query: str) -> List[float]:
    """単一クエリの埋め込みを取得。"""
    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")

テンプレート 2: 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]]:
    """次元削減オプション付きで OpenAI から埋め込みを取得。"""
    # 大規模リストのバッチ処理
    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 次元削減
            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]:
    """単一の埋め込みを取得。"""
    return get_embeddings([text], **kwargs)[0]


# Matryoshka 埋め込みを使用した次元削減
def get_reduced_embedding(text: str, dimensions: int = 512) -> List[float]:
    """次元を削減した埋め込みを取得 (Matryoshka)。"""
    return get_embedding(
        text,
        model="text-embedding-3-small",
        dimensions=dimensions
    )

テンプレート 3: Sentence Transformers によるローカル埋め込み

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

class LocalEmbedder:
    """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:
        """正規化オプション付きでテキストを埋め込み。"""
        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:
        """検索モデル向けの適切なプレフィックス付きでクエリを埋め込み。"""
        # BGE およびのような似たモデルはクエリプレフィックスが有効
        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:
        """インデックス化用にドキュメントを埋め込み。"""
        return self.embed(documents)


# 指示付き E5 モデル
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 はクエリに 'query:' プレフィックスが必須。"""
        return self.model.encode(f"query: {query}")

    def embed_document(self, document: str) -> np.ndarray:
        """E5 はドキュメントに 'passage:' プレフィックスが必須。"""
        return self.model.encode(f"passage: {document}")

テンプレート 4: チャンキング戦略

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]:
    """トークン数でテキストをチャンク。"""
    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]:
    """文でテキストをチャンク、サイズ制限を尊重。"""
    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]]:
    """Markdown をヘッダーでチャンク、階層を保持。"""
    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 スタイルの再帰的スプリッター。"""
    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 == "":
            # 文字レベルのスプリット
            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)

                # まだ大きすぎる場合は再帰的にスプリット
                if len(chunk_text) > chunk_size and remaining_separators:
                    chunks.extend(split_text(chunk_text, remaining_separators))
                else:
                    chunks.append(chunk_text)

                # オーバーラップで新しいチャンクを開始
                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)

テンプレート 5: ドメイン別埋め込みパイプライン

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:
    """ドメイン別埋め込み用パイプライン。"""

    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:
        """デフォルト前処理。"""
        # 過剰なホワイトスペースを削除
        text = re.sub(r'\s+', ' ', text)
        # 特殊文字を削除 (ドメイン向けにカスタマイズ)
        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]:
        """ベクトルストレージ用にドキュメントを処理。"""
        processed = []

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

            # 前処理
            cleaned = self.preprocess(content)

            # チャンク化
            chunks = chunk_by_tokens(
                cleaned,
                self.chunk_size,
                self.chunk_overlap
            )

            # 埋め込みを作成
            embeddings = await self.embeddings.aembed_documents(chunks)

            # レコードを作成
            for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
                metadata = {"document_id": doc_id, "chunk_index": i}

                # 指定したメタデータフィールドを追加
                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


# コード別パイプライン
class CodeEmbeddingPipeline:
    """コード埋め込み用の特化したパイプライン。"""

    def __init__(self):
        # Voyage のコード特化モデルを使用
        self.embeddings = VoyageAIEmbeddings(model="voyage-code-3")

    def chunk_code(self, code: str, language: str) -> List[dict]:
        """tree-sitter を使用して関数/クラスでコードをチャンク。"""
        try:
            import tree_sitter_languages
            parser = tree_sitter_languages.get_parser(language)
            tree = parser.parse(bytes(code, "utf8"))

            chunks = []
            # 関数とクラス定義を抽出
            self._extract_nodes(tree.root_node, code, chunks)
            return chunks
        except ImportError:
            # シンプルなチャンキングにフォールバック
            return [{"text": code, "type": "module"}]

    def _extract_nodes(self, node, source_code: str, chunks: list):
        """関数/クラス定義を再帰的に抽出。"""
        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:
        """関数/クラスノードから名前を抽出。"""
        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]:
        """周囲のコンテキスト付きでコードを埋め込み。"""
        if context:
            combined = f"Context: {context}\n\nCode:\n{chunk}"
        else:
            combined = chunk
        return await self.embeddings.aembed_query(combined)

テンプレート 6: 埋め込み品質評価

import numpy as np
from typing import List, Dict

def evaluate_retrieval_quality(
    queries: List[str],
    relevant_docs: List[List[str]],  # クエリごとの関連ドキュメント ID リスト
    retrieved_docs: List[List[str]],  # クエリごとの取得ドキュメント ID リスト
    k: int = 10
) -> Dict[str, float]:
    """埋め込み品質を検索向けに評価。"""

    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:
    """埋め込みセット間の類似度行列を計算。"""
    if metric == "cosine":
        # 正規化してドット積を計算
        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]]:
    """複数の埋め込みモデルを検索品質で比較。"""
    results = {}

    for model_name, embed_fn in models.items():
        # すべてのテキストを埋め込み
        doc_embeddings = np.array(embed_fn(texts))

        retrieved_per_query = []
        for query in queries:
            query_embedding = np.array(embed_fn([query])[0])
            # 類似度を計算
            similarities = compute_embedding_similarity(
                query_embedding.reshape(1, -1),
                doc_embeddings,
                metric="cosine"
            )[0]
            # 上位 k インデックスを取得
            top_k_indices = np.argsort(similarities)[::-1][:k]
            retrieved_per_query.append([str(i) for i in top_k_indices])

        # 関連インデックスを文字列 ID に変換
        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

ベストプラクティス

すべき事

• ユースケースにモデルを合わせる: コード vs 散文 vs 多言語
• 思慮深くチャンク化: セマンティック境界を保持
• 埋め込みを正規化: コサイン類似度検索向け
• リクエストをバッチ処理: 1 つ 1 つより効率的
• 埋め込みをキャッシュ: 静的コンテンツの再計算を回避
• Claude アプリに Voyage AI を使用: Anthropic による推奨

すべきでない事

• トークン制限を無視しない: 切り詰めで情報が失われる
• 埋め込みモデルを混ぜない: 互換性のないベクトル空間
• 前処理をスキップしない: ゴミ入れればゴミ出し
• 過剰にチャンク化しない: 重要なコンテキストを失う
• メタデータを忘れない: フィルタリングとデバッグに必須