벡터 데이터베이스 완전 비교 - Chroma, Milvus, Pinecone, Qdrant, Weaviate, pgvector

pip install chromadb

import chromadb
 
client = chromadb.Client()  # 인메모리
# client = chromadb.PersistentClient(path="./chroma_db")  # 영구 저장
 
collection = client.create_collection("documents")
 
collection.add(
    ids=["doc1", "doc2", "doc3"],
    documents=[
        "Apache Spark는 분산 데이터 처리 프레임워크입니다",
        "Kafka는 분산 이벤트 스트리밍 플랫폼입니다",
        "Kubernetes는 컨테이너 오케스트레이션 도구입니다"
    ],
    metadatas=[
        {"category": "compute"},
        {"category": "messaging"},
        {"category": "orchestration"}
    ]
)
 
results = collection.query(
    query_texts=["대용량 데이터 병렬 처리"],
    n_results=2
)
print(results["documents"])

pip install pinecone

from pinecone import Pinecone, ServerlessSpec
 
pc = Pinecone(api_key="YOUR_API_KEY")
 
pc.create_index(
    name="documents",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
 
index = pc.Index("documents")
 
index.upsert(vectors=[
    {"id": "doc1", "values": [0.1, 0.2, ...], "metadata": {"category": "compute"}},
    {"id": "doc2", "values": [0.3, 0.4, ...], "metadata": {"category": "messaging"}},
])
 
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=3,
    include_metadata=True,
    filter={"category": {"$eq": "compute"}}
)

# Docker로 실행
docker compose up -d  # docker-compose.yml 필요
 
pip install pymilvus

from pymilvus import MilvusClient
 
client = MilvusClient(uri="http://localhost:19530")
 
client.create_collection(
    collection_name="documents",
    dimension=1536,
    metric_type="COSINE"
)
 
data = [
    {"id": 1, "vector": [0.1, 0.2, ...], "text": "Spark는...", "category": "compute"},
    {"id": 2, "vector": [0.3, 0.4, ...], "text": "Kafka는...", "category": "messaging"},
]
client.insert(collection_name="documents", data=data)
 
results = client.search(
    collection_name="documents",
    data=[[0.1, 0.2, ...]],
    limit=3,
    output_fields=["text", "category"]
)

docker run -p 6333:6333 qdrant/qdrant
 
pip install qdrant-client

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, 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={"category": "compute", "text": "Spark는..."}),
        PointStruct(id=2, vector=[0.3, 0.4, ...], payload={"category": "messaging", "text": "Kafka는..."}),
    ]
)
 
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=3,
    query_filter={"must": [{"key": "category", "match": {"value": "compute"}}]}
)

docker run -p 8080:8080 -p 50051:50051 semitechnologies/weaviate
 
pip install weaviate-client

import weaviate
 
client = weaviate.connect_to_local()
 
collection = client.collections.create(
    name="Document",
    vectorizer_config=weaviate.classes.config.Configure.Vectorizer.none(),
    properties=[
        weaviate.classes.config.Property(name="text", data_type=weaviate.classes.config.DataType.TEXT),
        weaviate.classes.config.Property(name="category", data_type=weaviate.classes.config.DataType.TEXT),
    ]
)
 
collection.data.insert_many([
    {"text": "Spark는 분산 데이터 처리 프레임워크입니다", "category": "compute"},
    {"text": "Kafka는 분산 이벤트 스트리밍 플랫폼입니다", "category": "messaging"},
])
 
results = collection.query.near_vector(
    near_vector=[0.1, 0.2, ...],
    limit=3,
    return_metadata=weaviate.classes.query.MetadataQuery(distance=True)
)

-- PostgreSQL에 pgvector 확장 설치
CREATE EXTENSION vector;
 
-- 테이블 생성
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    text TEXT,
    category TEXT,
    embedding vector(1536)
);
 
-- 인덱스 생성 (HNSW)
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
 
-- 데이터 삽입
INSERT INTO documents (text, category, embedding)
VALUES ('Spark는 분산 처리 프레임워크입니다', 'compute', '[0.1, 0.2, ...]');
 
-- 유사도 검색
SELECT text, category, 1 - (embedding <=> '[0.1, 0.2, ...]') AS similarity
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 3;

# Qdrant: 강력한 필터링 예시
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range
 
results = client.search(
    collection_name="documents",
    query_vector=query_vector,
    limit=5,
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="engineering")),
            FieldCondition(key="date", range=Range(gte="2025-01-01")),
        ],
        must_not=[
            FieldCondition(key="status", match=MatchValue(value="archived")),
        ]
    )
)

# Weaviate: 네이티브 하이브리드 검색
results = collection.query.hybrid(
    query="Spark 성능 최적화",        # BM25 키워드 검색
    vector=[0.1, 0.2, ...],          # 벡터 검색
    alpha=0.7,                        # 0=키워드만, 1=벡터만, 0.7=벡터 70%
    limit=5
)

# Chroma
from langchain_chroma import Chroma
vectorstore = Chroma.from_documents(docs, embeddings, persist_directory="./db")
 
# Pinecone
from langchain_pinecone import PineconeVectorStore
vectorstore = PineconeVectorStore.from_documents(docs, embeddings, index_name="my-index")
 
# Milvus
from langchain_milvus import Milvus
vectorstore = Milvus.from_documents(docs, embeddings, connection_args={"host": "localhost", "port": "19530"})
 
# Qdrant
from langchain_qdrant import QdrantVectorStore
vectorstore = QdrantVectorStore.from_documents(docs, embeddings, url="http://localhost:6333", collection_name="docs")
 
# Weaviate
from langchain_weaviate import WeaviateVectorStore
vectorstore = WeaviateVectorStore.from_documents(docs, embeddings, client=weaviate_client)
 
# pgvector
from langchain_postgres import PGVector
vectorstore = PGVector.from_documents(docs, embeddings, connection="postgresql://...", collection_name="docs")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
 
# 어떤 벡터 DB든 동일한 RAG 체인
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
 
prompt = ChatPromptTemplate.from_template("""
컨텍스트를 참고하여 질문에 답하세요.
 
컨텍스트: {context}
질문: {question}
""")
 
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)
 
answer = rag_chain.invoke("Spark 성능 최적화 방법은?")