بناء أنظمة RAG للإنتاج: من النموذج الأولي إلى التوسع

مقدمة

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

يستعرض هذا الدليل كل مكوّن من نظام RAG إنتاجي، مع توصيات عملية وأمثلة برمجية.

فهم خط أنابيب RAG

يتكوّن نظام RAG إنتاجي من عدة مراحل:

1. معالجة المستندات: تحميل وتنظيف المستندات المصدر
2. التقسيم: تقسيم المستندات إلى وحدات قابلة للاسترجاع
3. التضمين: تحويل الأجزاء إلى تمثيلات متجهة
4. التخزين: فهرسة المتجهات في قاعدة بيانات متجهات
5. الاسترجاع: إيجاد الأجزاء ذات الصلة لاستعلام معيّن
6. التوليد: استخدام السياق المسترجع لتوليد إجابة
7. التقييم: قياس جودة الاسترجاع والتوليد

تقدّم كل مرحلة مفاضلات تؤثر على الدقة والزمن والتكلفة.

استراتيجيات التقسيم

يُعدّ التقسيم إلى حد بعيد أكثر القرارات تأثيرًا في نظام RAG. الهدف هو إنشاء أجزاء صغيرة بما يكفي لاسترجاعها بدقة ولكن كبيرة بما يكفي لتوفير سياق ذي معنى.

التقسيم ذو الحجم الثابت

النهج الأبسط يقسّم النص إلى أجزاء بحجم ثابت مع تداخل:

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""]
)

chunks = splitter.split_text(document_text)

المزايا: بسيط، متوقع، سريع العيوب: قد يقسّم في منتصف الجملة، يفقد الحدود الدلالية

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

يقسّم التقسيم الدلالي النص بناءً على المعنى بدلًا من عدد الأحرف. يجمّع الجمل ذات التضمينات المتشابهة معًا:

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95
)

chunks = splitter.split_text(document_text)

المزايا: يحافظ على التماسك الدلالي، دقة استرجاع أفضل العيوب: أبطأ، أكثر تكلفة (يتطلب تضمين كل جملة)

التقسيم الواعي بالمستندات

للمستندات المنظّمة (markdown, HTML, الكود)، استخدم مقسّمات واعية بالبنية:

from langchain_text_splitters import MarkdownHeaderTextSplitter

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
)

chunks = md_splitter.split_text(markdown_text)

يحافظ هذا على بنية المستند، مبقيًا العناوين ومحتواها معًا.

اختيار نموذج التضمين

يحدّد نموذج التضمين مدى تطابق نظام الاسترجاع للاستعلامات بالأجزاء ذات الصلة.

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

النموذج	الأبعاد	نقاط القوة	التكلفة
OpenAI text-embedding-3-small	1536	أداء عام جيد	0.02$/1M رمز
OpenAI text-embedding-3-large	3072	دقة عالية	0.13$/1M رمز
Cohere embed-v3	1024	متعدد اللغات، قوي	0.10$/1M رمز
BGE-large-en-v1.5	1024	مفتوح المصدر، قابل للاستضافة الذاتية	مجاني
Nomic Embed	768	مفتوح المصدر، سياق طويل	مجاني

التوصيات

• النماذج الأولية: OpenAI text-embedding-3-small يوفر أفضل توازن بين التكلفة والأداء
• الإنتاج على نطاق واسع: استضافة BGE أو Nomic ذاتيًا للقضاء على تكاليف الرموز
• متعدد اللغات: Cohere embed-v3 يتعامل مع لغات متعددة بشكل جيد
• أقصى دقة: OpenAI text-embedding-3-large مع تقليل الأبعاد

اختيار قاعدة بيانات المتجهات

Pinecone

Pinecone قاعدة بيانات متجهات مُدارة بالكامل ومُحسّنة للاستخدام الإنتاجي:

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")
index = pc.Index("rag-production")

index.upsert(
    vectors=[{
        "id": "chunk-1",
        "values": embedding,
        "metadata": {"source": "doc1.pdf", "page": 1}
    }]
)

results = index.query(
    vector=query_embedding,
    top_k=5,
    include_metadata=True,
    filter={"source": {"$eq": "doc1.pdf"}}
)

