Oct 19, 2025

의미 기반 검색(Semantic Search) 구현 가이드

Semantic Search란?

의미 기반 검색(Semantic Search)은 단어의 의미를 이해하여 검색하는 기술입니다.

전통적 키워드 검색:

검색: "강아지 사료"
결과: "강아지", "사료" 단어가 포함된 문서만 검색
놓침: "반려견 먹이", "애견 식품", "puppy food"

의미 기반 검색:

검색: "강아지 사료"
이해: "반려동물 먹이 관련 정보"
결과: "강아지 사료", "반려견 먹이", "애견 식품", "puppy food" 모두 검색!

작동 원리

1. 벡터 임베딩

텍스트를 의미를 담은 숫자 벡터로 변환합니다.

from openai import OpenAI

client = OpenAI()

# 텍스트 → 벡터
def get_embedding(text):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# 예시
vec_dog = get_embedding("강아지")
vec_puppy = get_embedding("puppy")
vec_car = get_embedding("자동차")

# 유사도
from sklearn.metrics.pairwise import cosine_similarity

similarity(vec_dog, vec_puppy)  # 0.85 - 높음!
similarity(vec_dog, vec_car)    # 0.12 - 낮음

2. 벡터 저장

모든 문서를 임베딩하여 벡터 데이터베이스에 저장합니다.

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 문서 준비
documents = [
    "강아지 사료 판매합니다",
    "반려견 건강식품 추천",
    "고양이 간식 저렴하게",
    "애견 훈련 프로그램"
]

# 임베딩 및 저장
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_texts(
    texts=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

3. 의미 검색

쿼리를 임베딩하고 유사한 벡터를 찾습니다.

# 검색
query = "개 먹이 구매"
results = vectorstore.similarity_search(query, k=3)

for doc in results:
    print(doc.page_content)

# 출력:
# "강아지 사료 판매합니다"
# "반려견 건강식품 추천"
# "애견 훈련 프로그램"

실전 구현

기본 시스템

import openai
from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

class SemanticSearchEngine:
    def __init__(self, docs_path):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = None
        self.load_documents(docs_path)

    def load_documents(self, path):
        # 1. 문서 로드
        loader = DirectoryLoader(path, glob="**/*.txt")
        documents = loader.load()

        # 2. 분할
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        chunks = splitter.split_documents(documents)

        # 3. 임베딩 및 저장
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory="./search_db"
        )

    def search(self, query, k=5):
        results = self.vectorstore.similarity_search(query, k=k)
        return [doc.page_content for doc in results]

# 사용
engine = SemanticSearchEngine("./documents")
results = engine.search("How to train a puppy?")

고급 검색 기법

1. 하이브리드 검색

키워드 검색과 의미 검색을 결합합니다.

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# BM25 (키워드)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 벡터 (의미)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 앙상블
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # 의미 검색에 더 높은 가중치
)

results = ensemble_retriever.get_relevant_documents(query)

2. MMR (Maximum Marginal Relevance)

관련성과 다양성을 모두 고려합니다.

# 유사하지만 다양한 결과 반환
results = vectorstore.max_marginal_relevance_search(
    query="AI applications",
    k=5,
    fetch_k=20  # 후보 20개 중 다양한 5개 선택
)

3. 메타데이터 필터링

조건과 의미를 동시에 고려합니다.

# 메타데이터와 함께 저장
documents_with_meta = [
    {
        "content": "강아지 사료 추천",
        "metadata": {"category": "food", "year": 2024}
    },
    {
        "content": "강아지 훈련 가이드",
        "metadata": {"category": "training", "year": 2023}
    }
]

vectorstore = Chroma.from_texts(
    texts=[d["content"] for d in documents_with_meta],
    metadatas=[d["metadata"] for d in documents_with_meta],
    embedding=embeddings
)

