Vector Search — 벡터 유사도 검색

"강아지"  → [0.23, -0.15, 0.87, 0.42, ...]  (1024차원)
"개"      → [0.25, -0.13, 0.85, 0.40, ...]  ← 가까움! (유사함)
"자동차"  → [-0.71, 0.34, -0.28, 0.55, ...] ← 멂! (다름)

from databricks.vector_search.client import VectorSearchClient

vsc = VectorSearchClient()

# Endpoint 생성
vsc.create_endpoint(
    name="vs-endpoint-prod",
    endpoint_type="STANDARD"  # STANDARD 또는 STORAGE_OPTIMIZED
)

index = vsc.create_delta_sync_index(
    endpoint_name="vs-endpoint-prod",
    index_name="catalog.schema.docs_index",
    source_table_name="catalog.schema.documents",
    primary_key="doc_id",
    embedding_source_column="content",           # 임베딩할 텍스트 컬럼
    embedding_model_endpoint_name="databricks-gte-large-en",  # 임베딩 모델
    pipeline_type="TRIGGERED",  # TRIGGERED: 수동/스케줄 | CONTINUOUS: 실시간
    columns_to_sync=["doc_id", "content", "title", "source", "updated_at"]
)

# 사전 준비: 임베딩 벡터가 포함된 Delta 테이블
# | doc_id | content         | embedding (array<float>) |
# |--------|-----------------|--------------------------|
# | 1      | "환불 절차..."   | [0.23, -0.15, 0.87, ...] |

index = vsc.create_delta_sync_index(
    endpoint_name="vs-endpoint-prod",
    index_name="catalog.schema.docs_index_self",
    source_table_name="catalog.schema.documents_with_embeddings",
    primary_key="doc_id",
    embedding_vector_column="embedding",   # 이미 계산된 벡터 컬럼
    embedding_dimension=1024,              # 벡터 차원 수
    pipeline_type="TRIGGERED"
)
| 단계 | 처리 | 결과 |
|------|------|------|
| 1 | 벡터 검색 | 의미적으로 유사한 Top-20 |
| 2 | 키워드 검색 (BM25) | 키워드 매칭 Top-20 |
| 3 | 결과 병합 (RRF) | 중복 제거, 점수 융합 → Top-20 |
| 4 | Reranker | LLM 기반 재순위화 → Top-5 |
| 5 | LLM 답변 생성 | 최종 응답 |python
# Direct Access 인덱스 생성
index = vsc.create_direct_access_index(
    endpoint_name="vs-endpoint-prod",
    index_name="catalog.schema.docs_direct",
    primary_key="doc_id",
    embedding_dimension=1024,
    schema={
        "doc_id": "string",
        "content": "string",
        "category": "string",
        "embedding": "array<float>"
    }
)

# 벡터 삽입 (upsert)
index.upsert([
    {
        "doc_id": "doc-001",
        "content": "환불은 구매 후 14일 이내에 가능합니다.",
        "category": "정책",
        "embedding": [0.23, -0.15, 0.87, ...]  # 직접 계산한 벡터
    },
    {
        "doc_id": "doc-002",
        "content": "배송은 영업일 기준 2-3일 소요됩니다.",
        "category": "배송",
        "embedding": [0.11, 0.42, -0.33, ...]
    }
])

# 벡터 삭제
index.delete(["doc-001"])

results = vsc.get_index(
    endpoint_name="vs-endpoint-prod",
    index_name="catalog.schema.docs_index"
).similarity_search(
    query_text="반품 및 환불 절차가 어떻게 되나요?",
    columns=["doc_id", "title", "content", "source"],
    num_results=5
)

for doc in results['result']['data_array']:
    print(f"Score: {doc[-1]:.4f} | Title: {doc[1]}")

results = index.similarity_search(
    query_text="데이터 보안 정책",
    columns=["doc_id", "title", "content", "category"],
    filters={"category": "보안"},
    num_results=3
)

