📌 יום 7: בואו נכתוב סוכן קידוד

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

https://github.com/ynonp/pydanticai-demos/tree/main/07-minicoder

✏ מה זה סוכן קידוד

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

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

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

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

✏ איך זה עובד

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

1. כלי קריאת קובץ.
2. כלי כתיבת קובץ מאפס.
3. כלי שינוי טקסט בקובץ.
4. כלי הפעלת פקודה משורת הפקודה (bash).

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

✏ לולאת הסוכן

הלולאה הראשית של הסוכן היא:

    while True:
        try:
            user_input = input("you> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break

        if not user_input:
            continue
        if user_input in ("exit", "quit"):
            break

        try:
            result = agent.run_sync(user_input, deps=deps, message_history=history)
            history = result.all_messages()
            print(f"\n{result.output}\n")
        except Exception as exc:  # keep the session alive on any single failure
            print(f"\n[error: {exc}]\n")

והיא בסך הכל קוראת פקודה מהמשתמש, מפעילה run_sync, שומרת את התוצאה וממשיכה לאיטרציה הבאה. פרמטר התלויות deps מכיל את תיקיית העבודה בה מותר לסוכן לעבוד:

work_dir = Path(sys.argv[1]).resolve()
work_dir.mkdir(parents=True, exist_ok=True)

deps = Deps(work_dir=work_dir)

בתוך הכלים נוודא שכל מה שאנחנו עושים יתבצע באותה תיקיית עבודה.

✏ כלי קריאת קובץ

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

def read_file(
    ctx: RunContext[Deps], file: str, offset: int = 0, limit: int = MAX_OUTPUT
) -> str:
    """Read a file and return its text from ``offset`` for ``limit`` chars."""
    path = _resolve(ctx.deps.work_dir, file)
    if not path.is_file():
        return f"Error: no such file: {file}"
    text = path.read_text(encoding="utf-8", errors="replace")
    return text[offset : offset + limit]

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

✏ כתיבת קובץ

הכלי השני לכתיבת קובץ נראה כך:

def write_file(ctx: RunContext[Deps], file: str, new_content: str) -> str:
    """Create ``file`` or replace its entire contents with ``new_content``."""
    path = _resolve(ctx.deps.work_dir, file)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(new_content, encoding="utf-8")
    return f"Wrote {len(new_content)} chars to {file}"

הוא לוקח שם קובץ ותוכן וכותב הכל לקובץ. נשים לב שכתיבת קובץ שלם מכניסה את כל תוכן הקובץ ל Context בתור פרמטר של כלי שימשיך ללוות אותנו בהיסטוריית ההודעות. במערכות רבות מסננים מהיסטוריית ההודעות את הפעלות הכלים כדי לא להעמיס על הקונטקסט. אם הסוכן ירצה לדעת מה יש בקובץ שישתמש בכלי read file.

✏ עריכת קובץ

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

def edit_file(

    ctx: RunContext[Deps], file: str, existing_text: str, replacement_text: str
) -> str:
    """Replace the single, exact occurrence of ``existing_text`` in ``file``."""
    path = _resolve(ctx.deps.work_dir, file)
    if not path.is_file():
        return f"Error: no such file: {file}"
    text = path.read_text(encoding="utf-8")
    count = text.count(existing_text)
    if count == 0:
        return f"Error: existing_text not found in {file}"
    if count > 1:
        return (
            f"Error: existing_text appears {count} times in {file}; "
            "add more surrounding context to make it unique"
        )
    path.write_text(text.replace(existing_text, replacement_text), encoding="utf-8")
    return f"Edited {file}"

✏ הפעלת פקודת מערכת

הכלי האחרון, bash, מפעיל פקודת מערכת. פה יש לנו שני אתגרים:

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

זה קוד הכלי:

def bash(ctx: RunContext[Deps], command: str) -> str:
    """Run a shell command in the project directory and return its output.

    Returns the first 10,000 chars of combined stdout+stderr. The full
    output is saved to a temp file whose path is reported when truncated,
    so it can be read in full later with read_file.
    """
    try:
        proc = subprocess.run(
            command,
            shell=True,
            cwd=ctx.deps.work_dir,
            capture_output=True,
            text=True,
            timeout=BASH_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        return f"Error: command timed out after {BASH_TIMEOUT}s"

    output = proc.stdout + proc.stderr
    header = f"(exit code {proc.returncode})\n"
    if len(output) <= MAX_OUTPUT:
        return header + output

    with tempfile.NamedTemporaryFile(
        mode="w", delete=False, suffix=".txt", prefix="minicoder-", encoding="utf-8"
    ) as fh:
        fh.write(output)
        tmp_path = fh.name
    return (
        header
        + output[:MAX_OUTPUT]
        + f"\n\n[output truncated; full output saved to {tmp_path} — "
        "read it with read_file]"
    )

✏ עכשיו אתם

1. הפעילו את הסוכן על המכונה שלכם ובנו בעזרתו משחק. האם הוא הצליח? נסו להחליף מודל ובדקו כיצד מודלים שונים מתמודדים עם המשימה.
2. הוסיפו לסוכן "זכרון" כך שהסוכן יכתוב פרטים חשובים שהוא מגלה על הקוד לקובץ טקסט. הוסיפו את קובץ הטקסט הזה לפרומפט באופן אוטומטי. האם הזכרון עוזר לסוכן לכתוב קוד טוב יותר?
3. הוסיפו תמיכה בניהול מספר שיחות במקביל. פקודת /new פותחת שיחה חדשה, פקודת /list מראה את כל השיחות ופקודת /resume חוזרת לשיחה אחרת.

📌 יום 8: בוחן אינטרקטיבי מהטלגרם

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

קוד הדוגמה בקישור:

https://github.com/ynonp/pydanticai-demos/tree/main/08-interactice-telegram-quiz-agent

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

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

✏ אחסון השיחות

נקודת ההתחלה לסקירה היום היא הקובץ store.py שמנהל את השיחות. הקובץ בסך הכל שומר Dictionary בזכרון עם כל השיחות של כל המשתמשים. בעתיד אפשר יהיה להעביר את זה לבסיס נתונים מסודר או ל redis. זה המימוש:

class InMemoryKVStore:
    """A ``KVStore`` backed by a plain dict. State is lost when the process exits."""

    def __init__(self) -> None:
        self._data: dict[str, Any] = {}

    def get(self, key: str, default: Any = None) -> Any:
        return self._data.get(key, default)

    def set(self, key: str, value: Any) -> None:
        self._data[key] = value

    def has(self, key: str) -> bool:
        return key in self._data

    def delete(self, key: str) -> None:
        self._data.pop(key, None)

✏ המח של הבוט - הקובץ quiz

בקובץ quiz.py נמצא המוח של הבוט. הקובץ מגדיר בסך הכל שתי פונקציות:

1. התחלת בוחן
2. טיפול בתשובה

פונקציית התחלת הבוחן מושכת את המאמר ובונה ממנו את השאלה הראשונה:

async def start_quiz(
    store: KVStore, agent: Agent[None, QuizTurn], chat_id: int | str, url: str
) -> list[str]:
    """Fetch the article and ask the first question."""
    try:
        title, text = fetch_article(url)
    except ValueError as exc:
        return [f"⚠️ {exc}"]

    result = await agent.run(SEED_PROMPT.format(article=text))
    turn = result.output

    store.set(
        _key(chat_id),
        {
            "title": title,
            "q_number": 1,
            "total_score": 0,
            "status": "active",
            "pai_history": _dump_history(result),
        },
    )

    return [
        f"📖 Quizzing you on: *{title}*\nSend /start anytime to begin a new one.",
        f"Question 1/{TOTAL_QUESTIONS}:\n{turn.next_question}",
    ]

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

async def handle_answer(
    store: KVStore, agent: Agent[None, QuizTurn], chat_id: int | str, text: str
) -> list[str]:
    """Evaluate the user's answer and ask the next question (or wrap up)."""
    session = store.get(_key(chat_id))
    if not session or session.get("status") != "active":
        return [
            "Send me an article link (starting with http) and I'll quiz you "
            "on it."
        ]

    q_number = session["q_number"]
    final = q_number >= TOTAL_QUESTIONS

    prompt = text
    if final:
        prompt += FINAL_CONTROL.format(n=q_number, total=TOTAL_QUESTIONS)

    history = _load_history(session)
    result = await agent.run(prompt, message_history=history)
    turn = result.output

    total_score = session["total_score"] + turn.score
    session["total_score"] = total_score
    session["pai_history"] = _dump_history(result)

    feedback_line = f"📝 {turn.feedback} (Score: {turn.score}/10)"

    if final or turn.next_question is None:
        session["status"] = "finished"
        store.set(_key(chat_id), session)
        max_score = TOTAL_QUESTIONS * 10
        scorecard = (
            f"🏁 Quiz complete!\n\n"
            f"Final score: *{total_score}/{max_score}*\n\n"
            f"Send another article link to play again."
        )
        return [feedback_line, scorecard]

    session["q_number"] = q_number + 1
    store.set(_key(chat_id), session)
    return [
        feedback_line,
        f"Question {q_number + 1}/{TOTAL_QUESTIONS}:\n{turn.next_question}",
    ]

✏ קוד הסוכן

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

class QuizTurn(BaseModel):
    """One step of the conversational quiz."""

    feedback: str = Field(
        description="Assessment of the user's previous answer; empty for the "
        "first question.",
    )
    score: int = Field(
        ge=0,
        le=10,
        description="Score for the previous answer, 0 for the first question.",
    )
    next_question: str | None = Field(
        description="The next question to ask; null when the quiz is over.",
    )

✏ טלגרם

בשביל החיבור לטלגרם מפייתון השתמשתי בספריית python-telegram-bot. הקוד שמתחבר לטגרם שמור בקובץ main.py.

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

app.add_handler(CommandHandler("start", on_start))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_text))

