Gids: RAG met embeddings

Retrieval-augmented generation (RAG) beantwoordt vragen op basis van je eigen documenten. Je splitst de documenten in chunks, embed elke chunk één keer, en op het moment van de vraag embed je de vraag, zoek je de dichtstbijzijnde chunks, rerank je ze eventueel en stuur je de beste naar een chatmodel met de opdracht ze te citeren.

Alles hieronder gebruikt de base URL https://api14.com/api/v1, je sk-api14-... key en het officiële pakket openai. Kosten komen uit je vooruitbetaalde USD-saldo en staan in usage.cost bij elke response.

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. Documenten in chunks splitsen

Chunks moeten klein genoeg zijn om specifiek te zijn en groot genoeg om op zichzelf begrijpelijk te zijn. Een goed startpunt is 500 tot 1.000 tekens met enige overlap, gesplitst op alinea-grenzen:

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:]  # neem context mee naar de volgende chunk
        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)]

Bewaar de bron (bestandsnaam, URL, paginanummer) bij elke chunk, je hebt die nodig voor bronvermelding.

2. Embedden met POST /embeddings

Stuur veel chunks per request. Eén request met 100 invoeren telt als één request voor je rate limit:

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

Bewaar de vectoren samen met het model-id. Vectoren van verschillende modellen zijn niet vergelijkbaar, dus als je EMBED_MODEL wijzigt, moet je het corpus opnieuw embedden. Zie Embeddings.

3. Zoeken met cosinusgelijkenis

Voor een paar duizend chunks is een lineaire scan in het geheugen snel genoeg. Daarboven zet je de vectoren in een vectordatabase of een Postgres-tabel met 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. Reranken met POST /rerank

Zoeken met embeddings is goed in recall maar grof in volgorde. Een reranker leest de vraag en elke kandidaat samen en scoort relevantie nauwkeuriger. Haal 20 tot 100 kandidaten op, rerank en houd de beste 3 tot 8. De OpenAI-SDK heeft geen rerank-methode, dus roep het aan via HTTP (httpx wordt met openai meegeïnstalleerd):

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 wijst naar de documents-array die je hebt verstuurd. Zie Rerank.

5. Antwoorden met bronvermelding

Nummer de chunks in de prompt en vraag het model die nummers te citeren. Zeg dat het dat moet melden wanneer de context het antwoord niet bevat:

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

Heeft je app de bronvermeldingen als data nodig, gebruik dan Structured outputs met een schema als { "answer": string, "citations": integer[] } in plaats van haakjes te parsen.

Dezelfde pipeline in 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;
}

Kosten schatten

Een RAG-systeem heeft drie kostenposten, alle in USD en alle vermeld in usage.cost:

Om vooraf te schatten, neem je de prijzen van het model uit GET /models/{id} (pricing.prompt en pricing.completion zijn USD per token) en ga je uit van ongeveer 4 tekens per token voor Engelse tekst:

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 chunks + instructies, kort antwoord
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

De chataanroep is meestal de grootste post. Om die te verlagen stuur je minder en kortere chunks (reranken maakt dat veilig), houd je de systeemprompt vast aan het begin zodat prompt caching kan werken, en stel je max_tokens in. Controleer de werkelijke kosten achteraf met usage.cost of Generatie opzoeken.