Werkplaats / De werkplaats
In jouw eigen tempo
← RAG
MODULE 01 · LES 03

Bouw eerst exact zoeken en meet daarna de benadering

Maak een eigen flat index, bewaak de embeddingruimte en onderzoek hoeveel resultaten een coarse index mist.

75 min3 checkpoints
DIT GA JE ONTDEKKEN
  • Je bouwt een vectorindex met stabiele IDs en een expliciete embeddingruimte.
  • Je vergelijkt benaderend zoeken met de exacte top-k en telt de onderzochte kandidaten.

Probeer ook echte tekstvectoren in het embedding-lab.

Een vraagvector kan met iedere passagevector worden vergeleken. Bij drie passages is dat gemakkelijk; bij veel passages wil je minder vergelijkingen doen. Een vectorindex bewaart daarom vectoren en IDs en organiseert het zoeken.

We beginnen met een flat index die alles vergelijkt. Daarna verdelen we de ruimte in buckets en meten we welke exacte buren verdwijnen. Dit is een kleine eigen NumPy-index; de bronverwijzing naar Faiss biedt verdere context en is geen snelheidsmeting van onze code.

1. Bewaar kopieën en het embeddingcontract

Een index bevat exact ids, vectors en space. De vectoren zijn genormaliseerd; de IDs koppelen ze aan passages. space benoemt in deze oefening het model, de revisie, pooling en normalisatie die bij elkaar horen.

x = np.array([[4.,3.],[3.,4.],[1.,0.]])
ids = ['a','b','c']
space = 'demo-v1'
index = build_flat_index(ids,x,space)
x[0] = [0,5]
ids[0] = 'gewijzigd'
print(index['ids'])  # ['a', 'b', 'c']
print(index['vectors'][0])  # [0.8 0.6]

Maak build_flat_index met een nieuwe ID-lijst en vectorarray. Het voorbeeld wijzigt daarna de invoer; de index blijft de eerdere waarden bewaren. Zonder kopieën kan een latere bewerking ongemerkt je zoekresultaten veranderen.

De normalisatie- en zoekhelpers uit de vorige les zijn gegeven. De applicatie moet zelf een juiste space-string vastleggen; die string inspecteert geen modelgewichten. Verander je bronversies, chunking of encoder, bereken dan de betreffende vectoren opnieuw en bewaar hun metadata.

01
CHECKPOINTIndexeer vectoren samen met hun ruimte-ID

2. Bereken eerst de exacte top-k

Voor query [1,1] is de eenheidsrichting ongeveer [.7071,.7071]. Documenten a=[.8,.6] en b=[.6,.8] krijgen beide ongeveer .9899; c=[1,0] krijgt .7071. De exacte top-twee bestaat dus uit a en b.

hits = search_flat(index,[1,1],space,2)
print([h['id'] for h in hits])
# ['a', 'b']
try:
    search_flat(index,[1,1],'anders',2)
except ValueError:
    print('Andere embeddingruimte')

Maak search_flat met controle van het indexcontract, de querydimensie en space. Een gelijke dimensie is onvoldoende bij een andere embeddingruimte. Dezelfde querytekst kan met een ander model immers een andere vectorbetekenis krijgen.

De flat referentie berekent scores tegen alle N vectoren met D dimensies: het scorewerk groeit met N×D. Onze wrapper valideert en normaliseert opnieuw. Een geteste productie-index kan zulke interne invarianten bij het bouwen bewaken.

Exact zoeken betekent hier exact binnen de opgeslagen vectorruimte. Het bewijst geen inhoudelijke relevantie van de passages. Bewaar die twee evaluatievragen afzonderlijk.

02
CHECKPOINTGebruik exact zoeken als controleerbare referentie

3. Kies buckets en tel wat je overslaat

Neem twee gegeven centroiden: oost=[1,0] en noord=[0,1]. Iedere documentvector gaat naar het centrum met het grootste inproduct. Dat plaatst a en c in oost, en b in noord.

Voor eenheidsvectoren u en v geldt:

‖u−v‖2=2−2u·v

Daarom correspondeert maximaal inproduct hier met minimale kwadratische afstand. Dit sluit aan op k-means, maar deze functie traint geen centroiden: ze zijn invoer. Bij een centrumtie kiest de laagste centroidindex.

coarse = build_coarse_index(
    index,[[1,0],[0,1]])
print(coarse['buckets'])
# [[0, 2], [1]]
for n in [1,2]:
    r = search_coarse(
        coarse,[1,1],space,2,n)
    found = [h['id'] for h in r['hits']]
    print(r['probed'],r['candidates'],
          found)
# [0] 2 ['a', 'c']
# [0, 1] 3 ['a', 'b']

