نظرة عامة على هندسة RAG

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

لماذا RAG؟

المشكلة	كيف يحلها RAG
هلوسات LLM	ترسيخ الاستجابات في حقائق مسترجعة
المعرفة القديمة	استخدام بيانات حالية من مصادرك
الاستجابات العامة	توفير سياق خاص بالمجال
حدود الرموز	استرجاع المعلومات ذات الصلة فقط

هندسة خط أنابيب RAG

┌─────────────────────────────────────────────────────────────────┐
│                       خط أنابيب RAG                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────────────── مرحلة الفهرسة ──────────────────────┐   │
│  │                                                          │   │
│  │   المستندات ──▶ التقسيم ──▶ التضمين ──▶ قاعدة المتجهات  │   │
│  │                                                          │   │
│  └──────────────────────────────────────────────────────────┘   │
│                                                                  │
│  ┌──────────────────── مرحلة الاستعلام ─────────────────────┐   │
│  │                                                          │   │
│  │   الاستعلام ──▶ التضمين ──▶ الاسترجاع ──▶ إعادة الترتيب  │   │
│  │                              │                           │   │
│  │                              ▼                           │   │
│  │                    أفضل K مستند                          │   │
│  │                              │                           │   │
│  │                              ▼                           │   │
│  │   ┌─────────────────────────────────────────────────┐   │   │
│  │   │     السياق + الاستعلام ──▶ LLM ──▶ الاستجابة    │   │   │
│  │   └─────────────────────────────────────────────────┘   │   │
│  │                                                          │   │
│  └──────────────────────────────────────────────────────────┘   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

المرحلة 1: فهرسة المستندات

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

class ChunkingStrategy:
    """نهج مختلفة لتقسيم المستندات."""

    @staticmethod
    def fixed_size(text: str, chunk_size: int = 500, overlap: int = 50) -> list:
        """قطع بحجم ثابت بسيطة مع تداخل."""
        chunks = []
        start = 0
        while start < len(text):
            end = start + chunk_size
            chunks.append(text[start:end])
            start = end - overlap
        return chunks

    @staticmethod
    def semantic(text: str, max_tokens: int = 500) -> list:
        """التقسيم على حدود دلالية (فقرات، أقسام)."""
        # التقسيم على أسطر جديدة مزدوجة (فقرات)
        paragraphs = text.split("\n\n")

        chunks = []
        current_chunk = []
        current_tokens = 0

        for para in paragraphs:
            para_tokens = len(para) // 4  # تقدير تقريبي
            if current_tokens + para_tokens > max_tokens:
                chunks.append("\n\n".join(current_chunk))
                current_chunk = [para]
                current_tokens = para_tokens
            else:
                current_chunk.append(para)
                current_tokens += para_tokens

        if current_chunk:
            chunks.append("\n\n".join(current_chunk))

        return chunks

مقايضات حجم القطعة

حجم القطعة	الإيجابيات	السلبيات
صغير (100-200 رمز)	استرجاع دقيق	قد يفقد السياق
متوسط (300-500 رمز)	متوازن	خيار افتراضي جيد
كبير (500-1000 رمز)	سياق كامل	قد يتضمن معلومات غير ذات صلة

توليد التضمينات

from openai import OpenAI

class EmbeddingPipeline:
    def __init__(self, model: str = "text-embedding-3-small"):
        self.client = OpenAI()
        self.model = model

    async def embed_chunks(self, chunks: list) -> list:
        """تضمين قطع متعددة بكفاءة."""
        # تضمين دفعات للكفاءة
        response = self.client.embeddings.create(
            model=self.model,
            input=chunks
        )
        return [item.embedding for item in response.data]

    async def embed_query(self, query: str) -> list:
        """تضمين استعلام واحد."""
        response = self.client.embeddings.create(
            model=self.model,
            input=query
        )
        return response.data[0].embedding

المرحلة 2: معالجة الاستعلام

تدفق الاسترجاع

class RAGPipeline:
    def __init__(self, vector_store, llm, embedding_model):
        self.vector_store = vector_store
        self.llm = llm
        self.embedder = embedding_model

    async def query(self, user_query: str, top_k: int = 5) -> dict:
        # الخطوة 1: تضمين الاستعلام
        query_embedding = await self.embedder.embed_query(user_query)

        # الخطوة 2: استرجاع المستندات المتشابهة
        results = await self.vector_store.search(
            embedding=query_embedding,
            top_k=top_k
        )

        # الخطوة 3: بناء السياق
        context = "\n\n---\n\n".join([
            f"المصدر: {r.metadata.get('source', 'غير معروف')}\n{r.content}"
            for r in results
        ])

        # الخطوة 4: توليد الاستجابة
        prompt = f"""أجب بناءً على السياق المقدم فقط.
إذا لم تكن الإجابة في السياق، قل "ليس لدي معلومات عن ذلك."

السياق:
{context}

السؤال: {user_query}

الإجابة:"""

        response = await self.llm.complete(prompt)

        return {
            "answer": response,
            "sources": [r.metadata for r in results],
            "context_used": context
        }

المقاييس الرئيسية لأنظمة RAG

المقياس	ما يقيسه	الهدف
دقة الاسترجاع	% المستندات المسترجعة ذات الصلة	> 80%
استدعاء الاسترجاع	% المستندات ذات الصلة التي تم استرجاعها	> 70%
دقة الإجابة	صحة الاستجابة النهائية	> 90%
زمن الاستجابة	الوقت من الاستعلام إلى الاستجابة	< 3 ثوانٍ
الإخلاص	الإجابة مرتكزة على المستندات المسترجعة	> 95%

بعد ذلك، سنتعمق في اختيار قاعدة البيانات المتجهة والمقايضات. :::