본문 바로가기

AI Agent

임베딩 모델 완전 가이드 — text-embedding 선택과 RAG 적용

반응형

RAG를 만들었는데 검색 품질이 나쁩니다. 청킹도 바꿔보고 프롬프트도 바꿨는데 여전합니다. 임베딩 모델이 문제일 수 있습니다. 선택 기준부터 실전 적용까지 정리했습니다.

[핵심 요약]
→ 임베딩: 텍스트를 의미 기반 숫자 벡터로 변환하는 것
→ 역할: RAG에서 "관련 문서를 찾는" 핵심 엔진
→ 모델 선택이 RAG 품질의 40~60%를 결정
→ 주요 모델: Qwen3-Embedding, OpenAI, Cohere, bge-m3, Voyage AI
→ 한국어: Qwen3-Embedding-8B, bge-m3, Cohere multilingual 추천
→ 평가: MTEB 벤치마크 기준 + 실제 도메인 테스트 필수
→ 비용: 로컬(무료) vs API(편리) — 볼륨에 따라 선택

임베딩이 뭔지 30초 정리

# 임베딩 = 텍스트 → 의미 공간의 좌표

texts = [
    "고양이는 귀여운 동물입니다",
    "강아지는 사람을 좋아합니다",
    "파이썬으로 웹 개발을 합니다"
]

# 임베딩 후 (실제로는 1024~4096차원 벡터)
embeddings = {
    "고양이는 귀여운 동물입니다": [0.23, -0.45, 0.89, ...],
    "강아지는 사람을 좋아합니다": [0.21, -0.41, 0.87, ...],  # 동물 관련이라 유사
    "파이썬으로 웹 개발을 합니다": [-0.67, 0.23, -0.12, ...]  # 전혀 다른 방향
}

# 코사인 유사도: 벡터 간 각도로 의미 유사도 측정
# 고양이 ↔ 강아지: 0.94 (매우 유사)
# 고양이 ↔ 파이썬: 0.12 (관련 없음)
[RAG에서 임베딩의 역할]

1. 색인 단계:
   문서 청크들 → 임베딩 → 벡터 DB 저장

2. 검색 단계:
   사용자 질문 → 임베딩 → 벡터 DB에서 유사 청크 검색

3. 생성 단계:
   검색된 청크 + 질문 → LLM → 답변

임베딩 모델이 나쁘면:
→ 관련 없는 청크가 검색됨
→ LLM이 엉뚱한 컨텍스트로 답변
→ RAG 전체 품질 저하

실전 1 — 주요 임베딩 모델 비교

# 2026년 4월 기준 주요 모델 비교
models = {
    "Qwen3-Embedding-8B": {
        "provider":    "Alibaba Qwen (오픈소스, Apache 2.0)",
        "dims":        "32~4096 (유연하게 설정)",
        "context":     "32K 토큰",
        "mteb_multi":  70.58,  # ← MTEB 멀티링구얼 현재 1위
        "price":       "무료 (로컬 실행)",
        "한국어":       "매우 좋음 (100개+ 언어)",
        "특징":         "멀티링구얼 MTEB 1위, 인스트럭션 지원, 리랭커 세트"
    },
    "Qwen3-Embedding-0.6B": {
        "provider":    "Alibaba Qwen (오픈소스, Apache 2.0)",
        "dims":        "32~1024",
        "context":     "32K 토큰",
        "mteb_multi":  "높음",
        "price":       "무료 (CPU 실행 가능)",
        "한국어":       "좋음",
        "특징":         "초경량, GPU 없이 실행, 빠른 추론"
    },
    "text-embedding-3-large": {
        "provider":    "OpenAI",
        "dims":        3072,
        "mteb_eng":    64.6,
        "price":       "$0.13/1M tokens",
        "한국어":       "보통",
        "특징":         "고차원, 높은 영어 정확도, Matryoshka 지원"
    },
    "text-embedding-3-small": {
        "provider":    "OpenAI",
        "dims":        1536,
        "mteb_eng":    62.3,
        "price":       "$0.02/1M tokens",
        "한국어":       "보통",
        "특징":         "가성비 최강, 대부분 케이스 충분"
    },
    "embed-multilingual-v3.0": {
        "provider":    "Cohere",
        "dims":        1024,
        "mteb_multi":  64.5,
        "price":       "$0.10/1M tokens",
        "한국어":       "좋음",
        "특징":         "검색 특화, int8 압축 지원, input_type 구분"
    },
    "voyage-3-large": {
        "provider":    "Voyage AI",
        "dims":        1024,
        "mteb_eng":    68.3,
        "price":       "$0.18/1M tokens",
        "한국어":       "좋음",
        "특징":         "영어 MTEB 최고 수준, 고품질"
    },
    "bge-m3": {
        "provider":    "BAAI (오픈소스)",
        "dims":        1024,
        "mteb_multi":  63.1,
        "price":       "무료 (로컬 실행)",
        "한국어":       "매우 좋음",
        "특징":         "검증된 다국어 모델, 8192 토큰, 안정적"
    }
}

