الدليل الشامل لأنظمة RAG: بناء أنظمة التوليد المعزز بالاسترجاع

ملاحظة: تستخدم أمثلة الكود في هذا الدليل LangChain 0.3+ و LlamaIndex 0.11+ و OpenAI SDK 1.x+. تغطي أمثلة قواعد البيانات المتجهية Chroma و Pinecone و pgvector. راجع دائماً التوثيق الرسمي لأحدث تغييرات الواجهات البرمجية.

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

ما هو RAG ولماذا هو مهم

RAG (التوليد المعزز بالاسترجاع) هو بنية معمارية يتم فيها تعزيز استجابة نموذج اللغة الكبير من خلال استرجاع المعلومات ذات الصلة أولاً من مصدر معرفة خارجي. يتم تمرير السياق المسترجع مع سؤال المستخدم، مما يربط إجابة النموذج ببياناتك الفعلية.

المشكلة التي يحلها RAG

لدى نماذج اللغة الكبيرة ثلاثة قيود أساسية يعالجها RAG:

1. انقطاع المعرفة: تعرف النماذج فقط ما تدربت عليه. مستنداتك الداخلية وبياناتك الحديثة ومعرفتك الخاصة غير مرئية لها.
2. الهلوسات: بدون ربط بالبيانات، تولد النماذج إجابات تبدو معقولة لكنها غير صحيحة بثقة عالية.
3. غياب الاستشهادات: لا تستطيع نماذج اللغة الأساسية الإشارة إلى المصادر. يمكّن RAG من تقديم استجابات قابلة للتتبع والتحقق.

RAG مقابل الضبط الدقيق مقابل هندسة المطالبات

قاعدة عامة: إذا كنت تحتاج أن يعرف النموذج حقائق محددة، استخدم RAG. إذا كنت تحتاج أن يتصرف بشكل مختلف، استخدم الضبط الدقيق. تستفيد معظم أنظمة الإنتاج من RAG أولاً، ثم إضافة الضبط الدقيق لاحقاً إذا لزم الأمر.

بنية RAG: خط الأنابيب الكامل

كل نظام RAG له مرحلتان: الفهرسة (غير متصلة) والاسترجاع + التوليد (وقت التشغيل).

خط أنابيب الفهرسة (غير متصل)

المستندات → التحميل → التقسيم → التضمين → التخزين في قاعدة بيانات متجهية

1. التحميل: استيعاب المستندات من مصادر متنوعة (PDF، صفحات ويب، قواعد بيانات، واجهات برمجية)
2. التقسيم: تقسيم المستندات إلى أجزاء أصغر مناسبة للتضمين
3. التضمين: تحويل كل جزء إلى متجه كثيف باستخدام نموذج تضمين
4. التخزين: حفظ المتجهات + البيانات الوصفية في قاعدة بيانات متجهية للبحث السريع بالتشابه

خط أنابيب الاسترجاع + التوليد (وقت التشغيل)

استعلام المستخدم → تضمين الاستعلام → البحث في قاعدة البيانات المتجهية → استرجاع أفضل K → إعادة الترتيب → التوليد مع LLM

1. تضمين الاستعلام: تحويل سؤال المستخدم إلى متجه باستخدام نفس نموذج التضمين
2. البحث: إيجاد أجزاء المستندات الأكثر تشابهاً عبر تشابه المتجهات
3. إعادة الترتيب (اختياري): إعادة تسجيل النتائج باستخدام مشفر متقاطع لدقة أفضل
4. التوليد: تمرير السياق المسترجع + السؤال إلى نموذج اللغة الكبير لتوليد الإجابة

RAG أساسي مع LangChain

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. تحميل المستندات
loader = PyPDFLoader("technical_manual.pdf")
docs = loader.load()

# 2. التقسيم إلى أجزاء
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = splitter.split_documents(docs)

# 3. التضمين والتخزين
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")

# 4. إنشاء المسترجع
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 5. بناء سلسلة RAG
template = """Answer the question based only on the following context.
If you cannot answer from the context, say "I don't have enough information."

Context: {context}

Question: {question}"""

prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 6. الاستعلام
answer = chain.invoke("How do I configure the backup system?")

نماذج التضمين: تحويل النص إلى متجهات

التضمينات هي أساس RAG. تحوّل النص إلى متجهات رقمية كثيفة حيث يُعيَّن التشابه الدلالي إلى قرب هندسي.

كيف تعمل التضمينات

