יישום חיפוש היברידי (וקטורי + טקסט מלא) עבור RAG

from collections import defaultdict

def reciprocal_rank_fusion(
    dense_results: list[tuple],  # [(doc_id, score), ...]
    sparse_results: list[tuple],
    k: int = 60  # RRF константа (обычно 60)
) -> list[tuple]:
    """
    RRF score = sum(1 / (k + rank_i)) по всем спискам
    k=60 стандартное значение (Cormack et al.)
    """
    scores = defaultdict(float)
    for rank, (doc_id, _) in enumerate(dense_results, 1):
        scores[doc_id] += 1 / (k + rank)
    for rank, (doc_id, _) in enumerate(sparse_results, 1):
        scores[doc_id] += 1 / (k + rank)
    return sorted(scores.items(), key=lambda x: -x[1])

Relative Score Fusion (RSF)—שילוב מנורמל:

def relative_score_fusion(
    dense_results: list[tuple],
    sparse_results: list[tuple],
    alpha: float = 0.5  # Вес dense
) -> list[tuple]:
    """Нормализует оценки в [0,1] и взвешивает"""
    scores = defaultdict(float)

    # Нормализация dense
    if dense_results:
        max_d = max(s for _, s in dense_results)
        min_d = min(s for _, s in dense_results)
        for doc_id, score in dense_results:
            norm = (score - min_d) / (max_d - min_d + 1e-8)
            scores[doc_id] += alpha * norm

    # Нормализация sparse
    if sparse_results:
        max_s = max(s for _, s in sparse_results)
        min_s = min(s for _, s in sparse_results)
        for doc_id, score in sparse_results:
            norm = (score - min_s) / (max_s - min_s + 1e-8)
            scores[doc_id] += (1 - alpha) * norm

    return sorted(scores.items(), key=lambda x: -x[1])

השוואת אלגוריתמי מיזוג

SPLADE: מקודד ספארס מתקדם

SPLADE (Sparse Lexical and Expansion Model) מייצר וקטורים ספארסיים עם הרחבה לקסיקלית—המודל לומד "להרחיב" את השאילתה עם מילים נרדפות ומונחים קשורים. לפי מדד BEIR, SPLADE עולה על BM25 פי 1.2–1.5 ב-NDCG@10.

from fastembed import SparseTextEmbedding

sparse_model = SparseTextEmbedding(
    model_name="prithivida/Splade_PP_en_v1"
)


def encode_sparse(text: str) -> dict:
    """Возвращает sparse вектор {token_id: weight}"""
    output = list(sparse_model.embed([text]))[0]
    return {
        "indices": output.indices.tolist(),
        "values": output.values.tolist(),
    }

SPLADE עולה על BM25 ברוב מדדי BEIR. עבור רוסית, אנו ממליצים על המודל from collections import defaultdict def reciprocal_rank_fusion( dense_results: list[tuple], # [(doc_id, score), ...] sparse_results: list[tuple], k: int = 60 # RRF константа (обычно 60) ) -> list[tuple]: """ RRF score = sum(1 / (k + rank_i)) по всем спискам k=60 стандартное значение (Cormack et al.) """ scores = defaultdict(float) for rank, (doc_id, _) in enumerate(dense_results, 1): scores[doc_id] += 1 / (k + rank) for rank, (doc_id, _) in enumerate(sparse_results, 1): scores[doc_id] += 1 / (k + rank) return sorted(scores.items(), key=lambda x: -x[1]) או גרסאות רב-לשוניות.

יישום עם Qdrant (דוגמה מעשית)

from qdrant_client import QdrantClient
from qdrant_client.models import (
    SparseVector,
    Prefetch,
    FusionQuery,
    Fusion,
    NamedVector,
    NamedSparseVector,
)
from fastembed import TextEmbedding, SparseTextEmbedding

dense_model = TextEmbedding("BAAI/bge-m3")  # Multilingual dense
sparse_model = SparseTextEmbedding("prithivida/Splade_PP_en_v1")

client = QdrantClient(url="http://localhost:6333")


def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
    # Dense embedding
    dense_vec = list(dense_model.embed([query]))[0].tolist()

    # Sparse embedding
    sparse_output = list(sparse_model.embed([query]))[0]
    sparse_vec = SparseVector(
        indices=sparse_output.indices.tolist(),
        values=sparse_output.values.tolist(),
    )

    results = client.query_points(
        collection_name="hybrid_docs",
        prefetch=[
            Prefetch(query=dense_vec, using="dense", limit=50),
            Prefetch(query=sparse_vec, using="sparse", limit=50),
        ],
        query=FusionQuery(fusion=Fusion.RRF),
        limit=top_k,
        with_payload=True,
    )

    return [
        {"text": r.payload["text"], "source": r.payload["source"], "score": r.score}
        for r in results.points
    ]

מקרה מעשי: השפעת alpha על איכות האחזור

מהניסיון שלנו: בפרויקט עם 12,000 מסמכים במאגר ידע ארגוני (חוזים, תקנות, שאלות נפוצות), בדקנו 400 שאילתות מסוגים שונים. תוצאות:

חיפוש היברידי RRF ללא reranker כבר עולה על dense+reranker. השילוב של חיפוש היברידי+reranker מניב את התוצאה הטובה ביותר. לשם השוואה, שימוש ב-SPLADE כמקודד ספארס נותן שיפור MRR@5 של כ-0.03–0.05 לעומת BM25 עם אותה שיטת מיזוג.

כיצד לכוונן מיזוג RRF על מערך הנתונים שלך?

k אופטימלי עבור RRF: k=60 הוא ערך חזק אמפירית. k קטן מדי (10–20) נותן משקל גדול למיקומים הראשונים. גדול מדי (100+) מאזן את ההבדלים בין מיקומים. על נתונים אמיתיים, בדוק k∈{20, 40, 60, 80} על קבוצת אימות. עבור RSF, כוונן alpha מ-0.3 עד 0.7 בצעדים של 0.1.

תהליך יישום חיפוש היברידי שלב אחר שלב

1. ביקורת על תכנית האחזור הנוכחית: ניתוח ה-embeddings שבשימוש, מחסן הווקטורים (vector DB), ומדדי איכות.
2. בחירה והגדרה של מקודד ספארס: התקנת SPLADE או מקודד ספארס אחר המתאים לשפה ולתחום שלך.
3. שילוב חיפוש כפול: הגדרת אינדוקס של וקטורים צפופים וספארסיים ב-Qdrant/Pinecone/Weaviate.
4. יישום מיזוג: פריסת RRF או RSF עם פרמטרים ראשוניים (k=60, alpha=0.5).
5. בדיקה ואופטימיזציה: הרצת השאילתות שלך, כוונון פרמטרים באמצעות מדדי MRR/NDCG.
6. תיעוד ומסירה: תיאור התהליך, הדרכת הצוות, מסירת קוד וקבצי הגדרות.

מה כלול בפרויקט

• קוד אינטגרציה לחיפוש היברידי לתוך מערכת ה-RAG שלך.
• קבצי הגדרות עבור Qdrant/Pinecone.
• תיעוד על התקנה ותפעול.
• הדרכת צוות (סמינר מקוון של שעתיים).
• אחריות לאיכות אחזור (מדדים נמדדים לפני/אחרי).
• תמיכה חודש לאחר סיום הפרויקט.

צור קשר לקבלת הערכת פרויקט. קבל ייעוץ על יישום חיפוש היברידי ושפר את איכות האחזור של מערכת ה-RAG שלך.