Semantic Search

# Semantic Search 구현 예제
from openai import OpenAI
import numpy as np
from typing import List

client = OpenAI()

def get_embedding(text: str) -> List[float]:
    """텍스트를 벡터로 변환"""
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    )
    return response.data[0].embedding

def cosine_similarity(a: List[float], b: List[float]) -> float:
    """코사인 유사도 계산"""
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 문서 데이터베이스 (실제로는 벡터 DB 사용)
documents = [
    "인공지능은 컴퓨터가 인간처럼 학습하고 판단하는 기술입니다.",
    "머신러닝은 데이터에서 패턴을 학습하는 AI의 한 분야입니다.",
    "딥러닝은 신경망을 여러 층으로 쌓아 복잡한 패턴을 학습합니다.",
    "자연어 처리는 컴퓨터가 인간의 언어를 이해하는 기술입니다.",
    "오늘 서울 날씨는 맑고 기온은 15도입니다."
]

# 문서 임베딩 (사전 계산)
doc_embeddings = [get_embedding(doc) for doc in documents]

def semantic_search(query: str, top_k: int = 3) -> List[tuple]:
    """Semantic Search 실행"""
    query_embedding = get_embedding(query)

    # 모든 문서와 유사도 계산
    scores = [
        (doc, cosine_similarity(query_embedding, emb))
        for doc, emb in zip(documents, doc_embeddings)
    ]

    # 유사도 높은 순으로 정렬
    scores.sort(key=lambda x: x[1], reverse=True)
    return scores[:top_k]

# 검색 예시
query = "AI가 스스로 배우는 방법"
results = semantic_search(query)

print(f"Query: {query}\n")
for doc, score in results:
    print(f"[{score:.3f}] {doc}")

# 출력:
# [0.892] 머신러닝은 데이터에서 패턴을 학습하는 AI의 한 분야입니다.
# [0.856] 인공지능은 컴퓨터가 인간처럼 학습하고 판단하는 기술입니다.
# [0.823] 딥러닝은 신경망을 여러 층으로 쌓아 복잡한 패턴을 학습합니다.

# 키워드 검색이었다면 "AI", "스스로", "배우는"이 없어서 결과가 없었을 것!

# === Hybrid Search 구현 (BM25 + Semantic) ===
from rank_bm25 import BM25Okapi
import numpy as np

class HybridSearch:
    def __init__(self, documents: List[str], alpha: float = 0.5):
        """
        alpha: semantic weight (1-alpha = BM25 weight)
        """
        self.documents = documents
        self.alpha = alpha

        # BM25 인덱스
        tokenized_docs = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)

        # Semantic 임베딩
        self.doc_embeddings = [get_embedding(doc) for doc in documents]

    def search(self, query: str, top_k: int = 5) -> List[tuple]:
        # BM25 점수
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_scores = bm25_scores / (bm25_scores.max() + 1e-8)  # 정규화

        # Semantic 점수
        query_emb = get_embedding(query)
        semantic_scores = np.array([
            cosine_similarity(query_emb, doc_emb)
            for doc_emb in self.doc_embeddings
        ])

        # Hybrid 점수 (가중 합)
        hybrid_scores = self.alpha * semantic_scores + (1 - self.alpha) * bm25_scores

        # 정렬 및 반환
        indices = np.argsort(hybrid_scores)[::-1][:top_k]
        return [(self.documents[i], hybrid_scores[i]) for i in indices]

# Pinecone 벡터 DB 사용 예시
from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")
index = pc.Index("my-semantic-index")

# 문서 업로드 (임베딩 + 메타데이터)
vectors = [
    {"id": f"doc_{i}", "values": get_embedding(doc), "metadata": {"text": doc}}
    for i, doc in enumerate(documents)
]
index.upsert(vectors=vectors)

# 쿼리
query_embedding = get_embedding("머신러닝 학습 방법")
results = index.query(vector=query_embedding, top_k=5, include_metadata=True)

for match in results.matches:
    print(f"[{match.score:.3f}] {match.metadata['text']}")

# === 문서 청킹 전략 ===
def chunk_documents(text: str, chunk_size: int = 500, overlap: int = 100) -> List[str]:
    """문서를 적절한 크기로 청킹 (overlap으로 문맥 유지)"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start += chunk_size - overlap  # 겹침 적용
    return chunks

# === 임베딩 모델 비교 ===
# text-embedding-3-small (OpenAI): 1536차원, $0.02/1M 토큰
# text-embedding-3-large (OpenAI): 3072차원, $0.13/1M 토큰
# multilingual-e5-large: 오픈소스, 한국어 지원 우수
# BGE-M3: 다국어 + sparse/dense 하이브리드

# === Reranking (2단계 검색) ===
# 1단계: 빠른 벡터 검색으로 후보 100개 추출
# 2단계: Cross-encoder로 정밀 재순위

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# 후보 문서 재순위
query = "파이썬 리스트 정렬 방법"
candidates = ["리스트.sort() 메서드 사용", "sorted() 함수 활용", "배열 병합 정렬"]
scores = reranker.predict([(query, doc) for doc in candidates])
# rerank된 결과 반환

# === 평가 메트릭 ===
def evaluate_search(queries, ground_truth, search_fn, k=10):
    """검색 품질 평가 (Recall@k, MRR)"""
    recalls = []
    mrrs = []

    for query, relevant_ids in zip(queries, ground_truth):
        results = search_fn(query, top_k=k)
        result_ids = [r['id'] for r in results]

        # Recall@k
        found = len(set(result_ids) & set(relevant_ids))
        recalls.append(found / len(relevant_ids))

        # MRR (Mean Reciprocal Rank)
        for i, rid in enumerate(result_ids):
            if rid in relevant_ids:
                mrrs.append(1 / (i + 1))
                break
        else:
            mrrs.append(0)

    return {
        "Recall@k": sum(recalls) / len(recalls),
        "MRR": sum(mrrs) / len(mrrs)
    }

# === 프로덕션 팁 ===
# 1. 배치 임베딩: GPU 활용으로 100배 이상 빠름
# 2. 인덱스 샤딩: 대규모 문서는 여러 인덱스로 분산
# 3. 캐싱: 자주 쓰는 쿼리 임베딩은 캐시
# 4. ANN 알고리즘: HNSW, IVF로 근사 검색 (99% 정확도, 10배 빠름)
# 5. 주기적 재인덱싱: 데이터 변경 시 임베딩 업데이트 필요