يحوّل نموذج التضمين النص إلى متجه ذي بُعد ثابت (مثلاً 1536 بُعداً لنموذج text-embedding-3-small من OpenAI). النصوص ذات المعنى المتشابه تنتج متجهات قريبة من بعضها في هذا الفضاء، مما يتيح البحث بالتشابه.

نماذج التضمين الشائعة

اختيار النموذج المناسب

• النماذج الأولية: text-embedding-3-small (رخيص، جودة جيدة، واجهة برمجية سهلة)
• الإنتاج (سحابي): text-embedding-3-large أو Cohere embed-v3
• مستضاف ذاتياً/الخصوصية: نماذج BGE أو E5 عبر sentence-transformers
• متعدد اللغات: Cohere embed-v3 أو multilingual-e5-large

# استخدام sentence-transformers للتضمينات المحلية
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-en-v1.5")

texts = ["How does photosynthesis work?", "Plants convert sunlight to energy"]
embeddings = model.encode(texts)

# تشابه جيب التمام
from sklearn.metrics.pairwise import cosine_similarity
similarity = cosine_similarity([embeddings[0]], [embeddings[1]])
print(f"Similarity: {similarity[0][0]:.3f}")  # ~0.85

أفضل ممارسات التضمين

1. استخدم نفس النموذج للفهرسة والاستعلام — النماذج المختلفة تنتج فضاءات متجهية غير متوافقة
2. طبّع المتجهات إذا لم تقم قاعدة بياناتك بذلك تلقائياً (تشابه جيب التمام يتطلب متجهات مطبعة)
3. قِس الأداء على بياناتك — تصنيفات لوحة MTEB لا تتنبأ دائماً بالأداء على محتوى خاص بالمجال
4. فكر في تقليل الأبعاد — نماذج OpenAI تدعم معلمة dimensions لتقليل تكاليف التخزين

قواعد البيانات المتجهية: اختيار المخزن المناسب

قواعد البيانات المتجهية مصممة خصيصاً لتخزين وفهرسة والاستعلام عن متجهات عالية الأبعاد بكفاءة.

جدول المقارنة

مثال Pinecone (مُدار)

from pinecone import Pinecone, ServerlessSpec

# التهيئة
pc = Pinecone(api_key="your-api-key")

# إنشاء فهرس
pc.create_index(
    name="rag-docs",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

index = pc.Index("rag-docs")

# إدراج متجهات مع بيانات وصفية
index.upsert(vectors=[
    {
        "id": "doc-1-chunk-0",
        "values": embedding_vector,
        "metadata": {
            "source": "manual.pdf",
            "page": 5,
            "section": "Installation",
            "text": "To install the software..."
        }
    }
])

# استعلام مع تصفية البيانات الوصفية
results = index.query(
    vector=query_embedding,
    top_k=5,
    include_metadata=True,
    filter={"source": {"$eq": "manual.pdf"}}
)

مثال pgvector (PostgreSQL)

-- تفعيل الإضافة
CREATE EXTENSION vector;

-- إنشاء جدول مع عمود متجهي
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    metadata JSONB,
    embedding vector(1536)
);

-- إنشاء فهرس HNSW للبحث السريع
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- استعلام: إيجاد أكثر 5 مستندات تشابهاً
SELECT id, content, metadata,
       1 - (embedding <=> $1::vector) AS similarity
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 5;

pgvector مثالي عندما تستخدم PostgreSQL بالفعل — لا حاجة لبنية تحتية جديدة، قوة SQL الكاملة لتصفية البيانات الوصفية، واتساق المعاملات مع بيانات تطبيقك.

استراتيجيات التقسيم: كيفية تقسيم مستنداتك

التقسيم هو أحد أكثر القرارات تأثيراً في RAG. التقسيم السيئ يؤدي إلى استرجاعات غير ذات صلة وسياق مقطوع وإجابات ضعيفة.

طرق التقسيم

التقسيم التكراري للأحرف (الأكثر شيوعاً)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,        # حجم الجزء المستهدف بالأحرف
    chunk_overlap=100,     # التداخل بين الأجزاء للحفاظ على السياق
    separators=[
        "\n\n",  # فواصل الفقرات أولاً
        "\n",    # فواصل الأسطر
        ". ",    # حدود الجمل
        " ",     # حدود الكلمات (الملاذ الأخير)
        ""       # مستوى الحرف (الملاذ الأخير المطلق)
    ],
    length_function=len,
)

