تنفيذ البحث الهجين

ادمج بحث BM25 بالكلمات المفتاحية مع بحث تشابه المتجهات باستخدام دمج الترتيب التبادلي (RRF) لنتائج مثلى.

نظرة عامة على البنية

                    الاستعلام
                      │
          ┌───────────┴───────────┐
          ▼                       ▼
    ┌──────────┐           ┌──────────┐
    │   BM25   │           │  بحث    │
    │  بحث    │           │ المتجهات │
    └────┬─────┘           └────┬─────┘
         │                      │
         │   النتائج + الترتيب  │
         └──────────┬───────────┘
                    ▼
            ┌──────────────┐
            │     RRF      │
            │     دمج      │
            └──────┬───────┘
                   ▼
            النتائج المدمجة

تنفيذ BM25

أولاً، أعد البحث بالكلمات المفتاحية مع BM25:

from rank_bm25 import BM25Okapi
import nltk
from nltk.tokenize import word_tokenize

nltk.download('punkt')

class BM25Retriever:
    def __init__(self, documents: list[str]):
        # تجزئة المستندات
        self.documents = documents
        self.tokenized_docs = [word_tokenize(doc.lower()) for doc in documents]

        # بناء فهرس BM25
        self.bm25 = BM25Okapi(self.tokenized_docs)

    def search(self, query: str, k: int = 10) -> list[tuple[int, float]]:
        """أرجع أزواج (فهرس_المستند، الدرجة)."""
        tokenized_query = word_tokenize(query.lower())
        scores = self.bm25.get_scores(tokenized_query)

        # احصل على أعلى k فهارس ودرجات
        top_indices = scores.argsort()[-k:][::-1]
        return [(idx, scores[idx]) for idx in top_indices]

تنفيذ بحث المتجهات

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

class VectorRetriever:
    def __init__(self, documents: list[str]):
        self.documents = documents
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

        # بناء فهرس المتجهات
        self.vectorstore = Chroma.from_texts(
            texts=documents,
            embedding=self.embeddings
        )

    def search(self, query: str, k: int = 10) -> list[tuple[int, float]]:
        """أرجع أزواج (فهرس_المستند، الدرجة)."""
        results = self.vectorstore.similarity_search_with_score(query, k=k)

        # اربط النتائج بالفهارس
        doc_to_idx = {doc: idx for idx, doc in enumerate(self.documents)}
        return [(doc_to_idx[r[0].page_content], r[1]) for r in results]

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

RRF يدمج الترتيبات دون الحاجة لتطبيع الدرجات:

def reciprocal_rank_fusion(
    rankings: list[list[tuple[int, float]]],
    k: int = 60
) -> list[tuple[int, float]]:
    """
    ادمج ترتيبات متعددة باستخدام RRF.

    المعاملات:
        rankings: قائمة ترتيبات، كل منها [(معرف_المستند، الدرجة), ...]
        k: ثابت RRF (افتراضي 60)

    يرجع:
        الترتيب المدمج [(معرف_المستند، درجة_rrf), ...]
    """
    rrf_scores = {}

    for ranking in rankings:
        for rank, (doc_id, _) in enumerate(ranking):
            if doc_id not in rrf_scores:
                rrf_scores[doc_id] = 0
            # صيغة RRF: 1 / (k + rank)
            rrf_scores[doc_id] += 1 / (k + rank + 1)

    # رتب حسب درجة RRF تنازلياً
    sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

لماذا يعمل RRF:

• مستقل عن الدرجات: لا حاجة لتطبيع أنظمة تسجيل مختلفة
• قائم على الموقع: المستندات المرتبة عالياً بأنظمة متعددة تُعزز
• الثابت k: يتحكم في مدى تفضيل المستندات الأعلى ترتيباً

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