모델 MTEB 점수 가격 한국어 추천 상황

Qwen3-Embedding-8B 70.58 (멀티 1위) 무료 ★★★★★ 최고 성능 + 무료
voyage-3-large 68.3 (영어) $0.18/1M ★★★★ 영어 최고 API
text-embedding-3-small 62.3 (영어) $0.02/1M ★★★ 영어 가성비
embed-multilingual-v3.0 64.5 (멀티) $0.10/1M ★★★★ 한국어 API
bge-m3 63.1 (멀티) 무료 ★★★★★ 검증된 로컬
Qwen3-Embedding-0.6B 높음 무료 ★★★★ 초경량 로컬

실전 2 — Qwen3-Embedding 로컬 실행

pip install sentence-transformers torch
from sentence_transformers import SentenceTransformer
import numpy as np

class Qwen3Embedder:
    """Qwen3-Embedding 래퍼"""

    def __init__(self, model_size: str = "8B", truncate_dim: int = None):
        model_name = f"Qwen/Qwen3-Embedding-{model_size}"
        print(f"모델 로딩: {model_name}")

        kwargs = {}
        if truncate_dim:
            kwargs["truncate_dim"] = truncate_dim  # 차원 축소 (Matryoshka 지원)

        self.model = SentenceTransformer(model_name, **kwargs)
        print("로딩 완료")

    def embed_documents(self, texts: list[str]) -> np.ndarray:
        """문서 임베딩 (색인용)"""
        return self.model.encode(
            texts,
            normalize_embeddings=True,
            batch_size=32,
            show_progress_bar=len(texts) > 100
        )

    def embed_query(
        self,
        query:       str,
        instruction: str = None
    ) -> np.ndarray:
        """
        쿼리 임베딩 (검색용)
        instruction 추가 시 1~5% 성능 향상
        """
        if instruction:
            query = f"Instruct: {instruction}\nQuery: {query}"

        return self.model.encode(
            query,
            normalize_embeddings=True
        )


# 기본 사용
embedder = Qwen3Embedder(model_size="8B")

# 문서 색인
docs = [
    "파이썬에서 리스트 중복 제거하는 법",
    "자바스크립트 비동기 처리 방법",
    "한국어 자연어 처리 기술",
    "데이터베이스 인덱스 최적화"
]
doc_embeddings = embedder.embed_documents(docs)

# 쿼리 검색 (인스트럭션 포함)
query     = "파이썬 리스트 중복 없애기"
query_emb = embedder.embed_query(
    query,
    instruction="코드 관련 기술 질문을 검색하기 위한 쿼리"
)

# 유사도 계산
similarities = np.dot(doc_embeddings, query_emb)
ranked       = np.argsort(similarities)[::-1]

for idx in ranked:
    print(f"점수: {similarities[idx]:.3f} — {docs[idx]}")

# 점수: 0.912 — 파이썬에서 리스트 중복 제거하는 법
# 점수: 0.341 — 자바스크립트 비동기 처리 방법
# 점수: 0.287 — 한국어 자연어 처리 기술
# 점수: 0.198 — 데이터베이스 인덱스 최적화


