Anleitung: RAG mit Embeddings

Retrieval-Augmented Generation (RAG) beantwortet Fragen aus Ihren eigenen Dokumenten. Sie zerlegen die Dokumente in Chunks, betten jeden Chunk einmal ein, und zur Fragezeit betten Sie die Frage ein, finden die ähnlichsten Chunks, ranken sie optional neu und senden die besten an ein Chat-Modell mit der Anweisung, sie zu zitieren.

Alles Folgende nutzt die Base URL https://api14.com/api/v1, Ihren sk-api14-...-Schlüssel und das offizielle Paket openai. Die Kosten stammen aus Ihrem vorausbezahlten USD-Guthaben und werden bei jeder Antwort in usage.cost zurückgegeben.

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. Dokumente in Chunks zerlegen

Chunks sollten klein genug sein, um spezifisch zu sein, und groß genug, um für sich Sinn zu ergeben. Ein guter Startpunkt sind 500 bis 1.000 Zeichen mit etwas Überlappung, getrennt an Absatzgrenzen:

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:]  # Kontext in den nächsten Chunk übernehmen
        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)]

Behalten Sie die Quelle (Dateiname, URL, Seitenzahl) bei jedem Chunk, Sie brauchen sie für die Quellenangaben.

2. Einbetten mit POST /embeddings

Senden Sie viele Chunks pro Anfrage. Eine Anfrage mit 100 Eingaben zählt als eine Anfrage für Ihr 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

Speichern Sie Vektoren und Modell-ID zusammen. Vektoren verschiedener Modelle sind nicht vergleichbar; wenn Sie EMBED_MODEL ändern, müssen Sie den Korpus neu einbetten. Siehe Embeddings.

3. Suchen mit Kosinus-Ähnlichkeit

Für einige tausend Chunks genügt ein linearer Scan im Speicher. Darüber hinaus legen Sie die Vektoren in eine Vektordatenbank oder eine Postgres-Tabelle mit 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. Neu ranken mit POST /rerank

Die Embedding-Suche ist gut im Recall, aber grob in der Reihenfolge. Ein Reranker liest Frage und jeden Kandidaten gemeinsam und bewertet die Relevanz genauer. Holen Sie 20 bis 100 Kandidaten, ranken Sie neu und behalten Sie die besten 3 bis 8. Das OpenAI-SDK hat keine Rerank-Methode, rufen Sie den Endpunkt daher per HTTP auf (httpx wird mit openai installiert):

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 verweist auf das von Ihnen gesendete documents-Array. Siehe Rerank.

5. Antworten mit Quellenangaben

Nummerieren Sie die Chunks im Prompt und bitten Sie das Modell, diese Nummern zu zitieren. Weisen Sie es an, dies zu sagen, wenn der Kontext die Antwort nicht enthält:

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

Wenn Ihre App die Quellenangaben als Daten braucht, verwenden Sie Strukturierte Ausgaben mit einem Schema wie { "answer": string, "citations": integer[] }, statt Klammern zu parsen.

Dieselbe 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 abschätzen

Ein RAG-System hat drei Kostenposten, alle in USD und alle in usage.cost ausgewiesen:

Für eine Schätzung vor dem Lauf nehmen Sie die Preise des Modells aus GET /models/{id} (pricing.prompt und pricing.completion sind USD pro Token) und rechnen bei englischem Text mit etwa 4 Zeichen pro Token:

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 + Anweisungen, kurze Antwort
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

Der Chat-Aufruf ist meist der größte Posten. Um ihn zu senken, senden Sie weniger und kürzere Chunks (Reranking macht das sicher), halten Sie den System-Prompt fest am Anfang, damit Prompt Caching greifen kann, und setzen Sie max_tokens. Prüfen Sie die tatsächlichen Kosten anschließend mit usage.cost oder Generation abrufen.