Guide : RAG avec embeddings

La génération augmentée par récupération (RAG) répond à des questions à partir de vos propres documents. Vous découpez les documents en fragments, vectorisez chaque fragment une fois, et au moment de la question vous vectorisez la question, trouvez les fragments les plus proches, les reclassez éventuellement, et envoyez les meilleurs à un modèle de chat avec la consigne de les citer.

Tout ce qui suit utilise l'URL de base https://api14.com/api/v1, votre clé sk-api14-... et le paquet officiel openai. Les coûts sont prélevés sur votre solde prépayé en USD et renvoyés dans usage.cost à chaque réponse.

import os
from openai import OpenAI

client = OpenAI(base_url="https://api14.com/api/v1", api_key=os.environ["API14_API_KEY"])
EMBED_MODEL = "openai/text-embedding-3-small"
CHAT_MODEL = "openai/gpt-5-mini"

1. Découper les documents

Les fragments doivent être assez petits pour être précis et assez grands pour avoir du sens seuls. Un bon point de départ : 500 à 1 000 caractères avec un peu de chevauchement, découpés aux limites de paragraphes :

def chunk(text, size=800, overlap=150):
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, current = [], ""
    for p in paragraphs:
        if current and len(current) + len(p) > size:
            chunks.append(current)
            current = current[-overlap:]  # reporte du contexte dans le fragment suivant
        current = f"{current}\n\n{p}".strip()
    if current:
        chunks.append(current)
    return chunks

docs = {"handbook.md": open("handbook.md").read(), "faq.md": open("faq.md").read()}
chunks = [{"source": name, "text": c} for name, text in docs.items() for c in chunk(text)]

Conservez la source (nom de fichier, URL, numéro de page) avec chaque fragment : vous en avez besoin pour les citations.

2. Vectoriser avec POST /embeddings

Envoyez beaucoup de fragments par requête. Une requête de 100 entrées compte pour une seule requête dans votre limite de débit :

def embed(texts):
    vectors = []
    for i in range(0, len(texts), 100):
        r = client.embeddings.create(model=EMBED_MODEL, input=texts[i:i + 100])
        vectors += [d.embedding for d in sorted(r.data, key=lambda d: d.index)]
    return vectors

for c, v in zip(chunks, embed([c["text"] for c in chunks])):
    c["vector"] = v

Stockez les vecteurs avec l'identifiant du modèle. Les vecteurs de modèles différents ne sont pas comparables : si vous changez EMBED_MODEL, vous devez revectoriser le corpus. Voir Embeddings.

3. Chercher par similarité cosinus

Pour quelques milliers de fragments, un balayage linéaire en mémoire suffit. Au-delà, placez les vecteurs dans une base de données vectorielle ou une table Postgres avec pgvector.

import math

def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))

def search(question, k=20):
    q = embed([question])[0]
    return sorted(chunks, key=lambda c: cosine(q, c["vector"]), reverse=True)[:k]

4. Reclasser avec POST /rerank

La recherche par embeddings est bonne pour le rappel mais approximative pour l'ordre. Un reranker lit la question et chaque candidat ensemble et note la pertinence plus précisément. Récupérez 20 à 100 candidats, reclassez, et gardez les 3 à 8 meilleurs. Le SDK OpenAI n'a pas de méthode rerank : appelez-le donc en HTTP (httpx est installé avec openai) :

import httpx

def rerank(question, candidates, top_n=5):
    r = httpx.post(
        "https://api14.com/api/v1/rerank",
        headers={"Authorization": f"Bearer {os.environ['API14_API_KEY']}"},
        json={
            "model": "cohere/rerank-v3.5",
            "query": question,
            "documents": [c["text"] for c in candidates],
            "top_n": top_n,
        },
        timeout=30,
    )
    r.raise_for_status()
    return [candidates[x["index"]] for x in r.json()["results"]]

index renvoie à la position dans le tableau documents que vous avez envoyé. Voir Rerank.

5. Répondre avec des citations

Numérotez les fragments dans le prompt et demandez au modèle de citer ces numéros. Dites-lui de l'indiquer quand le contexte ne contient pas la réponse :

def answer(question):
    context = rerank(question, search(question))
    sources = "\n\n".join(f"[{i + 1}] ({c['source']})\n{c['text']}" for i, c in enumerate(context))
    r = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": (
                "Answer only from the numbered sources. Cite them like [1] or [2][3] after each claim. "
                "If the sources do not contain the answer, say you don't know."
            )},
            {"role": "user", "content": f"Sources:\n\n{sources}\n\nQuestion: {question}"},
        ],
    )
    return r.choices[0].message.content, context, r.usage.cost

Si votre application a besoin des citations sous forme de données, utilisez les Sorties structurées avec un schéma comme { "answer": string, "citations": integer[] } plutôt que d'analyser des crochets.

Le même pipeline en Node

import OpenAI from "openai";

const client = new OpenAI({ baseURL: "https://api14.com/api/v1", apiKey: process.env.API14_API_KEY });

const embed = async (input) =>
  (await client.embeddings.create({ model: "openai/text-embedding-3-small", input })).data.map((d) => d.embedding);

const cosine = (a, b) => {
  let dot = 0, na = 0, nb = 0;
  for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; na += a[i] ** 2; nb += b[i] ** 2; }
  return dot / Math.sqrt(na * nb);
};

async function answer(question, chunks /* [{ source, text, vector }] */) {
  const [q] = await embed([question]);
  const candidates = chunks.map((c) => ({ ...c, score: cosine(q, c.vector) }))
    .sort((a, b) => b.score - a.score).slice(0, 20);

  const ranked = await client.post("/rerank", {
    body: { model: "cohere/rerank-v3.5", query: question, documents: candidates.map((c) => c.text), top_n: 5 },
  });
  const context = ranked.results.map((r) => candidates[r.index]);

  const sources = context.map((c, i) => `[${i + 1}] (${c.source})\n${c.text}`).join("\n\n");
  const res = await client.chat.completions.create({
    model: "openai/gpt-5-mini",
    messages: [
      { role: "system", content: "Answer only from the numbered sources and cite them like [1]. If they do not contain the answer, say so." },
      { role: "user", content: `Sources:\n\n${sources}\n\nQuestion: ${question}` },
    ],
  });
  return res.choices[0].message.content;
}

Estimer le coût

Un système RAG a trois postes de coût, tous en USD et tous rapportés dans usage.cost :

Pour estimer avant de lancer, prenez les prix du modèle dans GET /models/{id} (pricing.prompt et pricing.completion sont en USD par token) et comptez environ 4 caractères par token pour du texte anglais :

import httpx

def price(model_id):
    return httpx.get(f"https://api14.com/api/v1/models/{model_id}").json()["pricing"]

corpus_tokens = sum(len(c["text"]) for c in chunks) / 4
print("One-off embedding:", corpus_tokens * float(price(EMBED_MODEL)["prompt"]), "USD")

p = price(CHAT_MODEL)
prompt_tokens, answer_tokens = 5 * 800 / 4 + 200, 300  # 5 fragments + instructions, réponse courte
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

L'appel de chat est généralement le poste le plus important. Pour le réduire, envoyez moins de fragments et plus courts (le reclassement le rend sûr), gardez le prompt système fixe au début pour que le cache de prompt puisse s'appliquer, et définissez max_tokens. Vérifiez ensuite les coûts réels avec usage.cost ou la consultation d'une génération.