chunks = splitter.split_documents(documents)

التقسيم الدلالي

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# التقسيم بناءً على تشابه التضمين بين الجمل المتتالية
splitter = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=90,
)

chunks = splitter.split_documents(documents)

إرشادات حجم الأجزاء

• صغيرة جداً (< 200 رمز): تفقد السياق، تسترجع أجزاء مبتورة
• كبيرة جداً (> 2000 رمز): تخفف الصلة، تهدر نافذة السياق
• النقطة المثلى (400-800 رمز): سياق كافٍ ليكون مفيداً، محدد بما يكفي ليكون ذا صلة
• أضف تداخلاً دائماً (50-100 رمز): يمنع قطع الجمل وفقدان المعلومات عند الحدود

إثراء البيانات الوصفية

أضف بيانات وصفية لكل جزء لاسترجاع وتصفية أفضل:

for i, chunk in enumerate(chunks):
    chunk.metadata.update({
        "chunk_index": i,
        "source_file": "manual.pdf",
        "section_title": extract_section_title(chunk),
        "doc_type": "technical",
        "created_at": "2026-01-15",
    })

الاسترجاع وإعادة الترتيب: إيجاد أفضل سياق

جودة الاسترجاع تحدد مباشرة جودة الإجابة. الاسترجاع الضعيف يعني أن نموذج اللغة الكبير يحصل على سياق غير ذي صلة وينتج إجابات سيئة.

استراتيجيات الاسترجاع

1. البحث الأساسي بالتشابه

أبسط نهج — إيجاد أقرب K متجهات:

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

2. الصلة الهامشية القصوى (MMR)

يوازن بين الصلة والتنوع لتجنب النتائج المكررة:

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,
        "fetch_k": 20,      # جلب 20 مرشحاً
        "lambda_mult": 0.7,  # 0=أقصى تنوع، 1=أقصى صلة
    }
)

3. البحث الهجين (دلالي + كلمات مفتاحية)

يجمع بين تشابه المتجهات ومطابقة الكلمات المفتاحية BM25 للحصول على أفضل ما في العالمين:

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# مسترجع الكلمات المفتاحية
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# المسترجع المتجهي
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# الجمع بأوزان متساوية
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # اضبط بناءً على حالة الاستخدام
)

إعادة الترتيب

تستخدم إعادة الترتيب نموذج مشفر متقاطع لإعادة تسجيل النتائج المسترجعة لدقة أفضل. المشفرات المتقاطعة أكثر دقة من المشفرات الثنائية (نماذج التضمين) لأنها تعالج الاستعلام والمستند معاً.

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# استخدام معيد ترتيب Cohere
reranker = CohereRerank(
    model="rerank-english-v3.0",
    top_n=3  # إرجاع أفضل 3 بعد إعادة الترتيب
)

# لف المسترجع الأساسي مع إعادة الترتيب
retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

تحويل الاستعلام

أحياناً لا يتطابق استعلام المستخدم مع لغة المستندات. حوّل الاستعلامات لتحسين الاسترجاع:

# استعلام متعدد: توليد عدة استعلامات بحث من سؤال واحد
from langchain.retrievers.multi_query import MultiQueryRetriever

multi_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.3),
)

# المسترجع يولد 3 تنويعات للاستعلام ويجمع النتائج
docs = multi_retriever.invoke("How does the authentication system work?")

تقييم أنظمة RAG

لا يمكنك تحسين ما لا تقيسه. تقييم RAG يخبرك أين يفشل خط الأنابيب ويوجه التحسين.

إطار RAGAS

RAGAS (تقييم التوليد المعزز بالاسترجاع) هو إطار التقييم القياسي لأنظمة RAG. يوفر أربعة مقاييس رئيسية:

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# إعداد مجموعة بيانات التقييم
eval_data = {
    "question": ["How do I reset my password?"],
    "answer": ["To reset your password, go to Settings > Security > Reset Password..."],
    "contexts": [["The password reset feature is in Settings > Security..."]],
    "ground_truth": ["Navigate to Settings, then Security, click Reset Password..."],
}

dataset = Dataset.from_dict(eval_data)

# تشغيل التقييم
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# {'faithfulness': 0.95, 'answer_relevancy': 0.90,
#  'context_precision': 0.85, 'context_recall': 0.80}

بناء مجموعة بيانات التقييم

مجموعة بيانات تقييم جيدة تحتاج:

