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

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

لماذا الاسترجاع الهجين؟

نوع البحث	نقاط القوة	نقاط الضعف
كثيف (متجه)	الفهم الدلالي، المرادفات	يفوت المطابقات التامة، الكيانات
متناثر (BM25)	الكلمات المفتاحية التامة، المصطلحات النادرة	لا فهم دلالي
هجين	أفضل ما في الاثنين	أكثر تعقيداً قليلاً

مثال على المشكلة

الاستعلام: "ما هي الآثار الجانبية للأسبرين؟"

المستند	نقاط BM25	نقاط المتجه	ذو صلة؟
"الأسبرين قد يسبب نزيف المعدة..."	عالية	عالية	نعم
"مسكنات الألم قد يكون لها آثار سلبية..."	منخفضة	عالية	نعم
"تناول الأسبرين مع الطعام..."	عالية	متوسطة	ربما

البحث الهجين يلتقط كل من مطابقات الكلمات المفتاحية والمطابقات الدلالية.

BM25 + الاسترجاع الكثيف

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, documents, embeddings, embedding_model):
        self.documents = documents
        self.embeddings = embeddings
        self.embedding_model = embedding_model

        # تهيئة BM25
        tokenized_docs = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)

    async def search(
        self,
        query: str,
        top_k: int = 10,
        alpha: float = 0.5  # الوزن للكثيف مقابل المتناثر
    ) -> list:
        # الاسترجاع المتناثر (BM25)
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_scores = self._normalize(bm25_scores)

        # الاسترجاع الكثيف (المتجه)
        query_embedding = await self.embedding_model.embed(query)
        dense_scores = np.array([
            self._cosine_similarity(query_embedding, emb)
            for emb in self.embeddings
        ])
        dense_scores = self._normalize(dense_scores)

        # دمج النقاط
        hybrid_scores = alpha * dense_scores + (1 - alpha) * bm25_scores

        # الحصول على أفضل k
        top_indices = np.argsort(hybrid_scores)[::-1][:top_k]

        return [
            {
                "document": self.documents[i],
                "score": hybrid_scores[i],
                "dense_score": dense_scores[i],
                "sparse_score": bm25_scores[i]
            }
            for i in top_indices
        ]

    def _normalize(self, scores: np.array) -> np.array:
        min_s, max_s = scores.min(), scores.max()
        if max_s - min_s == 0:
            return np.zeros_like(scores)
        return (scores - min_s) / (max_s - min_s)

    def _cosine_similarity(self, a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

دمج الرتبة التبادلية (RRF)

طريقة أكثر تطوراً لدمج النتائج:

class RRFRetriever:
    def __init__(self, retrievers: list, k: int = 60):
        self.retrievers = retrievers
        self.k = k  # ثابت RRF

    async def search(self, query: str, top_k: int = 10) -> list:
        # الحصول على النتائج من كل مسترجع
        all_results = []
        for retriever in self.retrievers:
            results = await retriever.search(query, top_k=top_k * 2)
            all_results.append(results)

        # حساب نقاط RRF
        doc_scores = {}
        for retriever_results in all_results:
            for rank, result in enumerate(retriever_results):
                doc_id = result["id"]
                # صيغة RRF: 1 / (k + الرتبة)
                rrf_score = 1 / (self.k + rank + 1)

                if doc_id not in doc_scores:
                    doc_scores[doc_id] = {
                        "document": result["document"],
                        "score": 0
                    }
                doc_scores[doc_id]["score"] += rrf_score

        # الترتيب حسب النقاط المجمعة
        ranked = sorted(
            doc_scores.values(),
            key=lambda x: x["score"],
            reverse=True
        )

        return ranked[:top_k]

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

بعد الاسترجاع الأولي، أعد الترتيب باستخدام مشفر متقاطع:

from sentence_transformers import CrossEncoder

class RerankedRetriever:
    def __init__(self, base_retriever, reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.retriever = base_retriever
        self.reranker = CrossEncoder(reranker_model)

    async def search(
        self,
        query: str,
        initial_k: int = 50,
        final_k: int = 10
    ) -> list:
        # الخطوة 1: الحصول على المرشحين الأوليين
        candidates = await self.retriever.search(query, top_k=initial_k)

        # الخطوة 2: تحضير الأزواج لإعادة الترتيب
        pairs = [(query, c["document"]) for c in candidates]

        # الخطوة 3: الحصول على نقاط إعادة الترتيب
        rerank_scores = self.reranker.predict(pairs)

        # الخطوة 4: الدمج مع النقاط الأصلية
        for i, candidate in enumerate(candidates):
            candidate["rerank_score"] = float(rerank_scores[i])
            # دمج موزون
            candidate["final_score"] = (
                0.3 * candidate["score"] +
                0.7 * candidate["rerank_score"]
            )

        # الخطوة 5: الترتيب حسب النقاط النهائية
        candidates.sort(key=lambda x: x["final_score"], reverse=True)

        return candidates[:final_k]

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

تحسين الاسترجاع بتوسيع الاستعلام الأصلي:

class QueryExpander:
    def __init__(self, llm):
        self.llm = llm

    async def expand(self, query: str) -> list:
        """توليد صياغات بديلة للاستعلام."""
        prompt = f"""أنشئ 3 طرق بديلة لطرح هذا السؤال.
أرجع الأسئلة فقط، سؤال واحد لكل سطر.

السؤال الأصلي: {query}

الأسئلة البديلة:"""

        response = await self.llm.complete(prompt)
        alternatives = response.strip().split("\n")

        return [query] + alternatives[:3]

# الاستخدام في الاسترجاع
async def search_with_expansion(query: str, retriever, expander) -> list:
    # توسيع الاستعلام
    expanded_queries = await expander.expand(query)

    # البحث مع جميع الاستعلامات
    all_results = []
    for q in expanded_queries:
        results = await retriever.search(q, top_k=5)
        all_results.extend(results)

    # إزالة التكرار والترتيب
    unique_results = deduplicate_by_id(all_results)
    return sorted(unique_results, key=lambda x: x["score"], reverse=True)[:10]

اختيار الاستراتيجية الصحيحة

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

بعد ذلك، سنتعلم كيفية توسيع أنظمة RAG للتعامل مع ملايين المستندات. :::