会社情報
採用情報
お問い合わせ
Article

Embedding Model選定ガイドと性能比較

高田晃太郎
Embedding Model選定ガイドと性能比較

100万チャンクを1536次元・float32で格納すると、生のベクトルだけで約6.1GBになります¹。ここにHNSWのグラフやメタデータが重なると、メモリはさらに増えます。ベクトル次元が増えるほど内積計算はほぼ比例して重くなるため²、選ぶ埋め込みモデル（Embedding Model）は推論コストとレイテンシに直結します。検索・推薦の研究コミュニティでは、多くの領域でタスク特化の埋め込みが有利と報告される一方³、一般モデルが勝るタスクも確認されています⁴。RAGでも同様に、モデルの選定とインデックス設計が回答精度とユーザー待ち時間のバランスを左右します。本稿では、CTOやリーダーが意思決定できる粒度で、比較観点、実装、評価法までを具体的に整理します。

なお、容量を抑えたいだけならfloat16で約半分、int8相当の量子化でさらに圧縮できる可能性がありますが、精度と再現率の変化は必ず実データで確認してください²。

何を基準にEmbeddingを選ぶか：RAG特性とモデル設計

RAGの埋め込みには、質問とパッセージの意味距離が安定して測れること（同義の表現でも近づく）、言語やドメインのズレに強いこと（多言語や専門用語に頑健）、そして現実的なコストで運用できることが求められます。一般に、検索用途で事前に指示最適化されたモデルは質問と文書で別々の表現空間を最適化しており、クエリ側に専用のプレフィックスを付ける設計が多く見られます。たとえばe5系はquery:とpassage:の指示を前置する前提で学習されており、無指示で使うと再現率が下がることがあります⁵。類似度の取り方は前提と一致させるのが要点で、コサイン類似度（方向の近さ）を使う場合はL2正規化を揃える、内積（長さも含むスコア）で評価するなら正規化を外す、といった整合性が重要です⁶。

多言語対応は、社内ドキュメントが日英混在する場合に無視できません。多言語モデルは英語特化に比べて僅かに遅く大きい傾向はありますが、翻訳を挟まない一貫性のメリットが得られます。逆に英語強度が高いモデルを日本語で使う場合、クエリを英訳してから検索し、上位候補を再ランキングで日本語文脈に合わせる手法は現実的な選択肢です。ドメイン適応は二段構えが現実的で、まず汎用の高性能モデルでベースラインを出し、その上に再ランキングや軽量な蒸留で領域最適化を重ねると、メンテナンス性と再現性を両立しやすくなります。上位候補としては、OpenAIやCohereのAPI系、そしてJina、BGE、E5、Voyageなどのローカル・ホステッド系が実用レンジにあります。最終判断は、業務言語、法務・医療などの専門語の頻度、予算とSLA、そしてトークン上限や次元数に対するインデックスの制約を合わせて、実データでのベンチマークで下します。

補足として、次元数は「高ければ良い」とは限りません。モデルの想定次元でまず評価し、必要ならPCAや学習済みの線形射影で384〜768次元に落として再評価すると、精度をほぼ保ったままメモリとレイテンシを下げられる可能性があります（要ベンチ）。

小さく早く確かめる：再現率と安定性を測る最短ループ

まずは数百〜数千のQ&A対を用意し、Recall@k、MRR、nDCGを計測します。Recall@kは「正解が上位kに1つでも含まれる率」、MRRは「正解が上位に現れるほど高い平均順位スコア」、nDCGは「上位に正解が多いほど高い正規化順位スコア」と捉えると直感的です。クエリの難易度を揃え、チャンク分割とオーバーラップ、正規化有無、クエリ・パッセージの指示文有無を振ると、構成が与える差分が明確になります。APIモデルとローカルモデルを同じ評価器で揃えて測ることで、トレードオフが可視化されます⁷。公開ベンチマーク（BEIR/MTEB）を参考に、まずは自社QA対に近いサブセットから始めると回りが早いです⁸。

このハーネスは、API/ローカル双方の埋め込みでランキングとメトリクスをざっと比較するための最小構成です。実運用では、ここに「言語別」「カテゴリ別」などの切り口を足して分解能を上げます。

# コード例1: 埋め込み品質の簡易評価ハーネス（Recall@k, nDCG）
import time
import math
import numpy as np
from typing import List, Dict, Tuple

