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