AI Engineering Academy · Leçon

Indexer : créer et stocker les segments

Créez un embedding pour chaque segment à l’aide de l’API d’embeddings d’OpenAI, puis insérez les vecteurs obtenus avec leurs métadonnées dans une base vectorielle afin de construire un index consultable de vos documents.

Leçon 3 sur 413 étapes

Indexer : créer et stocker les segments est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

L’étape d’indexation : présentation

Après avoir chargé et segmenté vos documents, vous arrivez à l’étape d’indexation : convertir les éléments de texte en embeddings vectoriels et les stocker dans une base de données vectorielle interrogeable. Il s’agit de la dernière étape hors ligne avant de pouvoir répondre aux requêtes. La qualité de vos embeddings ainsi que l’efficacité de votre stratégie de stockage et d’indexation déterminent directement la rapidité et la précision de votre système RAG au moment des requêtes.

Générer des embeddings avec l’API OpenAI

L’approche la plus courante consiste à appeler l’API d’embeddings d’OpenAI avec le texte de vos éléments. Le modèle text-embedding-3-small produit des vecteurs de 1 536 dimensions et coûte 0,02 $ par million de jetons, ce qui est extrêmement peu coûteux pour la plupart des charges de travail. Envoyez plusieurs textes dans un seul appel d’API (jusqu’à 2 048 entrées) afin de maximiser le débit. La réponse contient un vecteur d’embedding par texte d’entrée, dans le même ordre.

from openai import OpenAI

client = OpenAI()

def embed_batch(texts, model='text-embedding-3-small'):
    response = client.embeddings.create(
        model=model,
        input=texts  # up to 2048 texts per call
    )
    return [item.embedding for item in response.data]

# Embed one batch of 100 chunk texts
texts = [chunk['text'] for chunk in chunks[:100]]
vectors = embed_batch(texts)
print(f'Embedding dimension: {len(vectors[0])}')
print(f'Embedded {len(vectors)} chunks')

Traitement par lots pour gagner en efficacité

Lors de l’indexation de milliers d’éléments, l’efficacité est importante. Traitez les éléments par lots de 100 à 500 afin d’équilibrer le débit et l’utilisation de la mémoire. Suivez votre position pour pouvoir reprendre après un échec sans générer à nouveau les embeddings déjà traités. Consignez régulièrement la progression. Pour 100 000 éléments traités par lots de 500, vous effectuerez 200 appels d’API, ce qui prend généralement quelques minutes.

def embed_all_chunks(chunks, batch_size=200):
    embedded = []
    total = len(chunks)
    for i in range(0, total, batch_size):
        batch = chunks[i:i+batch_size]
        texts = [c['text'] for c in batch]
        vectors = embed_batch(texts)
        for chunk, vector in zip(batch, vectors):
            embedded.append({
                **chunk,
                'embedding': vector
            })
        if (i // batch_size) % 10 == 0:
            print(f'Progress: {min(i+batch_size, total)}/{total}')
    return embedded

Gérer les limites de débit pendant l’indexation

L’API d’embeddings d’OpenAI applique des limites de débit mesurées en jetons par minute (TPM). Les tâches d’indexation importantes atteignent ces limites et reçoivent une RateLimitError. Implémentez un retrait exponentiel avec temporisation aléatoire : lorsqu’une erreur de limite de débit se produit, attendez un bref intervalle aléatoire avant de réessayer, en doublant l’attente à chaque échec suivant. Cela répartit les nouvelles tentatives et empêche tous les processus parallèles de solliciter l’API au même moment.

import time
import random
from openai import RateLimitError

def embed_batch_with_retry(texts, max_retries=5):
    for attempt in range(max_retries):
        try:
            return embed_batch(texts)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'Rate limited. Waiting {wait:.1f}s...')
            time.sleep(wait)
    return []

Insérer les vecteurs dans Pinecone

Après avoir généré les représentations vectorielles, insérez-les dans le magasin de vecteurs. L'insertion ou la mise à jour consiste à insérer de nouveaux vecteurs ou à mettre à jour ceux qui existent déjà si le même ID existe — le processus est conçu pour être idempotent. Dans Pinecone, chaque enregistrement inséré ou mis à jour contient l'ID du vecteur, les valeurs de la représentation vectorielle et un dictionnaire de métadonnées regroupant les champs que vous souhaitez filtrer ou afficher ultérieurement. Effectuez les insertions ou mises à jour par lots de 100 enregistrements maximum par appel afin d'optimiser le débit.

import pinecone

pc = pinecone.Pinecone(api_key='YOUR_KEY')
index = pc.Index('rag-index')