try:
    from openai import OpenAI
    openai_client = OpenAI()
except Exception:
    openai_client = None

try:
    from sentence_transformers import SentenceTransformer
except Exception:
    SentenceTransformer = None

def l2_normalize(x: np.ndarray) -> np.ndarray:
    norm = np.linalg.norm(x, axis=1, keepdims=True) + 1e-12
    return x / norm

def embed_api(texts: List[str], model: str = "text-embedding-3-small", normalize: bool = True) -> np.ndarray:
    if openai_client is None:
        raise RuntimeError("OpenAI client not configured")
    resp = openai_client.embeddings.create(model=model, input=texts)
    arr = np.array([d.embedding for d in resp.data], dtype=np.float32)
    return l2_normalize(arr) if normalize else arr

def embed_local(texts: List[str], model_name: str = "intfloat/multilingual-e5-large", normalize: bool = True) -> np.ndarray:
    if SentenceTransformer is None:
        raise RuntimeError("sentence-transformers not available")
    model = SentenceTransformer(model_name)
    vecs = model.encode(texts, batch_size=64, normalize_embeddings=normalize)
    return vecs.astype(np.float32)

def recall_at_k(rankings: List[List[int]], ground_truth: Dict[int, List[int]], k: int = 10) -> float:
    hit = 0
    for qi, cand in enumerate(rankings):
        gt = set(ground_truth.get(qi, []))
        hit += 1 if len(gt.intersection(cand[:k])) > 0 else 0
    return hit / len(rankings)

def ndcg_at_k(rankings: List[List[int]], ground_truth: Dict[int, List[int]], k: int = 10) -> float:
    def dcg(rel):
        return sum(r / math.log2(i + 2) for i, r in enumerate(rel))
    scores = []
    for qi, cand in enumerate(rankings):
        gt = set(ground_truth.get(qi, []))
        rel = [1 if d in gt else 0 for d in cand[:k]]
        ideal = sorted(rel, reverse=True)
        scores.append(dcg(rel) / (dcg(ideal) or 1))
    return float(np.mean(scores))

# ダミーデータ例（実データでは自社QA対に置き換え）
queries = ["query: 労務規程の休暇付与", "query: passage chunking ベストプラクティス"]
passages = ["passage: 年次有給休暇の…", "passage: チャンクは重なり…", "passage: unrelated"]

# 埋め込み取得（APIとローカルのいずれかを使用）
q_vecs = embed_local(queries, normalize=True)
p_vecs = embed_local(passages, normalize=True)

# コサイン類似度でランク付け（L2正規化済みなので内積=コサイン）
sim = q_vecs @ p_vecs.T
rankings = [list(np.argsort(-sim[i]).tolist()) for i in range(len(queries))]

# 疑似の正解集合（各クエリに対して該当パッセージのインデックス）
ground_truth = {0: [0], 1: [1]}
print({
    "recall@10": recall_at_k(rankings, ground_truth, k=2),
    "ndcg@10": ndcg_at_k(rankings, ground_truth, k=2)
})

コストとレイテンシを支配するもの：次元数、量子化、インデックス

次元数d、件数N、データ型のバイト数bが決まると、ベクトルだけでN×d×bのメモリが必要です。1536次元・float32なら1ベクトルあたり約6KB、100万件で約6.1GBです¹。ここにHNSWのリンク（M、efConstruction）やメタデータ、フィルタ用の倒立インデックスが積み増しされます。HNSWのMは各ノードの近傍数、efConstructionは構築時の探索幅を表し、一般にMやefを上げるとRecallが上がる代わりにメモリと構築時間、検索時のefSearchを上げるとレイテンシが伸びます。APIベースのモデルは埋め込み生成時に推論コストが発生し、ローカル推論はGPUメモリとスループットで費用対効果が決まります⁷。総コストは、初回バッチ埋め込み、増分の再埋め込み、ストレージ、クエリ当たりのレイテンシSLAのためのレプリカ数で見積もると全体像が掴めます。

メモリ圧縮は量子化が有力です。Scalar QuantizationやPQ（Product Quantization）を使うとメモリとスループットが改善しますが、類似度の歪みによる微小な精度低下が生じるため、Recallの変化を許容範囲で評価します²。インデックスはHNSWが運用の素直さで広く使われ、フィルタ条件が多いならIVF+HNSWやDiskANN系も検討に値します¹。スコアはコサインでも内積でもよく、学習時の前提と合わせることが重要です⁶。必要に応じて、学習済み空間を壊さない範囲で低次元へ射影することで、インデックスサイズと検索レイテンシを同時に抑えられるケースもあります。

