Guida: RAG con gli embeddings

La retrieval-augmented generation (RAG) risponde a domande a partire dai tuoi documenti. Dividi i documenti in chunk, calcoli una sola volta l'embedding di ogni chunk e, al momento della domanda, calcoli l'embedding della domanda, trovi i chunk più vicini, facoltativamente li riordini e invii i migliori a un modello di chat con l'istruzione di citarli.

Tutto ciò che segue usa il base URL https://api14.com/api/v1, la tua chiave sk-api14-... e il pacchetto ufficiale openai. I costi vengono dal tuo saldo prepagato in USD e sono restituiti in usage.cost in ogni risposta.

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. Dividere i documenti in chunk

I chunk devono essere abbastanza piccoli da essere specifici e abbastanza grandi da avere senso da soli. Un buon punto di partenza sono 500-1,000 caratteri con un po' di sovrapposizione, divisi ai confini dei paragrafi:

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:]  # porta il contesto nel chunk successivo
        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)]

Conserva la fonte (nome del file, URL, numero di pagina) insieme a ogni chunk: ti serve per le citazioni.

2. Calcolare gli embeddings con POST /embeddings

Invia molti chunk per richiesta. Una richiesta con 100 input conta come una sola richiesta rispetto al tuo limite di frequenza:

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

Salva insieme i vettori e l'id del modello. I vettori di modelli diversi non sono confrontabili, quindi se cambi EMBED_MODEL devi rigenerare gli embeddings dell'intero corpus. Vedi Embeddings.

3. Cercare con la similarità coseno

Per qualche migliaio di chunk, una scansione lineare in memoria è abbastanza veloce. Oltre, metti i vettori in un database vettoriale o in una tabella Postgres con 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. Riordinare con POST /rerank

La ricerca con gli embeddings è buona nel recupero ma approssimativa nell'ordinamento. Un reranker legge insieme la domanda e ogni candidato e assegna un punteggio di rilevanza più preciso. Recupera 20-100 candidati, riordinali e tieni i primi 3-8. L'SDK di OpenAI non ha un metodo rerank, quindi chiamalo via HTTP (httpx viene installato insieme a 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 punta all'interno dell'array documents che hai inviato. Vedi Rerank.

5. Rispondere con le citazioni

Numera i chunk nel prompt e chiedi al modello di citare quei numeri. Digli di dirlo quando il contesto non contiene la risposta:

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

Se la tua app ha bisogno delle citazioni come dati, usa gli Output strutturati con uno schema come { "answer": string, "citations": integer[] } invece di analizzare le parentesi.

La stessa 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;
}

Stimare i costi

Un sistema RAG ha tre voci di costo, tutte in USD e tutte riportate in usage.cost:

Per stimare prima di eseguire, prendi i prezzi del modello da GET /models/{id} (pricing.prompt e pricing.completion sono USD per token) e assumi circa 4 caratteri per token per il testo inglese:

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

La chiamata di chat è di solito la voce più grande. Per ridurla, invia meno chunk e più brevi (il rerank lo rende sicuro), mantieni il prompt di sistema fisso all'inizio in modo che si applichi il prompt caching, e imposta max_tokens. Verifica poi i costi reali con usage.cost o con la Ricerca di una generazione.