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

    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)