# 차원 축소 (저장 공간 절약)
embedder_512 = Qwen3Embedder(model_size="8B", truncate_dim=512)
# 4096차원 → 512차원, 저장 공간 87% 절감
# Qwen3-Embedding-0.6B — CPU 실행 (GPU 없을 때)
embedder_lite = Qwen3Embedder(model_size="0.6B")
# → GPU 없어도 실행 가능
# → 속도: 8B 대비 10배 이상 빠름
# → 품질: 8B보다 낮지만 bge-m3와 유사 수준
[Qwen3-Embedding 시리즈 전체]

모델                      파라미터  VRAM    컨텍스트  특징
Qwen3-Embedding-0.6B     0.6B     ~1.5GB  32K      CPU 가능, 초경량
Qwen3-Embedding-4B       4B       ~8GB    32K      균형형
Qwen3-Embedding-8B       8B       ~16GB   32K      MTEB 멀티 1위

+ Qwen3-Reranker 시리즈 (0.6B, 4B, 8B) 별도 제공
→ 임베딩 + 리랭커 조합으로 검색 품질 추가 향상

실전 3 — API 임베딩 구현

import openai
import cohere
import numpy as np

# ===== OpenAI =====
openai_client = openai.OpenAI()

def embed_openai(
    texts: list[str],
    model: str = "text-embedding-3-small"
) -> list[list[float]]:
    response = openai_client.embeddings.create(
        input=texts,
        model=model,
        encoding_format="float"
    )
    return [item.embedding for item in response.data]


# ===== Cohere (한국어 권장) =====
co = cohere.Client("YOUR_COHERE_KEY")

def embed_cohere(
    texts:      list[str],
    input_type: str = "search_document"  # 문서: document, 쿼리: query
) -> list[list[float]]:
    """
    input_type 구분이 중요!
    색인 시: "search_document"
    검색 시: "search_query"
    """
    response = co.embed(
        texts=texts,
        model="embed-multilingual-v3.0",
        input_type=input_type,
        embedding_types=["float"]
    )
    return response.embeddings.float


# ===== 배치 처리 =====
def embed_in_batches(
    texts:      list[str],
    embed_fn,
    batch_size: int = 100
) -> list[list[float]]:
    """대량 텍스트 배치 임베딩"""
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        embeddings = embed_fn(batch)
        all_embeddings.extend(embeddings)
        print(f"진행: {min(i + batch_size, len(texts))}/{len(texts)}")
    return all_embeddings

실전 4 — RAG 파이프라인 구현

import anthropic
import json
import numpy as np
from pathlib import Path

