Split View: Qdrant 벡터DB 운영 및 사용 방법 총정리 — 컬렉션 설계부터 RAG 연동까지

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, HnswConfigDiff, OptimizersConfigDiff
from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig, ScalarType

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

client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,              # OpenAI text-embedding-3-small dimensions
        distance=Distance.COSINE,
    ),
    hnsw_config=HnswConfigDiff(
        m=16,                   # Graph connections
        ef_construct=128,       # Search width during index build
    ),
    optimizers_config=OptimizersConfigDiff(
        indexing_threshold=20000,
    ),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(type=ScalarType.INT8, quantile=0.99, always_ram=True),
    ),
)

# Scalar Quantization: float32 -> int8 (4x memory reduction)
# Suitable for general text search, recall loss < 1%

# Binary Quantization: float32 -> 1-bit (32x memory reduction)
# Very aggressive. Only use with high dimensions (1536+). Requires oversampling
from qdrant_client.models import BinaryQuantization, BinaryQuantizationConfig

client.update_collection(
    collection_name="documents",
    quantization_config=BinaryQuantization(
        binary=BinaryQuantizationConfig(
            always_ram=True,
        ),
    ),
)

from qdrant_client.models import PointStruct
import openai

def get_embedding(text: str) -> list[float]:
    resp = openai.embeddings.create(model="text-embedding-3-small", input=text)
    return resp.data[0].embedding

# Single upsert
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=get_embedding("Qdrant is a vector DB written in Rust."),
            payload={"title": "Qdrant Introduction", "category": "database", "tenant_id": "team-alpha"},
        ),
    ],
)

# Batch upsert — process in batches of 100
BATCH_SIZE = 100
for i in range(0, len(points), BATCH_SIZE):
    client.upsert(collection_name="documents", points=points[i : i + BATCH_SIZE])

from qdrant_client.models import SearchParams

results = client.search(
    collection_name="documents",
    query_vector=get_embedding("vector database performance optimization"),
    limit=10,
    score_threshold=0.7,
    search_params=SearchParams(
        hnsw_ef=128,       # Search-time ef value (controls accuracy)
        exact=False,       # True for brute-force (accurate but slow)
    ),
    with_payload=True,
    with_vectors=False,    # Don't return vectors (reduce response size)
)

for result in results:
    print(f"ID: {result.id}, Score: {result.score:.4f}")
    print(f"  Title: {result.payload['title']}")

curl -X POST "http://localhost:6333/collections/documents/points/search" \
  -H "Content-Type: application/json" \
  -d '{
    "vector": [0.1, 0.2, ...],
    "limit": 10,
    "with_payload": true,
    "params": {
      "hnsw_ef": 128
    }
  }'

from qdrant_client.models import PointIdsList, FilterSelector, Filter, FieldCondition, MatchValue

# Update payload
client.set_payload(collection_name="documents", payload={"category": "vector-database"}, points=[1, 2, 3])

# Delete by ID
client.delete(collection_name="documents", points_selector=PointIdsList(points=[10, 11, 12]))

# Delete by filter
client.delete(
    collection_name="documents",
    points_selector=FilterSelector(
        filter=Filter(must=[FieldCondition(key="category", match=MatchValue(value="deprecated"))])
    ),
)

from qdrant_client.models import PayloadSchemaType, TextIndexParams, TokenizerType

# Keyword index (exact matching)
client.create_payload_index(collection_name="documents", field_name="category", field_schema=PayloadSchemaType.KEYWORD)

# Integer index (range queries)
client.create_payload_index(collection_name="documents", field_name="page_count", field_schema=PayloadSchemaType.INTEGER)

# Text index (full-text search with multilingual tokenizer)
client.create_payload_index(
    collection_name="documents",
    field_name="content",
    field_schema=TextIndexParams(type="text", tokenizer=TokenizerType.MULTILINGUAL, min_token_len=2, max_token_len=20),
)

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

# Search for documents where category is "database" and page_count >= 10
results = client.search(
    collection_name="documents",
    query_vector=get_embedding("vector index optimization"),
    query_filter=Filter(
        must=[
            FieldCondition(
                key="category",
                match=MatchValue(value="database"),
            ),
            FieldCondition(
                key="page_count",
                range=Range(gte=10),
            ),
        ],
        must_not=[
            FieldCondition(
                key="status",
                match=MatchValue(value="archived"),
            ),
        ],
    ),
    limit=5,
)

from qdrant_client.models import SparseVector, SparseVectorParams, Prefetch, FusionQuery, Fusion

# Create collection with dense + sparse vector configuration
client.create_collection(
    collection_name="hybrid_docs",
    vectors_config={"dense": VectorParams(size=1536, distance=Distance.COSINE)},
    sparse_vectors_config={"sparse": SparseVectorParams()},
)

# Hybrid search with Reciprocal Rank Fusion
results = client.query_points(
    collection_name="hybrid_docs",
    prefetch=[
        Prefetch(query=get_embedding("vector search performance"), using="dense", limit=20),
        Prefetch(query=SparseVector(indices=[1, 42, 1337], values=[0.9, 0.3, 0.7]), using="sparse", limit=20),
    ],
    query=FusionQuery(fusion=Fusion.RRF),
    limit=10,
)

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_qdrant import QdrantVectorStore
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 1. Split documents
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = text_splitter.split_documents(documents)

# 2. Create vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = QdrantVectorStore.from_documents(
    documents=chunks,
    embedding=embeddings,
    url="http://localhost:6333",
    collection_name="rag_documents",
    force_recreate=True,
)

# 3. Build RAG chain
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 5,
        "score_threshold": 0.7,
    },
)

llm = ChatOpenAI(model="gpt-4o", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
)

# 4. Query
response = qa_chain.invoke({"query": "How do I tune HNSW index parameters in Qdrant?"})
print(response["result"])

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
import qdrant_client

# 1. Qdrant client & vector store
qclient = qdrant_client.QdrantClient(host="localhost", port=6333)

vector_store = QdrantVectorStore(
    client=qclient,
    collection_name="llamaindex_docs",
)

# 2. Configure settings
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.llm = OpenAI(model="gpt-4o", temperature=0)
Settings.chunk_size = 512
Settings.chunk_overlap = 50

# 3. Load and index documents
documents = SimpleDirectoryReader("./data").load_data()

index = VectorStoreIndex.from_documents(
    documents,
    vector_store=vector_store,
)

# 4. Query engine
query_engine = index.as_query_engine(
    similarity_top_k=5,
)

response = query_engine.query("How can I combine filtering and vector search in Qdrant?")
print(response)

# Check metrics
curl http://localhost:6333/metrics

# Key metrics
# qdrant_grpc_responses_total        - gRPC request count
# qdrant_rest_responses_total        - REST request count
# qdrant_collection_points_total     - Points per collection
# qdrant_collection_search_latency   - Search latency

# Create and list snapshots
client.create_snapshot(collection_name="documents")
snapshots = client.list_snapshots(collection_name="documents")
for snap in snapshots:
    print(f"Name: {snap.name}, Size: {snap.size}")

# Download snapshot via REST API
curl -o snapshot.tar "http://localhost:6333/collections/documents/snapshots/snapshot-2026-03-09.snapshot"

# Full storage snapshot (all collections)
curl -X POST "http://localhost:6333/snapshots"

# Recommended: use gRPC
client = QdrantClient(
    host="localhost",
    port=6333,
    grpc_port=6334,
    prefer_grpc=True,
)