📌 יום 9 - חיפוש בבלוג באמצעות RAG וחיפוש טקסט מלא

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

✏ אתגר הקונטקסט

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

למנגנון הזה בדיוק אנחנו קוראים RAG שזה ראשי תיבות של Retrieval-augmented generation. מילת המפתח של RAG היא Retrieval, המערכת מושכת את המידע הרלוונטי, מוסיפה אותו לחלון הקונטקסט (זה ה Augmented) ומפעילה את המודל כדי לייצר תשובה.

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

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

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

1. סקריפט שמוריד את הפוסטים מהבלוג ומאנדקס אותם לבסיס נתונים המיועד לחיפוש בשם Meilisearch.
2. קוד פייתון שמקבל שאלת משתמש ומחפש ב Meilisearch איזה פוסטים עשויים להיות רלוונטים לשאלה זו (איזה פוסטים מכילים מילים מהשאלה).
3. קוד סוכן שמקבל פוסטים מהבלוג ושאלת משתמש ועונה על השאלה לפי מה שכתוב בפוסטים.

שלושת החלקים מהווים מערכת RAG וילמדו אותנו על האתגרים בבניית מערכת כזו.

✏ יצירת המידע

שלב ראשון במערכת הוא יצירת מאגר המידע. בדוגמה שלנו נתחיל עם הסקריפט download-data.py שמוריד את כל הפוסטים מהאתר לקבצי markdown מקומיים. זה קל כי אני ממילא מפרסם את הפוסטים גם ב HTML וגם ב Markdown ולכן הקוד צריך לרוץ על דף או דפי אינדקס מהבלוג, לאסוף את הקישורים לפוסטים ואז להוריד אותם בלי להעמיס על השרת. זאת הפונקציה המרכזית:

async def run(count: int) -> None:
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

    async with httpx.AsyncClient(
        headers={"User-Agent": USER_AGENT},
        follow_redirects=True,
        timeout=30.0,
    ) as client:
        slugs = await collect_slugs(client, count, semaphore)
        if len(slugs) < count:
            print(f"warning: only {len(slugs)} posts found (requested {count})")

        # The slug list is the download queue; the semaphore enforces at most
        # MAX_CONCURRENCY concurrent requests.
        await asyncio.gather(
            *[download_post(client, slug, semaphore) for slug in slugs]
        )

    print(f"done: downloaded {len(slugs)} posts to {DATA_DIR}/")

הסקריפט לוקח מהמשתמש מספר פוסטים להוריד ושומר את כולם בתיקיית data.

✏ שמירת הפוסטים בבסיס נתונים לחיפוש

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

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

def build_documents() -> list[dict[str, str]]:
    """Read every ``.md`` file in ``data`` and build Meilisearch documents."""
    docs: list[dict[str, str]] = []
    for path in sorted(DATA_DIR.glob("*.md")):
        content = path.read_text(encoding="utf-8")
        docs.append(
            {
                "id": path.stem,
                "filename": path.name,
                "title": _extract_title(content),

                "content": content,
            }
        )
    return docs

ובהמשך אנחנו קוראים לפונקציה של מיילי שנקראת add_documents כדי לשמור את המסמכים בבסיס הנתונים:

docs = build_documents()
task = index.add_documents(docs)

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

