Vector Database 엔지니어 커리어 가이드: Pinecone·Weaviate·Milvus·pgvector 완전 비교와 RAG 시대의 필수 역량

사용자 질문
    ↓
질문을 벡터로 변환 (Embedding Model)
    ↓
벡터 DB에서 유사한 문서 벡터 검색 (ANN Search)
    ↓
검색된 문서를 컨텍스트로 LLM에 전달
    ↓
LLM이 컨텍스트 기반으로 답변 생성

from openai import OpenAI

client = OpenAI()

# 텍스트를 벡터로 변환
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Vector databases are essential for RAG applications"
)

# 결과: 1536차원의 float 배열
embedding = response.data[0].embedding
print(f"Dimension: {len(embedding)}")  # 1536
print(f"Sample: {embedding[:5]}")  # [0.0123, -0.0456, 0.0789, ...]

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def euclidean_distance(a, b):
    return np.linalg.norm(a - b)

def dot_product(a, b):
    return np.dot(a, b)

# 예시
v1 = np.array([1.0, 2.0, 3.0])
v2 = np.array([1.1, 2.1, 2.9])
v3 = np.array([-1.0, -2.0, -3.0])

print(f"v1-v2 cosine: {cosine_similarity(v1, v2):.4f}")   # 0.9998 (매우 유사)
print(f"v1-v3 cosine: {cosine_similarity(v1, v3):.4f}")   # -1.0000 (정반대)
print(f"v1-v2 L2: {euclidean_distance(v1, v2):.4f}")       # 0.1732

# HNSW 주요 파라미터
hnsw_params = {
    "M": 16,              # 각 노드의 최대 연결 수 (높을수록 정확, 메모리 증가)
    "ef_construction": 200, # 인덱스 구축 시 탐색 범위 (높을수록 품질 향상)
    "ef_search": 100       # 검색 시 탐색 범위 (높을수록 정확, 느림)
}

# IVF-Flat 주요 파라미터
ivf_params = {
    "nlist": 1024,    # 클러스터 수 (보통 sqrt(N) ~ 4*sqrt(N))
    "nprobe": 16      # 검색할 클러스터 수 (높을수록 정확, 느림)
}

# 예: 1536차원 벡터 → PQ 압축
# 원본: 1536 * 4 bytes = 6,144 bytes
# PQ(m=96, nbits=8): 96 bytes (64배 압축)

from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")

# 인덱스 생성
pc.create_index(
    name="my-index",
    dimension=1536,
    metric="cosine",
    spec={
        "serverless": {
            "cloud": "aws",
            "region": "us-east-1"
        }
    }
)

# 벡터 업서트
index = pc.Index("my-index")
index.upsert(
    vectors=[
        ("id1", [0.1, 0.2, ...], {"text": "example document"}),
        ("id2", [0.3, 0.4, ...], {"text": "another document"})
    ],
    namespace="my-namespace"
)

# 검색
results = index.query(
    vector=[0.15, 0.25, ...],
    top_k=10,
    include_metadata=True,
    namespace="my-namespace"
)

import weaviate
from weaviate.classes.config import Configure, Property, DataType

client = weaviate.connect_to_local()

# 컬렉션 생성 (하이브리드 검색 지원)
collection = client.collections.create(
    name="Document",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(),
    properties=[
        Property(name="text", data_type=DataType.TEXT),
        Property(name="source", data_type=DataType.TEXT),
    ]
)

# 데이터 삽입 (자동 벡터화)
collection.data.insert(
    properties={
        "text": "Vector databases power modern AI",
        "source": "blog"
    }
)

# 하이브리드 검색
results = collection.query.hybrid(
    query="vector search for AI",
    alpha=0.75,  # 0=BM25 only, 1=vector only
    limit=10
)

client.close()

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 연결
connections.connect("default", host="localhost", port="19530")

# 스키마 정의
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536)
]
schema = CollectionSchema(fields)