# 필터링 검색
results = vectorstore.similarity_search(
    query="강아지 관련 정보",
    k=5,
    filter={"category": "food", "year": {"$gte": 2024}}
)

실시간 검색 최적화

import asyncio
from concurrent.futures import ThreadPoolExecutor

class FastSemanticSearch:
    def __init__(self):
        self.vectorstore = Chroma(...)
        self.executor = ThreadPoolExecutor(max_workers=4)
        self.cache = {}

    async def search_async(self, query):
        # 캐시 확인
        if query in self.cache:
            return self.cache[query]

        # 병렬 검색
        loop = asyncio.get_event_loop()
        results = await loop.run_in_executor(
            self.executor,
            self.vectorstore.similarity_search,
            query
        )

        self.cache[query] = results
        return results

# 사용
search_engine = FastSemanticSearch()
results = await search_engine.search_async("AI trends")

응용 사례

1. 문서 검색 시스템

class DocumentSearchSystem:
    def __init__(self):
        self.vectorstore = None

    def index_documents(self, pdf_files):
        from langchain.document_loaders import PyPDFLoader

        all_docs = []
        for pdf in pdf_files:
            loader = PyPDFLoader(pdf)
            docs = loader.load()
            all_docs.extend(docs)

        splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
        chunks = splitter.split_documents(all_docs)

        self.vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings())

    def search_with_context(self, query):
        # 관련 문서 검색
        docs = self.vectorstore.similarity_search(query, k=3)

        # LLM으로 답변 생성
        context = "\n\n".join([doc.page_content for doc in docs])

        from openai import OpenAI
        client = OpenAI()

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {query}"
            }]
        )

        return {
            "answer": response.choices[0].message.content,
            "sources": docs
        }

2. 제품 추천

class ProductRecommender:
    def __init__(self, products):
        # 제품 설명을 임베딩
        descriptions = [p['description'] for p in products]

        self.vectorstore = Chroma.from_texts(
            texts=descriptions,
            metadatas=[{"id": p['id'], "price": p['price']} for p in products],
            embedding=OpenAIEmbeddings()
        )

    def recommend(self, user_query, budget=None):
        # 필터 조건
        filter_dict = {}
        if budget:
            filter_dict["price"] = {"$lte": budget}

        # 검색
        results = self.vectorstore.similarity_search(
            query=user_query,
            k=5,
            filter=filter_dict if filter_dict else None
        )

        return results

# 사용
recommender = ProductRecommender(products)
recommendations = recommender.recommend(
    "편안한 운동화",
    budget=100000
)

3. 코드 검색

class CodeSearchEngine:
    def __init__(self, code_repo_path):
        from langchain.document_loaders import DirectoryLoader

        # 코드 파일 로드
        loader = DirectoryLoader(
            code_repo_path,
            glob="**/*.py",
            show_progress=True
        )
        documents = loader.load()

        # 함수 단위로 분할
        from langchain.text_splitter import PythonCodeTextSplitter
        splitter = PythonCodeTextSplitter(chunk_size=500)
        chunks = splitter.split_documents(documents)

        self.vectorstore = Chroma.from_documents(
            chunks,
            OpenAIEmbeddings()
        )

    def search_code(self, natural_language_query):
        """자연어로 코드 검색"""
        results = self.vectorstore.similarity_search(
            natural_language_query,
            k=5
        )
        return results

# 사용
code_search = CodeSearchEngine("./my_project")
results = code_search.search_code("함수 중 이메일 검증하는 코드")

성능 최적화

1. 임베딩 캐싱

import hashlib
import pickle

class CachedEmbeddings:
    def __init__(self, base_embeddings):
        self.base = base_embeddings
        self.cache = {}

    def embed_documents(self, texts):
        results = []
        for text in texts:
            hash_key = hashlib.md5(text.encode()).hexdigest()

            if hash_key in self.cache:
                results.append(self.cache[hash_key])
            else:
                emb = self.base.embed_query(text)
                self.cache[hash_key] = emb
                results.append(emb)

        return results

    def embed_query(self, text):
        return self.embed_documents([text])[0]

