المشروع النهائي: ابنِ نظام RAG خاصاً بمستنداتك

النتيجة: بنهاية هذا الدرس سيكون لديك خدمة RAG منشورة — تطبيق FastAPI يستوعب مستنداتك الشخصية (PDF، markdown، نصوص)، يسترجع المقاطع ذات الصلة عبر البحث الهجين مع RRF، ويجيب على الأسئلة مع استشهادات مضمّنة تعيدك إلى ملفات المصدر.

هذا المشروع النهائي يجمع كل ما تعلمته عبر الوحدات الست: استراتيجية التضمين (الوحدة 2)، التقطيع (الوحدة 3)، البحث الهجين + إعادة الترتيب (الوحدة 4)، التقييم (الوحدة 5)، وتقوية الإنتاج (الوحدة 6).

ما ستشحنه فعلياً:

$ curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"q": "ماذا قلت عن خارطة طريق Q3 في ملاحظات اجتماعي؟"}'

{
  "answer": "وفقاً لملاحظاتك من 2026-03-14، أولويات خارطة طريق Q3 هي: (1) شحن تكامل MCP [cite:1]، (2) إطلاق الموقع العربي [cite:2]، و(3) تقليص تكاليف البنية التحتية بنسبة 30% [cite:1].",
  "citations": [...]
}

---

الجزء 0 — اختر مجموعة مستنداتك (5 دقائق)

اختر شيئاً تهتم به فعلاً:

• تصدير Notion/Obsidian
• ويكي شركتك الداخلي (تصدير markdown)
• مجلد ملاحظات اجتماعات
• حديقة معرفة شخصية
• أرشيف مدونتك

استهدف 20–200 مستنداً في المرة الأولى. قليل جداً ولن يكون RAG ممتعاً؛ كثير جداً وسيصبح أول استيعاب طويلاً للتصحيح.

---

الجزء 1 — إعداد Supabase مع pgvector (10 دقائق)

1.1 إنشاء مشروع

1. اذهب إلى https://supabase.com، سجل (الطبقة المجانية كافية لحوالي 50 ميغا من المحتوى النصي + التضمينات)
2. أنشئ مشروعاً جديداً. احفظ Project URL وservice role key للخطوة التالية.

1.2 تطبيق المخطط

في Supabase ← SQL Editor، الصق وشغّل هذا. ينشئ جدول chunks واحداً بعمود vector (للبحث الدلالي) وعمود tsvector (للبحث بالكلمات المفتاحية) — نصفا الاسترجاع الهجين الذي تعلمته في الوحدة 4.

-- schema.sql
create extension if not exists vector;

create table chunks (
  id uuid primary key default gen_random_uuid(),
  source_path text not null,
  chunk_index int not null,
  content text not null,
  embedding vector(3072) not null,     -- text-embedding-3-large
  tsv tsvector generated always as (to_tsvector('english', content)) stored,
  created_at timestamptz default now()
);

create index on chunks using hnsw (embedding vector_cosine_ops);
create index on chunks using gin (tsv);
create index on chunks (source_path);

create or replace function hybrid_search(
  query_embedding vector(3072),
  query_text text,
  match_count int default 10,
  rrf_k int default 60
)
returns table (id uuid, source_path text, content text, score float)
language sql stable as $$
  with dense as (
    select id, row_number() over (order by embedding <=> query_embedding) as rank
    from chunks order by embedding <=> query_embedding limit match_count * 3
  ),
  sparse as (
    select id, row_number() over (order by ts_rank(tsv, plainto_tsquery('english', query_text)) desc) as rank
    from chunks where tsv @@ plainto_tsquery('english', query_text) limit match_count * 3
  ),
  fused as (
    select coalesce(d.id, s.id) as id,
           coalesce(1.0 / (rrf_k + d.rank), 0) + coalesce(1.0 / (rrf_k + s.rank), 0) as rrf_score
    from dense d full outer join sparse s on d.id = s.id
  )
  select c.id, c.source_path, c.content, f.rrf_score as score
  from fused f join chunks c on c.id = f.id
  order by f.rrf_score desc limit match_count;
$$;

لماذا هذا المخطط: HNSW للبحث التقريبي السريع عن الأقرب، GIN لمطابقة الكلمات المفتاحية بأسلوب BM25، RRF لدمج الترتيبين في نتيجة واحدة. بالضبط النمط من الوحدة 4 الدرس 2.

---

الجزء 2 — خط الاستيعاب (15 دقيقة)

هيكل المشروع:

