אתמול ראינו איך לבצע חיפוש טקסט מלא ואת האתגרים שלו. היום נמשיך עם RAG והפעם נבנה מנוע חיפוש וקטורי ונראה באיזה מקרים מנוע כזה יכול לחזק את התוצאות.

1. מחיפוש טקסט מלא לחיפוש וקטורי

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

אחד הרעיונות הראשונים שאנשים העלו כשהתחילו לשלב חיפושים ומודלי שפה היה להשתמש במודל השפה כדי לחפש בערימה גדולה של טקסט, הרעיון הוא כזה:

1. כמו שמאמנים מודל שפה לגלות המשך טבעי לשיחה, נוכל לאמן מודל שפה לגלות קשר סמנטי בין מילים ומשפטים.

2. משפטים שיש ביניהם קשר סמנטי, כלומר הם מופיעים הרבה פעמים יחד קרוב באותם טקסטים יקודדו לוקטורים שהמרחק ביניהם קטן. מילים ומשפטים שאין ביניהם קשר או שיש קשר חלש יקודדו לוקטורים שהמרחק ביניהם גדול.

3. המערכת שלנו תיקח טקסט גדול בו אנחנו רוצים לחפש, תחלק אותו למשפטים או קטעים קצרים ותקודד כל קטע קצר לוקטור. כשמשתמש יגיע עם שאלה נקודד גם את השאלה של המשתמש לוקטור ואז נמצא במאגר המידע שלנו את הוקטורים שהכי קרובים לשאלת המשתמש. אלה המשפטים שיש להם את הקשר הסמנטי הכי חזק לשאלת המשתמש.

הרעיון הזה אולי נשמע הגיוני אבל יש בו המון בעיות. בתור התחלה לא ברור איך לחלק טקסט ארוך למשפטים קצרים כדי לקודד אותם לוקטורים, לא ברור למה שהשאלה תהיה קשורה סמנטית דווקא למשפט מסוים או לקבוצת משפטים, ומי בכלל מאמן את המודל הסמנטי ולפי איזה קורפוס טקסטואלי. יש המון שיטות להתמודד עם האתגרים האלה אבל כמו עם חיפוש טקסט מלא אנחנו לא מדברים כאן על נוסחת קסם לחיפוש מדויק אלא על אוסף של טכניקות שאולי יעבדו ואולי לא יעבדו ובדרך כלל מערכות אמיתיות ישלבו בין חיפוש וקטורי, חיפוש טקסט מלא ואינדקס חיפוש על המידע כדי באמת להצליח לענות לשאלות של משתמשים.

2. מה אנחנו בונים

בדוגמה היום אנחנו רוצים לראות איך עובד חיפוש וקטורי בלי יותר מדי ליפול בבורות של מערכת כזו ולכן לקחתי את החיפוש הסמנטי הכי קל שמצאתי - חיפוש בטקסט ארוך שלעולם לא משתנה ויש לו חלוקה טבעית לפסוקים, זהו התנ"ך. בדוגמה היום נכתוב מערכת שלוקחת שאלה של משתמש ומחפשת בתנ"ך קטעים בעלי דמיון סמנטי לשאלה ואז מעבירה את השאלה יחד עם הקטעים שכנראה רלוונטים למודל שפה כדי לקבל תשובה לשאלה.

המערכת עובדת באופן הבא:

1. תחילה עוברים על כל התנ"ך ומקודדים כל קבוצה של 6 פסוקים לוקטור. בין הקבוצות אני שומר חפיפה של שני פסוקים כדי שמידע סמנטי לא ילך לאיבוד במעבר בין קבוצות. את הוקטורים שומרים בבסיס נתונים Postgres עם הרחבה בשם pgvector שמאפשרת שמירת וקטורים והשוואה ביניהם. זה תהליך חד פעמי שקורה ביצירת המערכת.

2. כשהמערכת באוויר לוקחים שאלה מהמשתמשים, מקודדים גם אותה לוקטור ומחפשים 5 וקטורים קרובים לה. מושכים מבסיס הנתונים את הטקסטים שמתאימים לוקטורים אלה ושולחים הכל לג'מיני.

קוד הדוגמה המלא זמין בתיקיית הדוגמאות בקישור:

https://github.com/ynonp/pydanticai-demos/tree/main/10-bible-vectordb

המודל עצמו נקרא BAAI/bge-m3 ואפשר להתקין אותו כספריית פייתון מתוך Hugging Face. זה מודל סמנטי בעברית ולכן יתאים לטקסט שלנו. אחרי שיש לי את המודל מותקן קידוד של טקסט לוקטור הוא בסך הכל קריאת פונקציה בפייתון:

def embed_texts(texts: list[str]) -> list[list[float]]:
    """Embed a batch of texts, returning one 768-dim vector per text."""
    embeddings = get_model().encode(texts, show_progress_bar=True)
    return [vec.tolist() for vec in embeddings]

3. קידוד התנך לוקטורים

הקידוד לוקטורים ממומש בקובץ index.py.

הפונקציה המעניינת בקובץ נקראת build_chunks וזה הקוד שלה:

def build_chunks(
    verses: list[tuple[str, int, int, str]],
    chunk_size: int,
    overlap: int,
) -> list[dict]:
    """Group verses per (book, chapter) and split into overlapping windows."""
    if chunk_size <= overlap:
        raise ValueError(f"CHUNK_SIZE ({chunk_size}) must be greater than OVERLAP ({overlap})")
    step = chunk_size - overlap

    # Preserve file order while grouping by (book, chapter).
    chapters: dict[tuple[str, int], list[tuple[int, str]]] = {}
    order: list[tuple[str, int]] = []
    for book, chapter, verse, text in verses:
        key = (book, chapter)
        if key not in chapters:
            chapters[key] = []
            order.append(key)
        chapters[key].append((verse, text))

    chunks: list[dict] = []
    for book, chapter in order:
        entries = chapters[(book, chapter)]
        for start in range(0, len(entries), step):
            window = entries[start : start + chunk_size]
            if not window:
                continue
            first_verse = window[0][0]
            last_verse = window[-1][0]
            verses_range = str(first_verse) if first_verse == last_verse else f"{first_verse}-{last_verse}"
            chunk_text = " ".join(text for _, text in window)
            chunks.append(
                {
                    "book": book,
                    "chapter": chapter,
                    "verses_range": verses_range,
                    "chunk_text": chunk_text,
                }
            )
            # Stop once the window has consumed the tail of the chapter.
            if start + chunk_size >= len(entries):
                break
    return chunks

הפונקציה רצה על הטקסט ואוספת קבוצות של 6 פסוקים מכל קבוצה יוצרת chunk. כל chunk כזה מחזיק את הספר ממנו הוא נלקח, הפרק, איזה פסוקים הוא מכיל ואת הטקסט המלא.

בהמשך אני מריץ את הפקודה:

embeddings = embed_texts([c["chunk_text"] for c in chunks])

כדי ליצור את כל הוקטורים ואז את הלולאה הבאה כדי לשמור הכל בבסיס הנתונים:

    rows = [
        (
            chunk["book"],
            chunk["chapter"],
            chunk["verses_range"],
            chunk["chunk_text"],
            embedding,
        )
        for chunk, embedding in zip(chunks, embeddings)
    ]
    with conn.cursor() as cur:
        cur.execute(f"TRUNCATE {TABLE} RESTART IDENTITY")
        cur.executemany(
            f"INSERT INTO {TABLE} (book, chapter, verses_range, chunk_text, embedding) "
            "VALUES (%s, %s, %s, %s, %s)",
            rows,
        )
    conn.commit()

4. חיפוש וקטורי

החיפוש מבוצע בקובץ app.py בפונקציה inject_rag_context. זה הקוד הרלוונטי:

embedding = embed_query(query)
with ctx.deps.conn.cursor() as cur:
    cur.execute(
        f"""
        SELECT book, chapter, verses_range, chunk_text
        FROM {TABLE}
        ORDER BY embedding <=> %s
        LIMIT %s
        """,
        (embedding, ctx.deps.top_k),
    )
    hits = cur.fetchall()

את התוצאה שולחים לסוכן וזה כל הסיפור.

5. עכשיו אתם

1. הריצו את המערכת אצלכם. שאלו שאלות ובדקו האם הסוכן מצא את הקטעים הרלוונטים וענה תשובות מדויקות.

2. מצאו שאלות שהסוכן מצליח לפתור ושאלות אחרות שהוא לא מצליח. על כל שאלה שהוא לא הצליח נסו לחשוב איפה היתה הבעיה.

3. בדקו מודלים אחרים של Embedding. האם יש הבדל בתוצאות? נסו למצוא מודל טוב יותר מזה שאני מצאתי.

4. שלבו חיפוש טקסט מלא במערכת כך שהסוכן יקבל גם את הקטעים שקרובים סמנטית וגם 5 קטעים שמכילים הכי הרבה מילים משותפות עם השאלה. שימו לב איך זה משנה את התוצאות.