ToCode

📌 יום 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),

✏ קוד הסוכן הסוכן נמצא בקובץ 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?

📌 יום 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}",
    ]

    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 חוזרת לשיחה אחרת.

📌 יום 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(

    """Change the color of the shape with the given id (CSS hex string)."""
    shape = _find_shape(ctx.deps.canvas, id)
    if shape is None:
        return f"No shape with id {id}."
    shape.color = color
    return f"Recolored shape {id} to {color}."

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

yield emit({"type": "canvas", "state": deps.canvas.model_dump(mode="json")})

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

📌 יום 6: ציור עם סוכן במשחק האיקס עיגול למדנו שסוכנים חכמים יכולים לבצע פעולות על המסך. היום ניקח את הרעיון הזה צעד גדול קדימה ונבנה מערכת עם ממשק כפול - סוכן וממשק גרפי. משתמשים יכולים לעבוד עם המערכת בכל אחד משני הממשקים ושניהם תמיד מתואמים. ✏ מה אנחנו בונים האפליקציה שנבנה הוא אפליקציית צייר מבוסס תבניות. יש ציורים קטנים בצד שמאל של קשת בענן, חד קרן, בית, כוכב, עץ, לב, עיגול ופרח ופלטת צבעים בחלק העליון של העמוד. משתמשים יכולים לגרור תבנית ללוח העבודה המרכזי, שם הם יוכלו לגרור את הצורה, לשנות לה את הגודל או לבחור לה צבע אחר. עד פה אפליקציית Front End רגילה לגמרי. החידוש הוא שבנוסף לכל זה יש גם תיבת שיחה עם סוכן בתיבת הצד הימנית. משתמשים יכולים לשאול את הסוכן מה מצויר על הלוח, לשאול מה דעתו על הציור וגם לבקש שינויים למשל לבקש מהסוכן שיוסיף בית, חד קרן או שיסדר את הצבעים. לסוכן ולמשתמש יש גישה בדיוק לאותן הפעולות והם רואים תמיד את אותו הציור. כך נוצר שיתוף פעולה בין משתמשים לסוכנים חכמים שמוסיף קסם למערכות שלנו. ✏ איך מפעילים קוד הדוגמה נמצא בתיקיית הדוגמאות: https://github.com/ynonp/pydanticai-demos/tree/main/06-ai-painter והוא מורכב מאפליקציית Vue ומשרת Fastapi בפייתון. בשביל להפעיל נצטרך שני חלונות. בחלון אחד מפעילים את הפייתון:

uv run main.py

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

cd frontend
npm run dev

לאחר מכן נכנסים ל localhost:5173 כדי לראות את הממשק. ✏ איך זה עובד מנגנון שיתוף הפעולה בנוי משני חלקים: הראשון הוא שיתוף המידע והשני הוא הוא הפעלת הכלים. בכל הודעה מצד הלקוח לסוכן אנו מצרפים להודעה עותק של לוח הציור כלומר ה Endpoint של הודעה מקבל:

class ChatRequest(BaseModel):
    message: str
    canvas: CanvasState
    # Opaque pydantic-ai message history, exactly as returned by the previous
    # /api/chat response. Empty on a fresh page load.
    history: list[Any] = Field(default_factory=list)
    
class CanvasState(BaseModel):
    """The full canvas: its dimensions plus every shape on it."""

    width: int = CANVAS_WIDTH
    height: int = CANVAS_HEIGHT
    shapes: list[Shape] = Field(default_factory=list)    
    
class Shape(BaseModel):
    """A single shape placed on the canvas."""

    # Re-validate on assignment so the tools' `shape.size = ...` mutations go
    # through the same clamp as construction (keeps server + browser in sync).
    model_config = ConfigDict(validate_assignment=True)

    id: str
    type: ShapeType
    x: int = Field(description="Left position in pixels (0 = left edge).")
    y: int = Field(description="Top position in pixels (0 = top edge).")
    size: int = Field(default=DEFAULT_SHAPE_SIZE, description="Width/height in pixels.")
    color: str = Field(default=DEFAULT_SHAPE_COLOR, description="CSS color, e.g. '#ff0000'.")

    @field_validator("size")
    @classmethod
    def _clamp_size(cls, value: int) -> int:
        # Mirror the frontend store: never smaller than MIN_SHAPE_SIZE.
        return max(MIN_SHAPE_SIZE, value)

לוח הציור עובר לסוכן באמצעות הפונקציה add_canvas_state שמוגדרת בקובץ agent.py:

    @agent.instructions
    def add_canvas_state(ctx: RunContext[PainterDeps]) -> str:
        canvas = ctx.deps.canvas
        return (
            f"Canvas size: {canvas.width}x{canvas.height} pixels.\n"
            f"Current canvas state (JSON):\n{canvas.model_dump_json(indent=2)}"
        )

הפעולות שסוכן יכול לעשות עוברות באמצעות Tool Calls לדוגמה בשביל להוסיף צורה הסוכן יפעיל את הכלי:

@agent.tool
async def add_shape(
    ctx: RunContext[PainterDeps],
    type: ShapeType,
    x: int,
    y: int,
    size: int = DEFAULT_SHAPE_SIZE,
    color: str = DEFAULT_SHAPE_COLOR,
) -> str:
    """Add a new shape to the canvas. Returns the new shape's id."""
    shape = Shape(id=uuid.uuid4().hex, type=type, x=x, y=y, size=size, color=color)
    ctx.deps.canvas.shapes.append(shape)
    return f"Added {type} (id={shape.id}) at ({x}, {y})."

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

@agent.tool
async def recolor_shape(ctx: RunContext[PainterDeps], id: str, color: str) -> str:

    """Apply the human's move, then let the AI respond."""
    if game.is_over(BOARD):
        return state({"error": "Game is already over."})
    if not game.is_valid_move(BOARD, req.index):
        return state({"error": "Illegal move."})

    # 1. Human plays.
    BOARD[req.index] = game.HUMAN

    # 2. AI responds (if the game isn't already decided).
    ai_move = None
    if not game.is_over(BOARD):
        result = await agent.choose_move(BOARD)
        BOARD[result.index] = game.AI
        ai_move = {"index": result.index, "source": result.source}

    return state({"ai_move": ai_move})

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

def forced_move(index: int) -> TestModel:
    """Return a model that always asks to play the given cell."""
    return TestModel(custom_output_args={"index": index})


def play_human(board: list[str], index: int) -> None:
    assert game.is_valid_move(board, index)
    board[index] = game.HUMAN


def play_ai(board: list[str], index: int):
    result = build_agent(forced_move(index)).choose_move_sync(board)
    assert game.is_valid_move(board, result.index)
    board[result.index] = game.AI
    return result

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

def test_invalid_ai_move_uses_and_registers_random_fallback():
    board = game.new_board()
    play_human(board, 0)
    empty_before = set(game.empty_cells(board))

    result = play_ai(board, 9)

    assert result.source == "random"
    assert result.index in empty_before
    assert board[result.index] == game.AI
    assert board.count(game.AI) == 1

✏ עכשיו אתם 1. הריצו את המשחק על המכונה שלכם ושחקו נגד ג'מיני. 2. עדכנו את הוראות הסוכן כדי שג'מיני לא ינצח כל הזמן. 3. כתבו משחק בין שני סוכנים ושימו לב שהם תמיד יוצאים בתיקו (כי שניהם אלופים). 4. עדכנו את המשחק והפכו אותו לקשה יותר כדי שמשחק בין שני סוכנים יהיה מעניין - אפשר להגדיל את הלוח או לשחק על כמה לוחות במקביל. נסו לסבך אותם מספיק כדי שנראה הבדלים בין המודלים.

📌 יום 5: משחק איקס עיגול נגד הסוכן שיחות עם סוכן לא חייבות לכלול רק הודעות טקסט. בדוגמה של היום נבנה משחק איקס עיגול נגד סוכן ונראה איך להעביר את מצב המשחק בהודעה כדי שהסוכן יחליט על המהלך הבא. דוגמה זו מהווה בסיס לדוגמאות שנראה בהמשך בהן הסוכן משולב במערכת קיימת ומאפשר שני ממשקים: ממשק שיחה וממשק גרפי המשתלבים ביניהם. ✏ מה אנחנו בונים נבנה משחק איקס עיגול בו שחקן המחשב הוא סוכן חכם שמשתמש במודל של Gemini. המשחק כולל שני ממשקים: ממשק ווב וממשק מסוף למשחק משורת הפקודה. בשני המקרים המשחק יעבוד לפי אותו מנגנון: 1. מחכים למהלך של השחקן האנושי. 2. שולחים את לוח המשחק לסוכן בצירוף הוראות שמבקשות את המהלך הבא. 3. אם הסוכן לא מחזיר מהלך או מחזיר מהלך לא תקין נבחר מהלך באקראי מתוך המשבצות הריקות על הלוח. למערכת יהיה סט בדיקות שישתמש במחלקה TestModel כך שנוכל לראות שהכל עובד על מודל בדיקה בדיקה שאנחנו יודעים מראש מה יחזיר. ✏ קוד הסוכן אני מתחיל את הסיור בדוגמה בקובץ הסוכן agent.py. הקובץ מגדיר את מבנה הפלט שהוא בסך הכל מהלך במשחק:

class Move(BaseModel):
    """The move the AI wants to play."""

    index: int = Field(
        description="Index of the cell to play (0-8), counting left-to-right, "
        "top-to-bottom. Must be an empty cell.",
        ge=0,
        le=8,
    )

מהלך הוא רק מספר בין 0 ל-8 שמייצג את התא בו צריך לשחק. העטיפה ב Pydantic Model מאפשרת להוסיף לו תיאור וגם להגדיר גבולות. הקלאס המרכזי בקובץ הוא TicTacToeAgent:

class TicTacToeAgent:
    """A tic-tac-toe player backed by a Pydantic AI model."""

    def __init__(self, model: Model):
        self.model = model

    def _agent(self) -> Agent[None, Move]:
        return Agent(self.model, output_type=Move, instructions=INSTRUCTIONS)

    @staticmethod
    def _result(board: list[str], move: Move | None) -> MoveResult:
        """Use a legal agent move, otherwise fall back to a random one."""
        if move is not None and game.is_valid_move(board, move.index):
            return MoveResult(index=move.index, source="agent")
        return _random_fallback(board)

    async def choose_move(self, board: list[str]) -> MoveResult:
        """Ask the model for a move without blocking the current event loop."""
        try:
            move = (await self._agent().run(_render_board(board))).output
        except Exception:
            move = None
        return self._result(board, move)

    def choose_move_sync(self, board: list[str]) -> MoveResult:
        """Ask the model for a move from synchronous code."""
        try:
            move = self._agent().run_sync(_render_board(board)).output
        except Exception:
            move = None
        return self._result(board, move)

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

def main() -> None:
    load_dotenv()
    agent = build_agent(build_primary_model())
    board = game.new_board()
    print("Tic-Tac-Toe — you are X, the AI is O.\n")
    print(render(board))

    while not game.is_over(board):
        # 1. Human move.
        board[ask_human_move(board)] = game.HUMAN
        print("\n" + render(board) + "\n")
        if game.is_over(board):
            break

        # 2. AI move.
        print("AI is thinking...")
        result = agent.choose_move_sync(board)
        board[result.index] = game.AI
        note = "" if result.source == "agent" else "  (random fallback)"
        print(f"AI plays cell {result.index}.{note}\n")
        print(render(board) + "\n")

בממשק הווב (בקובץ server.py) הסיפור יותר מורכב כי צריך HTML, CSS ו JavaScript וכי המתנה למהלך היא אסינכרונית אבל גם שם הכל מתנקז לפונקציה אחת שמטפלת במהלך שמשתמש מבצע מתוך הממשק:

@app.post("/play")
async def play(req: PlayRequest) -> dict:

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

async def _chat_event_stream(
    conv_id: int, message: str
) -> AsyncIterator[bytes]:
    with SessionLocal() as db:
        conv = db.get(Conversation, conv_id)
        if conv is None:
            yield _emit({"type": "error", "content": "no such conversation"})
            return

        history = _load_history(conv)
        _save_user_message(db, conv, message)

        seen_results: set[str] = set()
        run_result = None
        async with agent.run_stream_events(
            message, message_history=history
        ) as events:
            async for event in events:
                payload = _event_payload(event, seen_results)
                if payload is not None:
                    yield _emit(payload)
                if isinstance(event, AgentRunResultEvent):
                    run_result = event.result

        if run_result is not None:
            _save_run_result(db, conv, run_result)

        if conv.title == "New conversation":
            title = await _generate_title(message)
            if title:
                conv.title = title
                db.commit()
                yield _emit({"type": "title", "title": conv.title})

        yield _emit({"type": "done"})

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

def _event_payload(event: object, seen_results: set[str]) -> dict | None:
    if isinstance(event, PartStartEvent):
        part = event.part
        if isinstance(part, TextPart) and part.content:
            return {"type": "text", "content": part.content}
        if isinstance(part, (NativeToolCallPart, ToolCallPart)):
            return {
                "type": "tool_call",
                "tool_name": part.tool_name,
                "tool_args": _stringify(part.args),
            }
        if isinstance(part, (NativeToolReturnPart, ToolReturnPart)):
            if part.tool_call_id not in seen_results:
                seen_results.add(part.tool_call_id)
                return {
                    "type": "tool_result",
                    "tool_name": part.tool_name,
                    "content": _stringify(part.content),
                }
    elif isinstance(event, PartDeltaEvent):
        if isinstance(event.delta, TextPartDelta):
            return {
                "type": "text",
                "content": event.delta.content_delta,
            }
    return None

✏ עכשיו אתם 1. הריצו את הדוגמה על המכונה שלכם עם מודלים שונים ושימו לב לשינויים בתשובות ובמהירות התגובה של כל מודל. 2. עדכנו את הקוד - הוסיפו כפתור "Compact" שיגרום לכיווץ כל ההודעות הישנות בשיחה להודעת תקציר אחת. שימו לב איך כפתור זה מאפשר לכם לשלוח פחות מידע גם בשיחות ארוכות. מערכות אמיתיות יבצעו כיווץ כזה בצורה אוטומטית. 3. סננו הפעלת כלים בהודעות ישנות - צרו Checkbox שאומר "שלחו גם הפעלת כלים". כשמכבים אותו סננו מהיסטוריית השיחה את כל הודעות הכלים (בקשה להפעלת כלים ואת התוצאה של הכלי). בדקו האם סינון כזה פוגע באיכות התוצאות? האם רק במצבים מסוימים? 4. הוסיפו Tool שיאפשר למודל להסתכל בהודעות משיחות אחרות. שימו לב מתי המודל משתמש בכלי זה, ואיך זה משפיע על התוצאות. 5. הוסיפו Tool של זכרון שמאפשר למודל "לשמור מידע בזכרון" ו"לטעון מידע מהזכרון". המידע בזכרון יהיה בסך הכל קובץ טקסט. שימו לב איך הזכרון משפיע על התוצאות ומתי המודל מפעיל את הכלי.

📌 יום 4: מימוש סוכן שיחה שלושת הדוגמאות שראינו במדריכים הקודמים היו עבור סוכנים של תהליכי עבודה, הסוכן מקבל טריגר עושה משהו ומסיים. בדוגמה היום ניקח צעד קדימה ונבנה סוכן שיחה אינטרקטיבי, סוכן שממשיך לקבל הודעות ומתיחס לכל היסטוריית השיחה בכל תשובה חדשה. ✏ למה זה קשה מודלים הם יצורים מאוד פשוטים, אתה נותן להם טקסט והם משלימים את ההודעה הבאה בשיחה. הם לא משתנים, הם לא שומרים זכרון, הם בסך הכל מכונת קלט-פלט. בשביל לנהל שיחה עם כזאת מכונה אנחנו צריכים קצת לרמות: 1. משתמש שולח הודעה, אנחנו מעבירים את ההודעה למכונה ומדפיסים למשתמש את התשובה. 2. משתמש מקליד הודעה שניה, ועכשיו מה? אם נשלח את ההודעה השניה בלבד למכונה אז נקבל השלמה לשיחה שבה הודעה אחת, רק ההודעה השניה. במקום זה בשביל לקבל הרגשה של שיחה אנחנו לוקחים את ההודעה הראשונה של המשתמש, את התשובה של המודל ואת ההודעה השניה ושולחים הכל בחבילה אחת. את התשובה של המודל נציג למשתמש. 3. משתמש שולח הודעה שלישית, הפעם המודל גם מחליט להפעיל שני כלים. הוא מקבל את התשובה של שני הכלים ואז מייצר תשובה למשתמש. בהודעה הבאה נצטרך לשלוח למודל את כל ההודעות של המשתמש, את בקשת הפעלת שני הכלים, את התוצאות של שני הכלים ואת ההודעה החדשה. לאורך זמן הודעות מצטברות בשיחות. בהתחלה סוכני שיחה חסמו את המשתמשים אחרי ששיחה התארכה מעבר לכמות הודעות מסוימת כי "ההשלמה" הבאה היתה צריכה המון טוקנים, היה צריך להעביר קלט מאוד ארוך למודל כדי לקבל תשובות. היום אנחנו רואים גישה יותר יצירתית בה הקוד מסנן חלק מההודעות, למשל אולי לא צריך את כל הפעלות הכלים והתשובות שלהם ומספיק איזה סיכום של זה, או שאולי לא צריך את כל ההודעות הישנות ואפשר להשתמש בקריאת LLM אחת כדי לסכם אותן ולתקצר את כל ההתחלה של השיחה. וכמובן שכל המידע הזה צריך להישמר באיזשהו בסיס נתונים שמישהו צריך לנהל. ✏ הקובץ db.py כדי לנהל ממשק שיחה נקודת התחלה טובה היא בסיס הנתונים. בדוגמה היום אנחנו רואים בסיס נתונים SQLite המנוהל דרך SQLAlchemy בקובץ db.py. הקובץ מגדיר טבלת שיחות וטבלת הודעות כשלכל שיחה יש נושא ומזהה (קל) וההודעות שייכות לשיחות. מעניין לראות את העמודות בטבלת ההודעות:

class Message(Base):
    """A single display unit rendered in the conversation view.

    ``kind`` selects which UI template renders the row:
      - ``text``        -> user / assistant chat bubble
      - ``tool_call``   -> "agent called a tool" card
      - ``tool_result`` -> "tool responded" card
    """

    __tablename__ = "messages"

    id: Mapped[int] = mapped_column(primary_key=True)
    conversation_id: Mapped[int] = mapped_column(ForeignKey("conversations.id"))

    role: Mapped[str] = mapped_column()  # "user" | "assistant" | "tool"
    kind: Mapped[str] = mapped_column(default="text")  # text|tool_call|tool_result
    content: Mapped[str] = mapped_column(Text, default="")

    tool_name: Mapped[str | None] = mapped_column(default=None)
    tool_args: Mapped[str | None] = mapped_column(Text, default=None)

    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    conversation: Mapped["Conversation"] = relationship(back_populates="messages")

אנחנו שומרים ב role את מי ששלח את ההודעה, kind את סוג ההודעה, content התוכן והפרמטרים הבאים עבור הפעלת כלים מה שם הכלי ומה הפרמטרים. בדוגמאות יותר מורכבות נצטרך לאפשר גם סוגי תוכן מגוונים בהודעות למשל הודעות תמונה או קבצים מצורפים. ✏ קוד הסוכן אני ממשיך לקוד הסוכן בקובץ chat_agent.py ופה לשמחתנו אין הרבה חדש. הדבר המעניין בקובץ הוא החלוקה לשני סוכנים, יש לנו את סוכן השיחה הרגיל ולידו סוכן ספציפי שמייצר את "שם" השיחה:

# A tiny dedicated agent used to auto-name conversations.
def build_title_agent() -> Agent[None, str]:
    provider = GoogleProvider(api_key=os.environ["GEMINI_API_KEY"])
    model = GoogleModel(MODEL_NAME, provider=provider)
    return Agent(
        model,
        instructions=(
            "Generate a very short title (3-5 words, no quotes, no trailing "
            "punctuation) summarizing the topic of the conversation."
        ),
    )

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

        # Provider-adaptive: native fetch when available, local fallback otherwise.
        capabilities.append(WebFetch(local=True, native=False))
    elif tool == "native":
        # Native-only. WebFetch itself raises UserError if the provider has no
        # native web-fetch tool (e.g. Gemini) — no fallback, no custom guard.
        capabilities.append(WebFetch(local=False))
    else:
        raise ValueError(f"Unknown tool backend: {tool}")

    return Agent(
        model,
        output_type=Digest,
        tools=tools,
        capabilities=capabilities,
        instructions=INSTRUCTIONS,
    )

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

PROGRAMMING_INDEX_SITES = [
    "https://reddit.com/r/programming",
    "https://news.ycombinator.com",
    "https://lobste.rs",
    "https://dev.to",
    "https://hashnode.com",
    "https://daringfireball.net",
    "https://arstechnica.com",
    "https://theverge.com/tech",
    "https://github.com/trending",
    "https://www.python.org/blogs/",
    "https://blog.rust-lang.org",
    "https://go.dev/blog",
    "https://nodejs.org/en/blog",
    "https://kubernetes.io/blog",
    "https://www.phoronix.com",
]

def build_digest(tool: str, max_articles: int) -> Digest:
    """Run the agent and return the produced digest."""
    agent = build_agent(tool)
    site_list = "\n".join(f"- {s}" for s in PROGRAMMING_INDEX_SITES)
    prompt = (
        f"Build today's programming digest with about {max_articles} articles.\n\n"
        f"Index pages you may browse:\n{site_list}"
    )
    return agent.run_sync(prompt).output

מה יקרה? אי אפשר לדעת. הסוכן מקבל רשימת אתרים וכלי לביקור באתר ומפה ממשיך בצורה חופשית לגמרי. הוא יוכל להכנס לאתרים ברשימה, לאתרים אחרים שאינם ברשימה, למאמרים מהם או להמציא דברים לא קשורים. זה הקסם וגם הסכנה בהפעלת כלים. ✏ עכשיו אתם 1. הריצו את הדוגמה אצלכם עם כל אחד מהכלים. שימו לב שתצטרכו מפתח API מ tavily. זה בחינם ונרשמים כאן Tavily 2. נסו לאלף את המודל - הגדירו כלי "סיום" שמקבל את ה Digest ובודק שיש בו לפחות 7 מאמרים. אם יש פחות החזירו כשלון למודל ובקשו תקציר טוב יותר. 3. נסו לשנות את הדינמיקה - במקום כלי "סיום" אחד הגדירו כלי "שמירת מאמר" ותנו למודל להפעיל אותו כמה פעמים שהוא רוצה בשביל לשמור מאמרים לתקציר היומי. האם יש הבדל במספר המאמרים? באיכות שלהם?

📌 יום 3: תקציר חדשות יומי ראינו כבר איך ליצור שאלות הבנה מדף אינטרנט, אבל מה קורה אם יש כמה דפי אינטרנט? בדוגמה של היום נראה איך לתת לסוכן שלנו את החופש לחקור לבד כדי שיוכל להחליט אוטונומית מהם המאמרים הכי חשובים ולתקצר אותם עבורנו. ✏ הפעלת כלים המנגנון הבסיסי ביותר של סוכנים חכמים נקרא הפעלת כלים. זכרו שמודלים הם בסך הכל יצורים טקסטואלים - אנחנו שולחים הודעת טקסט ומקבלים חזרה הודעת טקסט. על הבסיס הזה אנחנו צריכים לבנות מנגנון שיאפשר למודל גם לעשות דברים, וזה מנגנון הכלים. הרעיון פשוט: אם המודל מחזיר טקסט מיוחד שסיכמנו עליו מראש אנחנו ניקח את זה בתור בקשה להרצת קוד, נפעיל פונקציה שסיכמנו עליה מראש ונפנה למודל שוב עם התוצאה של אותה פונקציה. לאורך השנים מנגנון הפעלת כלים התפתח והוטמע בתוך ה API של המודלים כך שהיום הם יודעים לקבל רשימה ארוכה של כאלה פונקציות כחלק מהפרומפט, לקבל חתימה מלאה של הפונקציות ואף לבקש הפעלה של מספר פונקציות במקביל. יותר מזה, ספקי המודלים בנו פונקציות (כלים) שרצות על התשתית שלהן ואלה נקראים כלים מובנים. כשאנחנו פונים לספק מודלים למשל לגוגל בשביל לדבר עם ג'מיני, אנחנו יכולים לבקש לתת לג'מיני גישה לכלים מסוימים שרצים על התשתית של גוגל כחלק מה API. וכל התיאוריה הזאת מגיעה מובנית בתוך הספריה Pydantic AI Agents בצורה נוחה לשימוש: 1. אפשר להגדיר כל פונקציה בתור "כלי" והמודל יוכל להפעיל אותה. 2. פידנטיק AI מגדירה מספר כלים "מובנים", שהם בעצם פונקציות של הספריה שנכתבו כבר בשבילכם. 3. פידנטיק AI חושפים ממשק גישה לכלים המובנים של הספק. הנה דוגמה פשוטה מתוך התיעוד של פידנטיק להגדרת כלים והעברתם לסוכן:

import random

from pydantic_ai import Agent, RunContext

agent = Agent(
  'google:gemini-3-flash-preview',
  deps_type=str,
  instructions=(
      "You're a dice game, you should roll the die and see if the number "
      "you get back matches the user's guess. If so, tell them they're a winner. "
      "Use the player's name in the response."
  ),
)


@agent.tool_plain
def roll_dice() -> str:
  """Roll a six-sided die and return the result."""
  return str(random.randint(1, 6))


@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
  """Get the player's name."""
  return ctx.deps


dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.output)
#> Congratulations Anne, you guessed correctly! You're a winner!

