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

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

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

1. מה אנחנו בונים

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

2. אחסון השיחות

נקודת ההתחלה לסקירה היום היא הקובץ 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)

3. המח של הבוט - הקובץ 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}",
    ]

4. קוד הסוכן

הסוכן נמצא בקובץ 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.",
    )

5. טלגרם

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

6. עכשיו אתם

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

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

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

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

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