한국어 RAG 최적화 — 심화

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]
)

# 설치: pip install kss
import kss

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

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

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)

import mlflow
from sentence_transformers import SentenceTransformer

# 1. 모델 로드 및 MLflow에 로깅
model = SentenceTransformer("BM-K/KoSimCSE-roberta-multitask")

with mlflow.start_run():
    mlflow.sentence_transformers.log_model(
        model=model,
        artifact_path="kosimcse",
        registered_model_name="catalog.schema.kosimcse_embedding",
        input_example=["한국어 임베딩 테스트"],
    )

# 2. Model Serving Endpoint 생성
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import EndpointCoreConfigInput, ServedEntityInput

w = WorkspaceClient()
w.serving_endpoints.create_and_wait(
    name="kosimcse-embedding",
    config=EndpointCoreConfigInput(
        served_entities=[
            ServedEntityInput(
                entity_name="catalog.schema.kosimcse_embedding",
                entity_version="1",
                workload_size="Small",
                scale_to_zero_enabled=True,
            )
        ]
    ),
)

from langchain_databricks import DatabricksEmbeddings

# Databricks에서 기본 제공하는 다국어 임베딩
embeddings = DatabricksEmbeddings(
    endpoint="databricks-gte-large-en"  # 또는 커스텀 배포 엔드포인트
)

# 한국어 텍스트 임베딩
vectors = embeddings.embed_documents([
    "Databricks에서 RAG 파이프라인을 구축합니다",
    "데이터 레이크하우스 아키텍처 개요"
])

LLM에 전달되는 컨텍스트 (5개 문서):

  [문서 1] ← 높은 활용도 ✅  (컨텍스트의 시작)
  [문서 2] ← 보통
  [문서 3] ← 가장 낮은 활용도 ❌  (컨텍스트의 중간 = "사각지대")
  [문서 4] ← 보통
  [문서 5] ← 높은 활용도 ✅  (컨텍스트의 끝)

질문: "Databricks에서 Delta Lake 테이블의 OPTIMIZE 명령 실행 주기는?"

── Re-ranking 없이 (1차 검색 결과 그대로) ──
1위: Delta Lake 개요 및 특징 소개          ← 주제는 맞지만 답변 없음
2위: OPTIMIZE와 VACUUM 명령어 상세 가이드   ← 정답이 여기에! (2위)
3위: Delta Lake 트랜잭션 로그 구조          ← 관련은 있지만 답변 없음
4위: Databricks SQL로 테이블 관리하기       ← 간접 관련
5위: 파티셔닝 전략과 Z-ORDER               ← 간접 관련

── Re-ranking 후 (Cross-encoder가 재정렬) ──
1위: OPTIMIZE와 VACUUM 명령어 상세 가이드   ← 정답! 최상위로 올라옴 ✅
2위: 파티셔닝 전략과 Z-ORDER               ← OPTIMIZE와 직접 연관
3위: Delta Lake 개요 및 특징 소개
4위: Databricks SQL로 테이블 관리하기
5위: Delta Lake 트랜잭션 로그 구조

import re
from kiwipiepy import Kiwi

kiwi = Kiwi()

def preprocess_korean(text: str) -> str:
    """한국어 RAG를 위한 텍스트 전처리"""
    # 1. 불필요한 공백 정리
    text = re.sub(r'\s+', ' ', text).strip()

    # 2. 특수문자 정리 (문장부호는 유지)
    text = re.sub(r'[^\w\s가-힣a-zA-Z0-9.,!?·\-()]', '', text)

    # 3. 한자 → 한글 변환 (Kiwi 내장 기능)
    # Kiwi는 분석 시 자동으로 한자를 한글로 매핑

    return text