以下は、FAISSでHNSWと量子化（SQ/PQ相当）を使った簡易比較の骨子です。実データの一部で学習し、上位の一致率と速度を見ます。

# コード例2: FAISSでHNSWと量子化（SQ/PQ）を用いたインデックス
import faiss
import numpy as np

# 学習用サンプル（実データの一部で学習）
d = 768
nb = 20000
xb = np.random.randn(nb, d).astype('float32')

# L2正規化済みコサイン近似（内積）を想定
faiss.normalize_L2(xb)

# HNSW Flat（高精度・中メモリ）
hnsw = faiss.IndexHNSWFlat(d, 32)
hnsw.hnsw.efConstruction = 200
hnsw.add(xb)

# Scalar Quantizer（メモリ削減）
quantizer = faiss.IndexFlatIP(d)
sq = faiss.IndexIVFScalarQuantizer(quantizer, d, 1024, faiss.ScalarQuantizer.QT_8bit, faiss.METRIC_INNER_PRODUCT)
sq.nprobe = 16
sq.train(xb)
sq.add(xb)

# 検索の比較（上位一致や速度を別途計測してトレードオフを見る）
xq = np.random.randn(10, d).astype('float32')
faiss.normalize_L2(xq)
D1, I1 = hnsw.search(xq, 10)
D2, I2 = sq.search(xq, 10)
print({"hnsw_top1": I1[:,0].tolist()[:3], "sq_top1": I2[:,0].tolist()[:3]})

Qdrantを使う場合は、HNSWと量子化の設定をコレクション単位で与えられます。次の例は、コサイン距離でのHNSW＋INT8量子化を使い、属性フィルタと併用する最小構成です。

# コード例3: QdrantでHNSW + 量子化の実運用設定
from qdrant_client import QdrantClient
from qdrant_client.http import models as qm
import numpy as np

client = QdrantClient(host="localhost", port=6333)

client.recreate_collection(
    collection_name="docs",
    vectors_config=qm.VectorParams(size=768, distance=qm.Distance.COSINE),
    hnsw_config=qm.HnswConfig(m=32, ef_construct=200),
    optimizers_config=qm.OptimizersConfigDiff(memmap_threshold=20000),
    quantization_config=qm.ScalarQuantization(
        scalar=qm.ScalarQuantizationConfig(type=qm.ScalarType.INT8, always_ram=True)
    )
)

vecs = np.random.randn(1000, 768).astype("float32")
vecs = vecs / (np.linalg.norm(vecs, axis=1, keepdims=True) + 1e-9)
points = [qm.PointStruct(id=i, vector=vecs[i].tolist(), payload={"dept": "hr"}) for i in range(len(vecs))]
client.upsert(collection_name="docs", points=points)

hits = client.search(collection_name="docs", query_vector=vecs[0].tolist(), limit=5, query_filter=qm.Filter(must=[qm.FieldCondition(key="dept", match=qm.MatchValue(value="hr"))]))
print([h.id for h in hits])

多言語・ドメイン適応・ハイブリッド：実装の勘所

多言語環境では、翻訳を挟まずにそのまま検索できることが保守の簡潔さにつながります。英語強度が高いモデルを日本語で使う場合、クエリを英訳してから英語で検索し、再ランキングで日本語文脈を補う手法も現実解です。ドメイン適応は、まず汎用モデルで運用を開始し、トップkの再ランキングで品質を底上げし、余裕ができた時点で蒸留や軽量ファインチューニングに進むと安全です。再ランキングはクロスエンコーダ（クエリとパッセージを連結して文脈を評価するモデル）を用いて上位候補の局所順序を最適化するため、長文や曖昧なクエリに効きやすい一方で、kとモデルサイズに比例してレイテンシが増えるので、A/BでP90の遅延とRecallの積を見ながら最適点を探ります⁹。

以下は、e5系の指示最適化と素朴なチャンク分割を組み合わせた最小例です。実データでは、文単位・段落単位の境界を考慮しながらサイズとオーバーラップを調整します。

# コード例4: e5系の指示最適化とチャンク分割
from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("intfloat/multilingual-e5-large")