Maak build_coarse_index en search_coarse. Bij één probe wint oost de centroidtie en worden alleen a en c onderzocht. Bij twee probes worden alle drie kandidaten onderzocht en komt b weer in beeld.

Recall@2 tegenover de exacte top-twee is het aandeel exacte IDs dat je terugvindt. Eén probe levert één van twee op, dus .5. Twee probes herstellen beide exacte buren.

exact = {h['id'] for h in hits}
one = search_coarse(
    coarse,[1,1],space,2,1)
approx = {h['id'] for h in one['hits']}
print(len(exact & approx)/len(exact))
# 0.5

Buckets kunnen leeg zijn en minder dan k kandidaten opleveren. Geef daarom probed, het kandidaatenaantal en maximaal k hits terug. De index is een intern in-memory-object, geen veilig opslagformaat voor willekeurige bestanden.

ZES HANDGEMAAKTE VECTORENDe beste buur kan net over de groepsgrens liggen

Vergelijk twee exacte buren met zoeken in alleen de gekozen buckets. De punten zijn geen echte tekstembeddings.

oostnoordwestzuidquery 44.9°rand A · 44°rand B · 46°
Exacte top 2rand A · rand B
Gevonden top 2rand A · oost
Onderzochte kandidaten2 / 6Recall t.o.v. exact50%

De vier vierkanten zijn gegeven centroiden én corpuspunten. Twee extra punten liggen bij de grens van oost en noord. Meer buckets kost meer vergelijkingen; exacte buren zijn nog geen bewijs van inhoudelijke relevantie.

03
CHECKPOINTDoorzoek meer buckets en meet de gemiste buren

Wat neem je mee?

  • Kopieën en een expliciete embeddingruimte bewaken de indexbetekenis.
  • Exact zoeken geeft een referentie voor benaderingsverlies.
  • Meer probes kosten werk en kunnen gemiste buren terugbrengen.

Hierna combineer je deze vectorburen met een lexicale zoekroute.

DE VOLGENDE STAPCombineer woordmatches en vectorburen via hun rang
JOUW EXPERIMENT

Probeer het zelf

Python
main.py

Python start zodra jij code uitvoert.

UITVOER

Je uitvoer verschijnt hier.
Alles rekent in je eigen browser.

Jouw checkpoints0 / 3
01
Indexeer vectoren samen met hun ruimte-ID

Schrijf build_flat_index(ids,vectors,space). Gebruik normalize_rows; IDs uniek, passend bij N, niet-lege strings ≤100. space niet-leeg na strip() en ≤300. Geef exact ids/vectors/space met nieuwe lijst en array; anders ValueError.

Een hint

Normaliseer bij het bouwen en bewaar metadata buiten de vector. Dezelfde dimensie is onvoldoende als versiecontrole.

02
Gebruik exact zoeken als controleerbare referentie

Schrijf search_flat(index,query,space,k). Vereis exact ids/vectors/space in index, valideer via build_flat_index, weiger een andere space en zoek via cosine_search. Geef de top-k terug.

Een hint

Een andere ruimte krijgt ValueError, ook wanneer de query precies de juiste dimensie heeft.

03
Doorzoek meer buckets en meet de gemiste buren

Schrijf build_coarse_index(flat,centroids) en search_coarse(index,query,space,k,nprobe). Bouw: valideer flat, normaliseer 1–min(32,N) passende centroiden, bewaar centroids en buckets met rijindices. Zoek: juiste space/dimensie, exacte ints k=1–N en nprobe=1–C; sorteer centroidties op index en documentties op ID. Geef probed, candidates (aantal) en hits (maximaal k).

Een hint

Gebruik argmax(flat.vectors @ centers.T,axis=1) bij bouwen. Bij zoeken scoor je uitsluitend de rijindices uit de gekozen buckets. Behoud een lege hits-lijst als er geen kandidaten zijn.

Vastgelopen? Bekijk de voorbeeldoplossing
import numpy as np

def normalize_rows(vectors):
    try:
        raw_vectors = np.asarray(vectors)
        if np.iscomplexobj(raw_vectors):
            raise ValueError('Vectorwaarden moeten reëel zijn')
        vectors = np.asarray(raw_vectors, dtype=float)
    except (TypeError, OverflowError) as error:
        raise ValueError('Geen geldige reële vectorinvoer') from error
    if vectors.ndim != 2 or not 1 <= vectors.shape[0] <= 1000 or not 1 <= vectors.shape[1] <= 512 or not np.isfinite(vectors).all() or np.any(np.abs(vectors) > 1e6):
        raise ValueError('Geef een eindige begrensde matrix met maximaal 1000×512 waarden')
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    if np.any(norms < 1e-12):
        raise ValueError('Een nulvector heeft geen richting; vrijwel nul wordt hier geweigerd')
    return vectors / norms

