Vector Databases

W poprzedniej lekcji trzymaliśmy embeddingi w macierzy NumPy. Przy pięciu dokumentach to wystarcza, ale przy milionie macierz nie mieści się w pamięci, znika po restarcie programu, a przeszukanie jej za każdym razem trwa za długo. Potrzebujemy archiwum tropów, które przetrwa noc w obozie i odpowie w milisekundach.

Vector databases to specjalizowane bazy danych zoptymalizowane do przechowywania i wyszukiwania embeddingów. To jak biblioteka z magicznym katalogiem, który znajduje podobne książki na podstawie ich treści, a nie tytułu.

Popularne Vector Databases

Schemat grupuje popularne bazy według tego, gdzie działają:

1┌─────────────────────────────────────────────────────────┐
2│              Vector Databases Landscape                  │
3├─────────────────────────────────────────────────────────┤
4│                                                          │
5│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
6│  │   Pinecone   │  │    Qdrant    │  │   Chroma     │  │
7│  │   (Cloud)    │  │  (Self-host) │  │   (Local)    │  │
8│  └──────────────┘  └──────────────┘  └──────────────┘  │
9│                                                          │
10│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
11│  │   Weaviate   │  │    Milvus    │  │    FAISS     │  │
12│  │  (Semantic)  │  │  (Enterprise)│  │  (In-memory) │  │
13│  └──────────────┘  └──────────────┘  └──────────────┘  │
14│                                                          │
15└─────────────────────────────────────────────────────────┘

Pinecone działa wyłącznie w chmurze jako usługa zarządzana, Qdrant i Milvus możesz postawić na własnym serwerze, Chroma świetnie sprawdza się lokalnie przy prototypach, a FAISS to biblioteka od Meta, która trzyma indeks w pamięci procesu.

Chroma - lokalna vector database

Chroma to lokalna vector database dla aplikacji RAG, którą instalujesz jednym pip install chromadb. Najpierw tworzymy klienta, funkcję embeddingu i kolekcję, czyli odpowiednik tabeli:

1import chromadb
2from chromadb.utils import embedding_functions
3
4# Inicjalizacja klienta
5client = chromadb.Client()  # In-memory
6# lub: client = chromadb.PersistentClient(path="./chroma_db")  # Persistent
7
8# Konfiguracja funkcji embeddingu
9openai_ef = embedding_functions.OpenAIEmbeddingFunction(
10    api_key="twój-klucz",
11    model_name="text-embedding-3-small"
12)
13
14# Tworzenie kolekcji
15collection = client.create_collection(
16    name="python_safari",
17    embedding_function=openai_ef,
18    metadata={"description": "Dokumentacja kursu Python Safari"}
19)

chromadb.Client() trzyma dane tylko w pamięci, a PersistentClient zapisuje je w katalogu na dysku. Kolekcja sama wywoła openai_ef dla każdego dokumentu, więc nie musisz ręcznie liczyć embeddingów. Klucza API nie wpisuj w kod jak w przykładzie: w prawdziwym projekcie czytaj go ze zmiennej środowiskowej.

Teraz dodajemy dokumenty z metadanymi i od razu wyszukujemy:

1# Dodawanie dokumentów
2collection.add(
3    documents=[
4        "Python to język programowania wysokiego poziomu",
5        "RAG łączy retrieval z generowaniem tekstu",
6        "Vector databases przechowują embeddings"
7    ],
8    metadatas=[
9        {"topic": "python", "level": "beginner"},
10        {"topic": "ai", "level": "advanced"},
11        {"topic": "databases", "level": "intermediate"}
12    ],
13    ids=["doc1", "doc2", "doc3"]
14)
15
16# Wyszukiwanie
17results = collection.query(
18    query_texts=["Jak zacząć naukę programowania?"],
19    n_results=2,
20    where={"level": "beginner"}  # Filtrowanie po metadanych
21)
22
23print(results)