# 컬렉션 생성
collection = Collection("documents", schema)

# 인덱스 생성
index_params = {
    "metric_type": "COSINE",
    "index_type": "HNSW",
    "params": {"M": 16, "efConstruction": 256}
}
collection.create_index("embedding", index_params)

# 검색
collection.load()
results = collection.search(
    data=[[0.1, 0.2, ...]],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 128}},
    limit=10,
    output_fields=["text"]
)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(host="localhost", port=6333)

# 컬렉션 생성
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

# 데이터 삽입
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],
            payload={"text": "example document", "source": "blog"}
        )
    ]
)

# 검색 (필터링 포함)
results = client.search(
    collection_name="documents",
    query_vector=[0.15, 0.25, ...],
    query_filter={
        "must": [{"key": "source", "match": {"value": "blog"}}]
    },
    limit=10
)

-- pgvector 확장 설치
CREATE EXTENSION vector;

-- 벡터 컬럼이 있는 테이블 생성
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    text TEXT,
    embedding vector(1536)
);

-- HNSW 인덱스 생성
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- 벡터 검색 (코사인 유사도)
SELECT id, text, 1 - (embedding <=> query_embedding) AS similarity
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

-- 필터 + 벡터 검색 조합
SELECT id, text
FROM documents
WHERE metadata->>'source' = 'blog'
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

import chromadb

client = chromadb.Client()

# 컬렉션 생성
collection = client.create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}
)

# 데이터 삽입 (자동 임베딩 포함)
collection.add(
    documents=["Vector databases are essential", "RAG needs vector search"],
    metadatas=[{"source": "blog"}, {"source": "paper"}],
    ids=["doc1", "doc2"]
)

# 검색
results = collection.query(
    query_texts=["AI infrastructure"],
    n_results=5,
    where={"source": "blog"}
)

# 임베딩 모델 선택 의사결정 트리
def choose_embedding_model(requirements):
    if requirements["budget"] == "minimal":
        if requirements["quality"] == "high":
            return "E5-large-v2 or BGE-large"  # 무료 + 높은 성능
        else:
            return "nomic-embed-text"  # 무료 + 경량

    if requirements["multilingual"]:
        return "Cohere embed-v4"  # 다국어 최강

    if requirements["max_quality"]:
        return "text-embedding-3-large"  # OpenAI 최고 성능

    # 기본 추천
    return "text-embedding-3-small"  # 가성비 최고

# 차원 축소 예시
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Vector databases are essential",
    dimensions=256  # 1536 -> 256으로 축소 (메모리 6배 절약)
)

# Weaviate 하이브리드 검색 예시
results = collection.query.hybrid(
    query="kubernetes pod autoscaling",
    alpha=0.75,  # 0.75 = 벡터 75% + BM25 25%
    limit=10,
    return_metadata=["score", "explain_score"]
)

# alpha 값 가이드:
# alpha=1.0: 벡터 검색만 (의미적 검색)
# alpha=0.0: BM25만 (키워드 검색)
# alpha=0.5: 균등 혼합
# alpha=0.7-0.8: 대부분의 RAG 시스템에서 최적

def reciprocal_rank_fusion(vector_results, keyword_results, k=60):
    """
    RRF로 두 검색 결과를 결합
    k: 순위 감쇠 상수 (높을수록 하위 순위도 가중치 부여)
    """
    fused_scores = {}

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

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

    # 점수 기준 정렬
    sorted_results = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

from sentence_transformers import CrossEncoder

# Cross-Encoder 모델 로드
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

# 초기 검색 결과 리랭킹
query = "How does HNSW algorithm work?"
passages = [doc["text"] for doc in initial_results]

# 쿼리-문서 쌍의 점수 계산
pairs = [[query, passage] for passage in passages]
scores = reranker.predict(pairs)

# 점수 기준 재정렬
reranked = sorted(zip(passages, scores), key=lambda x: x[1], reverse=True)

