ガイド: 埋め込みによる RAG

検索拡張生成(RAG)は、自分のドキュメントに基づいて質問に答えます。ドキュメントをチャンクに分割し、各チャンクを一度だけ埋め込みます。質問時には、質問を埋め込み、最も近いチャンクを見つけ、必要に応じてリランクし、最良のものを、引用するよう指示を付けてチャットモデルに送信します。

以下のすべてで、ベース URL https://api14.com/api/v1、sk-api14-... のキー、公式の openai パッケージを使います。コストはプリペイドの USD 残高から差し引かれ、すべてのレスポンスの usage.cost で返されます。

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. ドキュメントを分割する

チャンクは、内容が具体的になるほど小さく、単独で意味が通るほど大きくします。出発点としては、段落の境界で分割し、多少の重なりを持たせた 500〜1,000 文字が適切です。

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:]  # 次のチャンクにコンテキストを引き継ぐ
        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)]

すべてのチャンクに出典(ファイル名、URL、ページ番号)を持たせてください。引用に必要になります。

2. POST /embeddings で埋め込む

1 回のリクエストで多くのチャンクを送ります。100 個の入力を含む 1 回のリクエストは、レート制限上は 1 リクエストとして数えられます。

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

ベクトルとモデル ID を一緒に保存してください。異なるモデルのベクトルは比較できないため、EMBED_MODEL を変更したらコーパスを再度埋め込む必要があります。埋め込みを参照してください。

3. コサイン類似度で検索する

数千チャンク程度であれば、メモリ上の線形スキャンで十分高速です。それ以上になる場合は、ベクトルをベクトルデータベース、または pgvector を使った Postgres のテーブルに格納してください。

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. POST /rerank でリランクする

埋め込み検索は再現率に優れますが、順序付けは大まかです。リランカーは質問と各候補をあわせて読み、関連度をより正確に採点します。20〜100 件の候補を取得してリランクし、上位 3〜8 件を残します。OpenAI SDK にはリランクのメソッドがないため、HTTP で呼び出します(httpx は 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 は、送信した documents 配列内の位置を指します。リランクを参照してください。

5. 引用付きで回答する

プロンプト内でチャンクに番号を付け、モデルにその番号を引用するよう指示します。コンテキストに答えが含まれていない場合は、そう答えるよう伝えます。

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

アプリで引用をデータとして使う場合は、角括弧を解析する代わりに、{ "answer": string, "citations": integer[] } のようなスキーマで構造化出力を使ってください。

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

コストの見積もり

RAG システムには 3 つのコスト項目があり、いずれも USD で、usage.cost に報告されます。

実行前に見積もるには、GET /models/{id} からモデルの価格を取得し(pricing.prompt と pricing.completion はトークンあたりの USD)、英語のテキストでは 1 トークンあたり約 4 文字と仮定します。

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 チャンク + 指示、短い回答
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

通常、最大の項目はチャットの呼び出しです。これを下げるには、より少なく短いチャンクを送り(リランクによって安全になります)、プロンプトキャッシュが適用されるようにシステムプロンプトを先頭で固定し、max_tokens を設定します。実際のコストは、その後 usage.cost または生成情報の取得で確認してください。