# 사용
cached_emb = CachedEmbeddings(OpenAIEmbeddings())
vectorstore = Chroma(embedding_function=cached_emb)

2. 양자화

벡터 크기를 줄여 저장 공간과 검색 속도를 개선합니다.

import numpy as np

def quantize_embeddings(embeddings, bits=8):
    """float32 → int8 변환"""
    emb_array = np.array(embeddings)

    # 정규화
    min_val = emb_array.min()
    max_val = emb_array.max()

    # 양자화
    scale = (2**bits - 1) / (max_val - min_val)
    quantized = ((emb_array - min_val) * scale).astype(np.uint8)

    return quantized, min_val, scale

def dequantize_embeddings(quantized, min_val, scale):
    """int8 → float32 복원"""
    return quantized.astype(np.float32) / scale + min_val

# 75% 메모리 절감

3. 인덱싱 전략

# HNSW (Hierarchical Navigable Small World)
# 더 빠른 검색을 위한 인덱스

import hnswlib

class HNSWSearch:
    def __init__(self, dim=1536):
        self.dim = dim
        self.index = hnswlib.Index(space='cosine', dim=dim)
        self.texts = []

    def add_documents(self, texts, embeddings):
        self.texts.extend(texts)

        if not self.index.get_current_count():
            # 초기화
            self.index.init_index(
                max_elements=len(embeddings),
                ef_construction=200,
                M=16
            )

        self.index.add_items(embeddings)

    def search(self, query_embedding, k=5):
        labels, distances = self.index.knn_query(query_embedding, k=k)

        results = []
        for idx in labels[0]:
            results.append(self.texts[idx])

        return results

평가 및 개선

검색 품질 평가

def evaluate_search(test_queries, expected_docs):
    """Precision@K, Recall@K 계산"""

    total_precision = 0
    total_recall = 0

    for query, expected in zip(test_queries, expected_docs):
        retrieved = vectorstore.similarity_search(query, k=10)
        retrieved_ids = [doc.metadata['id'] for doc in retrieved]

        # Precision: 검색된 것 중 관련 있는 비율
        relevant_retrieved = len(set(retrieved_ids) & set(expected))
        precision = relevant_retrieved / len(retrieved_ids)

        # Recall: 관련 있는 것 중 검색된 비율
        recall = relevant_retrieved / len(expected)

        total_precision += precision
        total_recall += recall

    avg_precision = total_precision / len(test_queries)
    avg_recall = total_recall / len(test_queries)

    return {
        "precision@10": avg_precision,
        "recall@10": avg_recall
    }

지속적 개선

class AdaptiveSearch:
    def __init__(self):
        self.vectorstore = None
        self.click_data = {}  # 클릭 데이터 저장

    def search(self, query):
        results = self.vectorstore.similarity_search(query, k=10)
        return results

    def record_click(self, query, clicked_doc_id):
        """사용자가 어떤 결과를 클릭했는지 기록"""
        if query not in self.click_data:
            self.click_data[query] = []

        self.click_data[query].append(clicked_doc_id)

    def rerank_by_popularity(self, query, results):
        """클릭 데이터로 재정렬"""
        if query not in self.click_data:
            return results

        clicks = self.click_data[query]

        # 클릭 횟수로 정렬
        sorted_results = sorted(
            results,
            key=lambda doc: clicks.count(doc.metadata['id']),
            reverse=True
        )

        return sorted_results

결론

Semantic Search는:

🎯 정확도 30-50% 향상
🌍 다국어 검색 가능
🤖 의미 이해
📈 사용자 만족도 증가

시작하기:

OpenAI Embeddings로 시작
Chroma로 프로토타입
성능 측정
점진적 최적화

의미 기반 검색으로 사용자 경험을 혁신하세요!