벡터 저장소 선택 및 벤치마킹

벡터 저장소 생태계

평가할 주요 기준

Pinecone: 간편한 관리형 서비스

import pinecone

pc = pinecone.Pinecone(api_key='YOUR_API_KEY')
index = pc.Index('my-index')

# Query with metadata filter
results = index.query(
    vector=[0.1, 0.2, 0.3],
    top_k=10,
    filter={'category': {'$eq': 'finance'}},
    include_metadata=True
)

pgvector: PostgreSQL의 임베딩

-- Create table with embedding column
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    category TEXT,
    embedding vector(1536)
);

-- Create HNSW index for fast ANN search
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops);

-- Query nearest neighbors with SQL filter
SELECT content, 1 - (embedding <=> '[0.1,0.2,...]')
FROM documents
WHERE category = 'finance'
ORDER BY embedding <=> '[0.1,0.2,...]'
LIMIT 10;

Chroma: 개발자 친화적인 로컬 우선 방식

import chromadb

client = chromadb.PersistentClient(path='./chroma_db')
collection = client.get_or_create_collection('my_docs')

# Add documents
collection.add(
    documents=['text one', 'text two'],
    metadatas=[{'source': 'doc1'}, {'source': 'doc2'}],
    ids=['id1', 'id2']
)

# Query
results = collection.query(
    query_texts=['search query'],
    n_results=5
)

Qdrant: 필터링과 페이로드 검색

from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

client = QdrantClient(url='http://localhost:6333')

# Search with payload filter
results = client.search(
    collection_name='documents',
    query_vector=[0.1, 0.2, 0.3],
    query_filter=Filter(
        must=[
            FieldCondition(
                key='category',
                match=MatchValue(value='finance')
            )
        ]
    ),
    limit=10
)

Weaviate: GraphQL과 멀티모달

import weaviate

client = weaviate.Client('http://localhost:8080')

# Near-text search using built-in vectorizer
result = client.query.get(
    'Document', ['content', 'category']
).with_near_text(
    {'concepts': ['financial analysis']}
).with_where({
    'path': ['category'],
    'operator': 'Equal',
    'valueString': 'finance'
}).with_limit(10).do()

FAISS: 대규모 메모리 내 검색

import faiss
import numpy as np

dimension = 1536
vectors = np.random.random((100000, dimension)).astype('float32')

# Normalize for cosine similarity
faiss.normalize_L2(vectors)

# Build HNSW index
index = faiss.IndexHNSWFlat(dimension, 32)  # M=32 neighbors
index.add(vectors)

# Search
query = np.random.random((1, dimension)).astype('float32')
faiss.normalize_L2(query)
D, I = index.search(query, k=10)  # top-10 results

벤치마크 테스트 만들기

import time
import numpy as np

def benchmark_store(store, queries, k=10):
    latencies = []
    for q in queries:
        start = time.perf_counter()
        store.search(q, k)
        latencies.append(time.perf_counter() - start)
    latencies.sort()
    n = len(latencies)
    print(f'p50: {latencies[n//2]*1000:.1f}ms')
    print(f'p95: {latencies[int(n*0.95)]*1000:.1f}ms')
    print(f'p99: {latencies[int(n*0.99)]*1000:.1f}ms')

재현율과 지연 시간의 상충 관계 측정하기

def compute_recall(approx_ids, exact_ids):
    '''Compute recall@K for one query'''
    return len(set(approx_ids) & set(exact_ids)) / len(exact_ids)

def benchmark_recall(index, brute_force, queries, k=10):
    recalls = []
    for q in queries:
        approx = index.search(q, k)
        exact = brute_force.search(q, k)
        recalls.append(compute_recall(approx, exact))
    print(f'Mean recall@{k}: {sum(recalls)/len(recalls):.3f}')

선택을 위한 의사 결정 체계

빠른 확인

레슨 요약