Każdy dokument potrzebuje unikalnego id. Parametr where filtruje po metadanych jeszcze przed porównaniem wektorów, więc tu przejdzie tylko dokument z level równym beginner. Wynik to słownik list, po jednej na każde pytanie.

Jeśli masz już własne wektory, na przykład z funkcji get_embedding z poprzedniej lekcji, przekaż je w parametrze embeddings, a Chroma pominie liczenie:

1# Własne wektory zamiast funkcji embeddingu kolekcji
2collection.add(
3    documents=["Słonie wędrują do wodopoju o świcie"],
4    embeddings=[get_embedding("Słonie wędrują do wodopoju o świcie")],
5    ids=["doc4"]
6)

Wektory muszą mieć ten sam wymiar co reszta kolekcji, dlatego używamy tego samego modelu text-embedding-3-small.

Qdrant - produkcyjny vector search

Qdrant to baza, którą uruchamiasz na serwerze albo w kontenerze Dockera. Praca z nią ma zawsze cztery kroki: połączenie z klientem, utworzenie kolekcji z konfiguracją wektorów, dodanie punktów i wyszukiwanie. Pierwsze dwa kroki wyglądają tak:

1from qdrant_client import QdrantClient
2from qdrant_client.models import Distance, VectorParams, PointStruct
3import numpy as np
4
5# Połączenie z Qdrant
6client = QdrantClient(host="localhost", port=6333)
7# lub: client = QdrantClient(":memory:")  # In-memory
8
9# Tworzenie kolekcji
10client.create_collection(
11    collection_name="documents",
12    vectors_config=VectorParams(
13        size=1536,  # Rozmiar embeddingu
14        distance=Distance.COSINE
15    )
16)

VectorParams mówi bazie, jak długie będą wektory i jaką miarą je porównywać. size musi pasować do modelu embeddingów, tu 1536 dla text-embedding-3-small.

Punkt w Qdrant to wektor plus payload, czyli dowolny słownik z danymi:

1# Dodawanie punktów
2def add_documents(texts: list[str], embeddings: list[list[float]]):
3    """Dodaje dokumenty do Qdrant."""
4    points = [
5        PointStruct(
6            id=i,
7            vector=embedding,
8            payload={"text": text}
9        )
10        for i, (text, embedding) in enumerate(zip(texts, embeddings))
11    ]
12
13    client.upsert(
14        collection_name="documents",
15        points=points
16    )

Metoda upsert wstawia nowe punkty albo nadpisuje istniejące o tym samym id, dlatego ponowne uruchomienie nie tworzy duplikatów.

Wyszukiwanie w aktualnym kliencie Pythona robimy metodą query_points. Starsza metoda search była oznaczona jako deprecated i nowe wersje qdrant-client już jej nie mają:

1# Wyszukiwanie
2def search(query_embedding: list[float], limit: int = 5):
3    """Wyszukuje podobne dokumenty."""
4    results = client.query_points(
5        collection_name="documents",
6        query=query_embedding,
7        limit=limit
8    ).points
9
10    return [
11        {
12            "text": hit.payload["text"],
13            "score": hit.score
14        }
15        for hit in results
16    ]

query_points zwraca obiekt z listą points, a każdy trafiony punkt ma score i payload.

Filtr działa jak where w Chromie, tylko ma bardziej formalną składnię:

1# Filtrowanie
2from qdrant_client.models import Filter, FieldCondition, MatchValue
3
4filtered_results = client.query_points(
5    collection_name="documents",
6    query=query_embedding,
7    query_filter=Filter(
8        must=[
9            FieldCondition(
10                key="category",
11                match=MatchValue(value="python")
12            )
13        ]
14    ),
15    limit=5
16).points.points

Uwaga na pułapkę: add_documents zapisuje w payloadzie tylko text, więc filtr po category nic nie znajdzie, dopóki nie dodasz tego pola do payloadu. query_embedding to wektor pytania, który masz już policzony.