my-rag/
├── schema.sql
├── ingest.py
├── retriever.py
├── generator.py
├── main.py
├── requirements.txt
└── .env.example

2.1 requirements.txt

anthropic==0.42.0
openai==1.58.0
fastapi==0.115.6
uvicorn[standard]==0.34.0
supabase==2.10.0
python-dotenv==1.0.1
pypdf==5.1.0
markdown-it-py==3.0.0
tiktoken==0.8.0

2.2 ingest.py — تحميل، تقطيع، تضمين، إدراج

import os, glob
from pathlib import Path
from openai import OpenAI
from supabase import create_client
from pypdf import PdfReader
from dotenv import load_dotenv
import tiktoken

load_dotenv()

openai = OpenAI()
supabase = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_SERVICE_KEY"])
enc = tiktoken.get_encoding("cl100k_base")

CHUNK_TOKENS = 500
CHUNK_OVERLAP = 75


def read_document(path: Path) -> str:
    if path.suffix.lower() == ".pdf":
        return "\n\n".join(p.extract_text() or "" for p in PdfReader(path).pages)
    return path.read_text(encoding="utf-8", errors="ignore")


def chunk(text: str, chunk_tokens: int = CHUNK_TOKENS, overlap: int = CHUNK_OVERLAP) -> list[str]:
    """نافذة منزلقة بالرموز — أبسط من الدلالية لكن قوية."""
    tokens = enc.encode(text)
    out, i = [], 0
    while i < len(tokens):
        out.append(enc.decode(tokens[i : i + chunk_tokens]))
        i += chunk_tokens - overlap
    return out


def embed_batch(texts: list[str]) -> list[list[float]]:
    """text-embedding-3-large يُرجع متجهات بـ 3072 بُعداً."""
    resp = openai.embeddings.create(model="text-embedding-3-large", input=texts)
    return [d.embedding for d in resp.data]


def ingest(docs_dir: str):
    paths = [
        Path(p) for p in glob.glob(f"{docs_dir}/**/*", recursive=True)
        if Path(p).is_file() and Path(p).suffix.lower() in {".md", ".txt", ".pdf"}
    ]
    print(f"Found {len(paths)} documents")

    for path in paths:
        text = read_document(path)
        chunks = chunk(text)
        if not chunks:
            continue

        for batch_start in range(0, len(chunks), 100):
            batch = chunks[batch_start : batch_start + 100]
            vectors = embed_batch(batch)
            rows = [{
                "source_path": str(path.relative_to(docs_dir)),
                "chunk_index": batch_start + i,
                "content": content,
                "embedding": vec,
            } for i, (content, vec) in enumerate(zip(batch, vectors))]
            supabase.table("chunks").insert(rows).execute()

        print(f"  ✓ {path.name}: {len(chunks)} chunks")


if __name__ == "__main__":
    import sys
    ingest(sys.argv[1] if len(sys.argv) > 1 else "./docs")

شغّله:

pip install -r requirements.txt
python ingest.py ./path/to/your/docs

---

الجزء 3 — الاسترجاع مع إعادة الترتيب (15 دقيقة)

3.1 retriever.py

import os, json, re
from anthropic import Anthropic
from openai import OpenAI
from supabase import create_client
from dotenv import load_dotenv

load_dotenv()

openai = OpenAI()
anthropic = Anthropic()
supabase = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_SERVICE_KEY"])


def retrieve(query: str, top_k: int = 5) -> list[dict]:
    query_vec = openai.embeddings.create(
        model="text-embedding-3-large", input=query,
    ).data[0].embedding

    result = supabase.rpc("hybrid_search", {
        "query_embedding": query_vec,
        "query_text": query,
        "match_count": 30,
    }).execute()

    candidates = result.data or []
    if not candidates:
        return []

    prompt = f"""Score each passage 0-10 for how well it answers the query.
Return JSON only: {{"scores": [int, ...]}} in the same order.

QUERY: {query}

PASSAGES:
""" + "\n\n".join(f"[{i}] {c['content'][:500]}" for i, c in enumerate(candidates))

    reply = anthropic.messages.create(
        model="claude-haiku-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}],
    )

    try:
        scores = json.loads(re.search(r"\{.*\}", reply.content[0].text, re.DOTALL).group(0))["scores"]
    except Exception:
        scores = [c.get("score", 0) for c in candidates]

    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:top_k]
    return [{**c, "rank": s} for c, s in ranked]

---

الجزء 4 — التوليد باستشهادات مضمّنة (10 دقائق)

4.1 generator.py

import os, re
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()
client = Anthropic()