def chunk_text(text: str, size: int = 500, overlap: int = 100):
    chunks = []
    i = 0
    while i < len(text):
        chunks.append(text[i:i+size])
        i += size - overlap
    return chunks

passages = [
    "社内労務規程の抜粋……",
    "RAGの設計メモ……",
]

p_inputs = [f"passage: {c}" for p in passages for c in chunk_text(p, 500, 100)]
p_vecs = model.encode(p_inputs, batch_size=64, normalize_embeddings=True)

query = "休暇の前倒し付与は可能ですか"
q_vec = model.encode([f"query: {query}"], normalize_embeddings=True)

scores = (q_vec @ p_vecs.T)[0]
rank = np.argsort(-scores)
print(rank[:5])

BM25とベクトル検索のハイブリッドは、キーワードが効くケースと同義表現が効くケースを両取りできます。OpenSearchではバージョンによりdense_vectorではなくknn_vectorタイプの設定が必要なことがある点に注意してください（knnクエリのパラメータ名も差分があります）。

# コード例5: OpenSearchでBM25とベクトルのハイブリッド検索
from opensearchpy import OpenSearch

client = OpenSearch(hosts=[{"host": "localhost", "port": 9200}])

mapping = {
  "mappings": {
    "properties": {
      "text": {"type": "text"},
      "vec": {"type": "dense_vector", "dims": 768, "index": True, "similarity": "cosine"}
    }
  }
}

idx = "docs"
if client.indices.exists(idx):
    client.indices.delete(idx)
client.indices.create(idx, body=mapping)

# インデクシング（省略: 事前にvecを生成して格納）

query = {
  "query": {
    "bool": {
      "should": [
        {"match": {"text": "休暇 付与"}},
        {"knn": {"vec": {"vector": [0.0]*768, "k": 50}}}
      ]
    }
  },
  "rescore": {
    "window_size": 50,
    "query": {"rescore_query": {"rank_feature": {"field": "_score"}}}
  },
  "size": 10
}
res = client.search(index=idx, body=query)
print([hit["_id"] for hit in res["hits"]["hits"]])

評価設計と意思決定：KPI、ベンチ、SLAをつなぐ

プロダクトKPIとモデルKPIを接続すると、選定の腹落ちが生まれます。検索精度はRecall@kやnDCG、クリック率やサポート解決率と相関しやすく、レイテンシは問い合わせ離脱の閾値に直結します。そこで、評価データセットを毎週更新し、チャンク分割、正規化、量子化、再ランキングの各パラメータをスイープしながら、P50/P95レイテンシ（中央値・95パーセンタイル）、スループット、インデックスサイズ、月間埋め込みコストのダッシュボードを維持します。API系はレート制限と一時的なエラーに備えたリトライとジッター付きバックオフを入れて、SLA（合意した応答水準）を守るための冗長化を決めます。ローカル推論はGPU占有時間とオートスケールの相性で見ると、稼働率の平準化が効いてきます。

この非同期ベンチは、並列度を上げたときのp50/p95とスループット、リトライ挙動を一度に可視化するための最小構成です。APIキーは環境変数から読み込み、失敗時は指数バックオフで再試行します。

# コード例6: 非同期ベンチハーネス（p50/p95レイテンシ、スループット、リトライ）
import asyncio
import time
import statistics
from typing import List
import os
import httpx

API_URL = "https://api.openai.com/v1/embeddings"
MODEL = "text-embedding-3-small"
HEADERS = {"Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY','')}"}

async def embed_session(client: httpx.AsyncClient, text: str) -> List[float]:
    payload = {"model": MODEL, "input": text}
    backoff = 0.5
    for _ in range(5):
        try:
            r = await client.post(API_URL, headers=HEADERS, json=payload, timeout=30)
            if r.status_code == 200:
                return r.json()["data"][0]["embedding"]
            if r.status_code in (429, 500, 503):
                await asyncio.sleep(backoff)
                backoff *= 2
                continue
            r.raise_for_status()
        except Exception:
            await asyncio.sleep(backoff)
            backoff = min(backoff*2, 8)
    raise RuntimeError("embed failed")

async def bench(texts: List[str], concurrency: int = 16):
    latencies = []
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        async def run(t):
            async with sem:
                t0 = time.perf_counter()
                _ = await embed_session(client, t)
                latencies.append(time.perf_counter() - t0)
        await asyncio.gather(*[run(t) for t in texts])
    p50 = statistics.median(latencies)
    p95 = sorted(latencies)[int(0.95*len(latencies))-1]
    tps = len(latencies) / sum(latencies)
    return {"p50": p50, "p95": p95, "throughput": tps}

