Przewodnik: RAG z embeddingami

Generowanie wspomagane wyszukiwaniem (RAG) odpowiada na pytania na podstawie Twoich własnych dokumentów. Dzielisz dokumenty na fragmenty, osadzasz każdy fragment jednorazowo, a w momencie pytania osadzasz pytanie, znajdujesz najbliższe fragmenty, opcjonalnie robisz ich rerank i wysyłasz najlepsze do modelu czatu z poleceniem, by je cytował.

Wszystko poniżej używa bazowego adresu URL https://api14.com/api/v1, Twojego klucza sk-api14-... i oficjalnego pakietu openai. Koszty pochodzą z Twojego przedpłaconego salda w USD i są zwracane w usage.cost w każdej odpowiedzi.

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. Podziel dokumenty na fragmenty

Fragmenty powinny być na tyle małe, by były konkretne, i na tyle duże, by miały sens same w sobie. Dobry punkt wyjścia to 500–1,000 znaków z pewnym nakładaniem się, dzielone na granicach akapitów:

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:]  # carry context into the next 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)]

Przechowuj źródło (nazwę pliku, adres URL, numer strony) przy każdym fragmencie, będzie potrzebne do cytowań.

2. Osadź przez POST /embeddings

Wysyłaj wiele fragmentów w jednym żądaniu. Jedno żądanie ze 100 wejściami liczy się jako jedno żądanie względem Twojego limitu zapytań:

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

Przechowuj wektory razem z identyfikatorem modelu. Wektorów z różnych modeli nie można porównywać, więc jeśli zmienisz EMBED_MODEL, musisz osadzić korpus od nowa. Zobacz Embeddingi.

3. Szukaj podobieństwem cosinusowym

Dla kilku tysięcy fragmentów wystarczająco szybki jest liniowy przegląd w pamięci. Powyżej tej liczby umieść wektory w bazie wektorowej lub tabeli Postgres z 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. Rerank przez POST /rerank

Wyszukiwanie po embeddingach dobrze znajduje kandydatów (recall), ale grubo ustala kolejność. Reranker czyta pytanie i każdego kandydata razem i dokładniej ocenia trafność. Pobierz 20–100 kandydatów, wykonaj rerank i zostaw 3–8 najlepszych. SDK OpenAI nie ma metody rerank, więc wywołaj ją przez HTTP (httpx jest instalowany razem z 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 wskazuje pozycję w wysłanej tablicy documents. Zobacz Rerank.

5. Odpowiadaj z cytowaniami

Ponumeruj fragmenty w prompcie i poproś model, by cytował te numery. Powiedz mu, żeby zaznaczył, gdy kontekst nie zawiera odpowiedzi:

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

Jeśli Twoja aplikacja potrzebuje cytowań jako danych, użyj Ustrukturyzowanego wyjścia ze schematem takim jak { "answer": string, "citations": integer[] } zamiast parsować nawiasy.

Ten sam potok w 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;
}

Szacowanie kosztu

System RAG ma trzy pozycje kosztów, wszystkie w USD i wszystkie raportowane w usage.cost:

Aby oszacować koszt przed uruchomieniem, pobierz ceny modelu z GET /models/{id} (pricing.prompt i pricing.completion to USD za token) i przyjmij około 4 znaków na token dla tekstu angielskiego:

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

Największą pozycją jest zwykle wywołanie czatu. Aby ją obniżyć, wysyłaj mniej i krótsze fragmenty (rerank sprawia, że jest to bezpieczne), trzymaj prompt systemowy stały na początku, aby zadziałał cache promptów, i ustaw max_tokens. Rzeczywiste koszty sprawdzisz później w usage.cost lub przez Wyszukiwanie generacji.