class EmbeddingRAG:
    """임베딩 기반 RAG 파이프라인"""

    def __init__(
        self,
        embedder,
        llm_client,
        chunk_size:    int = 500,
        chunk_overlap: int = 50,
        top_k:         int = 5
    ):
        self.embedder      = embedder
        self.llm           = llm_client
        self.chunk_size    = chunk_size
        self.chunk_overlap = chunk_overlap
        self.top_k         = top_k
        self.documents     = []

    def add_documents(self, texts: list[str], metadatas: list[dict] = None):
        """문서 추가 및 임베딩"""
        chunks = []
        metas  = []

        for i, text in enumerate(texts):
            doc_chunks = self._chunk_text(text)
            chunks.extend(doc_chunks)
            meta = metadatas[i] if metadatas else {}
            metas.extend([meta] * len(doc_chunks))

        print(f"{len(chunks)}개 청크 임베딩 중...")
        new_embeddings = self.embedder.embed_documents(chunks)

        for chunk, meta, emb in zip(chunks, metas, new_embeddings):
            self.documents.append({
                "id":        len(self.documents),
                "text":      chunk,
                "metadata":  meta,
                "embedding": emb.tolist() if hasattr(emb, 'tolist') else emb
            })

        print(f"총 {len(self.documents)}개 청크 저장됨")

    def _chunk_text(self, text: str) -> list[str]:
        """텍스트 청킹"""
        words   = text.split()
        chunks  = []
        current = []
        count   = 0

        for word in words:
            current.append(word)
            count += 1
            if count >= self.chunk_size:
                chunks.append(" ".join(current))
                current = current[-self.chunk_overlap:]
                count   = len(current)

        if current:
            chunks.append(" ".join(current))

        return chunks

    def search(self, query: str, top_k: int = None) -> list[dict]:
        """쿼리 유사 청크 검색"""
        k         = top_k or self.top_k
        query_emb = self.embedder.embed_query(query)
        query_arr = np.array(query_emb)

        scores = []
        for doc in self.documents:
            doc_arr    = np.array(doc["embedding"])
            similarity = float(
                np.dot(query_arr, doc_arr) /
                (np.linalg.norm(query_arr) * np.linalg.norm(doc_arr))
            )
            scores.append((doc, similarity))

        scores.sort(key=lambda x: x[1], reverse=True)
        return [
            {"text": d["text"], "metadata": d["metadata"], "score": s}
            for d, s in scores[:k]
        ]

    def query(self, question: str) -> str:
        """검색 + 생성 전체 파이프라인"""
        retrieved = self.search(question)

        context = "\n\n".join([
            f"[{i+1}] (유사도: {r['score']:.3f})\n{r['text']}"
            for i, r in enumerate(retrieved)
        ])

        response = self.llm.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=f"""다음 문서를 참고해서 질문에 답변하세요.
문서에 없는 내용은 "문서에 없는 내용입니다"라고 답하세요.

[참고 문서]
{context}""",
            messages=[{"role": "user", "content": question}]
        )
        return response.content[0].text

    def save_index(self, path: str):
        with open(path, "w", encoding="utf-8") as f:
            json.dump(self.documents, f, ensure_ascii=False)

    def load_index(self, path: str):
        with open(path, encoding="utf-8") as f:
            self.documents = json.load(f)
        print(f"인덱스 로드: {len(self.documents)}개 청크")


# 전체 사용 예시
embedder = Qwen3Embedder(model_size="8B")
llm      = anthropic.Anthropic()

rag = EmbeddingRAG(embedder=embedder, llm_client=llm)

rag.add_documents(
    texts=[
        "파이썬은 1991년에 귀도 반 로섬이 만든 프로그래밍 언어입니다...",
        "LangChain은 LLM 애플리케이션 개발을 위한 프레임워크입니다..."
    ],
    metadatas=[
        {"source": "python_intro.txt"},
        {"source": "langchain_intro.txt"}
    ]
)

rag.save_index("index.json")

answer = rag.query("파이썬은 누가 만들었나요?")
print(answer)

실전 5 — Qwen3-Reranker로 검색 품질 추가 향상

Qwen3-Embedding과 짝을 이루는 리랭커입니다. 1차 검색 결과를 재정렬해서 품질을 높입니다.

from sentence_transformers import CrossEncoder

class Qwen3Reranker:
    """Qwen3-Reranker 래퍼"""

    def __init__(self, model_size: str = "8B"):
        model_name  = f"Qwen/Qwen3-Reranker-{model_size}"
        self.model  = CrossEncoder(model_name, max_length=8192)

    def rerank(
        self,
        query:     str,
        documents: list[str],
        top_k:     int = 3
    ) -> list[tuple[int, float, str]]:
        """
        문서 재정렬
        Returns: [(원래인덱스, 점수, 텍스트), ...]
        """
        pairs  = [(query, doc) for doc in documents]
        scores = self.model.predict(pairs)

        results = sorted(
            zip(range(len(documents)), scores, documents),
            key=lambda x: x[1],
            reverse=True
        )
        return list(results[:top_k])


