دليل بناء نظام RAG ذاتي الاستضافة — من الصفر إلى الإنتاج

لماذا self-hosted أصلاً؟

قبل ما تكتب سطر كود واحد، لازم تفهم أنك بتتخذ قرار معماري بتعيش معه لفترة طويلة.

الـSaaS solutions (زي Pinecone + OpenAI) هتوصّلك للنتيجة في ساعات. لكن في سياقات معينة، هذا الاختيار مش متاح أو مش عاقل:

• البيانات الحساسة: عقود قانونية، سجلات طبية، كود مصدري proprietary — أي بيانات ما يُرسَلها لطرف ثالث
• التكلفة عند Scale: لما تبدأ تعالج ملايين الـchunks شهرياً، فارق التكلفة يصبح قرار تجاري لا تقني
• التخصيص العميق: custom embeddings لـdomain معين، أو pipeline logic مش موجودة في الـofferings الجاهزة
• الامتثال التنظيمي: بيئات تشترط أن البيانات لا تخرج من infrastructure معين

إذا ما كنت في إحدى هذه الحالات، الـSaaS الصح قد يكون الخيار الأذكى.

---

المتطلبات قبل تبدأ

المعرفة التقنية

✓ Python 3.10+
✓ Docker و Docker Compose
✓ فهم أساسي لـvector embeddings (مش لازم رياضيات عميقة)
✓ معرفة بـREST APIs

الأجهزة (الحد الأدنى للتطوير)

المكوّن	الحد الأدنى	الموصى به
RAM	16 GB	32 GB
Storage	20 GB SSD	100 GB NVMe
GPU	اختياري	NVIDIA 8GB VRAM+
CPU	4 cores	8+ cores

ملاحظة GPU: بدون GPU، الـembedding generation والـLLM inference ستكون أبطأ بشكل ملحوظ. Ollama يشتغل على CPU لكن latency ستكون بالثواني لا المللي ثانية.

---

اختيار الـStack: معايير لا تخمين

أربعة frameworks رئيسية في الساحة. كل واحد له منطق مختلف.

المقارنة

المعيار	LangChain	LlamaIndex	Haystack	Weaviate standalone
سهولة البداية	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐	⭐⭐
مرونة الـPipeline	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐
Production-readiness	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐
Debugging وضوح	⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐	N/A
حجم الـcommunity	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐
التوثيق	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐

متى تختار كل واحد؟

LangChain: Prototyping سريع، فريقك يحب abstractions جاهزة، ومش محتاج control دقيق على كل خطوة. تحذير: الـabstraction leak تحت الضغط مؤلم.

LlamaIndex: بياناتك معقدة الهيكل (PDFs متداخلة، graphs، databases متعددة). محسّن أكثر للـdata ingestion من LangChain.

Haystack: المختار في هذا الدليل. Pipeline declarative واضح، component-based architecture، وdebugability ممتازة. الأنسب للـproduction systems التي تحتاج maintainability.

Weaviate standalone: لما تبني نظامك الخاص بالكامل وتريد فقط vector store قوية.

---

الـStack المختار: Haystack + Weaviate + Ollama

Documents → Haystack Indexing Pipeline → Weaviate (Vector Store)
                                              ↑
User Query → Haystack Query Pipeline ─────────┘ → Ollama (LLM) → Answer

• Haystack: يدير الـpipelines (indexing + querying)
• Weaviate: يخزن الـvectors ويتولى الـsemantic search
• Ollama: يشغّل الـLLM محلياً (سنستخدم llama3.2 و nomic-embed-text)

---

Setup كامل خطوة بخطوة

الخطوة 1: هيكل المشروع

mkdir rag-local && cd rag-local

mkdir -p {data/raw,data/processed,src,tests,config}

touch docker-compose.yml
touch src/{indexer.py,retriever.py,pipeline.py}
touch requirements.txt
touch config/settings.py

الخطوة 2: Docker Compose

# docker-compose.yml
version: '3.8'

