Векторные базы данных

🎯 Цель

После прочтения этой главы:

• Знаете, что такое vector database (vector DB) и зачем он нужен
• Знаете разницу между Qdrant, ChromaDB, pgvector, Pinecone, Weaviate
• Знаете критерии выбора vector DB в production
• Можете построить hybrid search (vector + keyword)
• Можете управлять million-scale vector индексами

Что нужно изучить

• Vector embeddings — помните, из Месяца 4
• Similarity metrics — cosine, dot product, Euclidean
• Алгоритмы **ANN (Approximate Nearest Neighbor)** — HNSW, IVF
• Vector DB — Qdrant, ChromaDB, pgvector, Pinecone, Weaviate, Milvus
• Hybrid search — vector + BM25 (keyword)
• Reranking — переупорядочивание top-k результата через Cross-encoder
• Metadata filtering
• Sharding и indexing — million-scale

Что такое Vector DB и зачем он нужен?

Проблема

В классическом SQL: “name = ‘John’” — точный match. Но: “Python developer kerak” → “Python dasturchi izlanmoqda” — semantically одинаково, но в string разное.

Решение

Превратить текст в vector (embedding) и искать по cosine similarity.

"Python developer" → [0.12, 0.45, ..., -0.23]  (1536-dim)
"Python dasturchi" → [0.14, 0.47, ..., -0.21]
cosine_similarity > 0.95 — очень близко!

Что делает Vector DB?

1. Index — эффективное хранение миллионов векторов
2. Search — поиск K ближайших векторов к query vector (за ms)
3. Metadata — сохранение JSON вместе с каждым vector
4. Filtering — поиск с условием metadata.category = "tech"

ANN (Approximate NN) — почему «approximate»?

Поиск ближайшего среди миллиона векторов — O(N) операция, медленно. HNSW(Hierarchical Navigable Small Worlds) — O(log N) — на миллиардной scale.

Trade-off: 99% accuracy, но в 1000x быстрее.

Основные Vector DB

Comparison table

Рекомендации

• **Начало (prototype):**ChromaDB (внутри Python, no setup)
• **Backend dev (Postgres уже есть):**pgvector
• **Production (self-hosted):**Qdrant(лучшее качество/простая установка)
• **Production (managed):**Pinecone
• **Enterprise (millions+):**Milvus, Weaviate

Примеры кода

ChromaDB — самый простой старт

import chromadb

client = chromadb.Client()
# или persistent:
# client = chromadb.PersistentClient(path="./chroma_db")

collection = client.create_collection("docs")

# Add documents
collection.add(
    documents=["Python — язык высокого уровня", "JavaScript — для web"],
    metadatas=[{"category": "language"}, {"category": "language"}],
    ids=["doc1", "doc2"],
)
# ChromaDB автоматически делает embed (default: all-MiniLM-L6-v2)

# Query
results = collection.query(
    query_texts=["Python программирование"],
    n_results=5,
    where={"category": "language"},
)
print(results)

ChromaDB with custom embeddings

from chromadb.utils import embedding_functions

# OpenAI embeddings
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="...",
    model_name="text-embedding-3-small",
)

collection = client.create_collection(
    name="docs",
    embedding_function=openai_ef,
)

Qdrant — production grade

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

# Local Docker:
# docker run -p 6333:6333 qdrant/qdrant
client = QdrantClient(url="http://localhost:6333")

# 1. Create collection
client.create_collection(
    collection_name="docs",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

# 2. Add points
from openai import OpenAI
openai_client = OpenAI()

texts = ["Python is a programming language", "JavaScript is for web"]
embeddings = openai_client.embeddings.create(
    model="text-embedding-3-small",
    input=texts,
)

points = [
    PointStruct(
        id=i,
        vector=emb.embedding,
        payload={"text": text, "category": "language"},
    )
    for i, (text, emb) in enumerate(zip(texts, embeddings.data))
]

client.upsert(collection_name="docs", points=points)

# 3. Search
query_emb = openai_client.embeddings.create(
    model="text-embedding-3-small",
    input=["Python программирование"],
).data[0].embedding

results = client.search(
    collection_name="docs",
    query_vector=query_emb,
    limit=5,
    query_filter=Filter(
        must=[FieldCondition(key="category", match=MatchValue(value="language"))]
    ),
)

for r in results:
    print(f"Score: {r.score:.4f}, Text: {r.payload['text']}")

pgvector — Postgres extension

-- Install (one time)
CREATE EXTENSION vector;

-- Create table
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536),
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Create HNSW index
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Or IVFFlat
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

import psycopg
from psycopg.rows import dict_row

conn = psycopg.connect("dbname=mydb", row_factory=dict_row)

# Insert
embedding = openai.embeddings.create(input="text", model="text-embedding-3-small").data[0].embedding
conn.execute(
    "INSERT INTO documents (content, embedding, metadata) VALUES (%s, %s, %s)",
    ("Python is great", embedding, '{"category": "language"}'),
)

# Search (cosine distance)
query_emb = openai.embeddings.create(input="Python программирование", model="text-embedding-3-small").data[0].embedding

results = conn.execute("""
    SELECT id, content, metadata,
           1 - (embedding <=> %s::vector) AS similarity
    FROM documents
    WHERE metadata->>'category' = %s
    ORDER BY embedding <=> %s::vector
    LIMIT 5
""", (query_emb, "language", query_emb)).fetchall()

# Distance operators:
# <-> Euclidean (L2)
# <#> Negative dot product
# <=> Cosine distance

Hybrid search (Qdrant)