HEBREW_SYNONYMS: dict[str, list[str]] = {
    # Singular ↔ plural
    "טיפ": ["טיפים"],
    "כלי": ["כלים"],
    "שגיאה": ["שגיאות"],
    "באג": ["באגים"],
    "קוד": ["קודים"],
    "פיצ׳ר": ["פיצ׳רים"],
    "מודל": ["מודלים"],
    "סוכן": ["סוכנים"],
    "פרויקט": ["פרויקטים"],
    "שאלה": ["שאלות"],
    "תשובה": ["תשובות"],
    "פתרון": ["פתרונות"],
    "בעיה": ["בעיות"],
    "דוגמה": ["דוגמאות"],
    "קובץ": ["קבצים"],
    "שפה": ["שפות"],
    "כתבה": ["כתבות"],
    "פוסט": ["פוסטים"],
    # Common abbreviations / shortcuts
    "בינה מלאכותית": ["AI", "ai", "ב״מ"],
    "למידת מכונה": ["ML", "ml", "machine learning"],
    "עיבוד שפה טבעית": ["NLP", "nlp"],
    "ביג דאטה": ["big data"],
    # English terms that appear in Hebrew text
    "prompt": ["פרומפט", "פרומפטים", "הנדסת פרומפטים"],
    "agent": ["אייג׳נט", "אייג׳נטים"],
    "token": ["טוקן", "טוקנים"],
    "API": ["api", "ממשק"],
    "framework": ["פריימוורק", "פריימוורקים"],
    "library": ["ספרייה", "ספריות"],
    "debugging": ["דיבוג", "ניפוי שגיאות"],
    "refactoring": ["ריפקטור", "ריפקטורינג", "שכתוב קוד"],
    "testing": ["בדיקות", "טסטים", "טסט"],
    "code review": ["קוד ריוויו", "סקירת קוד"],
    "open source": ["קוד פתוח", "open-source"],
    "CLI": ["cli", "שורת פקודה"],
    "LLM": ["llm", "מודל שפה גדול", "מודלי שפה"],
    "RAG": ["rag"],
    "MCP": ["mcp"],
    "VSCode": ["vs code", "ויזואל סטודיו קוד", "vs-code"],
    "Git": ["git", "גיט"],
    "GitHub": ["github", "גיטהאב"],
    "Copilot": ["copilot", "קופיילוט"],
    "Claude": ["claude", "קלוד"],
    "Docker": ["docker", "דוקר"],
    "Python": ["python", "פייתון"],
    "JavaScript": ["javascript", "js", "ג׳אווה סקריפט"],
    "TypeScript": ["typescript", "ts"],
    "Ruby": ["ruby", "רובי"],
    "Rails": ["rails", "ריילס"],
    "React": ["react", "ריאקט"],
    "Vue": ["vue", "ויו"],
}

✏ חיפוש ושילוב התוצאות

ראינו בדוגמאות קודמות איך להגדיר לסוכן דף הוראות סטטי - כלומר טקסט של instructions שיסביר לסוכן מה צריך לעשות. בעבודה RAG דף ההוראות נבנה בצורה דינמית לפי השאילתה ולכן אנחנו משתמשים במנגנון נוסף של Pydantic AI Agents שנקרא Dynamic Instructions. מנגנון זה מאפשר לנו לשמור פונקציה בתור תוספת להוראות, פידנטיק יפעיל את הפונקציה ויוסיף את ערך ההחזר שלה לדף ההוראות. הפונקציה רצה בדיוק לפני שליחת ההודעה למודל ויש לה גישה לפרומפט ולתלויות של אותה בקשה.

נקרא את הקוד הבא מתוך הקובץ app.py:

@agent.instructions
async def inject_rag_context(ctx: RunContext[Deps]) -> str:
    """Search Meilisearch and attach the top matching posts to the prompt.

    The filenames are also stored in ``ctx.metadata['rag_sources']`` so the
    output validator can prepend them to the final answer.
    """
    query = _extract_query(ctx.prompt)
    if not query:
        return ""

    if ctx.metadata is None:
        ctx.metadata = {}

    results = ctx.deps.meili.index(INDEX_NAME).search(query, {"limit": ctx.deps.top_k})
    hits = results.get("hits", [])
    filenames = [hit["filename"] for hit in hits]
    ctx.metadata["rag_sources"] = filenames

    if not hits:
        return (
            "No relevant blog posts were found for this question. "
            "Answer based on your general knowledge, but mention that no posts matched."
        )

    sections = [f"--- {hit['filename']} ---\n{hit['content']}" for hit in hits]
    return (
        f"Relevant blog posts found by Meilisearch: {', '.join(filenames)}. "
        "Use the content below to answer the user's question. "
        "Start your response by listing the filenames of the posts you used, "

        "for example 'Based on: file1.md, file2.md'. Then answer the question.\n\n"
        + "\n\n".join(sections)
    )

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

✏ עכשיו אתם

בשביל להריץ את הדוגמה תצטרכו הפעם את דוקר מאחר ויש לנו גם את בסיס הנתונים meilisearch וגם את האפליקציה. צירפתי בתיקיית הדוגמה קובץ docker-compose.yml בו אפשר להשתמש.