def cosine_search(vectors, query, ids, k):
    vectors = normalize_rows(vectors)
    try:
        raw_query = np.asarray(query)
        if np.iscomplexobj(raw_query):
            raise ValueError('Vectorwaarden moeten reëel zijn')
        query = np.asarray(raw_query, dtype=float)
    except (TypeError, OverflowError) as error:
        raise ValueError('Geen geldige reële vectorinvoer') from error
    if query.shape != (vectors.shape[1],):
        raise ValueError('Query en corpus moeten dezelfde dimensie hebben')
    query = normalize_rows(query[None, :])[0]
    if not isinstance(ids, list) or len(ids) != len(vectors) or any(not isinstance(i, str) or not 1 <= len(i) <= 100 for i in ids) or len(set(ids)) != len(ids):
        raise ValueError('Unieke niet-lege IDs vereist')
    if type(k) is not int or not 1 <= k <= len(ids):
        raise ValueError('Kies 1 tot N resultaten')
    scores = vectors @ query
    order = sorted(range(len(ids)), key=lambda i: (-scores[i], ids[i]))[:k]
    return [{'id': ids[i], 'score': float(scores[i])} for i in order]

def build_flat_index(ids, vectors, space):
    normalized = normalize_rows(vectors)
    if not isinstance(ids, list) or len(ids) != len(normalized) or any(not isinstance(i, str) or not 1 <= len(i) <= 100 for i in ids) or len(set(ids)) != len(ids):
        raise ValueError('Unieke begrensde IDs vereist')
    if not isinstance(space, str) or not space.strip() or len(space) > 300:
        raise ValueError('Geef een niet-lege embeddingruimte-ID van maximaal 300 tekens')
    return {'ids': list(ids), 'vectors': normalized.copy(), 'space': space}

def search_flat(index, query, space, k):
    if not isinstance(index, dict) or set(index) != {'ids', 'vectors', 'space'}:
        raise ValueError('Geef een flat index')
    checked = build_flat_index(index['ids'], index['vectors'], index['space'])
    if not isinstance(space, str) or space != checked['space']:
        raise ValueError('Query en index horen bij verschillende embeddingruimtes')
    return cosine_search(checked['vectors'], query, checked['ids'], k)

def build_coarse_index(flat, centroids):
    if not isinstance(flat, dict) or set(flat) != {'ids', 'vectors', 'space'}:
        raise ValueError('Geef een flat index')
    flat = build_flat_index(flat['ids'], flat['vectors'], flat['space'])
    centers = normalize_rows(centroids)
    if not 1 <= len(centers) <= min(32, len(flat['ids'])) or centers.shape[1] != flat['vectors'].shape[1]:
        raise ValueError('Geef 1 tot min(32,N) passende centroidvectoren')
    labels = np.argmax(flat['vectors'] @ centers.T, axis=1)
    buckets = [np.flatnonzero(labels == i).tolist() for i in range(len(centers))]
    return {**flat, 'centroids': centers.copy(), 'buckets': buckets}

def search_coarse(index, query, space, k, nprobe):
    if not isinstance(index, dict) or set(index) != {'ids', 'vectors', 'space', 'centroids', 'buckets'}:
        raise ValueError('Geef een coarse index uit build_coarse_index')
    # The index is an internal in-memory object, not an untrusted serialized file.
    if not isinstance(space, str) or space != index['space']:
        raise ValueError('Embeddingruimte verschilt')
    if type(k) is not int or not 1 <= k <= len(index['ids']) or type(nprobe) is not int or not 1 <= nprobe <= len(index['centroids']):
        raise ValueError('Ongeldige k of nprobe')
    try:
        raw_query = np.asarray(query)
        if np.iscomplexobj(raw_query):
            raise ValueError('Vectorwaarden moeten reëel zijn')
        query = np.asarray(raw_query, dtype=float)
    except (TypeError, OverflowError) as error:
        raise ValueError('Geen geldige reële vectorinvoer') from error
    if query.shape != (index['vectors'].shape[1],):
        raise ValueError('Verkeerde querydimensie')
    query = normalize_rows(query[None, :])[0]
    center_scores = index['centroids'] @ query
    probes = sorted(range(len(center_scores)), key=lambda i: (-center_scores[i], i))[:nprobe]
    candidates = sorted({i for c in probes for i in index['buckets'][c]})
    scores = index['vectors'][candidates] @ query
    ranked = sorted(zip(candidates, scores), key=lambda item: (-item[1], index['ids'][item[0]]))[:k]
    return {'probed': probes, 'candidates': len(candidates),
            'hits': [{'id': index['ids'][i], 'score': float(score)} for i, score in ranked]}

Voer de code uit en verklaar voor jezelf waarom elke stap werkt.