from qdrant_client.models import SparseVectorParams, SparseVector

# Hybrid: vector (dense) + BM25 (sparse)
client.create_collection(
    collection_name="docs_hybrid",
    vectors_config={
        "dense": VectorParams(size=1536, distance=Distance.COSINE),
    },
    sparse_vectors_config={
        "bm25": SparseVectorParams(),
    },
)

# Search with reciprocal rank fusion
from qdrant_client.models import Prefetch, Fusion, FusionQuery

results = client.query_points(
    collection_name="docs_hybrid",
    prefetch=[
        Prefetch(query=dense_query, using="dense", limit=20),
        Prefetch(query=sparse_query, using="bm25", limit=20),
    ],
    query=FusionQuery(fusion=Fusion.RRF),
    limit=10,
)

Reranking — улучшение качества

from sentence_transformers import CrossEncoder

# Cross-encoder даёт большую точность (но медленнее)
reranker = CrossEncoder("BAAI/bge-reranker-base")

# 1. Получить top 50 через vector search
candidates = client.search(collection_name="docs", query_vector=q, limit=50)

# 2. Rerank через Cross-encoder
pairs = [(query_text, c.payload["text"]) for c in candidates]
scores = reranker.predict(pairs)

# 3. Top 5
reranked = sorted(zip(scores, candidates), key=lambda x: -x[0])[:5]

Интеграция с backend

RAG ingestion pipeline

from fastapi import FastAPI, UploadFile
from celery import Celery

celery_app = Celery("rag", broker="redis://localhost:6379")

@celery_app.task
def ingest_document(file_path: str, source_url: str = None):
    # 1. Load
    from langchain_community.document_loaders import PyPDFLoader
    docs = PyPDFLoader(file_path).load()
    
    # 2. Split
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
    chunks = splitter.split_documents(docs)
    
    # 3. Embed
    openai_client = OpenAI()
    embeddings = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=[c.page_content for c in chunks],
    )
    
    # 4. Store in Qdrant
    points = [
        PointStruct(
            id=uuid.uuid4().hex,
            vector=emb.embedding,
            payload={
                "text": chunk.page_content,
                "page": chunk.metadata.get("page", 0),
                "source": source_url or file_path,
            },
        )
        for chunk, emb in zip(chunks, embeddings.data)
    ]
    qdrant.upsert(collection_name="docs", points=points)
    
    return {"chunks_added": len(points)}

@app.post("/ingest")
async def ingest(file: UploadFile, source_url: str = None):
    path = f"/tmp/{uuid.uuid4().hex}_{file.filename}"
    with open(path, "wb") as f:
        f.write(await file.read())
    
    task = ingest_document.delay(path, source_url)
    return {"task_id": task.id}

Search endpoint

class SearchRequest(BaseModel):
    query: str
    top_k: int = 5
    filters: dict = {}

@app.post("/search")
async def search(req: SearchRequest):
    # 1. Embed query
    emb = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=[req.query],
    ).data[0].embedding
    
    # 2. Vector search
    results = qdrant.search(
        collection_name="docs",
        query_vector=emb,
        limit=req.top_k * 4,  # over-fetch для reranking
        query_filter=build_filter(req.filters) if req.filters else None,
    )
    
    # 3. Rerank
    if len(results) > req.top_k:
        pairs = [(req.query, r.payload["text"]) for r in results]
        scores = reranker.predict(pairs)
        reranked = sorted(zip(scores, results), key=lambda x: -x[0])[:req.top_k]
        results = [r for _, r in reranked]
    
    return {
        "results": [
            {
                "text": r.payload["text"],
                "score": r.score,
                "source": r.payload.get("source"),
                "page": r.payload.get("page"),
            }
            for r in results
        ]
    }

Ресурсы

• Qdrant docs — qdrant.tech
• pgvector GitHub — github.com/pgvector/pgvector
• ChromaDB docs — docs.trychroma.com
• Pinecone Learning Center — pinecone.io/learn
• HNSW paper — Yu. Malkov, D. Yashunin
• “Vector Databases: A First Principles Approach” — Roie Schwaber-Cohen

🏋️ Упражнения

🟢 Easy

1. Сохраните 100 документов в ChromaDB и выполните semantic search.
2. Запустите Qdrant в Docker, создайте простую collection.
3. Установите pgvector в Postgres, добавьте 50 векторов.

🟡 Medium

1. Hybrid search: dense + BM25 hybrid index в Qdrant.
2. Reranking: vector search → cross-encoder rerank — посмотрите разницу accuracy.
3. Metadata filtering: 1000+ документов в разных категориях — поиск с filter.

🔴 Hard

1. Production RAG ingestion: pipeline FastAPI + Celery + Qdrant из PDF/URL/Notion.
2. Multi-tenant vector DB: отдельный namespace/collection для каждого user.
3. Million-scale benchmark: 1M документов в Qdrant и pgvector — сравнение query latency и recall.

Capstone

notebooks/month-05/05_vector_db.ipynb:

• 1000+ документов на узбекском (Wikipedia, daryo.uz, kun.uz)
• Полный RAG index в Qdrant
• Hybrid search + reranking
• Multi-source ingestion pipeline

✅ Чек-лист

• Знаю, что такое Vector DB и зачем он нужен
• Разница Cosine similarity и Euclidean
• Intuition алгоритма HNSW
• Попробовал хотя бы 2 из ChromaDB, Qdrant, pgvector
• Что такое Hybrid search
• Pattern Reranking
• Metadata filtering
• RAG ingestion pipeline в FastAPI

Переходим к RAG Pipeline.