BM25 검색 & 청킹 전략

from kiwipiepy import Kiwi
from langchain_community.retrievers import BM25Retriever

kiwi = Kiwi()

def kiwi_tokenize(text: str) -> list[str]:
    """Kiwi 형태소 분석기로 명사/동사/형용사/외국어만 추출"""
    tokens = kiwi.tokenize(text)
    # NNG(일반명사), NNP(고유명사), VV(동사), VA(형용사), SL(외국어)
    return [t.form for t in tokens if t.tag in ('NNG', 'NNP', 'VV', 'VA', 'SL')]

# Kiwi 토크나이저를 사용하는 BM25 Retriever
bm25_retriever = BM25Retriever.from_documents(
    documents,
    preprocess_func=kiwi_tokenize,
    k=5
)

# "데이터브릭스에서 RAG를 구축하는 방법" →
# kiwi_tokenize → ["데이터브릭스", "RAG", "구축", "방법"]
results = bm25_retriever.invoke("데이터브릭스에서 RAG를 구축하는 방법")

from langchain.retrievers import EnsembleRetriever
from langchain_databricks import DatabricksVectorSearch

# Kiwi 기반 BM25
bm25_retriever = BM25Retriever.from_documents(
    documents, preprocess_func=kiwi_tokenize, k=5
)

# Databricks Vector Search (다국어 임베딩)
vs_retriever = DatabricksVectorSearch(
    endpoint="vs-endpoint",
    index_name="catalog.schema.ko_docs_index",
    columns=["content", "source"]
).as_retriever(search_kwargs={"k": 5})

# 앙상블
ensemble = EnsembleRetriever(
    retrievers=[bm25_retriever, vs_retriever],
    weights=[0.4, 0.6]
)

from kiwipiepy import Kiwi
from rank_bm25 import BM25Okapi
import numpy as np

kiwi = Kiwi()

def korean_tokenize(text):
    """Kiwi로 의미 형태소(명사, 동사, 형용사)만 추출"""
    tokens = kiwi.tokenize(text)
    return [t.form for t in tokens if t.tag in ('NNG', 'NNP', 'VV', 'VA')]

def reciprocal_rank_fusion(vector_results, bm25_results, k=60):
    """두 검색기의 결과를 RRF 알고리즘으로 결합합니다.

    Args:
        vector_results: 벡터 검색 결과 (doc_id 리스트, 관련도순 정렬)
        bm25_results:   BM25 검색 결과 (doc_id 리스트, 관련도순 정렬)
        k: 순위 완화 파라미터 (기본값 60, 논문 권장값)
    """
    scores = {}
    for rank, doc_id in enumerate(vector_results):
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
    for rank, doc_id in enumerate(bm25_results):
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

# 사용 예시
docs = {
    "doc_001": "Databricks는 데이터와 AI를 위한 통합 레이크하우스 플랫폼입니다",
    "doc_002": "Delta Lake는 ACID 트랜잭션과 스키마 관리를 지원합니다",
    "doc_003": "Unity Catalog로 테이블, 모델, 볼륨의 접근 권한을 관리합니다",
    "doc_004": "Vector Search로 임베딩 기반 유사도 검색을 수행합니다",
}

doc_ids = list(docs.keys())
tokenized = [korean_tokenize(text) for text in docs.values()]
bm25 = BM25Okapi(tokenized)

query = "데이터 플랫폼에서 접근 권한 관리"
bm25_scores = bm25.get_scores(korean_tokenize(query))
bm25_ranked = [doc_ids[i] for i in np.argsort(bm25_scores)[::-1]]

# 벡터 검색 결과 (Databricks Vector Search API로 조회)
vector_ranked = ["doc_001", "doc_004", "doc_003", "doc_002"]

fused = reciprocal_rank_fusion(vector_ranked, bm25_ranked, k=60)
for doc_id, score in fused[:5]:
    print(f"[RRF: {score:.4f}] {doc_id}: {docs[doc_id]}")

# 설치: pip install kss
import kss

text = """Databricks는 데이터와 AI를 위한 통합 플랫폼입니다.
Delta Lake를 기반으로 데이터 레이크하우스 아키텍처를 제공합니다.
Unity Catalog로 데이터 거버넌스를 통합 관리할 수 있습니다."""

sentences = kss.split_sentences(text)
for s in sentences:
    print(s)

from langchain.text_splitter import RecursiveCharacterTextSplitter

korean_splitter = RecursiveCharacterTextSplitter(
    separators=[
        "\n\n",    # 문단 구분
        "\n",      # 줄바꿈
        "다. ",    # 평서문 종결
        "요. ",    # 존댓말 종결
        "까? ",    # 의문문 종결
        ". ",      # 일반 마침표
        " ",       # 공백
    ],
    chunk_size=500,
    chunk_overlap=50,
    length_function=len,
)

chunks = korean_splitter.split_text(long_korean_text)