לא נפתח... ייפתר בהמשך

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

📌 יום 2 - פלט מובנה מסוכן

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

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

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

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

אפשר למצוא את קוד התוכנית המלא בגיטהאב בקישור:
https://github.com/ynonp/pydanticai-demos/tree/main/02-translator-agent

✏ הסוכן - הקובץ translator.py

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

class Translation(BaseModel):
    """A single translation of the source text into one language."""

    language: str
    translation: str


class TranslationResult(BaseModel):
    """Structured output: one entry per requested target language."""

    translations: list[Translation]

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

נמשיך להגדרת הסוכן:

def build_agent() -> Agent[None, TranslationResult]:
    load_dotenv()
    provider = GoogleProvider(api_key=os.environ["GEMINI_API_KEY"])
    model = GoogleModel("gemini-3-flash-preview", provider=provider)
    return Agent(
        model,
        output_type=TranslationResult,    # Structured Output
        instructions=(
            "You are a translator. Translate the given text into each of the "
            "requested target languages. Return one entry per language, using "
            "the language code you were given as the 'language' field and the "
            "translated text as the 'translation' field. Translate only; do not "
            "add comments or explanations."
        ),
    )

את המפתחות model ו instructions אתם כבר מכירים מהמדריך של אתמול. הפרמטר החדש היום הוא output_type. פרמטר זה מבקש מהסוכן להחזיר רק פלט שמתאים למבנה שבחרנו, במקרה שלנו TranslationResult.

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

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

from pydantic_ai import Agent, NativeOutput

from tool_output import Fruit, Vehicle

agent = Agent(
  'openai:gpt-5.2',
  output_type=NativeOutput(
      [Fruit, Vehicle], 
      name='Fruit_or_vehicle',
      description='Return a fruit or vehicle.'
  ),
)
result = agent.run_sync('What is a Ford Explorer?')
print(repr(result.output))
#> Vehicle(name='Ford Explorer', wheels=4)

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

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

def translate(text: str, languages: list[str]) -> TranslationResult:
    """Translate ``text`` into each language code in ``languages``."""
    agent = build_agent()
    prompt = (
        f"Text to translate:\n{text}\n\n"
        f"Target languages: {', '.join(languages)}"
    )

    return agent.run_sync(prompt).output

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

✏ ממשק שורת פקודה

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

from cli_args import parse_args, parse_languages
from translator import translate


def main():
    args = parse_args()
    languages = parse_languages(args.to)

    result = translate(args.text, languages)
    for item in result.translations:
        print(f"[{item.language}] {item.translation}")


if __name__ == "__main__":
    main()

הפונקציות parse_args ו parse_languages הן פונקציות עזר שמוגדרות בקובץ cli_args.py זה תוכנו:

import argparse


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse command line arguments for the translator CLI."""
    parser = argparse.ArgumentParser(
        description="Translate text into one or more languages.",
    )
    parser.add_argument(
        "--text",
        required=True,
        help="The text to translate.",
    )
    parser.add_argument(
        "--to",
        required=True,
        help="Comma-separated target language codes, e.g. 'es,fr'.",
    )
    return parser.parse_args(argv)


def parse_languages(value: str) -> list[str]:
    """Split a comma-separated language string into a clean list of codes."""
    return [lang.strip() for lang in value.split(",") if lang.strip()]

אפשר להפעיל את הקוד משורת הפקודה עם פקודה כמו:

uv run main.py --text 'hello world' --to es,fr

ולקבל תרגום לשתי השפות.

✏ ממשק ווב

הממשק השני לאותו סוכן הוא ממשק ה Web. גם הוא ישתמש באותה פונקציית עזר translate ויעביר את התוצאה לפורמט HTML. זה הקוד הרלוונטי מהקובץ app.py:

@app.post("/translate", response_class=HTMLResponse)
def translate_endpoint(
    request: Request,
    text: str = Form(...),
    languages: list[str] = Form(...),
):
    """Translate the submitted text and render the results template."""
    result = translate(text, languages)

    return templates.TemplateResponse(
        request,
        "result.html",
        {
            "text": text,
            "translations": result.translations,
        },
    )

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

✏ עכשיו אתם

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

class UserProfile(BaseModel):
    username: str = Field(
        ..., 
        title="Account Username", 
        description="The unique handle used to identify the user across the platform."
    )
    age: int = Field(
        None, 
        description="The age of the user in years. Optional."
    )

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

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

        # 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. נסו לשנות את הדינמיקה - במקום כלי "סיום" אחד הגדירו כלי "שמירת מאמר" ותנו למודל להפעיל אותו כמה פעמים שהוא רוצה בשביל לשמור מאמרים לתקציר היומי. האם יש הבדל במספר המאמרים? באיכות שלהם?

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

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

✏ חיבור ממשק משתמש

הקובץ האחרון והארוך ביותר בתוכנית הוא 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 של זכרון שמאפשר למודל "לשמור מידע בזכרון" ו"לטעון מידע מהזכרון". המידע בזכרון יהיה בסך הכל קובץ טקסט. שימו לב איך הזכרון משפיע על התוצאות ומתי המודל מפעיל את הכלי.

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

    """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. עדכנו את המשחק והפכו אותו לקשה יותר כדי שמשחק בין שני סוכנים יהיה מעניין - אפשר להגדיל את הלוח או לשחק על כמה לוחות במקביל. נסו לסבך אותם מספיק כדי שנראה הבדלים בין המודלים.

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

    """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. עדכנו את התוכנית כך שהסוכן יקבל הזדמנות לחוות דעה אחרי כל שינוי או ציור שאתם עושים על הלוח. השתמשו בסוכן אחר ללא כלים כדי שלא יקלקל את הפעולות שעכשיו עשיתם.

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