Un motore di ricerca tradizionale si basa spesso sulla corrispondenza lessicale tra parole: l’utente cerca una parola chiave e il sistema restituisce documenti che contengono esattamente quella parola, o varianti molto simili. Questo approccio funziona bene in molti casi, ma ha un limite importante: fatica a comprendere il significato.
Ad esempio, una ricerca per:
“come funziona la ricerca semantica”
potrebbe non restituire documenti che parlano di:
“embedding testuali e similarità tra frasi”
anche se i due concetti sono molto vicini.
I motori di ricerca vettoriali nascono per superare questo limite. Invece di cercare parole identiche, lavorano su rappresentazioni numeriche del significato, chiamate embedding.
Un embedding è un vettore di numeri reali che rappresenta un contenuto: una frase, un’immagine, un prodotto, un documento, un frammento di codice. Contenuti semanticamente simili avranno vettori vicini nello spazio vettoriale.
In pratica:
- il testo viene trasformato in un vettore;
- la ricerca avviene confrontando la distanza o la similarità tra vettori;
- i risultati più rilevanti sono quelli con vettori più vicini alla query.
Questo tipo di ricerca è alla base di molte applicazioni moderne:
- sistemi RAG per LLM;
- ricerca semantica su documentazione tecnica;
- raccomandazione di prodotti o contenuti;
- deduplicazione di testi;
- ricerca su immagini o audio;
- chatbot con memoria contestuale;
- motori di ricerca aziendali su documenti interni.
In questo tutorial costruiremo un mini motore di ricerca vettoriale in Python, partendo da zero. L’obiettivo non è solo usare una libreria esistente, ma capire come funzionano gli elementi fondamentali:
- normalizzazione dei vettori;
- similarità coseno;
- indice vettoriale;
- metadata filtering;
- persistenza;
- API HTTP;
- benchmark;
- evoluzione verso indici approssimati come HNSW.
Alla fine avremo un piccolo motore vettoriale funzionante, didattico ma sufficientemente solido per capire cosa succede dentro sistemi più complessi come Qdrant, Milvus, Weaviate, Pinecone, pgvector o Faiss.
Tutorial: costruire un motore di ricerca vettoriale in Python
Obiettivo del progetto
Vogliamo costruire un sistema in grado di:
- ricevere documenti testuali;
- trasformarli in embedding;
- memorizzarli in un indice vettoriale;
- cercare documenti simili a una query;
- filtrare i risultati tramite metadata;
- esporre il tutto tramite API HTTP;
- salvare i dati su disco.
Useremo:
- Python;
- NumPy per il calcolo vettoriale;
- FastAPI per esporre le API;
- Sentence Transformers per generare embedding testuali.
Il progetto sarà didattico, quindi privilegeremo la chiarezza. Alla fine vedremo anche cosa cambiare per avvicinarci a un sistema production-ready.
Prerequisiti
Per seguire il tutorial ti serviranno:
- Python 3.10 o superiore;
- familiarità con Python;
- conoscenze base di API HTTP;
- concetti base di algebra lineare;
- terminale o IDE a scelta.
Installeremo alcune librerie:
pip install numpy fastapi uvicorn sentence-transformersNota: sentence-transformers può scaricare modelli anche pesanti. Per il tutorial useremo un modello leggero, ma in alternativa puoi usare qualsiasi altro provider di embedding.
1. Concetti fondamentali
1.1 Embedding
Un embedding è una rappresentazione numerica di un contenuto.
Ad esempio, la frase:
“Il gatto dorme sul divano”
può essere trasformata in un vettore come:
[0.12, -0.34, 0.98, 0.05, ..., -0.44]La dimensione del vettore dipende dal modello. Nel nostro caso useremo:
sentence-transformers/all-MiniLM-L6-v2che produce embedding di dimensione:
384Frasi simili avranno vettori simili.
1.2 Similarità coseno
Per confrontare due vettori possiamo usare la similarità coseno.
Date due rappresentazioni:
A = [a1, a2, ..., an]
B = [b1, b2, ..., bn]la similarità coseno è:
cosine_similarity(A, B) = (A · B) / (||A|| * ||B||)Dove:
A · Bè il prodotto scalare;||A||e||B||sono le norme dei vettori.
Il risultato è normalmente compreso tra -1 e 1.
Valori più alti indicano maggiore similarità.
Se normalizziamo i vettori in modo che abbiano lunghezza unitaria, la similarità coseno diventa semplicemente:
A · BQuesto rende il calcolo molto più semplice e veloce.
1.3 Ricerca brute force
Il modo più semplice per cercare vettori simili è confrontare la query con tutti i vettori memorizzati.
Questo approccio si chiama brute force search.
Pro:
- semplice;
- preciso;
- ottimo per dataset piccoli o medi;
- perfetto per capire il funzionamento interno.
Contro:
- diventa lento con milioni di documenti;
- ogni query richiede il confronto con tutti i vettori.
Per questo nei sistemi reali si usano spesso indici approssimati, ad esempio:
- HNSW;
- IVF;
- Product Quantization;
- ScaNN;
- Faiss;
- Annoy.
Nel nostro tutorial partiremo da un indice brute force, poi vedremo come evolverlo.
2. Struttura del progetto
Creiamo una cartella:
vector-search-tutorial
cd vector-search-tutorialPossiamo organizzare il progetto così:
vector-search-tutorial/
│
├── app.py
├── data/
└── requirements.txtIl file principale sarà:
app.py3. Installazione delle dipendenze
Creiamo un ambiente virtuale:
python -m venv .venvAttiviamolo:
Su Linux/macOS:
source .venv/bin/activateSu Windows:
.venv\Scripts\activateInstalliamo le librerie:
pip install numpy fastapi uvicorn sentence-transformers4. Creare il motore vettoriale
Iniziamo dal cuore del sistema: un indice vettoriale in memoria.
Il nostro motore dovrà:
- accettare vettori;
- normalizzarli;
- memorizzarli;
- cercare i più simili a una query;
- supportare filtri sui metadata.
4.1 Import e classi base
In app.py:
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
import numpy as npDefiniamo il risultato di una ricerca:
@dataclass
class SearchHit:
id: str
score: float
metadata: dict[str, Any]Ogni risultato avrà:
id: identificativo del documento;score: punteggio di similarità;metadata: informazioni aggiuntive.
4.2 Classe VectorStore
Creiamo la classe principale:
class VectorStore:
def __init__(self, dim: int):
self.dim = dim
self.ids: list[str] = []
self.metadata: list[dict[str, Any]] = []
self.vectors: np.ndarray | None = Nonedim è la dimensione degli embedding.
Ad esempio:
store = VectorStore(dim=384)4.3 Normalizzazione dei vettori
Aggiungiamo un metodo per normalizzare i vettori:
def _normalize(self, vectors: np.ndarray | list[list[float]]) -> np.ndarray:
v = np.asarray(vectors, dtype=np.float32)
if v.ndim == 1:
v = v.reshape(1, -1)
if v.shape[1] != self.dim:
raise ValueError(
f"Attesa dimensione {self.dim}, ricevuta {v.shape[1]}"
)
norms = np.linalg.norm(v, axis=1, keepdims=True)
norms = np.where(norms == 0, 1.0, norms)
return v / normsQuesto metodo:
- converte l’input in un array NumPy;
- gestisce sia singoli vettori sia batch di vettori;
- controlla la dimensione;
- normalizza ogni vettore.
4.4 Inserimento documenti
Ora aggiungiamo il metodo add:
def add(self, items: Iterable[dict[str, Any]]) -> None:
item_list = list(items)
if not item_list:
return
ids: list[str] = []
vectors: list[np.ndarray] = []
metadata: list[dict[str, Any]] = []
for item in item_list:
if "id" not in item:
raise ValueError("Ogni documento deve avere un id")
if "vector" not in item:
raise ValueError("Ogni documento deve avere un vector")
ids.append(str(item["id"]))
vectors.append(np.asarray(item["vector"], dtype=np.float32))
metadata.append(item.get("metadata", {}))
matrix = np.vstack(vectors)
matrix = self._normalize(matrix)
if self.vectors is None:
self.vectors = matrix
else:
self.vectors = np.vstack([self.vectors, matrix])
self.ids.extend(ids)
self.metadata.extend(metadata)Ogni documento avrà questa forma:
{
"id": "doc-1",
"vector": [0.1, 0.2, ...],
"metadata": {
"title": "Introduzione ai vettori",
"category": "ai"
}
}4.5 Ricerca vettoriale
Adesso implementiamo la ricerca.
Vogliamo:
- normalizzare la query;
- calcolare la similarità con tutti i vettori;
- applicare eventuali filtri sui metadata;
- restituire i primi
krisultati.
def search(
self,
query_vector: np.ndarray | list[float],
k: int = 10,
where: dict[str, Any] | None = None,
) -> list[SearchHit]:
if self.vectors is None or len(self.ids) == 0:
return []
query = self._normalize(query_vector)
scores = self.vectors @ query.T
scores = scores.ravel()
if where:
mask = np.fromiter(
(self._match_metadata(meta, where) for meta in self.metadata),
dtype=bool,
count=len(self.metadata),
)
if not mask.any():
return []
candidate_idxs = np.where(mask)[0]
candidate_scores = scores[candidate_idxs]
else:
candidate_idxs = np.arange(len(self.ids))
candidate_scores = scores
k = min(k, len(candidate_idxs))
if k <= 0:
return []
if k < len(candidate_scores):
top_local = np.argpartition(-candidate_scores, kth=k - 1)[:k]
top_local = top_local[np.argsort(-candidate_scores[top_local])]
else:
top_local = np.argsort(-candidate_scores)
hits: list[SearchHit] = []
for local_idx in top_local:
global_idx = candidate_idxs[local_idx]
hits.append(
SearchHit(
id=self.ids[global_idx],
score=float(candidate_scores[local_idx]),
metadata=self.metadata[global_idx],
)
)
return hitsIl metodo _match_metadata serve per filtrare i documenti:
def _match_metadata(
self,
metadata: dict[str, Any],
where: dict[str, Any],
) -> bool:
for key, value in where.items():
if metadata.get(key) != value:
return False
return TrueEsempio di ricerca:
results = store.search(
query_vector=query_embedding,
k=5,
where={"category": "ai"}
)5. Persistenza dei dati
Un motore reale deve salvare i dati su disco.
Aggiungiamo due metodi: save e load.
5.1 Salvataggio
def save(self, directory: str | Path) -> None:
directory = Path(directory)
directory.mkdir(parents=True, exist_ok=True)
if self.vectors is not None:
np.save(directory / "vectors.npy", self.vectors)
else:
np.save(
directory / "vectors.npy",
np.empty((0, self.dim), dtype=np.float32),
)
with open(directory / "documents.jsonl", "w", encoding="utf-8") as f:
for doc_id, meta in zip(self.ids, self.metadata):
row = {
"id": doc_id,
"metadata": meta,
}
f.write(json.dumps(row, ensure_ascii=False) + "\n")I dati vengono salvati in due file:
data/
├── vectors.npy
└── documents.jsonlvectors.npy: contiene la matrice dei vettori;documents.jsonl: contiene id e metadata.
5.2 Caricamento
@classmethod
def load(cls, directory: str | Path) -> VectorStore:
directory = Path(directory)
vectors_path = directory / "vectors.npy"
documents_path = directory / "documents.jsonl"
if not vectors_path.exists() or not documents_path.exists():
raise FileNotFoundError("Directory di persistenza non valida")
vectors = np.load(vectors_path)
if vectors.ndim != 2:
raise ValueError("File vettori non valido")
dim = vectors.shape[1]
store = cls(dim=dim)
store.ids = []
store.metadata = []
with open(documents_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
obj = json.loads(line)
store.ids.append(str(obj["id"]))
store.metadata.append(obj.get("metadata", {}))
if len(store.ids) == 0:
store.vectors = None
else:
if vectors.shape[0] != len(store.ids):
raise ValueError("Numero di vettori e documenti non coerente")
store.vectors = vectors
return store6. Generare embedding testuali
Ora colleghiamo il motore a un modello di embedding.
Useremo Sentence Transformers.
Aggiungiamo:
from sentence_transformers import SentenceTransformerCarichiamo il modello:
MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
EMBEDDING_DIM = 384
model = SentenceTransformer(MODEL_NAME)Creiamo una funzione per generare embedding:
def embed_texts(texts: list[str]) -> np.ndarray:
return model.encode(
texts,
normalize_embeddings=True,
convert_to_numpy=True,
).astype(np.float32)Il modello restituisce vettori già normalizzati, ma il nostro VectorStore li normalizzerà comunque per sicurezza.
7. Esempio rapido senza API
Prima di creare l’API, facciamo una prova locale.
Aggiungiamo questo blocco alla fine del file, oppure in uno script separato:
if __name__ == "__main__":
documents = [
{
"id": "doc-1",
"text": "I database vettoriali memorizzano embedding e cercano documenti simili.",
"metadata": {"category": "ai"},
},
{
"id": "doc-2",
"text": "La ricerca semantica usa rappresentazioni numeriche del significato.",
"metadata": {"category": "ai"},
},
{
"id": "doc-3",
"text": "PostgreSQL può essere usato come database relazionale.",
"metadata": {"category": "database"},
},
{
"id": "doc-4",
"text": "Gli LLM possono usare documenti esterni tramite Retrieval Augmented Generation.",
"metadata": {"category": "llm"},
},
]
texts = [doc["text"] for doc in documents]
vectors = embed_texts(texts)
store = VectorStore(dim=EMBEDDING_DIM)
items = []
for doc, vector in zip(documents, vectors):
items.append(
{
"id": doc["id"],
"vector": vector,
"metadata": {
"text": doc["text"],
**doc["metadata"],
},
}
)
store.add(items)
query = "Come funziona la ricerca semantica?"
query_vector = embed_texts([query])[0]
results = store.search(query_vector, k=3)
for hit in results:
print(hit.id, hit.score, hit.metadata["text"])Esegui:
python app.pyDovresti vedere risultati simili a quelli più pertinenti alla query, anche se non contengono esattamente le stesse parole.
8. Esporre il motore con FastAPI
Ora trasformiamo il nostro motore in un servizio HTTP.
Aggiungiamo gli import:
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel, FieldDefiniamo i modelli delle richieste.
8.1 Modelli Pydantic
class DocumentIn(BaseModel):
id: str
text: str
metadata: dict[str, Any] = Field(default_factory=dict)
class SearchIn(BaseModel):
query: str
k: int = 10
where: dict[str, Any] | None = None8.2 Stato globale
Useremo due variabili globali:
store: VectorStore | None = None
DATA_DIR = Path("data")8.3 Lifespan dell’applicazione
FastAPI può eseguire codice all’avvio.
@asynccontextmanager
async def lifespan(app: FastAPI):
global store
if DATA_DIR.exists() and (DATA_DIR / "documents.jsonl").exists():
store = VectorStore.load(DATA_DIR)
else:
store = VectorStore(dim=EMBEDDING_DIM)
yield8.4 Creazione dell’app
app = FastAPI(
title="Mini Vector Search API",
lifespan=lifespan,
)8.5 Endpoint per aggiungere documenti
@app.post("/documents")
def add_documents(payload: list[DocumentIn]) -> dict[str, Any]:
if store is None:
raise RuntimeError("Store non inizializzato")
texts = [item.text for item in payload]
vectors = embed_texts(texts)
items = []
for doc, vector in zip(payload, vectors):
metadata = {
"text": doc.text,
**doc.metadata,
}
items.append(
{
"id": doc.id,
"vector": vector,
"metadata": metadata,
}
)
store.add(items)
store.save(DATA_DIR)
return {
"added": len(payload),
}8.6 Endpoint per cercare
@app.post("/search")
def search_documents(payload: SearchIn) -> dict[str, Any]:
if store is None:
raise RuntimeError("Store non inizializzato")
query_vector = embed_texts([payload.query])[0]
hits = store.search(
query_vector=query_vector,
k=payload.k,
where=payload.where,
)
return {
"query": payload.query,
"hits": [
{
"id": hit.id,
"score": hit.score,
"metadata": hit.metadata,
}
for hit in hits
],
}8.7 Endpoint di salute
Utile per monitoring:
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}9. Avviare il server
Avvia l’applicazione:
uvicorn app:app --reloadLa documentazione interattiva sarà disponibile su:
http://127.0.0.1:8000/docs10. Testare le API
Aggiungere documenti
Esempio di richiesta POST verso /documents:
[
{
"id": "doc-1",
"text": "I database vettoriali sono usati per la ricerca semantica.",
"metadata": {
"category": "ai"
}
},
{
"id": "doc-2",
"text": "La Retrieval Augmented Generation migliora le risposte degli LLM.",
"metadata": {
"category": "llm"
}
},
{
"id": "doc-3",
"text": "PostgreSQL supporta estensioni come pgvector.",
"metadata": {
"category": "database"
}
}
]Cercare documenti
Richiesta POST verso /search:
{
"query": "Come funziona la ricerca semantica?",
"k": 3
}Risposta simile a:
{
"query": "Come funziona la ricerca semantica?",
"hits": [
{
"id": "doc-1",
"score": 0.71,
"metadata": {
"text": "I database vettoriali sono usati per la ricerca semantica.",
"category": "ai"
}
}
]
}Cercare con filtro
{
"query": "database",
"k": 3,
"where": {
"category": "database"
}
}11. Benchmark: quanto è veloce il brute force?
Il nostro motore confronta la query con tutti i vettori.
Per capire le prestazioni, possiamo generare un dataset sintetico.
Aggiungi questo script in un file separato, ad esempio benchmark.py:
import time
import numpy as np
from app import VectorStore
def make_random_store(n: int, dim: int) -> VectorStore:
rng = np.random.default_rng(42)
vectors = rng.standard_normal((n, dim)).astype(np.float32)
store = VectorStore(dim=dim)
store.vectors = store._normalize(vectors)
store.ids = [str(i) for i in range(n)]
store.metadata = [{"index": i} for i in range(n)]
return store
def main():
n = 100_000
dim = 384
print("Creazione dataset...")
store = make_random_store(n, dim)
rng = np.random.default_rng(7)
query = rng.standard_normal(dim).astype(np.float32)
print("Ricerca...")
start = time.perf_counter()
results = store.search(query, k=10)
elapsed = time.perf_counter() - start
print(f"Tempo: {elapsed * 1000:.2f} ms")
print("Top result:", results[0].id)
if __name__ == "__main__":
main()Esegui:
python benchmark.pySu un portatile moderno, una ricerca brute force su 100.000 vettori da 384 dimensioni può richiedere qualche decina di millisecondi, a seconda dell’hardware.
Per dataset piccoli o medi può andare bene. Per milioni di documenti, però, serve un indice approssimato.
12. Evoluzione: indici approssimati
Il brute force è preciso ma non scala all’infinito.
Nei sistemi reali si usano spesso indici approssimati. Uno dei più diffusi è HNSW, cioè Hierarchical Navigable Small World.
HNSW costruisce una struttura a grafo che permette di trovare vicini approssimati molto velocemente.
Puoi usare ad esempio hnswlib:
pip install hnswlibEsempio:
import hnswlib
import numpy as np
dim = 384
n = 100_000
rng = np.random.default_rng(42)
vectors = rng.standard_normal((n, dim)).astype(np.float32)
ids = list(range(n))
index = hnswlib.Index(space="cosine", dim=dim)
index.init_index(
max_elements=n,
ef_construction=200,
M=16,
)
index.add_items(vectors, ids)
index.set_ef(50)
query = rng.standard_normal((1, dim)).astype(np.float32)
labels, distances = index.knn_query(query, k=10)
print(labels)
print(distances)Attenzione: in hnswlib, con space="cosine", i valori restituiti sono distanze, non similarità. Valori più bassi indicano risultati più simili.
13. Valutare la qualità della ricerca
Quando si passa da brute force a un indice approssimato, bisogna misurare la perdita di precisione.
Una metrica comune è la recall@k.
Esempio:
def recall_at_k(ann_ids: list[str], exact_ids: list[str]) -> float:
return len(set(ann_ids) & set(exact_ids)) / len(exact_ids)Se la brute force restituisce:
[doc-3, doc-9, doc-12]e l’indice approssimato restituisce:
[doc-3, doc-12, doc-7]la recall@3 sarà:
2/3 = 0.66Nei sistemi reali bisogna valutare:
- recall;
- latenza;
- throughput;
- uso di memoria;
- qualità percepita dall’utente.
14. Limiti del nostro motore didattico
Il motore che abbiamo costruito è utile per capire i concetti, ma non è ancora pronto per la produzione.
Mancano o sono semplificati:
- aggiornamento e cancellazione documenti;
- gestione concorrenza;
- indici approssimati;
- filtri più efficienti;
- transazioni e durabilità robusta;
- sharding;
- replica;
- autenticazione;
- rate limiting;
- validazione più forte dei metadata;
- supporto a embedding multi-modali;
- quantizzazione dei vettori;
- caching;
- observability.
15. Come migliorarlo
Ecco alcune evoluzioni interessanti per trasformarlo in un progetto più avanzato.
15.1 Upsert
Aggiungere un metodo per aggiornare un documento esistente:
def upsert(self, item: dict[str, Any]) -> None:
...Bisogna gestire:
- ricerca dell’id esistente;
- sostituzione del vettore;
- sostituzione dei metadata.
15.2 Delete
Aggiungere:
def delete(self, doc_id: str) -> None:
...Anche qui bisogna gestire:
- rimozione dall’array NumPy;
- aggiornamento degli id;
- aggiornamento dei metadata.
15.3 Filtri più avanzati
Invece di semplici filtri di uguaglianza, potresti supportare:
{
"category": "ai",
"year": {
"$gte": 2023
}
}Operatori possibili:
$eq;$ne;$gt;$gte;$lt;$lte;$in;$nin.
15.4 Indice HNSW interno
Potresti integrare:
- Faiss;
- hnswlib;
- Annoy;
- ScaNN.
In quel caso il tuo VectorStore diventerebbe un wrapper sopra un motore ANN.
15.5 Quantizzazione
Per ridurre la memoria puoi usare:
- Product Quantization;
- Scalar Quantization;
- Binary Quantization.
Utile quando hai milioni di vettori.
15.6 Multi-tenancy
Se vuoi usare il motore in un’applicazione con più utenti, puoi aggiungere un campo:
tenant_ide filtrare sempre le ricerche.
15.7 Hybrid search
Un motore moderno spesso combina:
- ricerca vettoriale;
- ricerca keyword;
- filtri strutturati;
- re-ranking.
Ad esempio:
score_finale = alpha * score_vettoriale + beta * score_bm2516. Considerazioni su RAG e LLM
Se vuoi usare questo motore per un sistema RAG, devi considerare anche altri aspetti.
Chunking
Non indicizzare interi documenti se sono lunghi.
Spezzali in chunk:
- per paragrafo;
- per sezione;
- per numero di token;
- con overlap.
Esempio:
chunk_size = 500 token
chunk_overlap = 50 tokenMetadata
Arricchisci ogni chunk con:
- titolo documento;
- fonte;
- autore;
- data;
- sezione;
- permessi;
- lingua;
- versione.
Retrieval quality
Non basta recuperare documenti simili. Spesso serve:
- re-ranking;
- filtraggio per data;
- deduplicazione;
- normalizzazione dei punteggi;
- valutazione con dataset di test.
17. Possibile versione production-ready
Per un uso production-ready, valuteremo:
- database vettoriale dedicato;
- indice HNSW persistito;
- replica dei dati;
- backup;
- autenticazione API;
- validazione input;
- logging strutturato;
- metriche Prometheus/OpenTelemetry;
- test di carico;
- valutazione automatica della retrieval quality;
- pipeline di embedding asincrona;
- cache delle query frequenti.
In molti casi conviene usare tecnologie mature come:
- Qdrant;
- Milvus;
- Weaviate;
- pgvector;
- Elasticsearch con dense vector;
- OpenSearch;
- LanceDB;
- Chroma;
- Faiss per ricerca in-process.
Il nostro motore resta però ottimo per capire cosa succede sotto il cofano.