1. הריצו את הדוגמה אצלכם אחרי הגדרת מפתח הגישה לג'מיני ב .env. שאלו את הסוכן שאלות ושימו לב באיזה פוסטים הוא משתמש כדי לענות.
2. חשבו: איך היינו מתחזקים את האינדקס במערכת פרודקשן? מה קורה אם תוכן פוסט מתעדכן? מה אם פוסט נמחק?
3. חשבו: מה קורה אם יש פוסט ארוך במיוחד - האם תמיד צריך לכלול את כל הפוסט כדי לענות על שאלת המשתמש? האם יהיה מספיק למודל להסתכל על פסקה בודדת?
4. החליפו את מנגנון ה RAG בגישה מבוססת כלים - תנו לסוכן כלי לחיפוש ב meilisearch ושאלו את אותן שאלות. האם קיבלתם תשובות טובות יותר? טובות פחות?
5. החליפו את מנגנון ה RAG בגישה מרובת סוכנים - בנו סוכן חיפוש שתפקידו לחפש ב meilisearch קטעים רלוונטים לשאלה (בעזרת כלי החיפוש). הפעילו את סוכן החיפוש במקום לבנות את השאילתה בעצמכם ואת התוצאות העבירו לסוכן השיחה. איך זה עבד?
6. בתיקיית docs של פידנטיק AI תוכלו למצוא את התיעוד של הספריה. כתבו סוכן שמחפש בתיעוד הספריה ועונה על שאלות לגביה.

📌 יום 10 - חיפוש בעזרת Vector DB

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

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

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

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

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

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

✏ מה אנחנו בונים

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

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

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]

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

הקידוד לוקטורים ממומש בקובץ 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()

✏ חיפוש וקטורי

החיפוש מבוצע בקובץ 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()

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

✏ עכשיו אתם

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

📌 יום 11 - סידור סרטים בקורס

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

✏ מה אנחנו בונים

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

1. חותכים את ההקלטה לקטעים של 30 דקות כדי שיהיה ל AI קל להתמודד איתם.
2. אנחנו לא יודעים עדיין איפה הכי הגיוני לסמן "שיעור" לכן נדאג לחפיפה של 5 דקות בין הקטעים הקצרים וכך שיעור לא ייפול בדיוק באמצע ביניהם.
3. נעלה כל קטע ל Gemini דרך Google File API.
4. נבקש מהסוכן שיעבור על כל קטע וייתן לי את ה Timestamps בהם הכי הגיוני לחתוך את הקטע לשיעורים.
5. נמזג ונחתוך את ההקלטה לפי הזמנים שקיבלנו מהסוכן ונצרף את הסיכומים.

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

https://github.com/ynonp/pydanticai-demos/blob/main/11-course-builder/build-course.py

נעבור יחד על החלקים המעניינים.

✏ מבנה הפלט

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

class Lesson(BaseModel):
    index: int = Field(description="1-based lesson number within this chunk")

קלאס השיעור המלא כולל כל ההסברים הוא הקלאס הראשון בקובץ:

class Lesson(BaseModel):
    """A single lesson extracted from the course video."""

    index: int = Field(description="1-based lesson number within this chunk")
    start_timestamp: str = Field(
        description="Start time relative to THIS CHUNK as HH:MM:SS, e.g. '00:02:30'"
    )
    end_timestamp: str = Field(
        description="End time relative to THIS CHUNK as HH:MM:SS, e.g. '00:14:45'"
    )
    is_break: bool = Field(
        default=False,
        description=(
            "True if this segment is a BREAK with no teaching content — e.g. "
            "students chatting, silence, instructor away, coffee/lunch break. "
            "False for a normal lesson."
        ),
    )
    slug: str = Field(
        description=(
            "URL-safe slug for the lesson, e.g. 'intro-to-pydantic'. "
            "If is_break is True, this is ignored (the slug 'breaktime' is "
            "applied automatically) — still provide a placeholder value."
        )
    )
    title: str = Field(description="Lesson title in the original language")
    summary_markdown_hebrew: str = Field(
        description=(
            "A full, self-contained lesson write-up in Hebrew markdown, written so "
            "that someone who never watched the video can read it and actually "
            "learn the material — NOT a table of contents or a bullet-point index "
            "of what was covered. Write it like an article/tutorial: "
            "Structure it as numbered sections ('### 1. <topic title>', "
            "'### 2. <topic title>', ...), one per sub-topic covered in the "
            "lesson, in the order they were taught. Under each heading, write "
            "full explanatory paragraphs in flowing Hebrew prose that actually "
            "teach the concept (what it is, why it matters, how it works) the "
            "way the instructor explained it — not short summaries or fragments. "
            "Include exact code shown in the video in fenced code blocks, "
            "including commands, filenames, and terminal output where relevant, "
            "each with a sentence or two explaining what the code does and why. "
            "Include any links or external references mentioned, and end with "
            "practical conclusions/recommendations if the instructor gave any. "

            "Be thorough, detailed, and written in the same didactic style as a "
            "professional programming course text (not a summary/recap). "
            "If is_break is True, skip all of this and just write a brief note "
            "that this was a break (e.g. 'הפסקה — אין תוכן לימודי בקטע זה.')."
        )
    )


