Embeddings i Vector Search

Embeddings to wektorowe reprezentacje tekstu, które pozwalają maszynom "rozumieć" znaczenie słów i zdań. To jak DNA każdego tekstu - unikalna sekwencja liczb reprezentująca jego semantykę!

Czym są embeddings?

Embedding to wektor liczb rzeczywistych reprezentujący tekst w przestrzeni semantycznej:

1from openai import OpenAI
2
3client = OpenAI()
4
5def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
6    """Generuje embedding dla podanego tekstu."""
7    response = client.embeddings.create(
8        model=model,
9        input=text
10    )
11    return response.data[0].embedding
12
13# Przykład
14text = "Python to świetny język programowania"
15embedding = get_embedding(text)
16
17print(f"Wymiar wektora: {len(embedding)}")  # 1536 dla text-embedding-3-small
18print(f"Pierwsze 5 wartości: {embedding[:5]}")

Modele embeddingów

Różne modele do różnych zastosowań:

1from sentence_transformers import SentenceTransformer
2
3# Modele open-source
4class EmbeddingModels:
5    """Różne modele embeddingów."""
6
7    def __init__(self):
8        # Szybki i lekki
9        self.minilm = SentenceTransformer('all-MiniLM-L6-v2')
10
11        # Większy, lepszy dla polskiego
12        self.multilingual = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
13
14    def embed_with_openai(self, texts: list[str]) -> list[list[float]]:
15        """OpenAI embeddings - najwyższa jakość."""
16        from openai import OpenAI
17        client = OpenAI()
18
19        response = client.embeddings.create(
20            model="text-embedding-3-large",  # 3072 wymiarów
21            input=texts
22        )
23        return [item.embedding for item in response.data]
24
25    def embed_with_sentence_transformers(self, texts: list[str]) -> list[list[float]]:
26        """Lokalne embeddings - bez kosztów API."""
27        return self.minilm.encode(texts).tolist()
28
29# Porównanie modeli
30models = EmbeddingModels()
31
32texts = [
33    "Uczenie maszynowe to dziedzina AI",
34    "Machine learning is a field of AI",
35    "Lubię jeść pizzę"
36]
37
38# Open-source embeddings
39embeddings = models.embed_with_sentence_transformers(texts)
40print(f"Lokalne embeddings: {len(embeddings[0])} wymiarów")

Metryki podobieństwa

1import numpy as np
2from typing import Callable
3
4def cosine_similarity(vec1: np.ndarray, vec2: np.ndarray) -> float:
5    """Podobieństwo kosinusowe - najczęściej używane."""
6    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
7
8def euclidean_distance(vec1: np.ndarray, vec2: np.ndarray) -> float:
9    """Odległość euklidesowa."""
10    return np.linalg.norm(vec1 - vec2)
11
12def dot_product(vec1: np.ndarray, vec2: np.ndarray) -> float:
13    """Iloczyn skalarny."""
14    return np.dot(vec1, vec2)
15
16# Demonstracja
17def demonstrate_similarity():
18    """Pokazuje jak działają metryki podobieństwa."""
19    from sentence_transformers import SentenceTransformer
20
21    model = SentenceTransformer('all-MiniLM-L6-v2')
22
23    sentences = [
24        "Kot siedzi na macie",           # 0
25        "Kotek leży na dywanie",         # 1 - podobne znaczenie
26        "Programowanie w Pythonie",       # 2 - inne znaczenie
27    ]
28
29    embeddings = model.encode(sentences)
30
31    print("Podobieństwo kosinusowe:")
32    print(f"  Zdanie 0 vs 1: {cosine_similarity(embeddings[0], embeddings[1]):.3f}")
33    print(f"  Zdanie 0 vs 2: {cosine_similarity(embeddings[0], embeddings[2]):.3f}")
34    print(f"  Zdanie 1 vs 2: {cosine_similarity(embeddings[1], embeddings[2]):.3f}")
35
36demonstrate_similarity()

Semantic Search

