Embeddingモデル選択完全ガイド2025：OpenAIからオープンソースまで

from openai import OpenAI
import numpy as np

client = OpenAI()

# テキスト → ベクトル変換
text = "人工知能が世界を変えている"
response = client.embeddings.create(
    input=text,
    model="text-embedding-3-small"
)
embedding = response.data[0].embedding
print(f"ベクトル次元数: {len(embedding)}")  # 1536

# 意味的類似度の計算
def cosine_similarity(a: list, b: list) -> float:
    a_np = np.array(a)
    b_np = np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

text1 = "機械学習はAIの一分野だ"
text2 = "ディープラーニングは人工知能技術だ"
text3 = "今日は天気が良い"

emb1 = client.embeddings.create(input=text1, model="text-embedding-3-small").data[0].embedding
emb2 = client.embeddings.create(input=text2, model="text-embedding-3-small").data[0].embedding
emb3 = client.embeddings.create(input=text3, model="text-embedding-3-small").data[0].embedding

print(f"AI関連の2文の類似度: {cosine_similarity(emb1, emb2):.3f}")  # ~0.85
print(f"AIと天気の類似度: {cosine_similarity(emb1, emb3):.3f}")     # ~0.30

from openai import OpenAI
client = OpenAI()

# text-embedding-3-small: 速くて安い、ほとんどのケースに十分
small_emb = client.embeddings.create(
    input="your text here",
    model="text-embedding-3-small"
).data[0].embedding
# 次元数: 1536 | コスト: $0.020/1Mトークン

# text-embedding-3-large: 最高品質、6.5倍高価
large_emb = client.embeddings.create(
    input="your text here",
    model="text-embedding-3-large"
).data[0].embedding
# 次元数: 3072 | コスト: $0.130/1Mトークン

# 次元削減サポート（Matryoshka Representation Learning）
reduced_emb = client.embeddings.create(
    input="your text here",
    model="text-embedding-3-small",
    dimensions=256  # 1536から256に削減 - ストレージコスト6倍削減
).data[0].embedding

import cohere
co = cohere.Client(api_key="your-key")

# 日本語を含む100言語以上をサポート
texts = [
    "人工知能が世界を変えている",
    "AI is changing the world",
    "AI가 세상을 바꾸고 있다"
]

# 注意：ドキュメントとクエリで異なるinput_typeを使用すること！
doc_response = co.embed(
    texts=texts,
    model="embed-multilingual-v3.0",
    input_type="search_document"  # インデックス作成用
)

query_response = co.embed(
    texts=["人工知能技術について教えてください"],
    model="embed-multilingual-v3.0",
    input_type="search_query"  # クエリ用 - 品質に影響する重要な設定！
)

# 圧縮サポート：ストレージコストを大幅に削減
compressed = co.embed(
    texts=texts,
    model="embed-multilingual-v3.0",
    input_type="search_document",
    embedding_types=["float", "int8", "binary"]
)
# binary: float32比32倍のストレージ削減！

from sentence_transformers import SentenceTransformer
import numpy as np

# BGE-M3: 2025年最高のオープンソース多言語埋め込みモデル
# 日本語、韓国語、中国語 + 英語で特に強力
bge_model = SentenceTransformer("BAAI/bge-m3")

texts = ["RAGは検索拡張生成技術です", "埋め込みはテキストをベクターに変換します"]
# BGE系モデルはドキュメントに"passage: "、クエリに"query: "プレフィックスを使用
doc_texts = [f"passage: {t}" for t in texts]
embeddings = bge_model.encode(doc_texts, normalize_embeddings=True)
print(f"Shape: {embeddings.shape}")  # (2, 1024)

# E5-mistral-7b: 7B LLMをエンコーダとして使用 - 最高品質オープンソース
# GPU RAM 16GB以上が必要
e5_model = SentenceTransformer("intfloat/e5-mistral-7b-instruct")
query = "Instruct: 関連ドキュメントを検索してください\nQuery: RAGの仕組みは？"
doc = "RAGは検索と言語モデル生成を組み合わせた技術です"
q_emb = e5_model.encode(query, normalize_embeddings=True)
d_emb = e5_model.encode(doc, normalize_embeddings=True)

# nomic-embed-text-v1.5: 優れた性能/サイズ比
nomic_model = SentenceTransformer(
    "nomic-ai/nomic-embed-text-v1.5",
    trust_remote_code=True
)
# 768次元、優れた性能、セルフホストが容易

# multilingual-e5-large: 日本語を含む多言語で安定した性能
ml_e5 = SentenceTransformer("intfloat/multilingual-e5-large")

MTEBのタスクタイプ：
- Retrieval（検索）: RAGで最重要。指標：nDCG@10
- STS（意味的テキスト類似度）: 文間の類似度。指標：Spearman相関係数
- Classification（分類）: テキスト分類。指標：Accuracy
- Clustering（クラスタリング）: 文書のグループ化。指標：V-measure
- Reranking（再ランキング）: 結果の並び替え。指標：MAP

RAGシステム構築時 → Retrievalスコアを最優先
テキスト分類時 → Classificationスコアを優先
意味的類似度計算時 → STSスコアを優先