def upsert_to_pinecone(embedded_chunks, batch_size=100):
    for i in range(0, len(embedded_chunks), batch_size):
        batch = embedded_chunks[i:i+batch_size]
        vectors = [
            (
                chunk['id'],
                chunk['embedding'],
                {
                    'text': chunk['text'],
                    'source': chunk['metadata']['source'],
                    'page': chunk['metadata'].get('page', 0)
                }
            )
            for chunk in batch
        ]
        index.upsert(vectors=vectors)
        print(f'Upserted {min(i+batch_size, len(embedded_chunks))}/{len(embedded_chunks)}')

Stockage dans pgvector

Avec pgvector, vous insérez directement les représentations vectorielles dans une table PostgreSQL à l'aide du SQL standard. Le type de données vector accepte une liste Python de nombres flottants sérialisée sous forme de chaîne. Après avoir inséré toutes les lignes, créez un index HNSW pour effectuer rapidement des requêtes approximatives de plus proches voisins. L'indexation d'une table existante contenant des millions de lignes peut prendre plusieurs minutes ; créez donc l'index après l'insertion en masse plutôt qu'avant.

import psycopg2
from psycopg2.extras import execute_values

def upsert_to_pgvector(conn, embedded_chunks):
    with conn.cursor() as cur:
        records = [
            (
                chunk['id'],
                chunk['text'],
                chunk['metadata']['source'],
                chunk['metadata'].get('page', 0),
                chunk['embedding']   # list of floats
            )
            for chunk in embedded_chunks
        ]
        execute_values(cur, '''
            INSERT INTO document_chunks (id, text, source, page, embedding)
            VALUES %s
            ON CONFLICT (id) DO UPDATE
            SET text = EXCLUDED.text, embedding = EXCLUDED.embedding
        ''', records)
    conn.commit()

Création de l'index HNSW

HNSW (Hierarchical Navigable Small World) est le type d'index qui permet d'effectuer rapidement une recherche approximative de plus proches voisins. Contrairement à la recherche par force brute (qui compare le vecteur de requête à chaque vecteur stocké), HNSW construit une structure de graphe multicouche qui réduit l'espace de recherche. Le paramètre m contrôle le nombre de connexions de chaque nœud (une valeur plus élevée améliore le rappel, mais utilise davantage de mémoire), tandis que ef_construction contrôle la qualité de l'index lors de sa création.

-- Build HNSW index after bulk insertion
CREATE INDEX CONCURRENTLY ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Set ef_search at query time to trade recall vs speed
SET hnsw.ef_search = 100;

-- Verify index was created
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'document_chunks';

Conception du schéma de métadonnées