כששולחים פקודת /start נפעיל את הפונקציה on_start. כששולחים טקסט כלשהו שאינו פקודה נפעיל את הפונקציה on_text. פונקציית ההתחלה שולחת הודעת פתיחה ופונקציית קריאת הטקסט בודקת אם הטקסט נראה כמו URL היא תתחיל שיחה חדשה עם הסוכן על המאמר שנשלח, וכל טקסט אחר נחשב בתור תשובה לשאלה שהסוכן אולי שלח.

✏ עכשיו אתם

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

https://paths.grasp.study/public-modules/2989b86f-e56c-4677-b75d-5042f0666827/lessons/13e343e2-1bb8-43f5-abc0-08695aaff682

2. הבוט מייצר שאלה חדשה אחרי כל הודעה. עדכנו את הקוד כך שהבוט יבנה את עשר השאלות מיד עם קבלת המאמר. חשבו: מה היתרונות והחסרונות של כל גישה?
3. האם משתמשים יכולים לשנות את התנהגות הבוט, למשל לייצר מספר שונה של שאלות או לגרום לבוט לקבל תשובות לא נכונות, או אפילו לחשוף מידע שלא אמור להיות נגיש להם? שחקו עם הבוט, נסו לבעוט בו ולמצוא נקודות תורפה.
4. מה יקרה אם תנסו לדחוף את הבוט כמו שהוא כתוב כרגע לשרת בענן של Vercel? על איזה שרתים אפשר להריץ את הבוט הזה? מה צריך לעדכן כדי להריץ אותו בתור AWS Lambda Function או Vercel Function?

📌 יום 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 או כל ספריה אחרת שאהבתם ובנו את הסוכן הראשון שלכם.

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