# 임베딩 + 리랭커 파이프라인
class HybridRAG(EmbeddingRAG):
    """임베딩 검색 + 리랭킹 파이프라인"""

    def __init__(self, embedder, reranker, llm_client, **kwargs):
        super().__init__(embedder, llm_client, **kwargs)
        self.reranker = reranker

    def query(self, question: str, top_k_retrieve: int = 20) -> str:
        # 1단계: 임베딩으로 넓게 검색 (20개)
        candidates = self.search(question, top_k=top_k_retrieve)

        # 2단계: 리랭커로 정밀 재정렬 (상위 5개)
        texts    = [c["text"] for c in candidates]
        reranked = self.reranker.rerank(question, texts, top_k=5)

        context = "\n\n".join([
            f"[{i+1}] (점수: {score:.3f})\n{text}"
            for i, (_, score, text) in enumerate(reranked)
        ])

        response = self.llm.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=f"다음 문서를 참고해서 답변하세요.\n\n{context}",
            messages=[{"role": "user", "content": question}]
        )
        return response.content[0].text


# 사용
embedder = Qwen3Embedder("8B")
reranker = Qwen3Reranker("8B")
llm      = anthropic.Anthropic()

rag = HybridRAG(
    embedder=embedder,
    reranker=reranker,
    llm_client=llm
)
[리랭킹이 효과적인 이유]

임베딩 검색 (Bi-Encoder):
→ 쿼리와 문서를 각각 독립적으로 인코딩
→ 빠르지만 세밀한 비교 불가

리랭킹 (Cross-Encoder):
→ 쿼리 + 문서를 함께 입력해서 세밀하게 비교
→ 느리지만 정확도 높음

→ 조합: 임베딩으로 후보 20개 → 리랭커로 5개로 압축
→ 속도 (임베딩) + 정확도 (리랭커) 동시 확보

실전 6 — 임베딩 품질 평가

import numpy as np

class EmbeddingEvaluator:
    """임베딩 모델 품질 평가"""

    def __init__(self, embedder):
        self.embedder = embedder

    def evaluate_retrieval(self, test_cases: list[dict]) -> dict:
        """
        검색 품질 평가

        test_cases 형식:
        [{"query": "질문", "relevant_docs": [0, 2], "documents": [...]}]
        """
        hit_at_1  = 0
        hit_at_3  = 0
        hit_at_5  = 0
        mrr_total = 0.0

        for case in test_cases:
            docs     = case["documents"]
            query    = case["query"]
            relevant = set(case["relevant_docs"])

            doc_embs  = self.embedder.embed_documents(docs)
            query_emb = self.embedder.embed_query(query)
            query_arr = np.array(query_emb)

            scores = [
                float(np.dot(query_arr, np.array(emb)) /
                      (np.linalg.norm(query_arr) * np.linalg.norm(emb)))
                for emb in doc_embs
            ]
            ranked = np.argsort(scores)[::-1]

            if ranked[0] in relevant:
                hit_at_1 += 1
            if any(r in relevant for r in ranked[:3]):
                hit_at_3 += 1
            if any(r in relevant for r in ranked[:5]):
                hit_at_5 += 1

            for rank, idx in enumerate(ranked, 1):
                if idx in relevant:
                    mrr_total += 1.0 / rank
                    break

        n = len(test_cases)
        return {
            "Hit@1": hit_at_1 / n,
            "Hit@3": hit_at_3 / n,
            "Hit@5": hit_at_5 / n,
            "MRR":   mrr_total / n
        }


# 모델 비교 실행
def compare_models(test_cases: list[dict]):
    models = {
        "Qwen3-Embedding-8B":    Qwen3Embedder("8B"),
        "Qwen3-Embedding-0.6B":  Qwen3Embedder("0.6B"),
        "bge-m3":                LocalEmbedder("BAAI/bge-m3"),
    }

    print("\n=== 모델 비교 ===")
    for name, embedder in models.items():
        evaluator = EmbeddingEvaluator(embedder)
        metrics   = evaluator.evaluate_retrieval(test_cases)
        print(f"\n{name}:")
        for metric, score in metrics.items():
            print(f"  {metric}: {score:.3f}")

실전 7 — Matryoshka 차원 축소

OpenAI와 Qwen3 모두 Matryoshka 방식을 지원합니다. 차원을 줄여도 품질이 크게 떨어지지 않습니다.

# Qwen3-Embedding 차원 축소
from sentence_transformers import SentenceTransformer

def get_truncated_embedder(dims: int = 512):
    return SentenceTransformer(
        "Qwen/Qwen3-Embedding-8B",
        truncate_dim=dims  # 4096 → 원하는 차원으로 축소
    )

