Guia: RAG com embeddings

A geração aumentada por recuperação (RAG) responde perguntas a partir dos seus próprios documentos. Você divide os documentos em trechos, gera o embedding de cada trecho uma vez e, na hora da pergunta, gera o embedding da pergunta, encontra os trechos mais próximos, opcionalmente os reordena e envia os melhores a um modelo de chat com a instrução de citá-los.

Tudo abaixo usa a base URL https://api14.com/api/v1, sua chave sk-api14-... e o pacote oficial openai. Os custos saem do seu saldo pré-pago em USD e são retornados em usage.cost em toda resposta.

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. Divida os documentos em trechos

Os trechos devem ser pequenos o bastante para serem específicos e grandes o bastante para fazer sentido sozinhos. Um bom ponto de partida é de 500 a 1,000 caracteres com alguma sobreposição, dividindo nos limites dos parágrafos:

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:]  # leva contexto para o próximo trecho
        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)]

Guarde a fonte (nome do arquivo, URL, número da página) junto com cada trecho: você precisa dela para as citações.

2. Gere embeddings com POST /embeddings

Envie vários trechos por requisição. Uma requisição com 100 entradas conta como uma única requisição no seu limite de taxa:

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

Guarde os vetores junto com o id do modelo. Vetores de modelos diferentes não são comparáveis, então, se você mudar EMBED_MODEL, precisa gerar os embeddings do corpus de novo. Veja Embeddings.

3. Busque por similaridade de cosseno

Para alguns milhares de trechos, uma varredura linear em memória é rápida o suficiente. Acima disso, coloque os vetores em um banco de dados vetorial ou em uma tabela Postgres com 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. Reordene com POST /rerank

A busca por embeddings é boa em recall, mas imprecisa na ordenação. Um reranker lê a pergunta e cada candidato juntos e pontua a relevância com mais precisão. Recupere de 20 a 100 candidatos, reordene e fique com os 3 a 8 melhores. O SDK da OpenAI não tem um método de rerank, então chame o endpoint por HTTP (httpx é instalado junto com 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 aponta para uma posição no array documents que você enviou. Veja Rerank.

5. Responda com citações

Numere os trechos no prompt e peça ao modelo que cite esses números. Diga a ele que avise quando o contexto não contiver a resposta:

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 o seu app precisa das citações como dados, use Saídas estruturadas com um schema como { "answer": string, "citations": integer[] } em vez de interpretar colchetes.

O mesmo pipeline em 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;
}

Como estimar o custo

Um sistema RAG tem três linhas de custo, todas em USD e todas informadas em usage.cost:

Para estimar antes de rodar, pegue os preços do modelo em GET /models/{id} (pricing.prompt e pricing.completion são USD por token) e considere cerca de 4 caracteres por token em textos em 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 trechos + instruções, resposta curta
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

A chamada de chat costuma ser a maior linha. Para reduzi-la, envie menos trechos e mais curtos (o rerank torna isso seguro), mantenha o prompt de sistema fixo no início para que o prompt caching possa ser aplicado e defina max_tokens. Confira os custos reais depois com usage.cost ou com a consulta de geração.