1from dataclasses import dataclass
2import numpy as np
3
4@dataclass
5class SearchResult:
6    """Wynik wyszukiwania semantycznego."""
7    text: str
8    score: float
9    index: int
10
11class SemanticSearch:
12    """Prosta implementacja wyszukiwania semantycznego."""
13
14    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
15        from sentence_transformers import SentenceTransformer
16        self.model = SentenceTransformer(model_name)
17        self.documents: list[str] = []
18        self.embeddings: np.ndarray | None = None
19
20    def index_documents(self, documents: list[str]) -> None:
21        """Indeksuje dokumenty."""
22        self.documents = documents
23        self.embeddings = self.model.encode(documents)
24        print(f"Zaindeksowano {len(documents)} dokumentów")
25
26    def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
27        """Wyszukuje najbardziej podobne dokumenty."""
28        if self.embeddings is None:
29            raise ValueError("Brak zaindeksowanych dokumentów!")
30
31        query_embedding = self.model.encode([query])[0]
32
33        # Oblicz podobieństwa
34        similarities = np.dot(self.embeddings, query_embedding)
35        similarities /= np.linalg.norm(self.embeddings, axis=1)
36        similarities /= np.linalg.norm(query_embedding)
37
38        # Top-K wyników
39        top_indices = np.argsort(similarities)[::-1][:top_k]
40
41        results = []
42        for idx in top_indices:
43            results.append(SearchResult(
44                text=self.documents[idx],
45                score=float(similarities[idx]),
46                index=int(idx)
47            ))
48
49        return results
50
51# Przykład użycia
52search = SemanticSearch()
53
54documents = [
55    "Python jest językiem programowania ogólnego przeznaczenia",
56    "Machine Learning wykorzystuje algorytmy do uczenia się z danych",
57    "RAG łączy wyszukiwanie z generowaniem tekstu",
58    "FastAPI to nowoczesny framework do budowania API",
59    "Docker konteneryzuje aplikacje",
60]
61
62search.index_documents(documents)
63
64results = search.search("Jak budować aplikacje webowe?")
65for r in results:
66    print(f"Score: {r.score:.3f} | {r.text}")

Batch Processing

1def batch_embed(texts: list[str], batch_size: int = 100) -> list[list[float]]:
2    """Embedowanie dużej ilości tekstów w partiach."""
3    from openai import OpenAI
4    client = OpenAI()
5
6    all_embeddings = []
7
8    for i in range(0, len(texts), batch_size):
9        batch = texts[i:i + batch_size]
10
11        response = client.embeddings.create(
12            model="text-embedding-3-small",
13            input=batch
14        )
15
16        batch_embeddings = [item.embedding for item in response.data]
17        all_embeddings.extend(batch_embeddings)
18
19        print(f"Przetworzono {min(i + batch_size, len(texts))}/{len(texts)}")
20
21    return all_embeddings
22
23# Cache'owanie embeddingów
24import hashlib
25import json
26from pathlib import Path
27
28class EmbeddingCache:
29    """Cache dla embeddingów."""
30
31    def __init__(self, cache_dir: str = ".embedding_cache"):
32        self.cache_dir = Path(cache_dir)
33        self.cache_dir.mkdir(exist_ok=True)
34
35    def _get_cache_key(self, text: str, model: str) -> str:
36        """Generuje klucz cache."""
37        content = f"{model}:{text}"
38        return hashlib.md5(content.encode()).hexdigest()
39
40    def get(self, text: str, model: str) -> list[float] | None:
41        """Pobiera embedding z cache."""
42        key = self._get_cache_key(text, model)
43        cache_file = self.cache_dir / f"{key}.json"
44
45        if cache_file.exists():
46            return json.loads(cache_file.read_text())
47        return None
48
49    def set(self, text: str, model: str, embedding: list[float]) -> None:
50        """Zapisuje embedding do cache."""
51        key = self._get_cache_key(text, model)
52        cache_file = self.cache_dir / f"{key}.json"
53        cache_file.write_text(json.dumps(embedding))

Embeddings to fundament każdego systemu RAG. W następnej lekcji poznasz vector databases - specjalizowane bazy danych do przechowywania i wyszukiwania wektorów!