한국어 RAG 최적화

"데이터브릭스" (원형)
→ "데이터브릭스가" (주격), "데이터브릭스를" (목적격)
→ "데이터브릭스에서" (처소격), "데이터브릭스의" (관형격)
→ "데이터브릭스로" (도구격), "데이터브릭스와" (공동격)
→ 영어: "Databricks" 하나의 형태로 모든 문법적 관계 표현

영어: "I   love   Databricks"   → 공백으로 나누면 3개 단어, 각각이 의미 단위
한국어: "데이터브릭스에서"         → 1개 어절이지만, 2개 형태소
       → "데이터브릭스" (고유명사, 의미 핵심) + "에서" (조사, 문법 기능)

형태소 분석 + 의미 형태소 추출:
  "데이터브릭스에서" → [데이터브릭스, 에서] → "데이터브릭스" ✓
  "데이터브릭스를"   → [데이터브릭스, 를]   → "데이터브릭스" ✓
  "데이터브릭스의"   → [데이터브릭스, 의]   → "데이터브릭스" ✓
  → 조사가 달라도 동일한 검색 토큰으로 통일!

입력: "데이터브릭스에서 머신러닝 파이프라인을 구축했습니다"

분석 결과:
  데이터브릭스 (NNP, 고유명사) — 의미 핵심 ✓
  에서       (JKB, 부사격 조사) — 문법 기능, 검색에 불필요
  머신러닝    (NNG, 일반명사) — 의미 핵심 ✓
  파이프라인   (NNG, 일반명사) — 의미 핵심 ✓
  을         (JKO, 목적격 조사) — 문법 기능, 검색에 불필요
  구축       (NNG, 일반명사) — 의미 핵심 ✓
  했         (XSA, 과거 시제 접사) — 문법 기능, 검색에 불필요
  습니다      (EF, 종결 어미) — 문법 기능, 검색에 불필요

→ 검색용 토큰: ["데이터브릭스", "머신러닝", "파이프라인", "구축"]

pip install kiwipiepy

from kiwipiepy import Kiwi

kiwi = Kiwi()

# 형태소 분석
result = kiwi.tokenize("Databricks에서 RAG 파이프라인을 구축합니다")
for token in result:
    print(f"{token.form}\t{token.tag}\t{token.start}-{token.end}")
# Databricks  NNP    0-11
# 에서         JKB    11-13
# RAG          SL     14-17
# 파이프라인    NNG    18-23
# 을           JKO    23-24
# 구축         NNG    25-27
# 합니다       XSV+EF 27-30

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