Semantic Search: Implementácia od nuly

Keyword search je mŕtvy. Semantic search nachádza to, čo používatelia skutočne hľadajú.

Čo je semantic search?

Keyword search:
  Query: "laptop" → Nájde len dokumenty obsahujúce "laptop"

Semantic search:
  Query: "laptop" → Nájde aj "notebook", "MacBook", "počítač na prácu"

Vyhľadávanie na základe významu, nie presných slov.

Architektúra

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Indexing  │     │    Query     │     │   Results   │
│             │     │              │     │             │
│ Documents   │     │ User Query   │     │ Ranked Docs │
│     ↓       │     │     ↓        │     │     ↑       │
│ Chunking    │     │ Embedding    │     │ Re-ranking  │
│     ↓       │     │     ↓        │     │     ↑       │
│ Embedding   │     │ Similarity   │────►│ Top K       │
│     ↓       │     │   Search     │     │             │
│ Vector DB   │◄────│              │     │             │
└─────────────┘     └──────────────┘     └─────────────┘

Step 1: Chunking

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(documents: list[str]) -> list[str]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " "]
    )

    chunks = []
    for doc in documents:
        chunks.extend(splitter.split_text(doc))

    return chunks

# Prečo chunking?
# - Embedding modely majú limit tokenov
# - Menšie chunky = presnejšie výsledky
# - Overlap zachováva kontext

Step 2: Embedding

from openai import OpenAI

client = OpenAI()

def get_embeddings(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )

    return [item.embedding for item in response.data]

# Batch processing
def embed_all_chunks(chunks: list[str], batch_size: int = 100):
    embeddings = []
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        embeddings.extend(get_embeddings(batch))
    return embeddings

Step 3: Vector Storage

import chromadb

# Inicializácia
client = chromadb.PersistentClient(path="./chroma_db")

collection = client.create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}
)

# Indexovanie
def index_documents(chunks: list[str], embeddings: list[list[float]]):
    collection.add(
        documents=chunks,
        embeddings=embeddings,
        ids=[f"chunk_{i}" for i in range(len(chunks))],
        metadatas=[{"source": "docs"} for _ in chunks]
    )

Step 4: Search

def semantic_search(query: str, top_k: int = 5) -> list[dict]:
    # Embed query
    query_embedding = get_embeddings([query])[0]

    # Search
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "distances", "metadatas"]
    )

    return [
        {
            "text": doc,
            "score": 1 - dist,  # Convert distance to similarity
            "metadata": meta
        }
        for doc, dist, meta in zip(
            results["documents"][0],
            results["distances"][0],
            results["metadatas"][0]
        )
    ]

Step 5: Hybrid Search

Kombinácia keyword + semantic:

from rank_bm25 import BM25Okapi

class HybridSearch:
    def __init__(self, chunks: list[str]):
        self.chunks = chunks
        # BM25 for keyword
        tokenized = [c.split() for c in chunks]
        self.bm25 = BM25Okapi(tokenized)
        # Vector for semantic
        self.embeddings = get_embeddings(chunks)

    def search(self, query: str, alpha: float = 0.5) -> list[dict]:
        # Keyword scores
        keyword_scores = self.bm25.get_scores(query.split())

        # Semantic scores
        query_emb = get_embeddings([query])[0]
        semantic_scores = [
            cosine_similarity(query_emb, emb)
            for emb in self.embeddings
        ]

        # Combine
        combined = [
            alpha * kw + (1 - alpha) * sem
            for kw, sem in zip(
                normalize(keyword_scores),
                normalize(semantic_scores)
            )
        ]

        # Rank
        ranked = sorted(
            enumerate(combined),
            key=lambda x: x[1],
            reverse=True
        )[:10]

        return [{"text": self.chunks[i], "score": s} for i, s in ranked]

Step 6: Re-ranking

from cohere import Client

cohere = Client(api_key="xxx")

def rerank(query: str, documents: list[str], top_k: int = 5):
    response = cohere.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=documents,
        top_n=top_k
    )

    return [
        {
            "text": documents[r.index],
            "score": r.relevance_score
        }
        for r in response.results
    ]

Kompletný pipeline

class SemanticSearchEngine:
    def __init__(self, documents: list[str]):
        self.chunks = chunk_documents(documents)
        self.embeddings = embed_all_chunks(self.chunks)
        self.index_documents()

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        # 1. Initial retrieval (fast, more results)
        candidates = semantic_search(query, top_k=20)

        # 2. Re-rank (slower, more accurate)
        reranked = rerank(
            query,
            [c["text"] for c in candidates],
            top_k=top_k
        )

        return reranked

Záver

Semantic search dramaticky zlepšuje používateľský zážitok. S modernými nástrojmi je implementácia otázkou hodín, nie mesiacov.