Generate embeddings for your data — Embeddings, RAG and Semantic Search

Objectives

Prepare your text data for embedding
Generate embeddings efficiently at scale
Store vectors for later use

Prepare your data

Your embedding quality depends directly on your data quality. Text riddled with stray spaces, control characters from a misconfigured export, or erratic line breaks produces a slightly off vector. That offset compounds on every search. Cleaning isn’t cosmetic: it’s the first and cheapest relevance optimization.

import re

def nettoyer_texte(texte: str) -> str:
    """Prepares text for embedding."""
    # Remove multiple spaces
    texte = re.sub(r"\s+", " ", texte)
    # Remove control characters
    texte = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", texte)
    # Trim
    texte = texte.strip()
    return texte

# Example
brut = "  Texte   avec   des   espaces\n\n\nen trop  "
propre = nettoyer_texte(brut)
print(propre)  # "Texte avec des espaces en trop"

Enrich context

An isolated documentation excerpt loses part of its meaning without its title. Take a paragraph that starts with “Download the package from the official site”: out of context, nothing indicates it refers to PostgreSQL, and the resulting vector will never match a query about PostgreSQL. By prefixing the content with its title and section, you reinject that information into the text the model sees.

def enrichir_document(titre: str, section: str, contenu: str) -> str:
    """Creates enriched text for better embedding."""
    return f"Titre : {titre}\nSection : {section}\n\n{contenu}"

texte_enrichi = enrichir_document(
    titre="Installation de PostgreSQL",
    section="Getting Started Guide",
    contenu="Download the package from the official site..."
)

Generate embeddings in batches

On a real corpus of hundreds or thousands of documents, processing must be chunked and resilient to network hiccups. The function below iterates through texts in slices, systematically sorts the response by index to ensure vector #7 matches document #7, and retries once after a pause if an error occurs.

from openai import OpenAI
import time
import json

client = OpenAI()

def generer_embeddings(
    textes: list[str],
    model: str = "text-embedding-3-large",
    batch_size: int = 100
) -> list[list[float]]:
    """Generates embeddings in batches with error handling."""
    tous_les_embeddings = []

    for i in range(0, len(textes), batch_size):
        batch = textes[i:i + batch_size]
        print(f"Processing batch {i // batch_size + 1}/"
              f"{(len(textes) - 1) // batch_size + 1}")

        try:
            response = client.embeddings.create(
                input=batch,
                model=model
            )
            # Sort by index to guarantee order
            batch_embeddings = sorted(
                response.data, key=lambda x: x.index
            )
            tous_les_embeddings.extend(
                [item.embedding for item in batch_embeddings]
            )
        except Exception as e:
            print(f"Error on batch {i}: {e}")
            # Retry after a pause
            time.sleep(5)
            response = client.embeddings.create(
                input=batch,
                model=model
            )
            batch_embeddings = sorted(
                response.data, key=lambda x: x.index
            )
            tous_les_embeddings.extend(
                [item.embedding for item in batch_embeddings]
            )

    return tous_les_embeddings

Sorting by index is a precaution you must never skip. A silent one-row shift is enough to associate every document with its neighbor’s vector. The symptom you’ll observe is “search returns garbage,” with no technical error to point you in the right direction.

Store embeddings

For a small corpus of a few thousand documents, a simple JSON file containing texts, metadata, and vectors works perfectly. It has the advantage of being human-readable when you need to debug a strange result.

import json

documents = [
    {"id": 1, "texte": "Premier document...", "metadata": {"source": "faq"}},
    {"id": 2, "texte": "Deuxième document...", "metadata": {"source": "docs"}},
]

# Generate embeddings
textes = [d["texte"] for d in documents]
embeddings = generer_embeddings(textes)

# Add vectors to documents
for doc, emb in zip(documents, embeddings):
    doc["embedding"] = emb

# Save
with open("corpus_embeddings.json", "w") as f:
    json.dump(documents, f)

Beyond that scale, JSON becomes a bottleneck: every number is written out in full, bloating the file and imposing costly parsing on every load. NumPy stores the same values in binary on 4 bytes and reloads instantly.

import numpy as np

# Convert to NumPy matrix
matrice = np.array(embeddings, dtype=np.float32)
print(f"Shape : {matrice.shape}")  # (n_docs, 3072)

# Save in binary format
np.save("embeddings.npy", matrice)

# Load
matrice_chargee = np.load("embeddings.npy")

Complete pipeline: from text files to embeddings

Let’s bring these elements together in a function that reads a folder, cleans each file, generates vectors, and splits storage in two: metadata in JSON (readable and editable) and the matrix in NumPy (compact and fast).

from openai import OpenAI
from pathlib import Path
import json
import numpy as np

client = OpenAI()

def pipeline_embeddings(dossier: str, model: str = "text-embedding-3-large"):
    """Complete pipeline: files → embeddings → storage."""
    dossier_path = Path(dossier)

    # 1. Read files
    documents = []
    for fichier in sorted(dossier_path.glob("*.txt")):
        contenu = fichier.read_text(encoding="utf-8")
        contenu = nettoyer_texte(contenu)
        if contenu:
            documents.append({
                "id": fichier.stem,
                "fichier": fichier.name,
                "texte": contenu
            })

    print(f"{len(documents)} documents found")

    # 2. Generate embeddings
    textes = [d["texte"] for d in documents]
    embeddings = generer_embeddings(textes, model=model)

    # 3. Save
    # Metadata in JSON
    meta = [{"id": d["id"], "fichier": d["fichier"]} for d in documents]
    with open("metadata.json", "w") as f:
        json.dump(meta, f, ensure_ascii=False, indent=2)

    # Vectors in NumPy
    np.save("embeddings.npy", np.array(embeddings, dtype=np.float32))

    print(f"Embeddings saved: {len(embeddings)} vectors "
          f"of {len(embeddings[0])} dimensions")

    return documents, embeddings

# Usage
documents, embeddings = pipeline_embeddings("./mes_documents/")

Note the sorted() on the glob: without it, file order depends on the filesystem, and your two output files risk falling out of sync after a simple machine change.

Estimate costs

Before launching a massive job, count tokens locally with tiktoken. The operation takes a few seconds and saves you from discovering too late that a poorly filtered corpus contained ten times the expected volume.

import tiktoken

encoder = tiktoken.get_encoding("cl100k_base")

def estimer_cout(textes: list[str], model: str = "text-embedding-3-large"):
    """Estimates embedding cost for a list of texts."""
    total_tokens = sum(len(encoder.encode(t)) for t in textes)

    # Approximate prices per million tokens (check current rates)
    prix = {
        "text-embedding-3-large": 0.13,
        "text-embedding-3-small": 0.02,
    }

    cout = total_tokens / 1_000_000 * prix.get(model, 0.13)
    print(f"Estimated tokens: {total_tokens:,}")
    print(f"Estimated cost: ${cout:.4f}")
    return total_tokens, cout

The prices in the dictionary are indicative and age quickly: always check current rates before presenting a quote to a client.

Summary

Clean and enrich your texts before generating embeddings
Use batch processing for large corpora
Store metadata in JSON and vectors in NumPy
Estimate your costs before launching a massive job