numpy pour embeddings

TL;DR — Un embedding est un vecteur dense de flottants (float32) qui place un morceau de texte dans un espace géométrique où « proche = sémantiquement similaire ». NumPy est l'outil de base pour les manipuler : on stocke N embeddings dans une matrice (N, d), on les normalise (norme L2 = 1), et la similarité cosinus se réduit alors à un simple produit matriciel embeddings @ query. Maîtriser dtype, le broadcasting, la normalisation et la vectorisation (pas de boucles Python) suffit à construire un moteur de recherche sémantique correct sur des centaines de milliers de vecteurs — et à comprendre exactement ce qu'une base vectorielle fait sous le capot avant d'en avoir réellement besoin. Pour un agent LLM, c'est la brique du RAG : récupérer les k passages les plus pertinents à injecter dans le contexte de Claude.

---

🧠 Mental model

Venant de TS/PHP, ton réflexe est de penser « liste d'objets, je .filter() / .map() ». Avec les embeddings, change de modèle mental : tu ne raisonnes plus sur des objets un par un, mais sur une grande feuille de calcul de nombres que tu transformes d'un bloc.

L'analogie la plus utile : un embedding est une coordonnée GPS dans un espace à 1024 dimensions. Deux villes proches sur la carte (Paris / Versailles) ont des coordonnées proches ; deux concepts proches (« chien » / « chiot ») ont des vecteurs proches. La « distance » entre deux textes n'est plus une comparaison de chaînes, c'est une distance géométrique.

NumPy est le tableur qui contient toutes ces coordonnées. Au lieu d'une boucle « pour chaque ligne, calcule la distance », tu poses une seule opération sur toute la matrice — comme une formule Excel tirée sur 100 000 lignes d'un coup.

            d dimensions (ex: 1024)
        ┌───────────────────────────────┐
doc 0   │ 0.02  -0.91   0.33  ...   0.10 │  ┐
doc 1   │ 0.45   0.12  -0.07  ...  -0.88 │  │  matrice
doc 2   │-0.31   0.50   0.61  ...   0.04 │  │  (N, d)
  ...   │  ...    ...    ...   ...   ... │  │
doc N-1 │ 0.77  -0.03   0.19  ...   0.42 │  ┘
        └───────────────────────────────┘
                       │
       query (1, d) ───┘   produit scalaire ligne par ligne
                       ▼
        scores = matrice @ query   →   (N,)   un score par doc
                       ▼
           argsort / argpartition  →   les k meilleurs

Le point clé : si tous les vecteurs sont normalisés (norme = 1), le produit scalaire EST la similarité cosinus. Tout le reste de cette leçon découle de cette identité.

---

1. Le vecteur, et pourquoi float32

Un embedding renvoyé par une API ou un modèle est une list[float]. La première erreur de débutant est de la garder telle quelle.

import numpy as np

# Ce que tu reçois d'une API d'embeddings (Voyage, OpenAI, un modèle local...)
raw: list[float] = [0.021, -0.913, 0.334, 0.102]  # en réalité, 256–3072 dims

# ❌ La mauvaise façon : garder une list[float]
# Chaque float Python = un objet boxé de 24+ octets, dispersé dans le heap.
# Aucune opération vectorisée possible, tout passe par des boucles Python.

# ✅ La bonne façon : un np.ndarray en float32
vec = np.asarray(raw, dtype=np.float32)
print(vec.shape, vec.dtype)  # (4,) float32

Pourquoi float32 et pas float64 (le défaut de NumPy) ?

• Mémoire : 1M de vecteurs de 1024 dims = 4 Go en float32, 8 Go en float64. Sur les embeddings, la précision float64 n'apporte rien (le bruit du modèle dépasse largement l'epsilon machine).
• Vitesse : moitié moins d'octets à déplacer entre la RAM et le CPU → débit mémoire doublé, et les instructions SIMD en traitent deux fois plus par cycle.
• Compatibilité : c'est le dtype qu'attendent FAISS, ONNX Runtime et la plupart des bases vectorielles. Le passer en float64 provoque des copies silencieuses coûteuses.

