Index a document corpus
On this page
Objectives
- Build a complete semantic search index
- Handle different document formats (text, PDF, Markdown)
- Update the index incrementally
Architecture of a semantic index
A semantic index relies on three elements that must stay in sync: the original documents stored with their metadata, the embeddings computed from their text, and the mapping that links each vector to its source document. This third piece is the most fragile—it’s almost always where indexes get corrupted, when a deletion shifts indices on one side but not the other.
The Document class below adds an MD5 hash calculated automatically at creation. We’ll use it later to detect what has already been indexed.
from dataclasses import dataclass, field
from openai import OpenAI
import numpy as np
import json
import hashlib
@dataclass
class Document:
id: str
texte: str
metadata: dict = field(default_factory=dict)
hash: str = ""
def __post_init__(self):
self.hash = hashlib.md5(self.texte.encode()).hexdigest()Indexing pipeline
Step 1: Load documents
Loading depends on the format, and the format determines the granularity. A plain text file becomes a single document; a Markdown file, however, often contains several distinct topics, and indexing it as one block would drown each topic in the average of the others. The second loader therefore splits on level-2 headings and reinjects the section title into the indexed text—this is the enrichment technique covered in lesson 3.
from pathlib import Path
def charger_fichiers_texte(dossier: str) -> list[Document]:
"""Load all text files from a folder."""
documents = []
for fichier in Path(dossier).rglob("*.txt"):
contenu = fichier.read_text(encoding="utf-8").strip()
if contenu:
documents.append(Document(
id=fichier.stem,
texte=contenu,
metadata={
"fichier": str(fichier),
"type": "txt",
"taille": len(contenu)
}
))
return documents
def charger_fichiers_markdown(dossier: str) -> list[Document]:
"""Load Markdown files, splitting by sections."""
documents = []
for fichier in Path(dossier).rglob("*.md"):
contenu = fichier.read_text(encoding="utf-8")
sections = contenu.split("\n## ")
for i, section in enumerate(sections):
if not section.strip():
continue
titre = section.split("\n")[0].strip("# ")
corps = "\n".join(section.split("\n")[1:]).strip()
if corps:
documents.append(Document(
id=f"{fichier.stem}_section_{i}",
texte=f"{titre}\n\n{corps}",
metadata={
"fichier": str(fichier),
"section": titre,
"type": "markdown"
}
))
return documentsStep 2: Build the index
The IndexSemantique class brings together batch construction, threshold-based search, and disk persistence. The seuil (threshold) parameter in the rechercher method deserves attention: without it, an off-topic query still returns the five least-bad documents, with scores around 0.15 that a naive caller will treat as valid answers. Filtering out anything below a floor score lets you honestly respond, “I found nothing.”
class IndexSemantique:
def __init__(self, model: str = "text-embedding-3-large"):
self.client = OpenAI()
self.model = model
self.documents: list[Document] = []
self.embeddings: np.ndarray | None = None
def construire(self, documents: list[Document], batch_size: int = 100):
"""Build the index from a list of documents."""
self.documents = documents
textes = [d.texte for d in documents]
tous_embeddings = []
for i in range(0, len(textes), batch_size):
batch = textes[i:i + batch_size]
response = self.client.embeddings.create(
input=batch,
model=self.model
)
batch_sorted = sorted(response.data, key=lambda x: x.index)
tous_embeddings.extend([e.embedding for e in batch_sorted])
print(f" Batch {i // batch_size + 1} processed "
f"({min(i + batch_size, len(textes))}/{len(textes)})")
self.embeddings = np.array(tous_embeddings, dtype=np.float32)
print(f"Index built: {len(self.documents)} documents, "
f"{self.embeddings.shape[1]} dimensions")
def rechercher(self, requete: str, k: int = 5,
seuil: float = 0.0) -> list[dict]:
"""Search for the k most relevant documents."""
query_emb = self.client.embeddings.create(
input=requete,
model=self.model
).data[0].embedding
scores = self.embeddings @ np.array(query_emb)
top_indices = np.argsort(scores)[-k:][::-1]
resultats = []
for idx in top_indices:
score = float(scores[idx])
if score >= seuil:
doc = self.documents[idx]
resultats.append({
"id": doc.id,
"texte": doc.texte[:200],
"score": score,
"metadata": doc.metadata
})
return resultats
def sauvegarder(self, chemin: str):
"""Save the index to disk."""
np.save(f"{chemin}_embeddings.npy", self.embeddings)
docs_serialisables = [
{"id": d.id, "texte": d.texte,
"metadata": d.metadata, "hash": d.hash}
for d in self.documents
]
with open(f"{chemin}_documents.json", "w") as f:
json.dump(docs_serialisables, f, ensure_ascii=False)
def charger(self, chemin: str):
"""Load an index from disk."""
self.embeddings = np.load(f"{chemin}_embeddings.npy")
with open(f"{chemin}_documents.json") as f:
docs = json.load(f)
self.documents = [
Document(id=d["id"], texte=d["texte"],
metadata=d["metadata"])
for d in docs
]The sauvegarder and charger methods write two files sharing the same prefix. Treat them as an inseparable pair: copying one without the other, or restoring a partial backup, produces an index where the vectors no longer match the documents—a silent failure that will only show up as absurd results.
Incremental updates
A corpus evolves: three documents modified on Monday, ten added on Wednesday. Recomputing fifteen thousand vectors for every change wastes time and money. This is where the MD5 hash comes in: it identifies content, not the file, so a renamed but unchanged document won’t be reindexed, while a modified document will produce a different hash.
def mettre_a_jour(self, nouveaux_docs: list[Document]):
"""Add documents to the existing index."""
# Filter out already indexed documents
hashes_existants = {d.hash for d in self.documents}
docs_a_ajouter = [
d for d in nouveaux_docs if d.hash not in hashes_existants
]
if not docs_a_ajouter:
print("No new documents to index")
return
textes = [d.texte for d in docs_a_ajouter]
response = self.client.embeddings.create(
input=textes,
model=self.model
)
nouveaux_vecteurs = np.array(
[e.embedding for e in sorted(response.data, key=lambda x: x.index)],
dtype=np.float32
)
self.documents.extend(docs_a_ajouter)
self.embeddings = np.vstack([self.embeddings, nouveaux_vecteurs])
print(f"{len(docs_a_ajouter)} documents added. "
f"Total: {len(self.documents)}")The np.vstack stacks the new vectors below the old ones in the same order as the extend on the document list: this symmetry preserves the mapping. Note, however, the limitation of this implementation—it adds, but does not delete or replace. A modified document therefore ends up present twice, in its old and new versions. Handling deletions requires removing the corresponding row from the matrix, an expensive operation in NumPy, and this is one of the reasons that push teams toward a proper vector database.
End-to-end usage
End to end, building an index over documentation takes just a few lines.
# Create and populate the index
index = IndexSemantique()
docs = charger_fichiers_markdown("./documentation/")
print(f"{len(docs)} sections loaded")
index.construire(docs)
index.sauvegarder("mon_index")
# Search
resultats = index.rechercher("how to configure authentication", k=3)
for r in resultats:
print(f"[{r['score']:.3f}] {r['id']}")
print(f" {r['texte'][:100]}...")Run it on your own documentation and ask three questions whose answers you know. If the right document doesn’t rank first, the culprit is almost always the chunking—sections too long or titles not reinjected—rather than the embedding model.
Summary
- A semantic index links documents, embeddings, and metadata
- Load documents from different formats (text, Markdown, etc.)
- Build the index in batches to handle large corpora
- Save and update incrementally to avoid recalculating everything