1. أسئلة: 50-100 سؤال تمثيلي سيطرحه مستخدموك
2. إجابات مرجعية: الإجابات الصحيحة/المتوقعة
3. مستندات المصدر: المستندات التي تحتوي على الإجابات

ابدأ بالتنسيق اليدوي، ثم وسّع بتوليد الأسئلة الاصطناعية:

from ragas.testset.generator import TestsetGenerator
from langchain_openai import ChatOpenAI

generator = TestsetGenerator.from_langchain(
    generator_llm=ChatOpenAI(model="gpt-4o"),
    critic_llm=ChatOpenAI(model="gpt-4o"),
)

testset = generator.generate_with_langchain_docs(
    documents=chunks,
    test_size=50,
)

ما يجب تحسينه بناءً على المقاييس

أنماط الإنتاج وأفضل الممارسات

الانتقال من النموذج الأولي إلى الإنتاج يتطلب اهتماماً بالأداء والموثوقية والتكلفة.

التخزين المؤقت

خزّن مؤقتاً الأسئلة المتكررة وسياقها المسترجع لتقليل زمن الاستجابة والتكلفة:

import hashlib

def get_cache_key(query: str) -> str:
    return hashlib.sha256(query.lower().strip().encode()).hexdigest()

# نمط تخزين مؤقت بسيط
cache = {}

def cached_rag(query: str):
    key = get_cache_key(query)
    if key in cache:
        return cache[key]

    result = rag_chain.invoke(query)
    cache[key] = result
    return result

للإنتاج، استخدم Redis أو ذاكرة تخزين مؤقت موزعة مماثلة مع انتهاء صلاحية TTL.

بث الاستجابات

ابث مخرجات نموذج اللغة الكبير لتجربة مستخدم أفضل:

from langchain_core.runnables import RunnablePassthrough

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
)

# بث الرموز أثناء توليدها
for chunk in chain.stream("How do I configure backups?"):
    print(chunk.content, end="", flush=True)

المراقبة والملاحظة

تتبع هذه المقاييس في الإنتاج:

• زمن استجابة الاسترجاع: وقت البحث في قاعدة البيانات المتجهية
• زمن استجابة التوليد: وقت استجابة نموذج اللغة الكبير
• درجات صلة الاسترجاع: هل درجات التشابه في اتجاه تنازلي؟
• ملاحظات المستخدمين: إعجاب/عدم إعجاب بالإجابات
• استخدام الرموز: التكلفة لكل استعلام

استخدم LangSmith أو Langfuse أو تسجيل مخصص لالتقاط تتبعات خط الأنابيب الكامل.

تحسين التكلفة

اعتبارات الأمان

1. تطهير المدخلات: منع حقن المطالبات من خلال استعلامات المستخدمين
2. التحكم في الوصول: ضمان أن المستخدمين يسترجعون فقط المستندات المصرح لهم برؤيتها
3. تصفية المعلومات الشخصية: إزالة البيانات الحساسة من السياق المسترجع قبل تمريرها لنماذج اللغة
4. تسجيل التدقيق: تسجيل الاستعلامات والمستندات المسترجعة للامتثال
5. تحديد المعدل: منع إساءة استخدام نقطة نهاية RAG

أنماط الفشل الشائعة

البدء

هل أنت جاهز لبناء أول نظام RAG لك؟ إليك مسار تعلم مُوصى به:

1. ابدأ بنموذج أولي: استخدم Chroma + تضمينات OpenAI + مجموعة مستندات صغيرة
2. أضف التقييم: أنشئ مجموعة اختبار من 20 سؤالاً وقِس درجات RAGAS
3. حسّن التقسيم: جرب أحجام واستراتيجيات تقسيم مختلفة
4. أضف البحث الهجين: اجمع البحث المتجهي مع BM25 لاسترجاع أفضل
5. أضف إعادة الترتيب: استخدم Cohere أو مشفر متقاطع لإعادة تسجيل النتائج
6. انتقل للإنتاج: أضف التخزين المؤقت والمراقبة وضوابط الوصول
7. وسّع النطاق: انتقل إلى قاعدة بيانات متجهية مُدارة وحسّن التكاليف

يتطور نظام RAG البيئي بسرعة. تقنيات جديدة مثل RAG الوكيلي (حيث يقرر الوكيل متى وكيف يسترجع) و Graph RAG (استخدام الرسوم البيانية المعرفية بجانب المتجهات) تستمر في دفع حدود ما هو ممكن.