⚠️ np.array(raw) sans dtype te donne du float64. Sois explicite : np.asarray(raw, dtype=np.float32). asarray (et non array) évite une copie si l'entrée est déjà un ndarray du bon type.

---

2. La matrice (N, d) : stocker plusieurs embeddings

On ne manipule jamais un vecteur isolé en production : on a un corpus. La structure canonique est une matrice 2D où chaque ligne est un document.

import numpy as np

def stack_embeddings(rows: list[list[float]]) -> np.ndarray:
    """Empile N embeddings en une matrice (N, d) contiguë en float32."""
    return np.asarray(rows, dtype=np.float32)

corpus = stack_embeddings([
    [0.02, -0.91, 0.33, 0.10],
    [0.45,  0.12, -0.07, -0.88],
    [-0.31, 0.50, 0.61, 0.04],
])
print(corpus.shape)  # (3, 4)  →  3 docs, 4 dimensions

Convention (N, d), jamais (d, N). C'est la convention de tout l'écosystème ML (scikit-learn, PyTorch, FAISS). La raison est physique : NumPy stocke en row-major (C order), donc une ligne — un document complet — est contiguë en mémoire. Parcourir les documents un par un est alors cache-friendly. Inverser les axes pénalise chaque opération.

# Inspecter la mémoire — réflexe à acquérir
print(corpus.flags["C_CONTIGUOUS"])  # True : prêt pour le calcul rapide
print(corpus.nbytes)                 # 48 octets (3 * 4 * 4)

---

3. La normalisation L2 : l'étape que tout le monde oublie

La similarité cosinus entre deux vecteurs a et b est :

cos(a, b) = (a · b) / (||a|| * ||b||)

Si tu pré-normalises chaque vecteur pour que sa norme vaille 1, alors ||a|| = ||b|| = 1 et la formule s'effondre en cos(a, b) = a · b. Un simple produit scalaire. C'est toute l'astuce du RAG performant.

import numpy as np

def l2_normalize(mat: np.ndarray, eps: float = 1e-12) -> np.ndarray:
    """Normalise chaque LIGNE pour qu'elle ait une norme L2 de 1.

    eps évite la division par zéro pour un vecteur nul (rare mais réel).
    """
    # axis=1 → norme par ligne ; keepdims=True → garde la forme (N, 1) pour le broadcasting
    norms = np.linalg.norm(mat, axis=1, keepdims=True)
    return mat / np.maximum(norms, eps)

corpus_n = l2_normalize(corpus)
# Vérification : chaque ligne a maintenant une norme de ~1.0
print(np.linalg.norm(corpus_n, axis=1))  # [1. 1. 1.]

Deux pièges classiques de débutant ici, qui sont tous deux des bugs silencieux :

# ❌ FAUX : axis=0 normalise par COLONNE (par dimension), pas par document.
# Le résultat n'a aucun sens — tes scores seront du bruit, sans erreur levée.
bad = mat / np.linalg.norm(mat, axis=0)

# ❌ FAUX : sans keepdims, norms a la forme (N,), pas (N, 1).
# Le broadcasting échoue ou — pire — s'aligne sur le mauvais axe.
norms = np.linalg.norm(mat, axis=1)      # shape (N,)
bad = mat / norms                        # ValueError, ou alignement silencieux erroné

# ✅ JUSTE : axis=1 (par ligne) + keepdims=True (forme (N, 1))

Le broadcasting est ce qui rend mat / norms possible sans boucle : NumPy étire (N, 1) sur (N, d) automatiquement en répétant virtuellement la colonne. Comprendre le broadcasting, c'est comprendre 80 % de NumPy.

🔑 Règle d'or : normalise tout à l'ingestion, une fois. Stocke les vecteurs déjà normalisés. Ne normalise jamais à chaque requête — c'est du calcul jeté.

---

4. La recherche par similarité : un produit matriciel, pas une boucle

C'est là que les ex-développeurs PHP/TS perdent le plus de temps : ils écrivent une boucle. Ne fais jamais ça.

import numpy as np

