Guía: RAG con embeddings

La generación aumentada por recuperación (RAG) responde preguntas a partir de tus propios documentos. Divides los documentos en fragmentos, calculas el embedding de cada fragmento una sola vez y, al recibir una pregunta, calculas su embedding, buscas los fragmentos más cercanos, los reordenas si quieres y envías los mejores a un modelo de chat con la instrucción de citarlos.

Todo lo que sigue usa la URL base https://api14.com/api/v1, tu clave sk-api14-... y el paquete oficial openai. Los costes salen de tu saldo prepago en USD y se devuelven en usage.cost en cada respuesta.

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. Divide los documentos en fragmentos

Los fragmentos deben ser lo bastante pequeños para ser específicos y lo bastante grandes para tener sentido por sí solos. Un buen punto de partida son 500-1.000 caracteres con algo de solapamiento, divididos por límites de párrafo:

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:]  # lleva contexto al siguiente fragmento
        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 fuente (nombre de archivo, URL, número de página) con cada fragmento; la necesitas para las citas.

2. Calcula los embeddings con POST /embeddings

Envía muchos fragmentos por petición. Una petición con 100 entradas cuenta como una sola petición para tu límite de tasa:

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

Guarda los vectores junto con el id del modelo. Los vectores de modelos distintos no son comparables, así que si cambias EMBED_MODEL debes volver a calcular los embeddings del corpus. Consulta Embeddings.

3. Busca con similitud del coseno

Para unos pocos miles de fragmentos, un recorrido lineal en memoria es suficientemente rápido. Más allá, guarda los vectores en una base de datos vectorial o en una tabla de 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. Reordena con POST /rerank

La búsqueda por embeddings es buena en recall pero imprecisa al ordenar. Un reranker lee la pregunta y cada candidato juntos y puntúa la relevancia con más precisión. Recupera de 20 a 100 candidatos, reordénalos y quédate con los 3-8 mejores. El SDK de OpenAI no tiene un método de rerank, así que llámalo por HTTP (httpx se instala con 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 apunta al array documents que enviaste. Consulta Rerank.

5. Responde con citas

Numera los fragmentos en el prompt y pide al modelo que cite esos números. Indícale que lo diga cuando el contexto no contenga la respuesta:

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 tu aplicación necesita las citas como datos, usa Salidas estructuradas con un esquema como { "answer": string, "citations": integer[] } en lugar de analizar corchetes.

El mismo flujo 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;
}

Estimar el coste

Un sistema RAG tiene tres líneas de coste, todas en USD y todas reportadas en usage.cost:

Para estimar antes de ejecutar, toma los precios del modelo de GET /models/{id} (pricing.prompt y pricing.completion son USD por token) y supón aproximadamente 4 caracteres por token en texto en inglés:

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

La llamada de chat suele ser la línea más grande. Para reducirla, envía menos fragmentos y más cortos (el rerank lo hace seguro), mantén el prompt de sistema fijo al principio para que se aplique la caché de prompts y establece max_tokens. Comprueba después los costes reales con usage.cost o con Consulta de generaciones.