# 차원별 트레이드오프
dimension_comparison = {
    4096: {"품질": "최상", "storage_1M_docs": "16GB", "검색속도": "1x"},
    1024: {"품질": "우수", "storage_1M_docs": "4GB",  "검색속도": "4x"},
    512:  {"품질": "양호", "storage_1M_docs": "2GB",  "검색속도": "8x"},
    256:  {"품질": "보통", "storage_1M_docs": "1GB",  "검색속도": "16x"},
}

# OpenAI text-embedding-3-large도 동일하게 지원
def embed_openai_truncated(texts: list[str], dims: int = 512):
    response = openai_client.embeddings.create(
        input=texts,
        model="text-embedding-3-large",
        dimensions=dims
    )
    return [item.embedding for item in response.data]

# 권장: 512차원 (성능/비용 최적 균형)

최종 선택 가이드

상황 추천 모델 이유

최고 성능 + 무료 Qwen3-Embedding-8B MTEB 멀티 1위, Apache 2.0
초경량 + 무료 Qwen3-Embedding-0.6B CPU 실행, GPU 불필요
한국어 + API Cohere multilingual-v3.0 input_type 구분, 검색 특화
영어 최고 성능 voyage-3-large 영어 MTEB 최상위
영어 가성비 text-embedding-3-small $0.02/1M, 충분한 품질
검증된 다국어 로컬 bge-m3 안정적, 커뮤니티 검증

마무리

✅ 임베딩 모델로 RAG 품질 높이는 순서

1순위: 모델 선택 (가장 큰 영향)
→ 한국어 문서 많으면 Qwen3-Embedding-8B 우선 고려
→ 영어 중심이면 voyage-3-large 또는 text-embedding-3-small

2순위: 리랭커 추가
→ Qwen3-Reranker 또는 Cohere Rerank
→ 검색 후보 20개 → 상위 5개로 압축

3순위: 청크 크기/오버랩 튜닝
→ 500~1000 토큰, 10~20% 오버랩

4순위: 하이브리드 검색 추가
→ Dense + BM25 결합

5순위: 프롬프트 최적화

[주의사항]
→ Qwen3-Embedding-8B: VRAM 16GB 이상 필요
→ 0.6B는 CPU 실행 가능하지만 속도 느림
→ 반드시 본인 도메인 데이터로 직접 평가 후 선택
→ MTEB 점수 ≠ 내 도메인 성능 (직접 테스트 필수)

관련 글:

 

https://cell-devlog.tistory.com/21

 

RAG 시스템에 맞는 벡터 DB는 뭔가 — ChromaDB vs Qdrant vs Pinecone vs Elasticsearch 완전 비교

RAG 시스템을 만들 때 이런 고민이 생깁니다."벡터 DB가 이렇게 많은데 뭘 써야 하지? 다들 자기가 제일 빠르다고 하는데."벤더 벤치마크는 전부 자기한테 유리하게 나와 있어요. 이번 글에서는 Chr

cell-devlog.tistory.com

https://cell-devlog.tistory.com/11

 

벡터 검색 정확도 올리는 법 — 임베딩 모델 선택부터 HNSW 튜닝, Reranking까지

벡터 검색을 붙여봤는데 결과가 기대보다 별로라는 경험, 한 번쯤 있으실 거예요."분명히 관련 있는 문서인데 왜 안 나오지?"이번 글에서는 벡터 검색 정확도를 높이는 방법을 임베딩 모델 선택

cell-devlog.tistory.com

https://cell-devlog.tistory.com/24

 

쿼리 재작성, 반복 검색, 멀티소스 라우팅 — Agentic RAG 동작 원리와 동적 검색 전략 완전 정리

RAG 시스템을 만들고 나면 이런 한계가 생겨요."단순한 질문은 잘 답하는데, '2024년 실적을 바탕으로 2025년 전략을 분석해줘' 같은 복잡한 질문은 엉뚱한 답이 나온다."이건 일반 RAG의 구조적 한계

cell-devlog.tistory.com

 


 

반응형