if __name__ == "__main__":
    texts = ["query: 休暇 付与"] * 64
    print(asyncio.run(bench(texts)))

ベンチの見方はシンプルに、Recallが十分か、P95がSLAに収まるか、インデックスサイズとコストが予算内かの三点を同時に満たせるかを確認します。Recallが不足する場合は、チャンクのオーバーラップ拡大やクエリ拡張で再現率を底上げしてから再ランキングのkを下げ、レイテンシを戻すのが安定した打ち手です。多言語ドメインで揺れる場合は、まず多言語モデルと翻訳＋英語特化モデルの二案を同一評価で比べ、上位案に絞って量子化や次元の削減を試し、微差であれば小さいモデルを選ぶと運用上の余白が増えます。

意思決定のチェックポイント

モデル選定は一度で終わりません。週次の評価でドリフトと回帰を監視し、失敗ケースを分析して、プレフィックス、正規化、インデックスのハイパーパラメータを都度調整します。最終的な採用は、精度・コスト・SLAの三角形の中で、将来の拡張余地が最も大きい点を選ぶのが合理的です。

まとめ：精度・コスト・SLAを一枚の絵に

埋め込みモデルの選定は、学術指標の優劣だけでは決まりません。自社データでのRecall@k、nDCG、再ランキング後の最終精度、P95レイテンシ、インデックスサイズと月次コストを同じキャンバスに重ねると、納得感のある解が見えてきます。まずはベースラインとして汎用の高品質モデルで動かし、指示と正規化を合わせ、チャンクとインデックスを整えて、必要に応じて量子化とハイブリッド検索を足します。次に、週次の評価で失敗パターンを潰しながら、より小さく速い構成へ安全に寄せていくと、精度とTCOのバランスが向上します。今日、どこから始めるかに迷うなら、この記事の評価ハーネスを自社データに差し替え、今週中に最初のベンチ結果をチームで共有してみてください。最適解は測定から始まります。

参考文献

  1. Weaviate Blog. ANN algorithms: HNSW & PQ — memory calculation examples. https://weaviate.io/blog/ann-algorithms-hnsw-pq#:~:text=Weaviate%20currently%20supports%20vectors%20of,4%20bytes%20%3D%20512%2C000%2C000%20bytes
  2. Weaviate Blog. ANN algorithms: HNSW & PQ — Product Quantization trade-offs. https://weaviate.io/blog/ann-algorithms-hnsw-pq#:~:text=As%20we%20should%20expect%2C%20Product,we%20shall%20save%20for%20later
  3. arXiv: MedEIR — a domain-specific medical embedding model (2025). https://arxiv.org/abs/2505.13482#:~:text=the%20semantic%20content%20of%20medical,MedEIR%2C%20a%20novel%20embedding%20model
  4. arXiv (2024). On constructing textual datasets; generalist vs. specialist embeddings. https://arxiv.org/abs/2401.01943#:~:text=constructing%20a%20textual%20dataset%20based,generalist%20models%20performed%20better%20than
  5. Hugging Face — multilingual-e5 README: instruction format (query:/passage:). https://huggingface.co/dragonkue/multilingual-e5-small-ko/blob/main/README.md#:~:text=1,to%20input%20texts
  6. Pinecone. The practitioner’s guide to e5 — normalization and similarity choice. https://www.pinecone.io/learn/the-practitioners-guide-to-e5/#:~:text=A%20good%20embedding%20model%20makes,well%20on%20benchmarks%20across%20languages
  7. arXiv (2024). The choice of embedding model and its impact on retrieval effectiveness and cost. https://arxiv.org/abs/2407.08275#:~:text=The%20choice%20of%20embedding%20model,only%20allows%20for%20a%20weak
  8. BEIR: A Heterogeneous Benchmark for Information Retrieval. https://arxiv.org/abs/2104.08663
  9. Eyka Blog. Reranking in RAG: enhancing accuracy with cross-encoders. https://eyka.com/blog/reranking-in-rag-enhancing-accuracy-with-cross-encoders/#:~:text=To%20achieve%20the%20best%20of,balance%20between%20efficiency%20and%20accuracy