def choose_embedding_model(
    languages: list,         # 例: ['ja'], ['ko', 'en'], ['ja', 'en', 'zh']
    budget: str,             # 'low', 'medium', 'high'
    privacy_required: bool,  # データを外部に送れないか
    scale: str,              # 'small'(<100k/日), 'medium', 'large'(>1M/日)
    use_case: str = 'rag'    # 'rag', 'classification', 'similarity'
) -> dict:
    """状況別の最適埋め込みモデルを推奨"""

    # プライバシーまたは大規模 → セルフホスト
    if privacy_required or scale == 'large':
        if 'ko' in languages or 'ja' in languages:
            return {
                "model": "BAAI/bge-m3",
                "reason": "最高のOSS多言語対応 + プライバシー保証",
                "est_monthly_cost": "GPUサーバー費用のみ（~$200-500/月）"
            }
        if use_case == 'rag' and budget == 'high':
            return {
                "model": "intfloat/e5-mistral-7b-instruct",
                "reason": "最高品質OSS、英語Retrieval最強",
                "est_monthly_cost": "GPUサーバー費用のみ（A100 ~$1000/月）"
            }
        return {
            "model": "nomic-ai/nomic-embed-text-v1.5",
            "reason": "軽量で優れた性能、セルフホストが容易",
            "est_monthly_cost": "GPUサーバー費用のみ（~$100-200/月）"
        }

    # 3言語以上 → Cohere
    if len(languages) > 2:
        return {
            "model": "cohere/embed-multilingual-v3.0",
            "reason": "100言語以上で均一な高品質サポート",
            "est_monthly_cost": "使用量ベース、一般的に$100-1000/月"
        }

    # コスト重視 + 英語中心
    if budget == 'low' and languages == ['en']:
        return {
            "model": "text-embedding-3-small",
            "reason": "最安値の商用オプション、ほとんどの英語タスクに十分",
            "est_monthly_cost": "100万リクエストで~$20/月"
        }

    # デフォルト：最高品質の商用モデル
    return {
        "model": "text-embedding-3-large",
        "reason": "最高の商用品質、インフラ負担なし",
        "est_monthly_cost": "100万リクエストで~$130/月"
    }

# 使用例
result = choose_embedding_model(
    languages=['ja', 'en'],
    budget='medium',
    privacy_required=True,
    scale='medium',
    use_case='rag'
)
print(f"推奨モデル: {result['model']}")
print(f"理由: {result['reason']}")

# text-embedding-3-smallの次元別パフォーマンス/コストトレードオフ
# (MTEB Retrievalスコア、近似値)
# 1536次元: 53.2（基準）
# 1024次元: 52.8 (-0.4)
#  512次元: 52.1 (-1.1)
#  256次元: 50.9 (-2.3)
# → 256次元にしても品質低下5%未満、ストレージコスト6倍削減！

# 100万ドキュメントでのストレージ比較:
# 1536次元: 100万 × 1536 × 4バイト = 6.14 GB
#  256次元: 100万 × 256 × 4バイト = 1.02 GB
# → ストレージコスト6倍削減 + 検索速度約3倍向上

from openai import OpenAI
client = OpenAI()

def get_reduced_embedding(text: str, dimensions: int = 256) -> list:
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-small",
        dimensions=dimensions
    )
    return response.data[0].embedding

from sentence_transformers import SentenceTransformer
import numpy as np

class EmbeddingPipeline:
    """本番対応の埋め込みパイプライン"""

    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.model = SentenceTransformer(model_name)
        self.model_name = model_name

    def embed_documents(self, texts: list, batch_size: int = 32) -> np.ndarray:
        """インデックス作成用のバッチ埋め込み"""
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            if "bge" in self.model_name.lower():
                batch = [f"passage: {t}" for t in batch]
            embeddings = self.model.encode(
                batch,
                normalize_embeddings=True,
                show_progress_bar=False
            )
            all_embeddings.append(embeddings)
        return np.vstack(all_embeddings)

    def embed_query(self, query: str) -> np.ndarray:
        """検索用のクエリ埋め込み"""
        if "bge" in self.model_name.lower():
            query = f"query: {query}"
        return self.model.encode(query, normalize_embeddings=True)

    def search(self, query: str, doc_embeddings: np.ndarray,
               documents: list, top_k: int = 5) -> list:
        """事前計算済みドキュメント埋め込みに対するセマンティック検索"""
        query_emb = self.embed_query(query)
        scores = np.dot(doc_embeddings, query_emb)
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [
            {"document": documents[i], "score": float(scores[i])}
            for i in top_indices
        ]

# 使用例
pipeline = EmbeddingPipeline("BAAI/bge-m3")

docs = [
    "RAGは検索と言語モデル生成を組み合わせた技術です",
    "埋め込みはテキストを高密度ベクターに変換します",
    "ベクターデータベースは近似最近傍探索を支援します"
]

doc_embeddings = pipeline.embed_documents(docs)
results = pipeline.search(
    "ベクターデータベースの仕組みは？",
    doc_embeddings,
    docs,
    top_k=2
)

for r in results:
    print(f"スコア: {r['score']:.3f} | {r['document']}")