# ❌ LA MAUVAISE FAÇON — idiomatique en PHP/JS, catastrophique en NumPy.
# Pour 100k docs × 1024 dims : ~plusieurs secondes, et ça scale linéairement mal.
def search_slow(corpus_n: np.ndarray, query_n: np.ndarray, k: int) -> list[int]:
    scores = []
    for i in range(corpus_n.shape[0]):          # boucle Python = mort par mille coupures
        scores.append(float(np.dot(corpus_n[i], query_n)))
    ranked = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)
    return ranked[:k]

# ✅ LA BONNE FAÇON — une seule opération vectorisée, déléguée à BLAS (C/Fortran).
# Pour 100k docs × 1024 dims : ~quelques millisecondes. 100–1000× plus rapide.
def search(corpus_n: np.ndarray, query_n: np.ndarray, k: int) -> tuple[np.ndarray, np.ndarray]:
    """Retourne (indices, scores) des k documents les plus similaires.

    Pré-condition : corpus_n et query_n sont DÉJÀ normalisés L2.
    """
    # corpus_n: (N, d) @ query_n: (d,)  →  scores: (N,)
    scores = corpus_n @ query_n  # produit scalaire de chaque ligne avec la requête

    # argpartition est O(N) : il place les k plus grands "quelque part" en tête,
    # sans trier les N-k autres. argsort serait O(N log N) sur tout le tableau.
    top_unsorted = np.argpartition(-scores, kth=k - 1)[:k]
    # On ne trie QUE ces k éléments, pas les N.
    top = top_unsorted[np.argsort(-scores[top_unsorted])]
    return top, scores[top]

Pourquoi le @ (produit matriciel) écrase la boucle :

• Il appelle BLAS (OpenBLAS / MKL / Accelerate), du code C/Fortran optimisé, multi-thread, vectorisé SIMD, conscient du cache CPU.
• Aucun aller-retour dans l'interpréteur Python par élément (le coût réel de search_slow).
• Une intention claire en une ligne.

Le détail qui sépare le junior du senior : argpartition plutôt que argsort. Tu veux le top-k de 1M de scores ? Trier le million entier (argsort, O(N log N)) pour ne garder que 10 résultats est du gâchis. argpartition (O(N)) partitionne pour isoler les k meilleurs, puis tu ne tries que ces k-là. Sur 1M de docs, c'est un facteur de plusieurs fois.

Note sur le signe : np.argpartition(-scores, ...) car argpartition/argsort trient en ordre croissant. Nier le score inverse l'ordre pour obtenir les plus grands (les plus similaires). Avec une distance (euclidienne), on ne nie pas — on veut les plus petites.

Recherche par lot (batch) : plusieurs requêtes d'un coup

Pour évaluer un jeu de tests ou traiter plusieurs requêtes utilisateur, ne boucle pas non plus :

def search_batch(corpus_n: np.ndarray, queries_n: np.ndarray, k: int) -> np.ndarray:
    """queries_n: (Q, d) → renvoie (Q, k) indices des top-k par requête."""
    sims = queries_n @ corpus_n.T          # (Q, d) @ (d, N) = (Q, N) matrice de scores
    return np.argpartition(-sims, kth=k - 1, axis=1)[:, :k]

Une seule multiplication matricielle (Q, N) calcule toutes les similarités croisées. C'est exactement ce que fait une base vectorielle, en plus indexé.

---

5. Lien concret avec un agent LLM (RAG) et le SDK Anthropic

Les embeddings ne sont qu'une moitié de l'histoire. Le RAG (Retrieval-Augmented Generation) consiste à : (1) embedder la question, (2) récupérer avec NumPy les k passages les plus pertinents, (3) les injecter dans le contexte de Claude pour qu'il réponde en s'appuyant dessus.

⚠️ Anthropic ne fournit pas d'endpoint d'embeddings. Le SDK anthropic sert à appeler le modèle (Claude). Pour générer les vecteurs, on utilise un fournisseur d'embeddings dédié (Voyage AI, recommandé par Anthropic, ou un modèle local type sentence-transformers). NumPy fait le pont entre les deux : il stocke les vecteurs et fait la recherche.