services:
  weaviate:
    image: semitechnologies/weaviate:1.24.1
    ports:
      - "8080:8080"
      - "50051:50051"
    environment:
      QUERY_DEFAULTS_LIMIT: 25
      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
      PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
      DEFAULT_VECTORIZER_MODULE: 'none'        # نحن نوفر الـvectors خارجياً
      ENABLE_MODULES: ''
      CLUSTER_HOSTNAME: 'node1'
    volumes:
      - weaviate_data:/var/lib/weaviate
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/v1/.well-known/ready"]
      interval: 10s
      timeout: 5s
      retries: 5

  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_models:/root/.ollama
    restart: unless-stopped
    # إذا عندك GPU: uncomment السطرين التاليين
    # deploy:
    #   resources:
    #     reservations:
    #       devices:
    #         - driver: nvidia
    #           count: 1
    #           capabilities: [gpu]

volumes:
  weaviate_data:
  ollama_models:

# شغّل الـservices
docker compose up -d

# تحقق أن كل شيء شغّال
docker compose ps
curl http://localhost:8080/v1/.well-known/ready   # يرجع: {"status":"200 OK"}
curl http://localhost:11434/api/tags               # يرجع قائمة الـmodels

الخطوة 3: تثبيت Python dependencies

# requirements.txt
haystack-ai>=2.3.0
weaviate-client>=4.5.0
ollama>=0.2.0
sentence-transformers>=2.7.0
python-dotenv>=1.0.0
pypdf>=4.0.0
tqdm>=4.66.0

python -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate
pip install -r requirements.txt

الخطوة 4: تحميل الـModels على Ollama

# LLM للـgeneration
docker exec -it rag-local-ollama-1 ollama pull llama3.2

# Embedding model (334MB فقط، ممتاز للـlocal use)
docker exec -it rag-local-ollama-1 ollama pull nomic-embed-text

# تحقق من التحميل
curl http://localhost:11434/api/tags | python3 -m json.tool

الخطوة 5: الإعدادات

# config/settings.py
from dataclasses import dataclass

@dataclass
class Config:
    # Weaviate
    weaviate_url: str = "http://localhost:8080"
    collection_name: str = "Documents"

    # Ollama
    ollama_url: str = "http://localhost:11434"
    embedding_model: str = "nomic-embed-text"
    llm_model: str = "llama3.2"

    # Chunking
    chunk_size: int = 512        # tokens
    chunk_overlap: int = 64      # tokens للـcontinuity
    split_by: str = "word"

    # Retrieval
    top_k: int = 5
    min_score: float = 0.65      # اضبط حسب domain بتاعك

config = Config()

الخطوة 6: Indexing Pipeline

# src/indexer.py
import weaviate
from weaviate.classes.config import Configure, Property, DataType
import ollama
from haystack import Document, Pipeline
from haystack.components.converters import PyPDFToDocument, TextFileToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.writers import DocumentWriter
from haystack.document_stores.types import DuplicatePolicy
from pathlib import Path
from tqdm import tqdm
import hashlib

from config.settings import config


class WeaviateDocumentStore:
    """Wrapper بسيط لـWeaviate يخزن documents مع vectors."""

    def __init__(self):
        self.client = weaviate.connect_to_local(
            host="localhost",
            port=8080,
        )
        self._ensure_collection()

    def _ensure_collection(self):
        if not self.client.collections.exists(config.collection_name):
            self.client.collections.create(
                name=config.collection_name,
                vectorizer_config=Configure.Vectorizer.none(),  # نحن نوفر الـvectors
                properties=[
                    Property(name="content", data_type=DataType.TEXT),
                    Property(name="source", data_type=DataType.TEXT),
                    Property(name="chunk_id", data_type=DataType.TEXT),
                ]
            )

    def add_documents(self, documents: list[dict]):
        collection = self.client.collections.get(config.collection_name)
        with collection.batch.dynamic() as batch:
            for doc in tqdm(documents, desc="Indexing"):
                batch.add_object(
                    properties={
                        "content": doc["content"],
                        "source": doc.get("source", "unknown"),
                        "chunk_id": doc["chunk_id"],
                    },
                    vector=doc["vector"]
                )

    def close(self):
        self.client.close()


def get_embedding(text: str) -> list[float]:
    response = ollama.embeddings(
        model=config.embedding_model,
        prompt=text
    )
    return response["embedding"]


def generate_chunk_id(content: str, source: str) -> str:
    return hashlib.md5(f"{source}:{content[:100]}".encode()).hexdigest()