Les métadonnées stockées avec chaque vecteur permettent une récupération filtrée très puissante. Concevez votre schéma de métadonnées avant l'indexation : l'ajout de nouveaux champs ultérieurement nécessite une réindexation. Incluez les champs sur lesquels vous effectuerez des filtres (service, type de document, plage de dates), ceux que vous afficherez dans les citations (titre, page, auteur) et ceux qui sont utiles au débogage (index du segment, nombre total de segments, date d'indexation). Gardez des valeurs de métadonnées simples : les chaînes, les nombres et les booléens s'indexent et se filtrent efficacement, contrairement aux objets imbriqués.

# Well-designed metadata schema
METADATA_SCHEMA = {
    # For filtering at retrieval time
    'department': 'HR',         # string
    'doc_type': 'policy',       # string
    'year': 2025,               # integer
    'is_active': True,          # boolean

    # For display in citations
    'title': 'Employee Handbook 2025',
    'author': 'HR Team',
    'page': 12,
    'source': 's3://docs/handbook_2025.pdf',

    # For debugging and updates
    'chunk_index': 3,
    'total_chunks': 24,
    'indexed_at': '2025-09-01T10:00:00Z'
}

Création de points de reprise pour les tâches d'indexation longues

L'indexation d'un corpus volumineux peut prendre des heures. Un plantage en cours de traitement fait perdre tous les progrès accomplis. Implémentez un fichier de point de reprise qui indique quels segments ont été indexés avec succès. Au redémarrage, ignorez les segments déjà indexés et reprenez là où vous vous étiez arrêté. La tâche d'indexation devient ainsi idempotente et peut être reprise en toute sécurité. Stockez le point de reprise sous forme d'ensemble d'ID de segments traités dans un fichier JSON ou une table de base de données.

import json
from pathlib import Path

CHECKPOINT_FILE = '/tmp/index_checkpoint.json'

def load_checkpoint():
    if Path(CHECKPOINT_FILE).exists():
        return set(json.loads(Path(CHECKPOINT_FILE).read_text()))
    return set()

def save_checkpoint(indexed_ids):
    Path(CHECKPOINT_FILE).write_text(json.dumps(list(indexed_ids)))

def index_with_checkpoint(chunks, index):
    done = load_checkpoint()
    remaining = [c for c in chunks if c['id'] not in done]
    print(f'Resuming: {len(done)} done, {len(remaining)} remaining')
    for chunk in remaining:
        upsert_to_pinecone([chunk], index)
        done.add(chunk['id'])
        save_checkpoint(done)

Vérification de l'exhaustivité de l'index

Après l'indexation, vérifiez que tous les segments ont bien été ajoutés au magasin de vecteurs. Comparez le nombre de segments produits par votre outil de découpage au nombre de vecteurs indiqué par l'index. Interrogez l'index avec le texte d'un document connu et vérifiez que le résultat attendu apparaît parmi les 5 premiers. Exécutez quelques requêtes connues de votre jeu de tests de référence et vérifiez que la précision atteint le niveau attendu. Ne supposez jamais que l'index est complet sans le vérifier.

def verify_index(index, chunks, sample_size=10):
    index_stats = index.describe_index_stats()
    total_vectors = index_stats.total_vector_count
    expected = len(chunks)
    print(f'Index vectors: {total_vectors}, Expected: {expected}')
    if total_vectors != expected:
        print('WARNING: mismatch — some chunks may not have been indexed')

    # Spot-check retrieval
    import random
    sample = random.sample(chunks, sample_size)
    for chunk in sample:
        vec = embed_batch([chunk['text']])[0]
        results = index.query(vector=vec, top_k=1, include_metadata=True)
        top_id = results.matches[0].id if results.matches else None
        if top_id != chunk['id']:
            print(f'WARNING: expected {chunk["id"]}, got {top_id}')

Alternatives locales pour les représentations vectorielles

Pour les données sensibles du point de vue de la confidentialité qui ne peuvent pas quitter votre infrastructure, utilisez des modèles de représentation vectorielle hébergés localement. La bibliothèque sentence-transformers fournit des modèles de haute qualité tels que all-MiniLM-L6-v2 (384 dimensions, 22 Mo, très rapide) et bge-large-en-v1.5 (1024 dimensions, meilleure qualité). Exécutez-les sur le CPU pour les charges de travail modérées ou sur le GPU pour les tâches d'indexation volumineuses. Les modèles locaux éliminent les coûts d'API et l'exfiltration des données, mais nécessitent de gérer les fichiers de modèles et les ressources de calcul.

from sentence_transformers import SentenceTransformer

# Load once at startup
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

def embed_locally(texts, batch_size=64):
    # encode() handles batching internally
    embeddings = model.encode(
        texts,
        batch_size=batch_size,
        show_progress_bar=True,
        convert_to_numpy=True
    )
    return embeddings.tolist()  # convert numpy array to Python list

vectors = embed_locally([c['text'] for c in chunks])

Vérification rapide

Vérifiez votre compréhension des concepts d'ingénierie de l'IA présentés dans cette leçon.

Récapitulatif de la leçon

Dans cette leçon, vous avez appris à générer des représentations vectorielles par lots avec l'API OpenAI et à gérer les limites de débit avec une temporisation exponentielle, à insérer ou mettre à jour des vecteurs dans Pinecone et pgvector avec des métadonnées, à créer des index HNSW pour effectuer rapidement des recherches approximatives de plus proches voisins, ainsi que les bonnes pratiques de production, notamment la reprise à partir de points de contrôle, la conception du schéma de métadonnées et la vérification de l'exhaustivité de l'index. Nous allons maintenant créer le pipeline de requête qui récupère les segments et génère des réponses étayées.

Gratuit pour commencer

Apprends Python avec un tuteur IA — gratuit

Écris et exécute du vrai code dans ton navigateur, obtiens de l'aide instantanée d'un tuteur IA disponible 24h/24, et reprends là où tu t'es arrêté sur le web ou dans l'app.

Cours
30
Leçons
120

Questions Fréquemment Posées

La leçon « Indexer : créer et stocker les segments » est-elle gratuite ?

Oui — le texte complet de « Indexer : créer et stocker les segments » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Indexer : créer et stocker les segments » ?

Créez un embedding pour chaque segment à l’aide de l’API d’embeddings d’OpenAI, puis insérez les vecteurs obtenus avec leurs métadonnées dans une base vectorielle afin de construire un index consulta… Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.

Combien de temps prend la leçon « Indexer : créer et stocker les segments » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Chargement des documents et extraction de texte
  2. Stratégies de segmentation : fixe, par phrases ou récursive
  3. Indexer : créer et stocker les segments
  4. Interroger, récupérer et générer
← Retour à AI Engineering Academy