دليل: RAG مع embeddings

يجيب التوليد المعزَّز بالاسترجاع (RAG) عن الأسئلة من مستنداتك أنت. تقسّم المستندات إلى مقاطع، وتحوّل كل مقطع إلى متجه مرة واحدة، وعند السؤال تحوّل السؤال إلى متجه، وتجد أقرب المقاطع، وتعيد ترتيبها اختياريًا، ثم ترسل أفضلها إلى نموذج محادثة مع تعليمة بالاستشهاد بها.

كل ما يلي يستخدم Base URL https://api14.com/api/v1 ومفتاحك sk-api14-... وحزمة openai الرسمية. تُخصم التكاليف من رصيدك المدفوع مسبقًا بالدولار وتُعاد في 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:]  # 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)]

احتفظ بالمصدر (اسم الملف أو URL أو رقم الصفحة) مع كل مقطع، فستحتاجه في الاستشهادات.

2. التحويل إلى متجهات عبر POST /embeddings

أرسل مقاطع كثيرة في الطلب الواحد. طلب واحد بـ 100 مدخل يُحتسب طلبًا واحدًا ضمن حد الطلبات:

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

احفظ المتجهات ومعرّف النموذج معًا. المتجهات الصادرة من نماذج مختلفة غير قابلة للمقارنة، فإذا غيّرت EMBED_MODEL يجب أن تعيد توليد متجهات المجموعة كلها. راجع Embeddings.

3. البحث بتشابه جيب التمام

لبضعة آلاف من المقاطع يكفي مسح خطي في الذاكرة. وما بعد ذلك ضع المتجهات في قاعدة بيانات متجهات أو جدول Postgres مع 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. إعادة الترتيب عبر POST /rerank

البحث بالمتجهات جيد في الاسترجاع (recall) لكنه تقريبي في الترتيب. يقرأ نموذج إعادة الترتيب السؤال وكل مرشح معًا ويقيّم الصلة بدقة أكبر. اجلب من 20 إلى 100 مرشح، وأعد ترتيبها، واحتفظ بأفضل 3 إلى 8. ليس في OpenAI SDK دالة rerank، فاستدعِها عبر 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 التي أرسلتها. راجع إعادة الترتيب (Rerank).

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 ثلاثة بنود تكلفة، كلها بالدولار وكلها تُعاد في usage.cost:

للتقدير قبل التشغيل، خذ أسعار النموذج من GET /models/{id} (pricing.prompt وpricing.completion بالدولار لكل رمز) وافترض نحو 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 chunks + instructions, short answer
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

استدعاء المحادثة هو عادةً أكبر البنود. لخفضه، أرسل مقاطع أقل وأقصر (وإعادة الترتيب تجعل ذلك آمنًا)، وأبقِ مطالبة النظام ثابتة في البداية لينطبق التخزين المؤقت للمطالبات، واضبط max_tokens. تحقق من التكاليف الفعلية لاحقًا عبر usage.cost أو البحث عن عملية توليد.