def build_indexing_pipeline() -> Pipeline:
    pipeline = Pipeline()

    pipeline.add_component("cleaner", DocumentCleaner(
        remove_empty_lines=True,
        remove_extra_whitespaces=True,
    ))

    pipeline.add_component("splitter", DocumentSplitter(
        split_by=config.split_by,
        split_length=config.chunk_size,
        split_overlap=config.chunk_overlap,
    ))

    # ربط المكوّنات
    pipeline.connect("cleaner", "splitter")

    return pipeline


def index_documents(file_paths: list[str]):
    store = WeaviateDocumentStore()
    pipeline = build_indexing_pipeline()

    all_chunks = []

    for file_path in file_paths:
        path = Path(file_path)
        print(f"Processing: {path.name}")

        # قراءة الملف حسب نوعه
        if path.suffix.lower() == ".pdf":
            converter = PyPDFToDocument()
            result = converter.run(sources=[path])
        else:
            converter = TextFileToDocument()
            result = converter.run(sources=[path])

        docs = result["documents"]

        # تنظيف وتقسيم
        pipeline_result = pipeline.run({"cleaner": {"documents": docs}})
        chunks = pipeline_result["splitter"]["documents"]

        # توليد الـembeddings وتجهيز للتخزين
        for chunk in chunks:
            vector = get_embedding(chunk.content)
            all_chunks.append({
                "content": chunk.content,
                "source": str(path.name),
                "chunk_id": generate_chunk_id(chunk.content, str(path.name)),
                "vector": vector,
            })

    # تخزين في Weaviate
    store.add_documents(all_chunks)
    store.close()

    print(f"\nDone: indexed {len(all_chunks)} chunks from {len(file_paths)} files")
    return len(all_chunks)


if __name__ == "__main__":
    import sys
    files = sys.argv[1:] or ["data/raw/sample.txt"]
    index_documents(files)

الخطوة 7: Query Pipeline

# src/retriever.py
import weaviate
import weaviate.classes as wvc
import ollama
from config.settings import config


class RAGRetriever:
    def __init__(self):
        self.client = weaviate.connect_to_local(host="localhost", port=8080)
        self.collection = self.client.collections.get(config.collection_name)

    def retrieve(self, query: str) -> list[dict]:
        """يسترجع الـchunks الأقرب للـquery."""
        query_vector = ollama.embeddings(
            model=config.embedding_model,
            prompt=query
        )["embedding"]

        results = self.collection.query.near_vector(
            near_vector=query_vector,
            limit=config.top_k,
            return_metadata=wvc.query.MetadataQuery(certainty=True),
        )

        chunks = []
        for obj in results.objects:
            certainty = obj.metadata.certainty or 0
            if certainty >= config.min_score:
                chunks.append({
                    "content": obj.properties["content"],
                    "source": obj.properties["source"],
                    "score": round(certainty, 4),
                })

        return chunks

    def close(self):
        self.client.close()


# src/pipeline.py
from src.retriever import RAGRetriever
import ollama

PROMPT_TEMPLATE = """استخدم المعلومات التالية فقط للإجابة على السؤال.
إذا لم تجد الإجابة في المعلومات المقدمة، قل "لا تتوفر لديّ معلومات كافية للإجابة".

المعلومات المرجعية:
{context}

السؤال: {question}

الإجابة:"""


def ask(question: str, verbose: bool = False) -> dict:
    retriever = RAGRetriever()

    # استرجاع الـchunks
    chunks = retriever.retrieve(question)
    retriever.close()

    if not chunks:
        return {
            "answer": "لم أجد معلومات ذات صلة في قاعدة البيانات.",
            "sources": [],
            "chunks_used": 0,
        }

    # بناء الـcontext
    context_parts = []
    for i, chunk in enumerate(chunks, 1):
        context_parts.append(f"[{i}] (من: {chunk['source']}, ثقة: {chunk['score']})\n{chunk['content']}")

    context = "\n\n".join(context_parts)

    if verbose:
        print(f"Retrieved {len(chunks)} chunks")
        print(f"Top chunk score: {chunks[0]['score']}")

    # توليد الإجابة
    prompt = PROMPT_TEMPLATE.format(context=context, question=question)

    response = ollama.generate(
        model=config.llm_model,
        prompt=prompt,
        options={"temperature": 0.1},  # منخفضة للـfactual accuracy
    )

    return {
        "answer": response["response"].strip(),
        "sources": list({c["source"] for c in chunks}),
        "chunks_used": len(chunks),
        "top_score": chunks[0]["score"] if chunks else 0,
    }