class ChunkOutline(BaseModel):
    """Lessons found within a single video chunk."""

    lessons: list[Lesson] = Field(description="Ordered list of lessons in this chunk")

✏ העלאת הקבצים לגוגל

לגוגל יש Google File API שמאפשר לנו לשמור קבצים כדי שג'מיני יוכל לקרוא אותם. המנגנון מותאם לסוכנים ולא דורש מאתנו למחוק את הקבצים, גוגל ימחקו אותם אוטומטית אחרי כמה שעות. מנגנונים דומים קיימים גם ב Claude וגם ב OpenAI. הפונקציה הבאה מעלה קובץ ושומרת את המזהה שלו לצורך העברה לסוכן בהמשך התוכנית:

def upload_chunk(chunk_path: str, chunk_index: int) -> UploadedFile:
    """Upload a single chunk to Google File API."""
    print(f"   📤 Uploading chunk {chunk_index} ({Path(chunk_path).name})...")
    client = genai.Client()
    t0 = time.time()

    uploaded = client.files.upload(
        file=chunk_path,
        config={"display_name": Path(chunk_path).name},
    )

    elapsed = time.time() - t0
    print(f"      ↑ {elapsed:.0f}s, state={uploaded.state.name}")

    while uploaded.state.name != "ACTIVE":
        time.sleep(5)
        uploaded = client.files.get(name=uploaded.name)

    print(f"      ✅ ACTIVE")
    return UploadedFile(
        file_id=uploaded.uri,
        provider_name="google",
        media_type=uploaded.mime_type or "video/mp4",
    )

אחרי העלאה Google File API צריך זמן לעבד את הקובץ וזו הסיבה ללולאת ההמתנה שאנחנו רואים שמחכה שהקובץ יהיה מוכן.

✏ פיענוח השיעורים

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

def build_agent() -> Agent:
    """Create the course-builder agent with Google Gemini."""
    model = GoogleModel("gemini-3-flash-preview")
    return Agent(
        model,
        output_type=ChunkOutline,
        system_prompt=(
            "You are an expert course builder and video content analyst. "
            "You receive a SEGMENT (chunk) of a longer course recording and must "
            "identify complete, self-contained lessons of 10-15 minutes each "
            "within this segment.\n\n"
            "CRITICAL RULES:\n"
            "- Timestamps MUST be relative to THIS CHUNK (00:00:00 = chunk start)\n"
            "- Only include lessons that are FULLY or MOSTLY contained in this chunk\n"
            "- If a lesson is cut off at the start or end, do NOT include it — "
            "the overlap with adjacent chunks will capture it\n"
            "- Each lesson should be 10-15 minutes of coherent content\n"
            "- Identify natural topic boundaries\n"
            "- Create descriptive, URL-safe English slugs (lowercase, hyphens)\n"
            "- Write comprehensive Hebrew markdown summaries including:\n"
            "  * Sub-topics covered\n"
            "  * ALL code snippets shown (in ``` code blocks)\n"
            "  * Links or references mentioned\n"
            "  * Key takeaways\n\n"
            "BREAK-TIME DETECTION:\n"
            "- Also detect BREAKS: segments with no teaching content, such as "
            "coffee/lunch breaks, silence, students chatting among themselves, "
            "the instructor stepping away, or any other non-lesson downtime.\n"
            "- Treat a break exactly like a lesson entry — give it a start/end "
            "timestamp — but set is_break=True and give it a short title such "
            "as 'הפסקה'. The slug field is ignored for breaks, just put any "
            "placeholder.\n"
            "- For a break's summary_markdown_hebrew, just write a brief note "
            "that this was a break (e.g. 'הפסקה — אין תוכן לימודי בקטע זה.'), "
            "no need for a full lesson write-up.\n"

            "- Only mark genuine downtime as a break — do not use it for slow "
            "or informal but still on-topic teaching."
        ),
    )