Voici un mini-pipeline RAG complet et typé. La partie « embedder » est abstraite derrière un protocole (tu branches Voyage, un modèle local, peu importe) ; la partie « récupérer » est du NumPy pur ; la partie « générer » utilise le SDK Anthropic avec streaming et output_config.effort.

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from typing import Protocol

import numpy as np
from anthropic import AsyncAnthropic


class Embedder(Protocol):
    """Toute source d'embeddings : Voyage, sentence-transformers, etc.

    Contrat : renvoie une matrice (len(texts), d) en float32, NON normalisée.
    """
    def embed(self, texts: list[str]) -> np.ndarray: ...


@dataclass(slots=True)
class VectorStore:
    """Index vectoriel en mémoire, normalisé à l'ingestion. ~le cœur d'une base vectorielle."""
    embedder: Embedder
    _matrix: np.ndarray | None = None        # (N, d) normalisé
    _docs: list[str] | None = None

    def index(self, docs: list[str]) -> None:
        mat = self.embedder.embed(docs).astype(np.float32, copy=False)
        norms = np.linalg.norm(mat, axis=1, keepdims=True)
        self._matrix = mat / np.maximum(norms, 1e-12)   # normalise UNE fois
        self._docs = docs

    def retrieve(self, query: str, k: int = 3) -> list[tuple[str, float]]:
        if self._matrix is None or self._docs is None:
            raise RuntimeError("index() doit être appelé avant retrieve()")
        q = self.embedder.embed([query])[0].astype(np.float32, copy=False)
        q /= max(float(np.linalg.norm(q)), 1e-12)        # normalise la requête
        scores = self._matrix @ q                          # (N,) similarités cosinus
        k = min(k, scores.shape[0])
        top = np.argpartition(-scores, kth=k - 1)[:k]
        top = top[np.argsort(-scores[top])]
        return [(self._docs[i], float(scores[i])) for i in top]