results = index.similarity_search(
    query_text="반품 시 배송비는 누가 부담하나요?",
    columns=["doc_id", "content"],
    num_results=20,  # 넓게 검색
    query_options={
        "reranker": {
            "model_name": "databricks-reranker",
            "columns": ["content"],
            "top_k": 5  # 상위 5개 선별
        }
    }
)

# 대규모 인덱스 상태 모니터링
index = vsc.get_index(
    endpoint_name="vs-endpoint-prod",
    index_name="catalog.schema.docs_index"
)

# 인덱스 상태 확인
status = index.describe()
print(f"상태: {status.get('status')}")
print(f"인덱싱된 문서 수: {status.get('num_docs')}")
print(f"마지막 동기화: {status.get('last_sync_time')}")

[사용자 질문]
| 단계 | 처리 | 결과 |
|------|------|------|
| 1 | 벡터 검색 | 의미적 유사 Top-20 |
| 2 | 키워드 검색 (BM25) | 키워드 매칭 Top-20 |
| 3 | 결과 병합 (RRF) | 중복 제거, 점수 융합 → Top-20 |
| 4 | Reranker | LLM 기반 재순위화 → Top-5 |
| 5 | LLM 답변 생성 | 최종 응답 |

# Reciprocal Rank Fusion (RRF) 구현 예시
def reciprocal_rank_fusion(vector_results, keyword_results, k=60):
    """두 검색 결과를 RRF 알고리즘으로 융합합니다."""
    scores = {}

    for rank, doc in enumerate(vector_results):
        doc_id = doc['doc_id']
        scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)

    for rank, doc in enumerate(keyword_results):
        doc_id = doc['doc_id']
        scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)

    # 점수 내림차순 정렬
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

# 1단계: 벡터 검색
vector_results = index.similarity_search(
    query_text="산업안전보건법 제44조 위반 시 벌칙",
    columns=["doc_id", "content"],
    num_results=20
)

# 2단계: 키워드 검색 (Delta 테이블에서 SQL로)
keyword_results = spark.sql("""
    SELECT doc_id, content
    FROM catalog.schema.documents
    WHERE content LIKE '%제44조%' AND content LIKE '%벌칙%'
    LIMIT 20
""").collect()

# 3단계: RRF 융합
fused = reciprocal_rank_fusion(vector_results, keyword_results)

# 4단계: Reranker로 최종 정렬 (상위 5개 선별)
final_results = index.similarity_search(
    query_text="산업안전보건법 제44조 위반 시 벌칙",
    columns=["doc_id", "content"],
    num_results=20,
    query_options={
        "reranker": {
            "model_name": "databricks-reranker",
            "columns": ["content"],
            "top_k": 5
        }
    }
)

# 한국어 최적화 청킹 예시 (재귀적 분리 + 형태소 인식)
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 한국어에 최적화된 구분자 순서
korean_splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ".", "!", "?", "。", " ", ""],
    chunk_size=500,        # 한국어는 영어 대비 정보 밀도가 높으므로 500자 권장
    chunk_overlap=50,      # 문맥 유지를 위한 오버랩
    length_function=len,
    is_separator_regex=False
)

# 청크 생성
chunks = korean_splitter.split_text(document_text)

# 비용 효율적인 인덱스 설정 예시
index = vsc.create_delta_sync_index(
    endpoint_name="vs-endpoint-prod",
    index_name="catalog.schema.docs_index",
    source_table_name="catalog.schema.documents",
    primary_key="doc_id",
    embedding_source_column="content",
    embedding_model_endpoint_name="databricks-gte-large-en",
    pipeline_type="TRIGGERED",    # 비용 절감: 수동 트리거
    columns_to_sync=[             # 최소한의 컬럼만 동기화
        "doc_id", "content", "title", "category"
        # ❌ "full_html", "raw_pdf" 등 대용량 컬럼 제외
    ]
)