if __name__ == "__main__":
    question = "ما هي أهم نقاط الوثيقة؟"
    result = ask(question, verbose=True)

    print(f"\nالسؤال: {question}")
    print(f"الإجابة: {result['answer']}")
    print(f"المصادر: {', '.join(result['sources'])}")

الخطوة 8: تشغيل النظام

# 1. ضع ملفاتك في data/raw/
echo "هذا مستند تجريبي يحتوي على معلومات مهمة عن المشروع." > data/raw/sample.txt

# 2. فهرسة المستندات
python -m src.indexer data/raw/sample.txt

# 3. اختبار الاسترجاع
python -m src.pipeline

# أو استخدامه كمكتبة
python3 -c "
from src.pipeline import ask
result = ask('ما محتوى المستند؟', verbose=True)
print(result['answer'])
"

---

اختبار الجودة (Evals)

النظام شغّال ≠ النظام مفيد. هذه الفجوة تقتل أنظمة RAG كثيرة.

3 مقاييس لا غنى عنها

1. Context Recall: هل الـretrieved chunks تحتوي على المعلومات اللازمة للإجابة؟

2. Answer Faithfulness: هل الإجابة مبنية على الـcontext فقط أم اخترع النموذج معلومات؟

3. Answer Relevancy: هل الإجابة تجيب على السؤال المطروح فعلاً؟

# tests/eval.py
"""
Eval بسيط بدون dependencies خارجية.
للـproduction: استخدم RAGAS أو DeepEval.
"""
from src.pipeline import ask
from src.retriever import RAGRetriever


TEST_CASES = [
    {
        "question": "ما هو موضوع المستند الرئيسي؟",
        "expected_keywords": ["مستند", "معلومات", "مشروع"],
        "should_answer": True,
    },
    {
        "question": "ما هو سعر البيتكوين اليوم؟",
        "expected_keywords": [],
        "should_answer": False,  # يجب أن يقول "لا أعلم"
    },
]


def evaluate():
    results = []

    for case in TEST_CASES:
        result = ask(case["question"])
        answer = result["answer"]

        # فحص أن النظام لم يخترع إجابة لسؤال خارج السياق
        if not case["should_answer"]:
            passed = "لا تتوفر" in answer or "لا أعلم" in answer
            results.append({
                "question": case["question"],
                "test": "no_hallucination",
                "passed": passed,
                "answer_preview": answer[:80],
            })
        else:
            # فحص وجود الـkeywords المتوقعة
            found = sum(1 for kw in case["expected_keywords"] if kw in answer)
            recall = found / len(case["expected_keywords"]) if case["expected_keywords"] else 1.0
            results.append({
                "question": case["question"],
                "test": "keyword_recall",
                "passed": recall >= 0.6,
                "score": round(recall, 2),
                "answer_preview": answer[:80],
            })

    # طباعة النتائج
    passed = sum(1 for r in results if r["passed"])
    print(f"\nEval Results: {passed}/{len(results)} passed\n")

    for r in results:
        status = "PASS" if r["passed"] else "FAIL"
        print(f"[{status}] {r['test']}: {r['question'][:50]}")
        if not r["passed"]:
            print(f"       Answer: {r['answer_preview']}...")

    return passed == len(results)


if __name__ == "__main__":
    import sys
    success = evaluate()
    sys.exit(0 if success else 1)

python -m tests.eval

للـproduction الجاد: استخدم RAGAS الذي يقيس هذه المقاييس تلقائياً باستخدام LLM كـjudge.

---

المشاكل الثلاث الأكثر شيوعاً وحلولها

المشكلة 1: الـRetrieval يرجع chunks غير ذات صلة

الأعراض: النظام يجيب بمعلومات خاطئة أو غير مرتبطة بالسؤال. الـscore مرتفع لكن المحتوى غلط.

السبب الغالب: chunk_size كبير جداً يخلط معلومات متعددة في chunk واحد.

# المشكلة: chunk_size = 1024 يجمع فقرات غير مترابطة
# الحل:
config.chunk_size = 256    # جرب قيماً بين 128-512
config.chunk_overlap = 32  # 10-15% من chunk_size