FAISS - szybkie wyszukiwanie w pamięci

FAISS to nie serwer, tylko biblioteka. Indeks żyje w pamięci Twojego procesu. Owińmy go w klasę, która pamięta też teksty dokumentów:

1import faiss
2import numpy as np
3from dataclasses import dataclass
4
5@dataclass
6class FAISSIndex:
7    """Wrapper dla FAISS."""
8
9    dimension: int
10    index: faiss.Index = None
11    documents: list[str] = None
12
13    def __post_init__(self):
14        # Różne typy indeksów
15        # Flat - dokładny, wolniejszy
16        self.index = faiss.IndexFlatL2(self.dimension)
17
18        # IVF - szybszy, przybliżony
19        # quantizer = faiss.IndexFlatL2(self.dimension)
20        # self.index = faiss.IndexIVFFlat(quantizer, self.dimension, 100)
21
22        self.documents = []
23
24    def add(self, embeddings: np.ndarray, documents: list[str]):
25        """Dodaje wektory do indeksu."""
26        embeddings = np.array(embeddings).astype('float32')
27        self.index.add(embeddings)
28        self.documents.extend(documents)

IndexFlatL2 porównuje pytanie z każdym wektorem, więc wynik jest dokładny. Zakomentowany IndexIVFFlat jest szybszy, ale przybliżony i przed dodaniem wektorów wymaga treningu metodą train. FAISS przyjmuje wyłącznie float32, stąd konwersja w add.

Wyszukiwanie zwraca odległości i numery najbliższych sąsiadów:

1    def search(self, query_embedding: np.ndarray, k: int = 5) -> list[tuple[str, float]]:
2        """Wyszukuje k najbliższych sąsiadów."""
3        query = np.array([query_embedding]).astype('float32')
4        distances, indices = self.index.search(query, k)
5
6        results = []
7        for idx, dist in zip(indices[0], distances[0]):
8            if 0 <= idx < len(self.documents):
9                results.append((self.documents[idx], float(dist)))
10
11        return results

Mniejsza odległość L2 oznacza większe podobieństwo, odwrotnie niż przy kosinusie. Gdy prosisz o więcej wyników, niż jest wektorów, FAISS zwraca indeks -1, dlatego sprawdzamy 0 <= idx.

Test na losowych wektorach pokazuje sam mechanizm:

1# Przykład użycia
2faiss_index = FAISSIndex(dimension=384)
3
4# Dodaj dokumenty
5embeddings = np.random.rand(100, 384).astype('float32')
6documents = [f"Dokument {i}" for i in range(100)]
7faiss_index.add(embeddings, documents)
8
9# Wyszukaj
10query = np.random.rand(384).astype('float32')
11results = faiss_index.search(query, k=5)

W prawdziwym kodzie zamiast losowych liczb podstawisz embeddingi z modelu o 384 wymiarach.

Pinecone - managed vector database

Pinecone to usługa w chmurze: nie instalujesz serwera, tylko tworzysz indeks przez API. Przy tworzeniu podajesz wymiar, metrykę i region:

1from pinecone import Pinecone, ServerlessSpec
2
3# Inicjalizacja
4pc = Pinecone(api_key="twój-klucz")
5
6# Tworzenie indeksu
7pc.create_index(
8    name="python-safari",
9    dimension=1536,
10    metric="cosine",
11    spec=ServerlessSpec(
12        cloud="aws",
13        region="us-east-1"
14    )
15)
16
17# Połączenie z indeksem
18index = pc.Index("python-safari")

ServerlessSpec oznacza indeks bezserwerowy, rozliczany za użycie.

Upsert przyjmuje listę słowników z id, wektorem i metadanymi:

1# Upsert (wstaw/aktualizuj)
2index.upsert(
3    vectors=[
4        {
5            "id": "doc1",
6            "values": [0.1, 0.2, ...],  # 1536 wartości
7            "metadata": {
8                "text": "Python to język programowania",
9                "category": "programming",
10                "level": 1
11            }
12        }
13    ],
14    namespace="tutorials"
15)

Zapis [0.1, 0.2, ...] to skrót myślowy: w Pythonie trzy kropki są obiektem Ellipsis, więc w prawdziwym kodzie wstawiasz pełną listę 1536 liczb. namespace dzieli indeks na niezależne przegrody.

Zapytanie łączy wektor z filtrem po metadanych:

1# Wyszukiwanie
2results = index.query(
3    vector=[0.1, 0.2, ...],
4    top_k=10,
5    include_metadata=True,
6    namespace="tutorials",
7    filter={
8        "category": {"$eq": "programming"},
9        "level": {"$lte": 3}
10    }
11)
12
13# Statystyki
14stats = index.describe_index_stats()
15print(f"Liczba wektorów: {stats['total_vector_count']}")

Operatory $eq i $lte znaczą "równe" oraz "mniejsze lub równe", a describe_index_stats podaje, ile wektorów leży w indeksie.

Hybrid Search - łączenie wektorów z BM25

Wektory świetnie łapią sens, ale potrafią przegapić dokładną nazwę, na przykład numer modelu. BM25 to klasyczny algorytm wyszukiwania po słowach kluczowych. Hybrid search łączy oba wyniki:

1from rank_bm25 import BM25Okapi
2import numpy as np
3
4class HybridSearch:
5    """Łączy semantic search z keyword search."""
6
7    def __init__(self, documents: list[str], embeddings: np.ndarray):
8        self.documents = documents
9        self.embeddings = embeddings
10
11        # BM25 dla keyword search
12        tokenized = [doc.lower().split() for doc in documents]
13        self.bm25 = BM25Okapi(tokenized)

Konstruktor dzieli każdy dokument na słowa i buduje z nich indeks BM25Okapi, a embeddingi dostaje gotowe.

Metoda search normalizuje oba wyniki do zakresu 0-1 i miesza je z wagą alpha:

1    def search(
2        self,
3        query: str,
4        query_embedding: np.ndarray,
5        alpha: float = 0.5,  # Waga semantic search
6        top_k: int = 5
7    ) -> list[tuple[str, float]]:
8        """Hybrid search z konfigurowalnymi wagami."""
9
10        # Semantic search scores
11        semantic_scores = np.dot(self.embeddings, query_embedding)
12        semantic_scores /= np.linalg.norm(self.embeddings, axis=1)
13        semantic_scores /= np.linalg.norm(query_embedding)
14
15        # Normalize to [0, 1]
16        semantic_scores = (semantic_scores - semantic_scores.min()) / (semantic_scores.max() - semantic_scores.min())
17
18        # BM25 scores
19        bm25_scores = np.array(self.bm25.get_scores(query.lower().split()))
20        if bm25_scores.max() > 0:
21            bm25_scores = bm25_scores / bm25_scores.max()
22
23        # Hybrid score
24        hybrid_scores = alpha * semantic_scores + (1 - alpha) * bm25_scores
25
26        # Top K
27        top_indices = np.argsort(hybrid_scores)[::-1][:top_k]
28
29        return [(self.documents[i], hybrid_scores[i]) for i in top_indices]

alpha=1 to czyste wyszukiwanie semantyczne, alpha=0 czyste BM25. Polecam zacząć od 0.5 i stroić wagę na własnych pytaniach testowych.

Vector databases to infrastruktura każdego systemu RAG. Do prototypu bierz Chromę, na produkcję Qdrant albo Pinecone. W następnej lekcji poznasz LlamaIndex - framework, który upraszcza budowanie aplikacji RAG i podłącza się do tych baz jedną linijką.

Zapamiętaj: baza wektorowa to obozowe archiwum tropów, które po odcisku znajduje najbardziej podobne ślady.