async def answer_with_rag(store: VectorStore, question: str) -> str:
    """Récupère le contexte avec NumPy, puis demande à Claude de répondre dessus."""
    passages = store.retrieve(question, k=3)
    context = "\n\n".join(f"[{i}] {text}" for i, (text, _) in enumerate(passages))

    client = AsyncAnthropic()  # lit ANTHROPIC_API_KEY depuis l'environnement
    system = (
        "Tu réponds UNIQUEMENT à partir des extraits fournis. "
        "Si la réponse n'y figure pas, dis-le explicitement. Cite les numéros [i] utilisés."
    )

    # Streaming : recommandé dès que la sortie peut être longue (évite les timeouts HTTP).
    # output_config.effort : "high" par défaut ; ici la tâche est simple → "low" suffit.
    async with client.messages.stream(
        model="claude-opus-4-8",
        max_tokens=1024,
        thinking={"type": "adaptive"},          # Claude décide quand/combien réfléchir
        output_config={"effort": "low"},        # tâche d'extraction simple → effort bas
        system=system,
        messages=[{
            "role": "user",
            "content": f"Contexte :\n{context}\n\nQuestion : {question}",
        }],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
        final = await stream.get_final_message()

    return "".join(b.text for b in final.content if b.type == "text")


# asyncio.run(answer_with_rag(store, "Quelle est la politique de remboursement ?"))

Ce qu'il faut retenir du lien NumPy ↔ Claude :

• NumPy est la couche de récupération, gratuite et locale. Pas besoin de payer un appel LLM pour trouver les passages pertinents.
• On filtre à k petit (3–8) avant d'appeler Claude. Chaque token de contexte se paie : claude-opus-4-8 est facturé 5 $/Mtok en entrée. Injecter 50 passages au lieu de 5 multiplie la facture sans gagner en qualité — souvent l'inverse (bruit).
• Streaming par défaut dès qu'une sortie peut être longue : messages.stream(...) + get_final_message() évite les timeouts et donne un affichage progressif.
• thinking: {type: "adaptive"} et output_config.effort (jamais budget_tokens, supprimé sur Opus 4.8) règlent l'effort de raisonnement. Pour une extraction guidée par le contexte, effort: "low" est largement suffisant et le moins cher.

💡 Pour le RAG en production, mets en cache le contexte volumineux et stable côté Claude (prompt caching) plutôt que de le renvoyer entier à chaque tour — voir la leçon dédiée. Et si Claude Haiku 4.5 (1/5 $ le Mtok) suffit à la qualité, c'est 5× moins cher en entrée que Opus pour de l'extraction simple.

---

⚙️ En production

Modes de défaillance

• dtype qui dérive vers float64. Une opération qui mélange un float32 et un scalaire Python ou un float64 peut upcaster silencieusement toute la matrice en float64, doublant la RAM. Vérifie : assert corpus.dtype == np.float32 après chaque transformation, et utilise .astype(np.float32, copy=False) aux frontières.
• Tableaux non contigus après slicing/transposition. corpus[::2] ou corpus.T créent une vue non contiguë ; un @ dessus déclenche une copie cachée. Force la contiguïté quand tu indexes une base : np.ascontiguousarray(mat).
• Vecteurs nuls. Un document vide peut produire un embedding nul → division par zéro à la normalisation → NaN qui contaminent ensuite tous les scores (NaN se propage). D'où le np.maximum(norms, eps). Détecte-les en amont : np.isnan(scores).any().
• Mismatch de dimensions. Mélanger des embeddings de deux modèles différents (1024 vs 1536 dims) dans la même matrice lève une erreur à l'empilement — mais mélanger deux versions du même modèle (même d, sémantique différente) passe silencieusement et ruine la pertinence. Versionne le modèle d'embedding avec l'index.

Performance et passage à l'échelle

• NumPy en mémoire scale jusqu'à ~1M de vecteurs confortablement (1024 dims × 1M × 4 o = 4 Go). Au-delà, une recherche brute force (Q, N) devient lente et la RAM explose → passe à un index ANN (Approximate Nearest Neighbors) : FAISS, hnswlib, ou une base vectorielle (Qdrant, pgvector). NumPy reste utile pour le prototypage et la vérité-terrain exacte contre laquelle valider l'index approximatif.
• argpartition vs argsort : sur 1M de scores pour un top-10, argpartition (O(N)) est nettement plus rapide qu'argsort (O(N log N)). Réflexe systématique.
• BLAS multi-thread : numpy via OpenBLAS/MKL utilise tous les cœurs pour le @. Si tu lances NumPy à l'intérieur de workers déjà parallélisés (Gunicorn, Celery), tu peux avoir une sur-souscription CPU. Plafonne avec OMP_NUM_THREADS / OPENBLAS_NUM_THREADS.
• np.memmap pour des matrices plus grosses que la RAM : tu mappes le fichier sur disque et l'OS pagine à la demande. Idéal pour un index immuable trop gros pour tenir en mémoire mais consulté par accès localisés.

Sécurité

• Ne jamais pickle un index venant d'une source non fiable : pickle.load exécute du code arbitraire. Pour sérialiser/charger une matrice d'embeddings, utilise np.save / np.load(path) (format .npy, données pures). Si tu dois charger du .npy non fiable, passe allow_pickle=False (le défaut) pour bloquer la désérialisation d'objets.
• Inversion d'embeddings. Un embedding n'est pas anonyme : des attaques peuvent reconstruire approximativement le texte source. Traite un index vectoriel de données personnelles avec le même soin que les données brutes (RGPD : droit à l'effacement → il faut pouvoir supprimer la ligne et réindexer).
• Injection via le contexte récupéré (RAG). Les passages récupérés sont des données non fiables injectées dans le prompt. Un document empoisonné peut contenir « ignore les instructions précédentes ». Garde les instructions opérateur dans le system (canal de confiance), encadre le contexte récupéré comme des données à analyser, et ne lui accorde jamais l'autorité d'instructions.

Observabilité

• Logue la distribution des scores du top-k, pas seulement les indices. Un meilleur score à 0.31 sur du cosinus normalisé = la requête n'a probablement aucun bon match (seuil de pertinence). Sentinelle utile contre les hallucinations de RAG.
• Surveille la latence de récupération séparément de la latence LLM. Si la récupération NumPy passe de 5 ms à 200 ms, c'est ton signal pour migrer vers un ANN.
• Sur l'appel Claude, exploite usage.input_tokens / cache_read_input_tokens pour suivre le coût réel du contexte injecté.

Les arbitrages senior

Choix	Quand	Coût
NumPy brute force exacte	< ~1M vecteurs, prototype, vérité-terrain	Linéaire en N, RAM = N×d×4 o
FAISS / hnswlib (ANN)	> 1M vecteurs, latence critique	Approximation (recall < 100 %), index à entretenir
Base vectorielle managée	Multi-tenant, filtres metadata, persistance	Latence réseau, coût, opérabilité
float32 (défaut)	Quasi toujours	—
float16 / quantization int8	RAM extrême, > 10M vecteurs	Perte de précision, recall en baisse

Le réflexe senior : commence par NumPy exact, mesure, et ne migre vers un ANN que lorsque les chiffres le justifient. Une base vectorielle prématurée est de la complexité gratuite.

---

🏋️ Exercices

Exercice 1 — Normaliser sans boucle, sans NaN

Objectif. Écris l2_normalize(mat: np.ndarray) -> np.ndarray qui normalise chaque ligne d'une matrice (N, d), gère les lignes nulles sans produire de NaN, et préserve le float32. Teste avec une matrice contenant une ligne entièrement à zéro.

Indice / Solution

norms = np.linalg.norm(mat, axis=1, keepdims=True) puis mat / np.maximum(norms, 1e-12). Le keepdims=True donne (N, 1) pour le broadcasting ; np.maximum évite la division par zéro. Vérifie assert not np.isnan(out).any() et assert out.dtype == np.float32. Si le dtype dérive, c'est qu'un scalaire float64 (l'eps) a upcasté — caste l'eps ou la sortie.

Exercice 2 — Recherche top-k vectorisée

Objectif. Implémente search(corpus_n, query_n, k) retournant les indices ET les scores des k documents les plus similaires, sans aucune boucle Python, avec argpartition (pas argsort sur tout le tableau). Vérifie l'égalité des résultats avec une implémentation naïve à boucle sur un petit corpus.

Indice / Solution

scores = corpus_n @ query_n ; top = np.argpartition(-scores, k-1)[:k] ; re-trie ces k : top = top[np.argsort(-scores[top])]. Le test d'équivalence : compare set(search(...)) au top-k d'un sorted(enumerate(scores), key=..., reverse=True). Attention aux égalités de scores qui peuvent réordonner — compare les scores triés, pas seulement les indices.

Exercice 3 — Du prototype au production-grade : VectorStore

Objectif. Construis une classe VectorStore (avec un Embedder factice qui renvoie des vecteurs aléatoires déterministes via np.random.default_rng(seed)) qui : indexe à l'ingestion (normalise une fois), persiste sur disque avec np.save, recharge avec np.load, et expose retrieve(query, k). Mesure le temps de retrieve sur 100 000 vecteurs de 768 dims avec time.perf_counter.

Indice / Solution

Stocke _matrix (normalisée) et _docs. Persistance : np.save("index.npy", self._matrix) et un json à part pour les docs. np.load("index.npy") au rechargement (laisse allow_pickle=False). Pour le benchmark, génère le corpus avec rng.standard_normal((100_000, 768), dtype=np.float32) puis normalise. Tu devrais voir une récupération en quelques millisecondes — c'est la baseline contre laquelle juger tout ANN.

Exercice 4 — Casse-le, puis répare-le : le piège du broadcasting

Objectif. On te donne ce code « qui marche » sur un petit jeu mais donne des résultats absurdes en vrai. Identifie le bug, explique pourquoi il ne lève pas d'erreur, corrige-le.

def normalize_broken(mat: np.ndarray) -> np.ndarray:
    norms = np.linalg.norm(mat, axis=0)   # ⚠️
    return mat / norms

Indice / Solution

axis=0 calcule la norme par colonne (par dimension), pas par document. La forme (d,) qui en résulte se broadcast sur (N, d) sans erreur car les dimensions s'alignent par la droite — d'où le bug silencieux : chaque dimension est mise à l'échelle globalement, les lignes ne sont pas normalisées, les similarités cosinus deviennent du bruit. Correction : axis=1, keepdims=True. Leçon : un broadcasting qui « passe » ne signifie pas qu'il fait ce que tu crois. Vérifie toujours np.linalg.norm(out, axis=1) ≈ 1.

Exercice 5 — Recherche par lot et matrice de similarité

Objectif. Implémente search_batch(corpus_n, queries_n, k) -> np.ndarray qui calcule en une seule multiplication matricielle les top-k pour Q requêtes simultanées. Ensuite, calcule la matrice de similarité intra-corpus (N, N) et identifie les paires de quasi-doublons (score > 0.95) — utile pour dédupliquer un corpus avant indexation.

Indice / Solution

sims = queries_n @ corpus_n.T donne (Q, N) ; np.argpartition(-sims, k-1, axis=1)[:, :k] pour le top-k par ligne. Pour l'intra-corpus : S = corpus_n @ corpus_n.T ; mets la diagonale à -inf (np.fill_diagonal(S, -np.inf)) pour ignorer l'auto-similarité, puis np.argwhere(S > 0.95). Attention : (N, N) est en O(N²) mémoire — pour 100k docs c'est 40 Go ! D'où l'intérêt de traiter par blocs (chunks) ou d'utiliser un ANN dédié au dédoublonnage à grande échelle.

Exercice 6 — Mini-RAG branché sur Claude

Objectif. Assemble les exercices 3 et le pipeline de la leçon : indexe 5–10 paragraphes, récupère le top-3 pour une question, et appelle claude-opus-4-8 en streaming avec output_config: {effort: "low"} pour répondre uniquement à partir du contexte. Ajoute une garde : si le meilleur score de récupération est sous un seuil (ex. 0.2), réponds « information non disponible » sans appeler Claude.

Indice / Solution

Réutilise VectorStore.retrieve. La garde de seuil économise un appel LLM inutile (et coûteux) quand rien n'est pertinent — un pattern de production essentiel contre les hallucinations. Pour l'appel : async with client.messages.stream(model="claude-opus-4-8", max_tokens=1024, thinking={"type": "adaptive"}, output_config={"effort": "low"}, system=..., messages=[...]) as stream: puis async for t in stream.text_stream. Le system doit interdire de répondre hors-contexte et exiger de citer les [i]. Récupère le texte final avec await stream.get_final_message().

---

🎤 En entretien

Q : Pourquoi normalise-t-on les embeddings avant une recherche par similarité cosinus, et que gagne-t-on ? R : Parce qu'avec des vecteurs de norme 1, la similarité cosinus se réduit à un simple produit scalaire — donc la recherche top-k devient une multiplication matricielle (N, d) @ (d,) déléguée à BLAS, au lieu d'un calcul de cosinus complet par paire. On normalise une fois à l'ingestion, jamais à la requête.

Q : argsort ou argpartition pour récupérer le top-10 sur 1M de scores ? Pourquoi ? R : argpartition, en O(N), car il isole les 10 meilleurs sans trier le million entier ; on ne trie ensuite que ces 10. argsort est O(N log N) sur tout le tableau — du travail jeté pour 999 990 éléments dont on ne veut pas l'ordre.

Q : Pourquoi float32 plutôt que float64 pour stocker des embeddings ? R : Moitié moins de RAM (4 Go vs 8 Go pour 1M×1024), débit mémoire et SIMD doublés, et c'est le dtype attendu par FAISS et les bases vectorielles. La précision float64 n'apporte rien : le bruit du modèle dépasse largement l'epsilon machine de float32.

Q : Jusqu'où une recherche brute force NumPy en mémoire tient-elle, et quand migrer ? R : Confortablement jusqu'à ~1M de vecteurs (recherche en quelques ms, RAM raisonnable). Au-delà, ou si la latence devient critique, on passe à un index ANN (FAISS, hnswlib, pgvector) qui échange un peu de recall contre une latence sous-linéaire. Le réflexe senior : commencer exact avec NumPy, mesurer, et ne migrer que quand les chiffres l'exigent — NumPy reste alors la vérité-terrain pour valider le recall de l'ANN.