الأفضل لـ: الفرق التي تريد إدارة بنية تحتية صفرية وتحتاج توسعًا أفقيًا.

Weaviate

Weaviate قاعدة بيانات متجهات مفتوحة المصدر بقدرات بحث هجين قوية:

import weaviate

client = weaviate.connect_to_local()

collection = client.collections.get("Document")

collection.data.insert({
    "content": chunk_text,
    "source": "doc1.pdf"
})

results = collection.query.near_text(
    query="كيف يعمل RAG؟",
    limit=5
)

الأفضل لـ: التطبيقات التي تحتاج بحثًا هجينًا (متجه + كلمات مفتاحية) واستضافة ذاتية.

pgvector

يوسّع pgvector قاعدة PostgreSQL بالبحث المتجهي، مما يتيح استخدام قاعدة بياناتك الحالية:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536),
    metadata JSONB
);

CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops);

INSERT INTO documents (content, embedding, metadata)
VALUES ($1, $2, $3);

SELECT content, metadata, embedding <=> $query_vector AS distance
FROM documents
ORDER BY embedding <=> $query_vector
LIMIT 5;

الأفضل لـ: التطبيقات التي تستخدم PostgreSQL بالفعل، ومجموعات البيانات الأصغر، والفرق التي تريد تجنب إدارة بنية تحتية منفصلة.

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

البحث الهجين

يجمع البحث المتجه والكلمات المفتاحية لتحسين دقة الاسترجاع بشكل كبير:

from langchain_community.retrievers import PineconeHybridSearchRetriever

retriever = PineconeHybridSearchRetriever(
    index=index,
    embeddings=OpenAIEmbeddings(),
    sparse_encoder=sparse_encoder,
    top_k=5,
    alpha=0.5  # التوازن بين البحث المتجهي (0) والكلمات المفتاحية (1)
)

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

استرجع مرشحين أكثر بالبحث الأولي، ثم أعد ترتيبهم بمُشفّر متقاطع:

from langchain_cohere import CohereRerank

compressor = CohereRerank(top_n=3)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

docs = compression_retriever.invoke(query)

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

حسّن الاسترجاع بتحويل استعلام المستخدم قبل البحث:

from langchain_openai import ChatOpenAI

def hyde_query(query: str) -> str:
    """استخدم HyDE - توليد إجابة افتراضية لتحسين الاسترجاع."""
    llm = ChatOpenAI(model="gpt-4o")
    prompt = f"ولّد إجابة تفصيلية لهذا السؤال: {query}"
    return llm.invoke(prompt).content

# استخدام الإجابة الافتراضية كاستعلام بحث
enhanced_query = hyde_query(user_query)
docs = retriever.invoke(enhanced_query)

التقييم

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

قِس ما إذا كان نظامك يسترجع المستندات الصحيحة باستخدام مقاييس مثل الدقة والاستدعاء:

from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness

eval_results = evaluate(
    dataset=eval_dataset,
    metrics=[context_precision, context_recall, faithfulness]
)

print(f"دقة السياق: {eval_results['context_precision']:.2f}")
print(f"استدعاء السياق: {eval_results['context_recall']:.2f}")
print(f"الإخلاص: {eval_results['faithfulness']:.2f}")

المقاييس الرئيسية للتتبع

• دقة السياق: هل الأجزاء المسترجعة ذات صلة فعلًا؟
• استدعاء السياق: هل تُسترجع جميع الأجزاء الضرورية؟
• الإخلاص: هل تلتزم الإجابة المولّدة بالسياق المسترجع؟
• ملاءمة الإجابة: هل تجيب الإجابة فعليًا على السؤال؟

اعتبارات الإنتاج

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

خاتمة

يتطلب بناء نظام RAG إنتاجي قرارات مدروسة في كل مرحلة. ابدأ بتقسيم بسيط وقاعدة بيانات متجهات مُدارة، ثم حسّن بناءً على مقاييس التقييم. أكبر التحسينات تأتي عادةً من تقسيم أفضل، وبحث هجين، وإعادة ترتيب — وليس من التبديل إلى LLM أكبر. قِس كل شيء، كرّر على نقاط الضعف، ودع مقاييس التقييم ترشد قراراتك.