נתחיל עם מה שקל: הפונקציות roll_dice ו get_player_name מוגדרות בתור פונקציות פייתון פשוטות עם הדקורטור @agent.tool או @agent.tool_plain. ההבדל בין השניים הוא ש tool יכול לקבל פרמטרים מהתוכנית וגם מהמודל ו tool_plain יכול לקבל פרמטרים רק מהמודל. כשהמודל יקבל את הודעת הפתיחה הוא יוכל להריץ את שתי הפונקציות. ככל הנראה הוא יריץ את הפונקציה roll_dice, יקבל מספר אקראי ואז לפי התוצאה ידפיס הודעה תוך שימוש בשם השחקנית שהוא יקבל מהפעלת הפונקציה get_player_name. משפט המפתח כאן הוא ככל הנראה. אין לנו דרך לקבוע למודל איזה כלים להפעיל או באיזה סדר. ✏ מה אנחנו בונים בדוגמה היום נבנה סוכן שירוץ על אתרי חדשות וידפיס תקציר חדשות טכנולוגיה יומי. לצורך יצירת התקציר המודל יוכל לבחור באיזה אתרים לבקר ואיזה מאמרים מהם לקרוא - וכן זה אומר שכל פעם שנפעיל את הסוכן נקבל תקציר אחר. ✏ הצגת הכלים שני הכלים בתוכנית מוגדרים בקובץ extractors.py. רציתי לנסות דרכים שונות לקבל מידע מהאינטרנט ולכן יצרתי שני כלים, אחד נקרא custom fetch והוא מקבל URL ומשתמש ב trafilatura כדי לקבל ממנו את הטקסט. כלי שני נקרא tavily fetch והוא משתמש בשירות רשת בשם tavily כדי לקחת את הטקסט מתוך URL. בנוסף לכלים שלנו התוכנית מאפשרת גם שימוש בכלי ה fetch המובנה של Gemini, ובכלי ה fetch המובנה של Pydantic AI שירוץ מהמכונה שלי. כך נראית פונקציית יצירת הסוכן בקובץ digest.py

def build_agent(tool: str) -> Agent[None, Digest]:
    """Build the digest agent, wiring up the selected extraction backend."""
    provider = GoogleProvider(api_key=os.environ["GEMINI_API_KEY"])
    model = GoogleModel("gemini-3-flash-preview", provider=provider)

    tools: list[Tool] = []
    capabilities: list = []

    if tool == "tavily":
        tools.append(Tool(make_tavily_fetch()))
    elif tool == "custom":
        tools.append(Tool(make_custom_fetch()))
    elif tool == "web_fetch_tool":