하이브리드 검색

from kiwipiepy import Kiwi

kiwi = Kiwi()

# 기본 형태소 분석
text = "Databricks에서 Unity Catalog 권한을 설정하는 방법"
tokens = kiwi.tokenize(text)
for token in tokens:
    print(f"{token.form}\t{token.tag}\t{token.start}~{token.end}")

# 출력:
# Databricks  SL   0~10
# 에서         JKB  10~12
# Unity       SL   13~18
# Catalog     SL   19~26
# 권한         NNG  27~29
# 을          JKO  29~30
# 설정         NNG  31~33
# 하           XSV  33~34
# 는          ETM  34~35
# 방법         NNG  36~38

# BM25용 토크나이저 함수
def kiwi_tokenizer(text: str) -> list[str]:
    """Kiwi 기반 한국어 토크나이저 - BM25 Retriever에 사용"""
    tokens = kiwi.tokenize(text)
    # 명사(NNG, NNP), 동사 어근(VV), 형용사 어근(VA), 외국어(SL) 추출
    meaningful_tags = {"NNG", "NNP", "NNB", "VV", "VA", "SL", "SN"}
    return [t.form for t in tokens if t.tag in meaningful_tags]

# 테스트
print(kiwi_tokenizer("Databricks에서 Unity Catalog 권한을 설정하는 방법"))
# ['Databricks', 'Unity', 'Catalog', '권한', '설정', '하', '방법']

Final_score(d) = Σ weight_i * (1 / (rank_i(d) + rank_constant))

문서 A: 벡터 검색 1등, 키워드 검색 3등
  → 0.5 * (1/(1+60)) + 0.5 * (1/(3+60)) = 0.5 * 0.0164 + 0.5 * 0.0159 = 0.01615

문서 B: 벡터 검색 5등, 키워드 검색 1등
  → 0.5 * (1/(5+60)) + 0.5 * (1/(1+60)) = 0.5 * 0.0154 + 0.5 * 0.0164 = 0.01590

문서 C: 벡터 검색 2등, 키워드 검색 2등
  → 0.5 * (1/(2+60)) + 0.5 * (1/(2+60)) = 0.5 * 0.0161 + 0.5 * 0.0161 = 0.01613

최종 순위: A > C > B (양쪽에서 모두 높은 순위를 받은 문서가 유리)

RRF_score(d) = Σ 1 / (k + rank_i(d))

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_databricks import DatabricksVectorSearch

# BM25 Retriever (키워드 기반)
bm25_retriever = BM25Retriever.from_documents(documents, k=5)

# Vector Search Retriever (의미 기반)
vs_retriever = DatabricksVectorSearch(
    endpoint="vs-endpoint",
    index_name="catalog.schema.index",
    columns=["content", "source"]
).as_retriever(search_kwargs={"k": 5})

# 앙상블 (RRF)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vs_retriever],
    weights=[0.4, 0.6]  # Vector Search에 더 높은 가중치
)

# 검색 실행
results = ensemble_retriever.invoke("Databricks에서 Delta Lake 사용법")
for doc in results:
    print(f"[{doc.metadata.get('source', 'N/A')}] {doc.page_content[:100]}...")

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_databricks import ChatDatabricks

llm = ChatDatabricks(endpoint="databricks-claude-sonnet-4")

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vs_retriever,
    llm=llm,
)

# "Delta Lake 성능 최적화" →
#   1. "Delta Lake 쿼리 속도를 높이는 방법"
#   2. "Delta Lake Z-Order와 파티셔닝 전략"
#   3. "Delta Lake 파일 최적화 설정"
results = multi_query_retriever.invoke("Delta Lake 성능 최적화 방법은?")

from databricks.vector_search.client import VectorSearchClient

vsc = VectorSearchClient()

# 1. Vector Search 엔드포인트 생성
vsc.create_endpoint(name="korean-rag-endpoint", endpoint_type="STANDARD")

# 2. Delta Sync 인덱스 생성 (한국어 지원 설정)
index = vsc.create_delta_sync_index(
    endpoint_name="korean-rag-endpoint",
    index_name="catalog.schema.korean_docs_index",
    source_table_name="catalog.schema.korean_docs",
    primary_key="doc_id",
    pipeline_type="TRIGGERED",
    embedding_source_column="content_raw",        # 원본 텍스트로 임베딩
    embedding_model_endpoint_name="databricks-bge-large-en",  # 또는 bge-m3
    columns_to_sync=["content_raw", "content_analyzed", "source", "metadata"]
)

from kiwipiepy import Kiwi
from langchain_community.retrievers import BM25Retriever
from langchain.schema import Document

kiwi = Kiwi()

def kiwi_tokenizer(text: str) -> list[str]:
    """Kiwi 기반 한국어 토크나이저"""
    tokens = kiwi.tokenize(text)
    meaningful_tags = {"NNG", "NNP", "NNB", "VV", "VA", "SL", "SN"}
    return [t.form for t in tokens if t.tag in meaningful_tags]

# 문서 로드 (Delta Lake에서 읽어온 청크들)
documents = [
    Document(page_content=row["content_raw"], metadata={"source": row["source"]})
    for row in spark.table("catalog.schema.korean_docs").collect()
]

# Kiwi 토크나이저를 적용한 BM25 Retriever
bm25_retriever = BM25Retriever.from_documents(
    documents,
    preprocess_func=kiwi_tokenizer,  # 한국어 형태소 분석 적용
    k=10
)

# 벡터 검색 + BM25 앙상블
from langchain.retrievers import EnsembleRetriever
from langchain_databricks import DatabricksVectorSearch

vs_retriever = DatabricksVectorSearch(
    endpoint="korean-rag-endpoint",
    index_name="catalog.schema.korean_docs_index",
    columns=["content_raw", "source"]
).as_retriever(search_kwargs={"k": 10})

# 최종 앙상블 Retriever
korean_hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vs_retriever],
    weights=[0.4, 0.6]  # 도메인에 따라 조정
)

# Databricks VS 내장 하이브리드 검색 (영어 콘텐츠에 적합)
results = index.similarity_search(
    query_text="How to set up Unity Catalog permissions",
    query_type="hybrid",    # Dense + Full Text Search 결합
    num_results=10,
    columns=["content_raw", "source"]
)

# Full Text Search만 사용 (키워드 검색만 필요한 경우)
results = index.similarity_search(
    query_text="DELTA_TABLE_NOT_FOUND",
    query_type="FULL_TEXT",  # 키워드 검색만
    num_results=10,
    columns=["content_raw", "source"]
)