def analyze_chunk(
    agent: Agent, chunk: dict, chunk_index: int, total_chunks: int
) -> list[Lesson]:
    """
    Analyze one chunk → list of lessons with timestamps converted to
    original video time.
    """
    offset = chunk["offset_seconds"]
    video_file = upload_chunk(chunk["path"], chunk_index)

    print(f"   🤖 Analyzing chunk {chunk_index}/{total_chunks - 1} "
          f"(offset={seconds_to_ts(offset)})...")
    t0 = time.time()

    result = agent.run_sync(
        [
            (
                f"This is chunk {chunk_index} of a longer course video. "
                f"The chunk starts at {seconds_to_ts(offset)} in the original video. "
                f"Identify all complete 10-15 minute lessons within this chunk. "
                f"Timestamps must be relative to THIS CHUNK (00:00:00 = chunk start). "
                f"Do NOT include lessons that are cut off at chunk boundaries — "
                f"adjacent overlapping chunks will capture them."
            ),
            video_file,
        ]
    )

    elapsed = time.time() - t0
    chunk_lessons = result.output.lessons
    print(f"      ✅ {len(chunk_lessons)} lessons found in {elapsed:.0f}s")

    # Convert chunk-relative timestamps to original video timestamps
    converted = []
    for lesson in chunk_lessons:
        rel_start = ts_to_seconds(lesson.start_timestamp)
        rel_end = ts_to_seconds(lesson.end_timestamp)
        abs_start = rel_start + offset
        abs_end = rel_end + offset
        converted.append(Lesson(
            index=0,  # will be re-indexed later
            start_timestamp=seconds_to_ts(abs_start),
            end_timestamp=seconds_to_ts(abs_end),
            is_break=lesson.is_break,
            # Force the canonical slug for breaks in code, rather than
            # trusting the model to use the right literal string.
            slug="breaktime" if lesson.is_break else lesson.slug,
            title=lesson.title,
            summary_markdown_hebrew=lesson.summary_markdown_hebrew,
        ))

    return converted

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

📌 יום 12 - לקחים מפיתוח סוכנים שכדאי לקחת לפרודקשן

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