SYSTEM = """أنت مساعد استرجاع معزز. أجب فقط باستخدام مقاطع السياق المقدمة.

القواعد:
1. كل عبارة واقعية يجب أن تستشهد بمقطع باستخدام [cite:N] حيث N رقم المقطع.
2. إذا لم يحتوِ السياق على الإجابة، رد: "لا أملك هذه المعلومة في مستنداتك."
3. لا تستخدم معرفة خارجية. إذا لم تكن في السياق، فليست صحيحة لهذا المستخدم.
4. انسخ الأرقام والتواريخ والاقتباسات حرفياً — لا تعيد صياغتها أبداً.
"""


def answer(query: str, chunks: list[dict]) -> dict:
    if not chunks:
        return {"answer": "لا أملك هذه المعلومة في مستنداتك.", "citations": []}

    context = "\n\n".join(
        f"[{i + 1}] ({c['source_path']}) {c['content']}"
        for i, c in enumerate(chunks)
    )

    reply = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=SYSTEM,
        messages=[{"role": "user", "content": f"CONTEXT:\n{context}\n\nQUERY: {query}"}],
    )
    text = reply.content[0].text

    cited_ids = {int(n) for n in re.findall(r"\[cite:(\d+)\]", text)}
    citations = [
        {"id": i + 1, "source": c["source_path"], "span": c["content"][:200] + "…"}
        for i, c in enumerate(chunks) if (i + 1) in cited_ids
    ]
    return {"answer": text, "citations": citations}

---

الجزء 5 — تطبيق FastAPI (5 دقائق)

# main.py
from fastapi import FastAPI
from pydantic import BaseModel
from retriever import retrieve
from generator import answer

app = FastAPI()


class AskBody(BaseModel):
    q: str
    top_k: int = 5


@app.post("/ask")
def ask(body: AskBody):
    chunks = retrieve(body.q, top_k=body.top_k)
    return answer(body.q, chunks)


@app.get("/health")
def health():
    return {"ok": True}

cp .env.example .env
# املأ مفاتيحك ثم:
uvicorn main:app --reload

---

الجزء 6 — النشر (اختياري، 10 دقائق)

أبسط مسار مجاني هو Railway:

1. ادفع مجلد my-rag/ إلى مستودع GitHub
2. https://railway.app ← New Project ← Deploy from GitHub
3. أضف متغيرات البيئة
4. Railway يكتشف requirements.txt و FastAPI تلقائياً

---

الجزء 7 — قائمة استكشاف الأخطاء

العرض	أول شيء تتحقق منه	السبب المعتاد
relation "chunks" does not exist	محرر SQL	المخطط لم يُطبّق — أعد تشغيل schema.sql
hybrid_search لا يُرجع شيئاً	select count(*) from chunks;	فشل الاستيعاب صامتاً — تحقق من حصة OpenAI
نفس المقطع يُستشهد كـ [cite:3] مراراً	top_k وترتيب إعادة الترتيب	إعادة الترتيب تعمل لكن كل 5 مقاطع من نفس المستند — ارفع top_k
النموذج يرد "لا أملك" للاستعلامات المغطاة	تحقق بـ SQL	حدود التقطيع قسمت الإجابة منتصف جملة — اضبط CHUNK_TOKENS
429 من Anthropic أثناء إعادة الترتيب	لوحة API	الطبقة المجانية — أعد ترتيب مقاطع أقل

---

نقطة تحقق — أنجز هذا قبل ادعاء الشهادة

1. شحن الاستيعاب. شغّل python ingest.py ./your-docs، تحقق من select count(*) from chunks;.
2. شحن الاسترجاع. POST إلى /ask بسؤال تعرف إجابته الصحيحة — تأكد أن الاستشهادات تشير إلى ملفات المصدر الصحيحة.
3. شحن اختبار خاطئ عمداً. اسأل عن شيء ليس في مستنداتك ("ما عاصمة البرازيل؟"). تأكد أن النموذج يقول "لا أملك هذه المعلومة."
4. انشر. اجعل الخدمة قابلة للوصول من URL عام.
5. التقط الإجابة + JSON الاستشهاد كإثبات عملك.

الآن لديك نظام RAG بنيته، يعمل على مستنداتك، مع استشهادات يمكنك مراجعتها. كل ما تعلمته من الوحدات الست السابقة حدث للتو في الإنتاج.

التالي: 06-next-steps — الخطوات التالية الحقيقية: توسيع هذا النظام بالتقييم (RAGAS من الوحدة 5)، قابلية المراقبة، والتحكم متعدد المستخدمين. :::