# 데이터 규모별 권장 인덱스 전략
indexing_strategy = {
    "small": {  # < 100K 벡터
        "index": "HNSW",
        "params": {"M": 16, "ef_construction": 200},
        "reason": "작은 규모에서 HNSW가 최고의 정확도 제공"
    },
    "medium": {  # 100K - 10M 벡터
        "index": "HNSW",
        "params": {"M": 32, "ef_construction": 256},
        "reason": "메모리가 허용하면 HNSW, 아니면 IVF-Flat"
    },
    "large": {  # 10M - 1B 벡터
        "index": "IVF-PQ",
        "params": {"nlist": 4096, "m": 96, "nbits": 8},
        "reason": "메모리 효율이 핵심, PQ로 압축 필수"
    },
    "massive": {  # > 1B 벡터
        "index": "IVF-PQ + 분산",
        "params": {"shards": 8, "replicas": 3},
        "reason": "단일 노드 불가, 분산 클러스터 필수"
    }
}

# 벡터 DB 핵심 모니터링 지표
monitoring_metrics = {
    "performance": {
        "query_latency_p50": "< 10ms",
        "query_latency_p95": "< 50ms",
        "query_latency_p99": "< 100ms",
        "queries_per_second": "> 1000 QPS",
        "index_build_time": "모니터링"
    },
    "quality": {
        "recall_at_10": "> 95%",
        "recall_at_100": "> 99%",
        "embedding_drift": "주기적 측정"
    },
    "operational": {
        "memory_usage": "< 80% capacity",
        "disk_usage": "< 70% capacity",
        "cpu_utilization": "< 70%",
        "index_freshness": "최신 데이터 반영 지연"
    }
}

# 멀티 테넌시 구현 방식별 비교
multi_tenancy = {
    "collection_per_tenant": {
        "isolation": "강함",
        "overhead": "높음 (테넌트당 컬렉션 생성)",
        "use_case": "소수의 대형 테넌트"
    },
    "namespace_per_tenant": {
        "isolation": "중간",
        "overhead": "중간",
        "use_case": "중간 규모 테넌트"
    },
    "metadata_filter": {
        "isolation": "약함",
        "overhead": "낮음",
        "use_case": "다수의 소형 테넌트"
    }
}

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 권장 청킹 전략
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # 청크 크기 (토큰 기준)
    chunk_overlap=50,     # 청크 간 겹침 (문맥 유지)
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len
)

chunks = splitter.split_text(document_text)

# RAG 검색 최적화 기법
class OptimizedRAGRetriever:
    def __init__(self, vector_db, reranker):
        self.vector_db = vector_db
        self.reranker = reranker

    def retrieve(self, query, top_k=5):
        # 1. 쿼리 확장 (Query Expansion)
        expanded_queries = self.expand_query(query)

        # 2. 하이브리드 검색 (벡터 + BM25)
        candidates = []
        for q in expanded_queries:
            results = self.vector_db.hybrid_search(q, limit=20)
            candidates.extend(results)

        # 3. 중복 제거
        unique_candidates = self.deduplicate(candidates)

        # 4. Cross-Encoder 리랭킹
        reranked = self.reranker.rerank(query, unique_candidates)

        # 5. 상위 K개 반환
        return reranked[:top_k]

    def expand_query(self, query):
        # HyDE (Hypothetical Document Embeddings) 또는
        # LLM 기반 쿼리 재작성
        return [query, self.generate_hypothetical_answer(query)]

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)

# RAG 파이프라인 평가
results = evaluate(
    dataset=eval_dataset,
    metrics=[
        faithfulness,        # 답변이 컨텍스트에 충실한가?
        answer_relevancy,    # 답변이 질문에 관련있는가?
        context_precision,   # 검색된 컨텍스트가 정확한가?
        context_recall       # 필요한 정보가 모두 검색되었는가?
    ]
)

print(results)
# faithfulness: 0.92
# answer_relevancy: 0.89
# context_precision: 0.85
# context_recall: 0.91