1. תעדו את כל השיחות - עובד על המכונה שלי זה לא מספיק. לא רק כי דברים עובדים אחרת בפרודקשן אלא שספציפית סוכנים באמת מחזירים תשובות שונות כל פעם שנפנה אליהם. לקוח יכול להתלונן על בעיה ולא יהיה לכם מושג איפה להתחיל לחפש אותה. יש אינסוף פלטפורמות ששומרות היסטוריית שיחות של סוכנים וגם אין בעיה לשמור את כל ההודעות בבסיס הנתונים אצלכם באפליקציה. מה שחשוב שמרו גישה לכל מה שקרה, עוד תצטרכו אותה.
2. שימו לב לטוקנים - כשאתם בודקים את הסוכן אתם יודעים בדיוק מה לכתוב כי אתם כתבתם את הפרומפט. המשתמשים שלכם הולכים להשתמש בטקסטים הרבה יותר ארוכים ולשאול שאלות לא רלוונטיות. שימו לב להכניס מנגנונים שבודקים כמה טוקנים שיחה מסוימת עולה לכם ואולי לעצור שיחות שלא הולכות לכיוון הנכון. טריק נפוץ כאן הוא מנגנון של Guardrails שם סוכן נוסף קורא את השיחה בשביל להבין אם היא פרודוקטיבית או שעדיף לחתוך אותה.
3. חשבו על העומס - שיחה עם סוכן תופסת Thread וחיבור רשת כי הגולשים מחכים לתשובה שמגיעה ב Streaming. זו תבנית עבודה יחסית חדשה ולכן עלינו לבדוק שכל השרשרת עובדת כמו שצריך איתה. שימו סביבת Staging תגייסו חברים ותעמיסו הודעות על השרת לפני שמגיעים למשתמשים אמיתיים כדי לראות מה נשבר. לכל חברות הענן יש היום פתרונות Deployment לסוכנים שווה לבדוק אותם לפני שאתם מקימים סביבה משלכם.
4. בנו מנגנוני Fallback - כל סוכן עם טמפרטורה מעל 0 הולך להחזיר תשובות שונות בכל הפעלה. הוא יכול לא להחזיר תשובה בלבד, להחזיר תשובה חלקית או להחזיר ממש שטויות. נסו להגדיר fallback למודל אחר אם משהו נכשל ותמיד וודאו מה קורה כשהמודל יענה תשובות אחרות ממה שציפיתם.
5. הוציאו כמה שיותר עבודה מחוץ למודל - קוד הוא דטרמניסטי, מודל לא. אם אני צריך לתרגם טקסט אין לי ברירה אני צריך מודל שפה, אבל בשביל להוריד רווחים מסוף או תחילת שורה אני לא צריך אותו. כמה שתסדרו יותר טוב את המידע לפני שתגיעו למודל תקבלו תוצאות טובות ומדויקות יותר.
6. השקיעו בבחירת מודלים ופרומפטים (ושימו לב שזה הולך יחד) - המודל הכי יקר הוא לא תמיד הכי טוב ולא רק בגלל מהירות. כשאתם משנים מודל סיכוי טוב שתצטרכו לדייק מחדש את הפרומפט. הגדירו סט של קלטי בדיקה והפעילו אותם מול קבוצה של פרומפטים ומודלים שהכנתם מראש. כשיגיע מודל חדש תוכלו להריץ אותו על אותו סט של קלטי בדיקה. בצורה כזאת תוכלו להבין איזה מודל באמת נותן את התוצאה המדויקת ביותר ל Use Case שלכם.
7. שימו לב להרשאות - סוכן חכם לא יודע לקבל החלטות בצורה אמינה ולכן תמיד כדאי להניח שהפלט שנקבל ממנו הוא זדוני. איזה כלים הסוכן יכול להפעיל? איזה בקרה יעבור המידע מהסוכן לפני שייכנס למערכת? ולא, להוסיף עוד סוכן לא יעזור. יש לנקות את הפלט של הסוכנים בכל מעבר מסוכן לקוד רגיל.
8. לא חייבים להעביר למודל את כל השיחה ולא חייבים לכתוב רק סוכן אחד - כשאתם כותבים סוכן שיחה שימו לב שאתם לא צריכים להעביר את כל היסטוריית השיחה ולא חייבים להמשיך שיחה שלמה עם אותו סוכן. תבנית שעבדה לי טוב היתה להריץ סוכן "מיון" בתחילת השיחה וכשהשיחה מקבלת כיוון להעביר את השליטה לסוכן ספציפי יותר, וגם לתת לכל סוכן אפשרות לזרוק את השיחה חזרה לסוכן המיון (העברת שיחה לסוכן היא בסך הכל הפעלת כלי).
9. נסיונות חוזרים - מותר לשלוח את אותה שאלה לסוכן כמה פעמים אבל רוב הזמן זה לא אפקטיבי. אם יש משהו ששבר את הסוכן פעם ראשונה הרבה פעמים הוא ישבור את הסוכן גם פעם שניה ובכל מקרה אתם לא רוצים להיות על 30% כשלונות. השתמשו בכשלונות כדי ללמוד על הסוכנים שכתבתם, על ניקוי הקלט והפרומפטים ולהתאים אותם כדי לקבל תשובה נכונה על אותו קלט שנכשל.
10. צאו לדרך, גם אם זה לא מושלם - סוכנים חכמים הם עדיין טריק חדש בספר וכל סוכן שתכתבו ילמד אתכם המון. כל יום שתחכו רק תצא עוד דרך חדשה לכתוב סוכנים ושוב תחזרו להתחלה. קחו את Pydantic AI Agents או כל ספריה אחרת שאהבתם ובנו את הסוכן הראשון שלכם.

יש לכם עוד? ספרו לי בתגובות בטלגרם.