# إذا مستمرة المشكلة، جرب sentence-based splitting بدلاً من word-based
pipeline.add_component("splitter", DocumentSplitter(
    split_by="sentence",    # بدلاً من "word"
    split_length=5,         # 5 جمل per chunk
    split_overlap=1,
))

بعد التغيير، أعد الفهرسة من الصفر — الـchunks القديمة لازم تُحذف:

# احذف الـcollection وأعد إنشاءها
client.collections.delete(config.collection_name)

---

المشكلة 2: الـLLM يخترع معلومات (Hallucination)

الأعراض: الإجابة منطقية لكنها غير موجودة في مستنداتك. verbose=True يظهر أن الـchunks المسترجعة لا تحتوي على الإجابة.

السبب: النموذج يكمل من معرفته العامة عندما يكون الـcontext ضعيفاً.

# حل 1: رفع الـtemperature لـ0 يقلل "الإبداع"
response = ollama.generate(
    model=config.llm_model,
    prompt=prompt,
    options={
        "temperature": 0.0,    # من 0.1 إلى 0.0
        "top_p": 0.9,
    },
)

# حل 2: تشديد الـprompt
PROMPT_TEMPLATE = """أنت مساعد يجيب فقط بناءً على المعلومات المقدمة.
لا تستخدم أي معرفة خارجية.
إذا لم تجد الإجابة في النص أدناه، أجب بـ"لا تتوفر لديّ معلومات كافية".

النص:
{context}

السؤال: {question}
الإجابة (بناءً على النص فقط):"""

# حل 3: رفع min_score لاستبعاد chunks ضعيفة الصلة
config.min_score = 0.75    # من 0.65 إلى 0.75

---

المشكلة 3: بطء شديد في الفهرسة

الأعراض: فهرسة 100 ملف تأخذ ساعات. الـCPU على 100%.

السبب: توليد الـembeddings بشكل sequential واحداً تلو الآخر.

# الحل: Batch processing مع ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

# Ollama thread-safe للـread operations
_ollama_lock = threading.Lock()


def get_embeddings_batch(texts: list[str], batch_size: int = 10) -> list[list[float]]:
    """توليد embeddings بشكل متوازٍ."""
    embeddings = [None] * len(texts)

    def embed_single(args):
        idx, text = args
        # Ollama يدعم concurrent requests بدون lock في معظم الحالات
        response = ollama.embeddings(model=config.embedding_model, prompt=text)
        return idx, response["embedding"]

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {
            executor.submit(embed_single, (i, text)): i
            for i, text in enumerate(texts)
        }
        for future in tqdm(as_completed(futures), total=len(texts), desc="Embedding"):
            idx, embedding = future.result()
            embeddings[idx] = embedding

    return embeddings

إذا عندك GPU: راجع إعدادات docker-compose.yml لتفعيل GPU passthrough — هذا وحده يحسّن السرعة 10-20x.

---

مقارنة Self-Hosted مقابل SaaS

المعيار	Self-Hosted	SaaS (Pinecone + OpenAI)
وقت الإعداد الأولي	ساعات-أيام	دقائق-ساعات
تكلفة التطوير	وقت المطور	منخفضة
تكلفة التشغيل (scale صغير)	تكلفة الـserver الثابتة	أرخص غالباً
تكلفة التشغيل (scale كبير)	أرخص بكثير	تتصاعد مع الاستخدام
خصوصية البيانات	كاملة	تعتمد على sla المزود
التخصيص	لا حدود	محدود بـAPI المزود
الصيانة والـupdates	عليك	على المزود
الـUptime والموثوقية	مسؤوليتك	مضمونة من المزود
الـLatency	يعتمد على أجهزتك	منخفضة ومستقرة
الامتثال التنظيمي	قابل للتخصيص الكامل	محدود بسياسات المزود

القاعدة العملية: إذا كنت تبني proof-of-concept أو startup في مراحله الأولى، ابدأ بـSaaS. انتقل لـself-hosted عندما يصبح أحد هذه المبررات حقيقياً: البيانات حساسة، التكلفة أصبحت ملحوظة، أو تحتاج تخصيصاً لا يوفره المزود.

---

الخطوات التالية

النظام الذي بنيناه هو نقطة البداية، وليس النهاية. الأشياء التالية ستحسّ