Embedding 모델 선택 완전 가이드: OpenAI부터 오픈소스까지 2025년 기준

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 관련 두 문장 유사도: {cosine_similarity(emb1, emb2):.3f}")  # ~0.85
print(f"AI vs 날씨 유사도: {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 tokens

# text-embedding-3-large: 최고 품질, 5배 비쌈
large_emb = client.embeddings.create(
    input="your text here",
    model="text-embedding-3-large"
).data[0].embedding
# 차원: 3072, 비용: $0.130/1M tokens

# 차원 축소 지원 (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が世界を変えている"
]

response = co.embed(
    texts=texts,
    model="embed-multilingual-v3.0",
    input_type="search_document"  # 문서 인덱싱용
    # input_type="search_query"   # 쿼리 임베딩용
)

# 쿼리는 다른 input_type 사용 (중요!)
query_response = co.embed(
    texts=["AI 발전에 대해 알려줘"],
    model="embed-multilingual-v3.0",
    input_type="search_query"  # 쿼리 전용
)

# 압축 지원: 저장 비용 대폭 절감
compressed_response = co.embed(
    texts=texts,
    model="embed-multilingual-v3.0",
    input_type="search_document",
    embedding_types=["float", "int8", "binary"]  # 다양한 정밀도
)
# binary: 원본 대비 32배 저장 공간 절감!

from sentence_transformers import SentenceTransformer
import torch

# BGE-M3: 2025년 최고의 오픈소스 다국어 임베딩 모델
model = SentenceTransformer("BAAI/bge-m3")

texts = ["인공지능이 세상을 바꾸고 있다", "AI is changing the world"]
embeddings = model.encode(texts, normalize_embeddings=True)
print(f"차원: {embeddings.shape}")  # (2, 1024)

# E5-mistral-7b: 7B LLM을 인코더로 사용, 최고 품질
# (GPU 메모리 최소 16GB 필요)
e5_model = SentenceTransformer("intfloat/e5-mistral-7b-instruct")
# 사용 시 쿼리에 "Instruct: ... \nQuery: " 접두어 필요
query = "Instruct: 관련 문서를 검색하시오\nQuery: 인공지능 기술 동향"
doc = "최근 AI 기술이 빠르게 발전하고 있다"
query_emb = e5_model.encode(query, normalize_embeddings=True)
doc_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 (Semantic Textual Similarity): 문장 간 유사도. Spearman 상관계수
- Classification: 텍스트 분류. Accuracy
- Clustering: 군집화. V-measure
- Reranking: 재순위 매기기. MAP

RAG 시스템 구축 시 → Retrieval 점수 최우선
텍스트 분류 시 → Classification 점수 우선
의미 유사도 계산 시 → STS 점수 우선

def choose_embedding_model(
    languages: list,        # 예: ['ko', 'en'], ['ja', 'en', 'zh']
    budget: str,            # 'low', 'medium', 'high'
    privacy_required: bool, # 데이터를 외부로 보내면 안 되는 경우
    scale: str,             # 'small'(<100k req/day), 'medium', 'large'(>1M req/day)
    use_case: str = 'rag'   # 'rag', 'classification', 'similarity'
) -> dict:
    """
    상황별 최적 임베딩 모델 추천

    Returns:
        dict: {'model': str, 'reason': str, 'estimated_monthly_cost': str}
    """

    # 프라이버시 또는 대규모 → 자체 호스팅
    if privacy_required or scale == 'large':
        if 'ko' in languages or 'ja' in languages:
            return {
                "model": "BAAI/bge-m3",
                "reason": "최고의 오픈소스 한국어/일본어 지원 + 프라이버시 보장",
                "estimated_monthly_cost": "GPU 서버 비용만 (~$200-500/월)"
            }
        if use_case == 'rag' and budget == 'high':
            return {
                "model": "intfloat/e5-mistral-7b-instruct",
                "reason": "최고 품질 오픈소스, 영어 Retrieval 최강",
                "estimated_monthly_cost": "GPU 서버 비용만 (A100 기준 ~$1000/월)"
            }
        return {
            "model": "nomic-ai/nomic-embed-text-v1.5",
            "reason": "가벼우면서 좋은 성능, 자체 호스팅 용이",
            "estimated_monthly_cost": "GPU 서버 비용만 (~$100-200/월)"
        }

    # 다국어 (3개 이상) → Cohere
    if len(languages) > 2:
        return {
            "model": "cohere/embed-multilingual-v3.0",
            "reason": "100개+ 언어 균일한 고품질 지원",
            "estimated_monthly_cost": "요청량에 따라 $100-1000/월"
        }

    # 비용 민감 + 영어 위주
    if budget == 'low' and 'en' in languages:
        return {
            "model": "text-embedding-3-small",
            "reason": "가장 저렴하고 대부분의 영어 태스크에 충분",
            "estimated_monthly_cost": "1M req 기준 ~$20/월"
        }

    # 기본: 최고 품질 상용
    return {
        "model": "text-embedding-3-large",
        "reason": "최고의 상용 임베딩 품질, 운영 부담 없음",
        "estimated_monthly_cost": "1M req 기준 ~$130/월"
    }

