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

1. למה זה קשה

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

1. משתמש שולח הודעה, אנחנו מעבירים את ההודעה למכונה ומדפיסים למשתמש את התשובה.

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

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

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

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

3. קוד הסוכן

אני ממשיך לקוד הסוכן בקובץ 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."
        ),
    )

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

4. חיבור ממשק משתמש

הקובץ האחרון והארוך ביותר בתוכנית הוא 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

5. עכשיו אתם

1. הריצו את הדוגמה על המכונה שלכם עם מודלים שונים ושימו לב לשינויים בתשובות ובמהירות התגובה של כל מודל.

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

3. סננו הפעלת כלים בהודעות ישנות - צרו Checkbox שאומר "שלחו גם הפעלת כלים". כשמכבים אותו סננו מהיסטוריית השיחה את כל הודעות הכלים (בקשה להפעלת כלים ואת התוצאה של הכלי). בדקו האם סינון כזה פוגע באיכות התוצאות? האם רק במצבים מסוימים?

4. הוסיפו Tool שיאפשר למודל להסתכל בהודעות משיחות אחרות. שימו לב מתי המודל משתמש בכלי זה, ואיך זה משפיע על התוצאות.

5. הוסיפו Tool של זכרון שמאפשר למודל "לשמור מידע בזכרון" ו"לטעון מידע מהזכרון". המידע בזכרון יהיה בסך הכל קובץ טקסט. שימו לב איך הזכרון משפיע על התוצאות ומתי המודל מפעיל את הכלי.