تعزيز الاستعلام

تحسين الاستعلام قبل الاسترجاع يؤثر بشكل كبير على النتائج. هذه التقنيات تسد الفجوة بين كيف يسأل المستخدمون وكيف تُكتب المستندات.

فجوة الاستعلام-المستند

المستخدمون يسألون أسئلة بشكل مختلف عن كيفية كتابة المستندات:

استعلام المستخدم: "لماذا تطبيقي بطيء؟"
المستند: "تقنيات تحسين الأداء تشمل..."

الفجوة: مفردات مختلفة، سؤال مقابل عبارة

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

ولّد تنويعات استعلام متعددة لتحسين الاستدعاء:

def expand_query(query: str, llm) -> list[str]:
    """ولّد تنويعات استعلام لتغطية أفضل."""
    prompt = f"""ولّد 3 استعلامات بحث بديلة لـ:
    "{query}"

    اشمل:
    1. نسخة معاد صياغتها
    2. نسخة أكثر تقنية
    3. نسخة أبسط

    أرجع الاستعلامات فقط، واحد في كل سطر."""

    response = llm.invoke(prompt)
    variations = response.content.strip().split('\n')

    # اشمل الاستعلام الأصلي
    return [query] + variations[:3]

# مثال
query = "لماذا تطبيقي بطيء؟"
expanded = expand_query(query, llm)
# ["لماذا تطبيقي بطيء؟",
#  "ما الذي يسبب مشاكل أداء التطبيق؟",
#  "استكشاف أخطاء زمن استجابة التطبيق",
#  "التطبيق يعمل ببطء"]

الاسترجاع متعدد الاستعلامات

ابحث مع جميع تنويعات الاستعلام:

class MultiQueryRetriever:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def search(self, query: str, k: int = 10) -> list[dict]:
        # وسّع الاستعلام
        queries = expand_query(query, self.llm)

        # استرجع لكل تنويع
        all_results = []
        seen_ids = set()

        for q in queries:
            results = self.retriever.search(q, k=k)
            for result in results:
                if result["id"] not in seen_ids:
                    all_results.append(result)
                    seen_ids.add(result["id"])

        # أعد ترتيب النتائج المدمجة
        return self._rerank(query, all_results, k)

HyDE (تضمينات المستند الافتراضي)

ولّد إجابة افتراضية، ثم ابحث عن مستندات مشابهة:

class HyDERetriever:
    def __init__(self, vectorstore, llm, embeddings):
        self.vectorstore = vectorstore
        self.llm = llm
        self.embeddings = embeddings

    def search(self, query: str, k: int = 5) -> list[dict]:
        # ولّد مستند افتراضي
        prompt = f"""اكتب إجابة مفصلة على هذا السؤال كما لو كنت
        تكتب توثيقاً:

        السؤال: {query}

        الإجابة:"""

        hypothetical_doc = self.llm.invoke(prompt).content

        # ضمّن المستند الافتراضي
        hyde_embedding = self.embeddings.embed_query(hypothetical_doc)

        # ابحث باستخدام التضمين الافتراضي
        results = self.vectorstore.similarity_search_by_vector(
            hyde_embedding,
            k=k
        )

        return results

# مثال
query = "كيف أنفذ تحديد المعدل؟"

# المستند الافتراضي المولد:
# "يمكن تنفيذ تحديد المعدل باستخدام خوارزمية دلو الرموز.
#  أولاً، حدد حجم الدلو ومعدل إعادة التعبئة..."

# هذا التضمين يطابق التوثيق أفضل من السؤال

لماذا يعمل HyDE:

• الأسئلة والمستندات لها تضمينات مختلفة
• الإجابة الافتراضية أقرب للتوثيق الفعلي
• تضمين الإجابة يطابق تضمينات المستندات أفضل

تفكيك الاستعلام

قسّم الاستعلامات المعقدة إلى استعلامات فرعية:

def decompose_query(query: str, llm) -> list[str]:
    """قسّم الاستعلام المعقد إلى استعلامات فرعية أبسط."""
    prompt = f"""حلل هذا الاستعلام وقسمه إلى استعلامات فرعية أبسط
    يمكن الإجابة عليها بشكل مستقل:

    الاستعلام: {query}

    إذا كان الاستعلام بسيطاً بالفعل، أرجعه كما هو.
    وإلا، أرجع 2-4 استعلامات فرعية، واحد في كل سطر.

    الاستعلامات الفرعية:"""

    response = llm.invoke(prompt)
    sub_queries = response.content.strip().split('\n')

    return sub_queries if len(sub_queries) > 1 else [query]