class HybridRetriever:
    def __init__(self, documents: list[str]):
        self.documents = documents
        self.bm25 = BM25Retriever(documents)
        self.vector = VectorRetriever(documents)

    def search(
        self,
        query: str,
        k: int = 10,
        bm25_weight: float = 0.5,
        vector_weight: float = 0.5
    ) -> list[dict]:
        """
        البحث الهجين يجمع BM25 وبحث المتجهات.

        المعاملات:
            query: استعلام البحث
            k: عدد النتائج للإرجاع
            bm25_weight: وزن نتائج BM25
            vector_weight: وزن نتائج المتجهات

        يرجع:
            قائمة مستندات مع الدرجات
        """
        # احصل على النتائج من كلا النظامين
        bm25_results = self.bm25.search(query, k=k*2)
        vector_results = self.vector.search(query, k=k*2)

        # طبق دمج RRF
        fused = reciprocal_rank_fusion([bm25_results, vector_results])

        # أرجع أعلى k مع محتوى المستند
        results = []
        for doc_id, score in fused[:k]:
            results.append({
                "content": self.documents[doc_id],
                "score": score,
                "doc_id": doc_id
            })

        return results

تنفيذ LangChain

باستخدام مسترجع المجموعة المدمج في LangChain:

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

# إنشاء المسترجعات
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# ادمج مع المجموعة
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]  # وزن متساوٍ
)

# البحث
results = ensemble_retriever.get_relevant_documents("OAuth authentication")

البحث الهجين الأصلي لقاعدة البيانات

العديد من قواعد البيانات المتجهة تدعم البحث الهجين أصلياً:

Qdrant

from qdrant_client import QdrantClient
from qdrant_client.models import models

client = QdrantClient("localhost", port=6333)

# البحث الهجين مع دمج مدمج
results = client.query_points(
    collection_name="documents",
    query=query_embedding,
    using="dense",  # حقل المتجهات
    with_payload=True,
    limit=10,
    query_filter=None,
    # تمكين الهجين مع المتجهات المتفرقة
    with_vectors=False,
).points

# أو استخدم متجهات Qdrant المتفرقة لبحث شبيه بـ BM25

Weaviate

import weaviate

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

# البحث الهجين
results = collection.query.hybrid(
    query="OAuth authentication",
    alpha=0.5,  # 0 = كلمات مفتاحية فقط، 1 = متجهات فقط، 0.5 = متوازن
    limit=10
)

Pinecone

from pinecone import Pinecone

pc = Pinecone(api_key="your-key")
index = pc.Index("hybrid-index")

# Pinecone الهجين مع متجهات كثيفة-متفرقة
results = index.query(
    vector=dense_embedding,
    sparse_vector={
        "indices": sparse_indices,
        "values": sparse_values
    },
    top_k=10,
    include_metadata=True
)

ضبط الأوزان الهجينة

def find_optimal_weights(
    queries: list[str],
    ground_truth: list[list[int]],
    retriever: HybridRetriever
) -> tuple[float, float]:
    """ابحث عن أوزان BM25/المتجهات المثلى."""
    best_weights = (0.5, 0.5)
    best_score = 0

    for bm25_w in [0.3, 0.4, 0.5, 0.6, 0.7]:
        vector_w = 1 - bm25_w

        total_recall = 0
        for query, truth in zip(queries, ground_truth):
            results = retriever.search(
                query,
                k=10,
                bm25_weight=bm25_w,
                vector_weight=vector_w
            )
            retrieved_ids = [r["doc_id"] for r in results]
            recall = len(set(retrieved_ids) & set(truth)) / len(truth)
            total_recall += recall

        avg_recall = total_recall / len(queries)
        if avg_recall > best_score:
            best_score = avg_recall
            best_weights = (bm25_w, vector_w)

    return best_weights

نصيحة التنفيذ: ابدأ بأوزان متساوية (0.5/0.5)، ثم اضبط بناءً على أنماط استعلاماتك. المجالات التقنية غالباً تستفيد من وزن BM25 أعلى؛ المجالات المفاهيمية تفضل وزن المتجهات.

التالي، لنستكشف استراتيجيات إعادة الترتيب لتحسين جودة النتائج أكثر. :::