# 사용 예시
result = choose_embedding_model(
    languages=['ko', 'en'],
    budget='medium',
    privacy_required=True,
    scale='medium',
    use_case='rag'
)
print(f"추천 모델: {result['model']}")
print(f"이유: {result['reason']}")

from openai import OpenAI
import numpy as np

client = OpenAI()

def compare_dimension_tradeoffs(text: str):
    """차원별 성능/비용 트레이드오프 분석"""

    dimensions_to_test = [256, 512, 1024, 1536]  # 1536이 기본
    results = {}

    for dim in dimensions_to_test:
        response = client.embeddings.create(
            input=text,
            model="text-embedding-3-small",
            dimensions=dim
        )
        emb = response.data[0].embedding
        results[dim] = {
            "embedding": emb,
            "storage_bytes": len(emb) * 4,  # float32 = 4 bytes
            "relative_storage": f"{dim/1536:.1%} of full"
        }
        print(f"차원 {dim}: 저장 {results[dim]['storage_bytes']} bytes "
              f"({results[dim]['relative_storage']})")

    return results

# 실제 성능 측정 (MTEB Retrieval 점수 기준)
# 모델: text-embedding-3-small
# 1536차원: 53.2 (기준)
# 1024차원: 52.8 (-0.4)
# 512차원:  52.1 (-1.1)
# 256차원:  50.9 (-2.3)
# → 256차원으로 줄여도 품질 4% 미만 감소, 저장 비용 6배 절감!

# pgvector 사용 시 실질적인 저장/검색 비용 비교:
# 문서 100만 건 기준:
# 1536차원: 100만 × 1536 × 4 bytes = 6.14 GB
# 256차원:  100만 × 256 × 4 bytes = 1.02 GB
# → 저장 비용 6배 절감 + 검색 속도 약 3배 향상

from sentence_transformers import SentenceTransformer
import numpy as np
from typing import Union
import time

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]
            # BGE 계열은 "passage: " 접두어 사용
            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}"  # BGE 계열 쿼리 접두어
        return self.model.encode(
            query,
            normalize_embeddings=True
        )

    def benchmark(self, texts: list) -> dict:
        """임베딩 속도 벤치마크"""
        start = time.time()
        embeddings = self.embed_documents(texts)
        elapsed = time.time() - start
        return {
            "total_texts": len(texts),
            "total_time_sec": elapsed,
            "texts_per_second": len(texts) / elapsed,
            "embedding_shape": embeddings.shape
        }

# 사용 예시
pipeline = EmbeddingPipeline("BAAI/bge-m3")

# 문서 임베딩
documents = [
    "RAG는 검색 증강 생성 기술입니다",
    "임베딩은 텍스트를 벡터로 변환합니다",
    "벡터 데이터베이스는 유사도 검색에 최적화되어 있습니다"
]
doc_embeddings = pipeline.embed_documents(documents)

# 쿼리 임베딩
query = "텍스트 검색 기술에 대해 알려주세요"
query_embedding = pipeline.embed_query(query)

# 유사도 계산
similarities = np.dot(doc_embeddings, query_embedding)
best_match_idx = np.argmax(similarities)
print(f"가장 관련 있는 문서: {documents[best_match_idx]}")
print(f"유사도 점수: {similarities[best_match_idx]:.3f}")