# مثال
query = "قارن OAuth و JWT لمصادقة API وأظهر التنفيذ"

sub_queries = decompose_query(query, llm)
# ["ما هو OAuth لمصادقة API؟",
#  "ما هو JWT لمصادقة API؟",
#  "كيف تنفذ OAuth؟",
#  "كيف تنفذ JWT؟"]

الموجه بالتراجع

ولّد استعلام أعم أولاً:

def step_back_query(query: str, llm) -> str:
    """ولّد استعلام أوسع وأعم."""
    prompt = f"""بالنظر لهذا الاستعلام المحدد، ولّد سؤالاً أوسع
    يوفر سياق خلفية مفيد:

    الاستعلام المحدد: {query}

    السؤال الأوسع:"""

    return llm.invoke(prompt).content.strip()

class StepBackRetriever:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def search(self, query: str, k: int = 5) -> list[dict]:
        # احصل على استعلام التراجع
        broad_query = step_back_query(query, self.llm)

        # استرجع لكليهما
        specific_results = self.retriever.search(query, k=k)
        broad_results = self.retriever.search(broad_query, k=k//2)

        # ادمج (السياق الواسع أولاً، ثم المحدد)
        return broad_results + specific_results

# مثال
query = "لماذا فهرس HNSW لديه استخدام ذاكرة أعلى من IVF؟"

step_back = "كيف تعمل خوارزميات فهرسة قاعدة البيانات المتجهة؟"
# السياق الواسع يساعد في الإجابة على السؤال المحدد

خط أنابيب تحويل الاستعلام

ادمج تقنيات متعددة:

class QueryTransformPipeline:
    def __init__(self, retriever, llm, embeddings):
        self.retriever = retriever
        self.llm = llm
        self.embeddings = embeddings

    def search(
        self,
        query: str,
        k: int = 5,
        use_expansion: bool = True,
        use_hyde: bool = False,
        use_decomposition: bool = False
    ) -> list[dict]:
        all_results = []
        seen = set()

        # الاستعلام الأصلي
        queries = [query]

        # توسيع الاستعلام
        if use_expansion:
            queries.extend(expand_query(query, self.llm))

        # تفكيك الاستعلام
        if use_decomposition:
            queries.extend(decompose_query(query, self.llm))

        # استرجع لجميع الاستعلامات
        for q in queries:
            if use_hyde:
                results = self._hyde_search(q, k=k)
            else:
                results = self.retriever.search(q, k=k)

            for r in results:
                if r["id"] not in seen:
                    all_results.append(r)
                    seen.add(r["id"])

        # أعد الترتيب بالاستعلام الأصلي
        return self._rerank(query, all_results, k)

اختيار تقنيات التعزيز

التقنية	الأفضل لـ	تأثير زمن الاستجابة
توسيع الاستعلام	عدم تطابق المفردات	+100-200ms
HyDE	استرجاع الأسئلة والأجوبة	+200-500ms
التفكيك	الاستعلامات المعقدة	+200-400ms
التراجع	الأسئلة التي تحتاج سياق	+100-200ms

البداية
  │
  ▼
استعلام حقائق بسيط؟
  │
  ├─ نعم → توسيع الاستعلام فقط
  │
  ▼
أسئلة وأجوبة على التوثيق؟
  │
  ├─ نعم → HyDE + توسيع
  │
  ▼
استعلام معقد متعدد الأجزاء؟
  │
  ├─ نعم → التفكيك
  │
  ▼
الاستعلام يحتاج سياق خلفية؟
  │
  ├─ نعم → الموجه بالتراجع
  │
  ▼
الافتراضي → توسيع الاستعلام (خط أساس جيد)

ملاحظة زمن الاستجابة: تعزيز الاستعلام يضيف استدعاءات LLM. خزّن تحويلات الاستعلام الشائعة مؤقتاً وفكر في المعالجة غير المتزامنة لأنظمة الإنتاج.

في الوحدة التالية، سنتعلم كيف نقيّم أنظمة RAG بشكل منهجي باستخدام RAGAS وأطر أخرى. :::