ToCode

מדריך: מה זה בעצם API הפעם אנחנו עם מדריך בסיסי שאני מקווה שיעזור לאנשי בדיקות ומנהלי מוצר כמו גם למפתחים להבין את ההגיון והמטרה של אחד המושגים החשובים בפיתוח מערכות ווב היום. איך עובדים דפי אינטרנט איפה מתחילים לדבר על APIs? האמת שלסיפור הזה יש הרבה התחלות אפשריות אבל בואו נבחר להתחיל עם דפי אינטרנט. נכון ממשקי פיתוח בין מערכות היו קיימים לפני האינטרנט אבל האופן בו דברים עובדים היום מאוד מושפע מטכנולוגיית האינטרנט. את האינטרנט, או לפחות את הטכנולוגיות שמפעילות אותו, המציא טים ברנרס לי בשנת 1991 והטכנולוגיה הראשונה שהוא המציא נקראת HTTP (ראשי תיבות של Hyptertext Transfer Protocol). זהו פרוטוקול, כלומר שפה, שנועדה להגדיר איך מחשבים מעבירים ביניהם דפים ברשת האינטרנט. הפרוטוקול מגדיר בגדול שכל מחשב יכול להחזיק "דפים" ולכל דף יש מזהה. המזהה נראה כמו שם קובץ על מערכת יוניקס לדוגמה /wiki/HTTP, /en/api/client-sdks או /arunsupe/semantic-grep. פרוטוקול HTTP מגדיר שאם מישהו רוצה לבקש דף ממחשב מרוחק הוא שולח הודעה שמורכבת מהמילה GET ואחריה שם הדף למשל:

GET /en/api/client-sdks/

טים ברנרס לי גם המציא דפדפן אינטרנט שאפשר לו לקרוא את הדפים האלה (שהיו כתובים בפורמט HTML) וקצת אחריו אנשים אחרים המציאו דפדפני אינטרנט נוספים והרחיבו את הפורמט של הדפים. ב 1996 ככל שיותר אנשים כתבו דפי אינטרנט עלה הצורך בביצוע פעולות דרך דפי האינטרנט והומצא אלמנט הטופס. בשביל להגיש טפסים טים ברנרס לי עדכן את פרוטוקול HTTP והוסיף פקודה בשם POST. בשביל להגיש טופס עדיין היה צריך לציין את הנתיב להגשה ובדרך כלל צורף לבקשה מידע מהטופס. כך נראתה בקשת POST שנוצרה מטופס:

POST /cgi-bin/subscribe.cgi HTTP/1.0
Host: www.example.com
Content-Type: application/x-www-form-urlencoded
Content-Length: 27

name=Alice+Smith&email=alice%40example.com

אנחנו שמים לב שמתחת לפקודת POST מופיע עוד בלוק של מפתחות וערכים, בלוק זה נקרא הכותרות של הבקשה ואפשר להעביר אותן גם בבקשות GET. אחרי סיום הכותרות יש שורה ריקה ואחריה שורת הערכים של הטופס וטקסט זה נקרא "גוף הבקשה". לאורך השנים נוספו לפרוטוקול HTTP פעלים נוספים והיום הפרוטוקול מונה את הפעלים GET, HEAD, OPTIONS, TRACE, POST, PUT, PATCH, DELETE. קוד בצד השרת, כלומר במחשב שמקבל את הבקשה, אחראי על טיפול בבקשה. הקוד מקבל את הפועל, את הנתיב, את הכותרות ואת גוף הבקשה וצריך לבצע פעולה ולהחזיר תשובה לדפדפן. לפעמים התשובה תהיה דף אינטרנט, לפעמים היא תהיה תמונה ולפעמים הפניה לעמוד אחר. נסו את זה - בפעם הבאה שאתם גולשים ברשת פתחו בדפדפן את חלון כלי הפיתוח ועברו לטאב Network. שם תוכלו לראות את כל בקשות ה HTTP שהדפדפן שלכם שולח לשרת ואת תשובות השרת לכל בקשה. עכשיו בואו נוסיף אפליקציה דפי אינטרנט כתובים בשפה שנקראת HTML, וכן גם אותה המציא טים ברנרס לי. השפה מורכבת מתגיות ותוכן כאשר התגיות מתארות את התפקיד של כל בלוק או שורה בתוכן. לדוגמה הבלוק:

<h1>hello world</h1>

מתאר שורת כותרת עם הטקסט hello world. הבלוק:

<h1>hello world</h1>
<p>HTML was also invented by Tim Berners Lee along with HTTP and the first web browser</p>

מגדיר שורת כותרת ואחריה פיסקת טקסט רגילה. התגית h1 היא קיצור של המילה header כדי לתאר שורת כותרת והתגית p היא קיצור של המילה paragraph ולכן מתארת פיסקה. עם המצאת שפת HTML טים ברנרס לי המציא גם כלי שנקרא דפדפן אינטרנט שיודע לקרוא מסמכים בשפת HTML ולהציג אותם בצורה גרפית, למשל את הכותרות הוא הציג בגופן גדול יותר. נקפוץ 15 שנה קדימה ונגיד שאנחנו רוצים לכתוב אפליקציה לטלפון. האפליקציה שלנו צריכה להציג חוקים של משחקי לוח פופולריים וכבר כתבנו את כל החוקים בצורה יפה בשפת HTML עבור דף האינטרנט שלנו. הבעיה היא שאת האפליקציה אנשים מתקינים על הטלפון ולא פותחים בתוך דפדפן - לא כל המידע שמופיע בדף האינטרנט רלוונטי גם לאפליקציה, וייתכן ובאפליקציה המפתחים ירצו להטמיע עיצוב או מבנה שונה מזה שמופיע ברשת. הפיתרון - JSON הפיתרון שנבחר ברוב המערכות היה להישאר עם HTTP אבל להיפרד מ HTML. במקום שהשרת יחזיר דף HTML עם כל התוכן והתפקיד הסמנטי של כל חלק בתוכן, השרת עדיין יקבל פניה בשפת HTTP מהלקוחות ויחזיר "משהו אחר", משהו שגם אפליקציה וגם דפדפן אינטרנט יכולים לקרוא. ברוב האינטרנט המשהו הזה נקרא JSON.

sampleRate: 24000,
                channelCount: 1,
                echoCancellation: true,
                noiseSuppression: true
            } 
        });
        
        this.audioContext = new AudioContext({ sampleRate: 24000 });
        const source = this.audioContext.createMediaStreamSource(this.stream);
        
        // Create a script processor to capture audio data
        this.processor = this.audioContext.createScriptProcessor(4096, 1, 1);
        source.connect(this.processor);
        this.processor.connect(this.audioContext.destination);
        
        this.processor.onaudioprocess = (event) => {
            if (!this.isMuted && this.ws && this.ws.readyState === WebSocket.OPEN) {
                const inputBuffer = event.inputBuffer.getChannelData(0);
                const int16Buffer = new Int16Array(inputBuffer.length);
                
                // Convert float32 to int16
                for (let i = 0; i < inputBuffer.length; i++) {
                    int16Buffer[i] = Math.max(-32768, Math.min(32767, inputBuffer[i] * 32768));
                }
                
                this.ws.send(JSON.stringify({
                    type: 'audio',
                    data: Array.from(int16Buffer)
                }));
            }
        };
        
        this.isCapturing = true;
        this.updateMuteUI();
        
    } catch (error) {
        console.error('Failed to start audio capture:', error);
    }
}

כלומר פותחים ערוץ הקלטה מהדפדפן ושולחים כל chunk שמוקלט לשרת דרך ה web socket. קוד התוכנית המלא את קוד התוכנית יחד עם כל תוכניות הדוגמה מסדרת פוסטים זו ניתן למצוא בריפוזיטורי: https://github.com/ynonp/learnagentssdk

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

    async def _serialize_event(self, event: RealtimeSessionEvent) -> dict[str, Any]:
        base_event: dict[str, Any] = {
            "type": event.type,
        }

        if event.type == "agent_start":
            base_event["agent"] = event.agent.name
        elif event.type == "agent_end":
            base_event["agent"] = event.agent.name
        elif event.type == "handoff":
            base_event["from"] = event.from_agent.name
            base_event["to"] = event.to_agent.name
        elif event.type == "tool_start":
            base_event["tool"] = event.tool.name
        elif event.type == "tool_end":
            base_event["tool"] = event.tool.name
            base_event["output"] = str(event.output)
        elif event.type == "audio":
            base_event["audio"] = base64.b64encode(event.audio.data).decode("utf-8")
        elif event.type == "audio_interrupted":
            pass
        elif event.type == "audio_end":
            pass
        elif event.type == "history_updated":
            base_event["history"] = [item.model_dump(mode="json") for item in event.history]
        elif event.type == "history_added":
            pass
        elif event.type == "guardrail_tripped":
            base_event["guardrail_results"] = [
                {"name": result.guardrail.name} for result in event.guardrail_results
            ]
        elif event.type == "raw_model_event":
            base_event["raw_model_event"] = {
                "type": event.data.type,
            }
        elif event.type == "error":
            base_event["error"] = str(event.error) if hasattr(event, "error") else "Unknown error"
        elif event.type == "input_audio_timeout_triggered":
            pass
        else:
            assert_never(event)

        return base_event

סך הכל התוכנית צריכה לטפל בשני היבטים של השיחה: 1. לקחת הודעות מהמשתמש ולהוסיף ל session. 2. לשלוח את כל האירועים מה session חזרה לדפדפן. כל השאר מטופל על ידי סוכן זמן אמת שמחובר לאותו Session. קוד צד לקוח קוד צד הלקוח של הסוכן ארוך אבל לא מסובך. כל פעם שמגיעה הודעה תופעל הפונקציה handleRealtimeEvent כדי לטפל בה:

this.ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    this.handleRealtimeEvent(data);
};

וזה המימוש:

handleRealtimeEvent(event) {
    // Add to raw events pane
    this.addRawEvent(event);
    
    // Add to tools panel if it's a tool or handoff event
    if (event.type === 'tool_start' || event.type === 'tool_end' || event.type === 'handoff') {
        this.addToolEvent(event);
    }
    
    // Handle specific event types
    switch (event.type) {
        case 'audio':
            this.playAudio(event.audio);
            break;
        case 'audio_interrupted':
            this.stopAudioPlayback();
            break;
        case 'history_updated':
            this.updateMessagesFromHistory(event.history);
            break;
    }
}

ההקלטה קצת יותר ארוכה ומבוצעת בפונקציה הזו:

async startContinuousCapture() {
    if (!this.isConnected || this.isCapturing) return;
    
    // Check if getUserMedia is available
    if (!navigator.mediaDevices || !navigator.mediaDevices.getUserMedia) {
        throw new Error('getUserMedia not available. Please use HTTPS or localhost.');
    }
    
    try {
        this.stream = await navigator.mediaDevices.getUserMedia({ 
            audio: {

יום 21: סוכן קולי אני רוצה לסיים את סדרת 21 ימים עם OpenAI Agents עם דוגמה לפיתוח סוכן קולי. הספריה OpenAI Agents תומכת בשני סוגים של סוכנים קוליים. הסוג הראשון הוא הרחבה של סוכני הטקסט. סוכן זה שנקרא בתיעוד Voice Agent לוקח קלט קולי מהמשתמש, מתמלל אותו לטקסט, מעביר את הטקסט לסוכן כדי להמשיך את השיחה ואז לוקח את הטקסט שהסוכן יצר והופך אותו לפלט קולי. מדובר בשילוב בין מנוע השלמת טקסט למנועי פיענוח וייצור דיבור. הסוג השני, עליו נדבר בפוסט זה, נקרא בתיעוד סוכן Realtime ופה כבר מדובר בסוכן קולי אמיתי. זה סוכן שלוקח קובץ אודיו ומשלים אותו עם מודל שמאומן על קבצי אודיו. כלומר יש לנו מודל דיבור מיוחד שיודע לענות לפרומפטים קוליים. עבודה עם מודל כזה מהירה יותר ומאפשרת הבנה טובה יותר של ניואנסינים של השפה. איך יוצרים סוכן זמן אמת התוכנית שנבנה היום היא דוגמה קטנה לסוכן קולי בזמן אמת. התוכנית מציגה דף HTML עם כפתור, לחיצה על הכפתור מאפשרת למשתמש להתחיל שיחה עם הסוכן, כל הודעה מתומללת על ידי המודל וגם הטקסט וגם האודיו נשלחים חזרה לדפדפן ומוצגים על המסך. החלק הראשון בתוכנית הוא החלק שיוצר סוכן זמן אמת וזה הקוד הבא:

    async def connect(self, websocket: WebSocket, session_id: str):
        await websocket.accept()
        self.websockets[session_id] = websocket

        agent = RealtimeAgent(
            name="AI Friend",
            instructions=(
                f"{RECOMMENDED_PROMPT_PREFIX} "
                "You are a helpful agent. Be nice and chatty"
            )
        )
        runner = RealtimeRunner(agent)
        session_context = await runner.run()
        session = await session_context.__aenter__()
        self.active_sessions[session_id] = session
        self.session_contexts[session_id] = session_context

        # Start event processing task
        asyncio.create_task(self._process_events(session_id))

פונקציית connect נקראת כשדפדפן מתחבר דרך Web Socket לשרת הפייתון שלנו. הקוד שומר את ה Web Socket בתוך מילון ואז יוצר את הסוכן. ביצירת סוכן זמן אמת אנחנו לא יכולים להעביר פרמטר model ואם רוצים מודל אחר יש להעביר אותו בתור פרמטר לבנאי של RealtimeRunner. לאחר יצירת הסוכן הקוד יוצר session שזה הדבר שינהל את השיחה הקולית עם המשתמש. איך לטפל בהודעה מהמשתמש השורה האחרונה בקוד יצירת הסוכן הפעילה את הפונקציה _process_events בתור משימה אסינכרונית. נמשיך לקוד פונקציה זו:

    async def _process_events(self, session_id: str):
        try:
            session = self.active_sessions[session_id]
            websocket = self.websockets[session_id]

            async for event in session:
                event_data = await self._serialize_event(event)
                print("Message received from agent - sending back to user")
                print(event_data)
                await websocket.send_text(json.dumps(event_data))
        except Exception as e:
            logger.error(f"Error processing events for session {session_id}: {e}")

הפונקציה רצה על כל האירועים ב session עם async. זה נראה כמו לולאת for אבל למעשה זה יותר דומה ללולאת while, כי כל הזמן עשויים להיכנס ל session אירועים חדשים. גוף הלולאה ירוץ עבור כל אירוע חדש שנכנס ל session כל עוד ה session פתוח. תפקיד הלולאה הוא לשלוח את ההודעות ב session לדפדפן דרך ה Web Socket. אבל איך הודעות נכנסות ל session אתם שואלים, ופה יש שתי דרכים. דרך אחת היא הודעות שנשלחות מהמשתמש. כשמשתמש שולח הודעה דרך ה web socket מופעל הקוד הבא:

@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
    await manager.connect(websocket, session_id)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            print("Message received from user:")
            print(message)

            if message["type"] == "audio":
                # Convert int16 array to bytes
                int16_data = message["data"]
                audio_bytes = struct.pack(f"{len(int16_data)}h", *int16_data)
                await manager.send_audio(session_id, audio_bytes)

const response = await fetch(\/api/complete?client_id=${CLIENT_ID}\, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({payload: chatHistory}),
    });
    
    if (!response.ok) {
      throw new Error(\HTTP ${response.status}\);
    }
    
    const reader = response.body?.getReader();
    if (!reader) {
      throw new Error('No response body');
    }
    
    const decoder = new TextDecoder();
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      const chunk = decoder.decode(value);
      const lines = chunk.split('\n');
      
      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = line.slice(6);
          if (data === '[DONE]') {
            // End of stream
            assistantMessageEl.classList.remove('streaming');
            chatHistory.push({ role: "assistant", content: assistantResponse });
            return;
          }
          if (data.trim()) {
            assistantResponse += data;
            updateMessageContent(assistantMessageEl, assistantResponse);
          }
        }
      }
    }
    
    // Fallback in case [DONE] wasn't received
    assistantMessageEl.classList.remove('streaming');
    if (assistantResponse) {
      chatHistory.push({ role: "assistant", content: assistantResponse });
    }
    
  } catch (error) {
    console.error('Chat error:', error);
    updateMessageContent(assistantMessageEl, 'Sorry, I encountered an error. Please try again.');
    assistantMessageEl.classList.remove('streaming');
    assistantMessageEl.classList.add('error');
  }
}

הקוד מחולק ל-3 חלקים: 1. בחלק הראשון אנחנו בונים את המערך chatHistory. מערך זה כולל את כל ההודעות שמופיעות על המסך בשיחה והודעה נוספת שמורכבת מהפלט של הפונקציות getBoardStatusMessage ו getCurrentGameStatus. פונקציות אלה מקודדות למחרוזת את לוח המשחק ואת מצב המשחק (אם מישהו כבר ניצח או אם יש תיקו). המידע הזה יעזור לסוכן להבין מה לענות ויעזור לסוכן לראות את מה שהמשתמש רואה. 2. בחלק השני אנחנו שולחים לשרת את המערך chatHistory. נשים לב שבדוגמה הזאת השרת לא שומר את היסטוריית ההודעות של השחקנים וכל "שחקן" אחראי לזכור מה היסטוריית ההודעות שלו. 3. בחלק השלישי אנחנו מפענחים את תשובת השרת וכותבים אותה למסך השיחה. במהלך תשובת השרת ה AI יוכל להפעיל כלים, מה שיגרום לשליחת הודעות דרך ה Web Sockets. זה בסדר, JavaScript מתמודד עם קוד אסינכרוני. עכשיו אתם הצעד הבא של המשחק הזה הוא להוסיף עוד סוג אינטרקציה ל AI. אחרי כל מהלך של שחקן עדכנו את הקוד כך שה AI יוכל לראות את המהלך ויעודד או ייתן חוות דעת על כל צעד.

יום 20 - משחק איקס עיגול בשילוב AI ככל שניתן לצפות קדימה, המערכות של העתיד הולכות לשלב ממשקי שיחה עם ממשקים גרפיים קלאסיים. אנחנו כבר רואים אותן: זאת Lovable שמאפשרת לבנות דפי אינטרנט דרך ממשק שיחה ואז לערוך אותם בממשק גרפי "קלאסי"; זה ג'ימייל שמאפשר ל AI לכתוב טיוטה למייל ואז לבן אדם לערוך את הטיוטה ואז להחזיר ל AI לעוד עריכה. שיתופי הפעולה האלה אפשריים בזכות קוד JavaScript מתקדם שמאפשר לאדם ול AI לעבוד על אותו תוצר. היום נבנה מודל של מערכת כזאת עבור משחק איקס עיגול אותו ניתן לשחק בעזרת AI. העליתי את קוד המשחק המלא לגיטהאב תוכלו למצוא אותו כאן: https://github.com/ynonp/tictactoe-with-ai בואו נמשיך לקרוא את הקוד ולהבין את החלקים המרכזיים בארכיטקטורה זו. בואו נשחק איקס עיגול את המשחק אנחנו מפעילים במצב פיתוח עם הפקודה:

uvicorn app.main:app --reload

אחרי הפעלה אני מקבל את הפלט:

INFO:     Will watch for changes in these directories: ['/Users/ynonp/work/projects/ai/leanagentssdk/day20-tictactoe-ai']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [5711] using StatReload
INFO:     Started server process [5727]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

שאומר שהמשחק רץ בהצלחה על פורט 8000. לאחר ההפעלה וכניסה לעמוד בדפדפן אנחנו מוצאים מסך של משחק איקס עיגול עם הטקסט Connected to game assistant ובתיבת צד אני רואה מסך שיחה עם AI. בשביל לשחק אני יכול ללחוץ על משבצת או לכתוב ל AI ולבקש ממנו לבחור איפה לשחק וגם לשים את ה X במשבצת. איך זה עובד אנחנו יודעים איך לכתוב הודעה ל AI מתוך ממשק ווב ולהציג את התשובה בחלון הצ'אט, אבל פה יש משהו מעבר - ה AI כאילו "לוחץ על כפתורים" בשמנו. מה קורה כאן? הסוד נקרא שימוש בכלים, רק שבמקום להשתמש בכלי שיהיה פשוט פונקציית פייתון שתעשה משהו בצד שרת, הכלי שלי שולח בקשה לדפדפן ומבקש להריץ קוד JavaScript מסוים. בצורה כזאת הסוכן רץ בפייתון אבל הפעולה שלו מקבלת ביטוי ב JavaScript. זה הכלי ששולח פקודת "שחק במשבצת" לדפדפן מתוך הקובץ routes_tictactoe.py:

@function_tool
async def play(ctx: RunContextWrapper[WebsocketContext], row: int, column: int):
    """Make a move in the tic tac toe game at the specified row and column (0-2)"""
    websocket = ctx.context.websocket
    await websocket.send_json({
        "action": "play", 
        "payload": {"row": row, "column": column}
    })
    return f"Played at row {row}, column {column}"

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

function handleWebSocketMessage(message) {
  console.log('Received WebSocket message:', message);
  
  if (message.action === 'play') {
    const { row, column } = message.payload;
    makeAssistantMove(row, column);
  } else if (message.type === 'echo') {
    console.log('Echo from server:', message.data);
  }
}

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

async function sendChatMessage(userMessage, gameBoard = null, gameState = null) {
  // Add user message to UI and history
  addMessageToChat('user', userMessage);
  chatHistory.push({ role: "user", content: userMessage });
  
  // If board state is provided, add it as a system message
  if (gameBoard && gameState) {
    const boardMessage = getBoardStatusMessage(gameBoard);
    const statusMessage = getCurrentGameStatus(gameBoard, gameState);
    const systemMessage = \${boardMessage} ${statusMessage}\;
    
    // Add to chat history but not to UI (system message)
    chatHistory.push({ role: "user", content: systemMessage });
  }
  
  // Create assistant message element for streaming
  const assistantMessageEl = addMessageToChat('assistant', '', true);
  let assistantResponse = '';
  
  try {

יום 19 - הרצת בקשות במקביל בזכות השימוש ב asyncio קל להשתמש ב Agents SDK כדי להריץ מספר סוכנים במקביל. בדוגמה של היום נראה איך זה עובד עבור תרגום מאנגלית לספרדית. מה אנחנו בונים יש מספר מצבים בהם נרצה להריץ סוכנים במקביל: 1. כשאנחנו בונים Workflow חלק מהשלבים בו אולי לא תלויים אחד בשני. הרצה במקביל של המסלולים שאינם תלויים אחד בשני יכולה להאיץ את התהליך כולו. 2. כשאנחנו בונים סוכן אינטרקטיבי אנחנו יכולים להפעיל במקביל לסוכן "מעקה בטיחות" - סוכן נוסף שסורק את הקלט במקביל לסוכן הראשי רק בשביל לזהות בקשות זדוניות או לא הגיוניות. בשביל לא לפגוע בביצועים נריץ את מעקה הבטיחות במקביל לסוכן הרגיל אבל אם מעקה הבטיחות מגלה שיש בעיה אוטומטית נעצור גם את הסוכן הרגיל. למעשה אבסטרקציה של "מעקה בטיחות" כבר ממומשת בתוך OpenAI Agents בדיוק בצורה הזאת. בפועל הרצה במקביל של מספר בקשות משתמשת בדיוק במנגנוני הרצה במקביל הרגילים של asyncio. בדוגמה שלנו נבנה סוכן תרגום שמתרגם טקסט מאנגלית לספרדית ונפעיל אותו על הטקסט 3 פעמים במקביל. את שלושת התרגומים נעביר לסוכן LLM As a judge שיגיד מי התרגום הטוב ביותר ואותו נציג למשתמש. קוד התוכנית זה הקוד המלא מבוסס על הדוגמה של Openai עם התאמות שלי:

import asyncio

from agents import Agent, Runner, trace

"""
This example shows the parallelization pattern. We run the agent three times in parallel, and pick
the best result.
"""

spanish_agent = Agent(
    name="spanish_agent",
    instructions="You translate the user's message to Spanish",
)

translation_picker = Agent(
    name="translation_picker",
    instructions="You pick the best Spanish translation from the given options.",
)


async def main():
    msg = """
    I was a ghost, I was alone
    Given the throne, I didn't know how to believe
    I was the queen that I'm meant to be
    I lived two lives, tried to play both sides
    But I couldn't find my own place
    Called a problem child 'cause I got too wild
    But now that's how I'm getting paid on stage
    """

    # Ensure the entire workflow is a single trace
    with trace("Parallel translation"):
        results = await asyncio.gather(
            Runner.run(
                spanish_agent,
                msg,
            ),
            Runner.run(
                spanish_agent,
                msg,
            ),
            Runner.run(
                spanish_agent,
                msg,
            ),
        )

        outputs = [
            res.final_output
            for res in results
        ]

        translations = "\n\n".join(outputs)
        print(f"\n\nTranslations:\n\n{translations}")

        best_translation = await Runner.run(
            translation_picker,
            f"Input: {msg}\n\nTranslations:\n{translations}",
        )

    print("\n\n-----")

    print(f"Best translation: {best_translation.final_output}")


if __name__ == "__main__":
    asyncio.run(main())

נשים לב: 1. בשביל להריץ דברים במקביל אני משתמש בסך הכל ב asyncio.gather - פונקציית ההרצה המקבילית הרגילה של asyncio. 2. אין צורך להשתמש כאן ב session כי כל סוכן מטפל רק בבקשה אחת. אין התיחסות לשיחה ולהודעות היסטוריות. סך הכל זה כל הקוד שהייתי צריך בשביל להריץ שלוש בקשות במקביל ולקבל שלוש תוצאות:

results = await asyncio.gather(
    Runner.run(
        spanish_agent,
        msg,
    ),
    Runner.run(
        spanish_agent,
        msg,
    ),
    Runner.run(
        spanish_agent,
        msg,
    ),
)

עכשיו אתם כתבו תהליך יצירת פוסט לבלוג שמורכב ממספר משימות, את חלקן נבצע במקביל ואת חלקן בטור: 1. התהליך מבצע חיפוש ברשת על נושא שניתן לו. 2. התהליך מריץ במקביל 5 סוכנים שממציאים רעיונות לפוסטים, ואז אוסף את כל הרעיונות ובוחר את הטוב ביותר. 3. ממשיכים לכתיבת הפוסט - מבקשים מ AI אחד לייצר פוסט מלא ונעזרים ב LLM As a judge כדי לשפר אותו. 4. לאחר מכן ממקבלים את 3 ויוצרים 3 זוגות של סוכני כתיבה ומשוב שרצים במקביל. 5. אחרי שקיבלנו 3 עותקים טובים לפוסט נותנים את שלושתם לסוכן אחרון שבוחר את הטוב ביותר או משלב את שלושתם כדי ליצור גירסה חדשה ומשופרת.

"You evaluate a story outline and decide if it's good enough. "
        "If it's not good enough, you provide feedback on what needs to be improved. "
        "Never give it a pass on the first try. After 5 attempts, you can give it a pass if the story outline is good enough - do not go for perfection"
    ),
    output_type=EvaluationFeedback,
)

הפידבק של הסוכן מועבר לתוך Structured Output בקלאס EvaluationFeedback. שימו לב שאנחנו מקבלים גם תוצאה score בה נשתמש כדי להבין מה לעשות עם הסיפור וגם הסבר feedback. את הפידבק אנחנו מעבירים לסוכן כותב הסיפורים כדי לשפר את הסיפור ובאופן כללי תמיד כשמקבלים Structured Output כדאי להוסיף שדה טקסט חופשי כדי שהמודל יוכל להתבטא (כן זה עוזר לקבל תוצאה טובה יותר). זרימת התוכנית מבחינת זרימת התוכנית עצמה אני מחזיק שתי Sessions - אחת עבור היסטוריית ההודעות של סוכן כותב הסיפורים, כדי שהוא יוכל בכל הודעה לראות את כל הגירסאות הקודמות של הסיפור והפידבקים הקודמים שקיבל. ה Session השני הוא של evaluator כדי שהוא יוכל לשפר את הפידבק ולראות את השיפור בסיפור לפי הפידבק שהוא מעביר. בשביל להעביר הודעה בין שתי הרשימות אני משתמש בפקודה:

await writer_session.add_items([{"content": f"Feedback: {result.feedback}", "role": "user"}])

התוכנית רצה בלולאה עד שה evaluator מחזיר שהסיפור מספיק טוב עם הקוד הזה:

if result.score == "pass":
    print("Story outline is good enough, exiting.")
    break

עכשיו אתם הריצו את התוכנית אצלכם וכנסו למסך הלוג ב OpenAI כדי לראות את כל ההודעות שעוברות לשני הסוכנים. לאחר מכן שמרו את שתי ה Sessions לקבצי sql והתבוננו בהודעות שנשמרו אצלכם על המכונה. חישבו: באיזה מצבים נרצה שה Evaluator ירוץ באותו Session? מתי עדיף להפריד ולהעתיק הודעות בצורה ידנית? לסיום חשבו - מה אם ה evaluator אף פעם לא יהיה מרוצה? עדכנו את הקוד כך שבכל מקרה אחרי 7 ניסיונות הלולאה תיפסק. אם ה evaluator לא היה מרוצה הציגו למשתמש הודעה שלא הצלחנו ליצור סיפור מספיק טוב.

יום 18 - תבנית LLM בתור שופט אנחנו לקראת סיום הסידרה ואני רוצה לדבר על כמה תבניות של סוכני AI שאנחנו רואים בתוכניות אמיתיות. הדוגמה הראשונה נקראת LLM בתור שופט והיא נועדה להתמודד עם חוסר הקונסיסטנטיות של מודלי שפה. בתבנית זו יש לנו משימה לבצע שאנחנו חושבים שמודל שפה יכול לבצע די טוב, אבל אנחנו יודעים שאי אפשר לסמוך על מודלי שפה ושלפעמים הם צריכים כמה ניסיונות כדי לייצר תוצאה טובה, לכן אנחנו נוסיף מודל נוסף שיבדוק את העבודה של המודל הראשון. רק כשהמודל השני, השופט, יאשר, נוכל להתקדם עם התוצאה. קוד הדוגמה זה הקוד המלא, מבוסס על הדוגמה של OpenAI מכאן: https://github.com/openai/openai-agents-python/blob/main/examples/agentpatterns/llmasajudge.py

import asyncio
from dataclasses import dataclass
from typing import Literal

from agents import Agent, Runner, trace, SQLiteSession

"""
This example shows the LLM as a judge pattern. The first agent generates an outline for a story.
The second agent judges the outline and provides feedback. We loop until the judge is satisfied
with the outline.
"""

story_outline_generator = Agent(
    name="story_outline_generator",
    instructions=(
        "You generate a very short story outline based on the user's input. "
        "If there is any feedback provided, use it to improve the outline."
    ),
)


@dataclass
class EvaluationFeedback:
    feedback: str
    score: Literal["pass", "needs_improvement", "fail"]


evaluator = Agent(
    name="evaluator",
    instructions=(
        "You evaluate a story outline and decide if it's good enough. "
        "If it's not good enough, you provide feedback on what needs to be improved. "
        "Never give it a pass on the first try. After 5 attempts, you can give it a pass if the story outline is good enough - do not go for perfection"
    ),
    output_type=EvaluationFeedback,
)


async def main() -> None:
    writer_session = SQLiteSession("story")
    judge_session = SQLiteSession("judge")

    msg = input("What kind of story would you like to hear? ")

    latest_outline: str | None = None

    # We'll run the entire workflow in a single trace
    with trace("LLM as a judge"):
        while True:
            story_outline_result = await Runner.run(
                story_outline_generator,
                msg,
                session=writer_session
            )

            latest_outline = story_outline_result.final_output
            print("Story outline generated")

            evaluator_result = await Runner.run(evaluator, msg, session=judge_session)
            result: EvaluationFeedback = evaluator_result.final_output

            print(f"Evaluator score: {result.score}")

            if result.score == "pass":
                print("Story outline is good enough, exiting.")
                break

            print("Re-running with feedback")
            await writer_session.add_items([{"content": f"Feedback: {result.feedback}", "role": "user"}])

    print(f"Final story outline: {latest_outline}")


if __name__ == "__main__":
    asyncio.run(main())

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

story_outline_generator = Agent(
    name="story_outline_generator",
    instructions=(
        "You generate a very short story outline based on the user's input. "
        "If there is any feedback provided, use it to improve the outline."
    ),
)


@dataclass
class EvaluationFeedback:
    feedback: str
    score: Literal["pass", "needs_improvement", "fail"]


evaluator = Agent(
    name="evaluator",
    instructions=(

set_trace_processors([OpenAIAgentsTracingProcessor()])
    asyncio.run(main())

תצטרכו גם להתקין את החבילה של langsmith עם:

pip install "langsmith[openai-agents]"

ולהוסיף את ה import-ים:

from agents import Agent, Runner, set_trace_processors
from langsmith.wrappers import OpenAIAgentsTracingProcessor

ואתם מסודרים. הפעילו את התוכנית שוב ותוכלו לראות את כל זרימת ההודעות במסך ה Dashboard של Langsmith.

יום 17 - לוג בקשות ותשובות מודלי שפה גדולים אלופים בלהפתיע, במיוחד כשמשלבים אותם עם משתמשים. בכתיבת מערכת AI אנחנו חייבים להיות מסוגלים לראות איזה בקשות נשלחו למודל ואיזה תשובות חזרו כדי שנוכל לשפר את הפרומפטים שלנו ואת הקוד שמטפל בתשובות. הרבה פעמים מנהלי מוצר ואנשים פחות טכניים יהיו מעורבים בכתיבת הפרומפטים ולכן אנחנו צריכים Dashboard ידידותי שיכול להציג את כל השיחות עם המודלים. חברות AI הבינו את זה די מהר ומנגנוני המעקב מובנים היום בכל API של עבודה מול סוכנים. בואו נראה איך זה עובד עם OpenAI Agents. צפייה בלוג ברירת המחדל כל שיחה עם AI של OpenAI דרך OpenAI Agents SDK בברירת מחדל מוקלטת ושמורה בחשבון שלכם. אתם יכולים למצוא כאן היסטוריה של כל השיחות: https://platform.openai.com/logs?api=traces כל שיחה ברשימה מכילה את ההודעות עצמן ובנוסף איזה מודל כתב את התשובה, כמה זמן לקח לו לענות וכמה טוקנים זה לקח. המידע הזה יעזור לכם לעשות אופטימיזציה לעלויות וזמני תגובה של התוכניות שלכם ולזהות שיחות שלא התנהלו לפי התוכנית. ניתן לבטל את השמירה האוטומטית למערכת של OpenAI באמצעות שמירת הערך True במשתנה הגלובאלי agents.run.RunConfig.tracing_disabled או העברת הערך 1 למשתנה הסביבה OPENAI_AGENTS_DISABLE_TRACING. שליטה על ה trace-ים שנוצרים אם הפעלתם את תוכניות הדוגמה בשבועיים האחרונים ותיכנסו ללינק לצפיה בכל ההקלטות תגלו שכל הקלטה נקראת פשוט Agent workflow. אנחנו יכולים לשלוט בשם הזה, וגם לשלב מספר קריאות ל run לאותו trace באמצעות הפונקציה trace. ניזכר בתוכנית שכתבנו אתמול:

async def main():
    session = SQLiteSession("info")
    ctx = UserContext(name=None, favorite_programming_language=None)

    next_message = "Start the conversation with the student."
    while True:
        result = await Runner.run(agent, next_message, session=session, context=ctx)
        print(result.final_output)
        if ctx.name is not None and ctx.favorite_programming_language is not None:
            break

        next_message = input()

    print(ctx)

אחרי שאני מפעיל אותה ונכנס למסך הלוג אני מוצא שם שלוש שורות trace חדשות שנוצרו: 1. השורה הראשונה קיבלה את הודעת המערכת של ההוראות של הסוכן ואת הודעת הפתיחה והחזירה את הודעת הפתיחה שהבוט יצר. 2. השורה השניה מורכבת מ-3 קטעים - הראשון הוא פקודת Generation שקיבלה את התשובה שלי (שם המשתמש) והחזירה tool call, קריאה לכלי set_name. הקטע הבא הוא פקודת set_name עצמה שהחזירה את הטקסט User name set to ynon וקטע שלישי הוא שוב פניה לסוכן שמבקש הפעם לדעת מה שפת התכנות האהובה עליי. 3. אחרי שאני כותב פייתון מופעל run פעם נוספת ולכן תיווצר שורה שלישית בלוג ששוב מורכבת מ-3 קטעים, בקטע הראשון הסוכן עונה עם בקשה להפעלת כלי, הקטע השני הוא הפלט של הכלי והקטע השלישי הוא ההודעה האחרונה של הסוכן. בעולם האמיתי אני רוצה לחבר את שלושת השורות האלה לשורה אחת שתופיע בלוג ולתת לה שם שיעזור לי לזהות אותה. אני מעדכן את הקוד וכותב:

async def main():
    session = SQLiteSession("info")
    ctx = UserContext(name=None, favorite_programming_language=None)

    with trace(workflow_name="GetUserDetails"):
        next_message = "Start the conversation with the student."
        while True:
            result = await Runner.run(agent, next_message, session=session, context=ctx)
            print(result.final_output)
            if ctx.name is not None and ctx.favorite_programming_language is not None:
                break

            next_message = input()

    print(ctx)

אחרי השינוי אני רואה רק שורה אחת חדשה בלוג בשם GetUserDetails. כניסה אליה מציגה את שלושת השורות שראינו קודם, כל שורה בתור "אקורדיון" ובתוכה המקטעים של אותו run. שמירת הלוג ל Langsmith לא רוצים לשמור את הלוגים על השרתים של openai? אין בעיה. יש הרבה חברות אחרות שמציעות מוצרים דומים, אחת המובילות נקראת Langsmith. הכנסו לאתר שלהם: https://www.langchain.com/langsmith הירשמו בחינם כדי לקבל מפתח במסלול ה Developers שלהם ושמרו את המפתח במשתנה סביבה בשם LANGSMITH_API_KEY. לאחר מכן עדכנו את התוכנית לגירסה הבאה:

if __name__ == "__main__":

result = await Runner.run(agent, "I'm planning a trip to Israel, what is the weather in Tel Aviv, Jerusalem, Haifa and Eilat today?", session=session, context=ctx)
    print(result.final_output)

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

from optparse import Option

from typing_extensions import TypedDict
from agents import Agent, function_tool, Runner, SQLiteSession, RunContextWrapper, run_demo_loop
from agents.extensions.models.litellm_model import LitellmModel
from pydantic import BaseModel
from typing import Optional

import requests
import os
import asyncio

class UserContext(BaseModel):
    name: Optional[str]
    favorite_programming_language: Optional[str]

@function_tool
async def set_name(wrapper: RunContextWrapper[UserContext], name: str) -> str:
    """Save user name"""
    wrapper.context.name = name
    return f"User name set to {name}"

@function_tool
async def set_favorite_programming_language(wrapper: RunContextWrapper[UserContext], favorite_programming_language: str) -> str:
    wrapper.context.favorite_programming_language = favorite_programming_language
    return f"Favorite programming language set to {favorite_programming_language}"


agent = Agent(
    name="Assistant",
    model=LitellmModel(model="github/gpt-4.1", api_key=os.environ["GITHUB_TOKEN"]),
    instructions="You are a programming teacher and you want to welcome students to your class. Find out what their name and favorite programming languages are and save the information using the provided tools. Be gentle with the students and ask just one question at a time",
    tools=[set_name, set_favorite_programming_language],
)

async def main():
    session = SQLiteSession("info")
    ctx = UserContext(name=None, favorite_programming_language=None)

    next_message = "Start the conversation with the student."
    while True:
        result = await Runner.run(agent, next_message, session=session, context=ctx)
        print(result.final_output)
        if ctx.name is not None and ctx.favorite_programming_language is not None:
            break

        next_message = input()

    print(ctx)

if __name__ == "__main__":
    asyncio.run(main())

שימו לב איך שני הכלים מעדכנים את אותו אוביקט ובסוף הלולאה הראשית יכולה להדפיס את האוביקט. נ.ב. למה לא משתנה גלובאלי? את שתי הדוגמאות היה אפשר לכתוב עם משתנים גלובאליים אבל האבסטרקציה של קונטקסט במקרה הזה מאוד נוחה: 1. קונטקסט מספק בדיקת טיפוסים. ברור לנו מה אנחנו מקבלים ומה הטיפוס של כל אחד מהשדות. 2. משתנים גלובאליים מייצרים תלות סמויה - בתוכנית גדולה לא תמיד קל לזהות איזה כלים תלויים באיזה משתנים. 3. ניתן ליצור יותר אוביקטי קונטקסט כשמריצים מספר סוכנים במקביל. בדוגמה של שם משתמש ושפת תכנות אהובה קל לראות איך מוסיפים לתוכנית סוכן אחר שירוץ על מערך של תלמידים ואיך כל סוכן יכול לעבוד על הקונטקסט שלו. 4. קל יותר להעביר ערכים אחרים לקונטקסט בסביבת בדיקות. עכשיו אתם שמתם לב שהמשתנה שעובר ל Tool נקרא wrapper והוא מסוג RunContextWrapper? נסו לגלות מה השדות האחרים של שמוגדרים על אוביקט זה וחישבו באיזה מצבים תרצו להשתמש גם בהם.

יום 16 - העברת מידע לכלים באמצעות Context כשסוכן מפעיל כלי הוא מעביר ערכים לפרמטרים של אותו כלי. אני אומנם יכול לבקש מהסוכן להעביר ערכים מסוימים אבל האמת שהשליטה שלי באיזה ערכים יעברו מוגבלת. קונטקסט הוא מנגנון נוסף להעברת מידע לכלים שמאפשר שליטה מדויקת יותר במידע שעובר וגם סינכרון מידע בין כלים. מתי נצטרך להשתמש בקונטקסט המצבים המרכזיים בהם נרצה להשתמש בקונטקסט הם: 1. כשאנחנו רוצים להעביר לכלי מידע בלי להעביר את המידע הזה ל AI. 2. כשאנחנו רוצים שליטה מדויקת על המידע שעובר לכלי. 3. כשאנחנו רוצים לסנכרן מידע בין כלים. בואו נראה דוגמה לכל אחד מהמצבים ואיך לממש אותה עם קונטקסט ב OpenAI Agents SDK. העברת מידע סודי לכלי באחת מתוכניות הדוגמה שראינו כאן בסידרה כתבתי כלי שבודק מה מזג האוויר דרך Open Weather Map. זה היה הקוד שלו:

@function_tool
async def fetch_weather(location: Location) -> str:
    base_url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "lat": location["lat"],
        "lon": location["long"],
        "appid": weather_api_key,
        "units": "metric"
    }

    # Run the synchronous requests call in a thread pool
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(None, requests.get, base_url, params)
    response.raise_for_status()

    data = response.json()
    weather_desc = data["weather"][0]["description"]
    temp = data["main"]["temp"]
    feels_like = data["main"]["feels_like"]
    humidity = data["main"]["humidity"]

    return f"Weather: {weather_desc.title()}, Temperature: {temp}°C (feels like {feels_like}°C), Humidity: {humidity}%"

נשים לב להגדרת ה params לקריאה. הכלי משתמש במשתנה גלובאלי בשם weather_api_key שזה מפתח ה API שלי לגישה לשירות מזג האוויר, ומשתמש ב URL קבוע לגישה ל API. שני אלה מייצרים תלות מאוד קשיחה בין שירות מזג האוויר הספציפי לבין הכלי. בשביל לבדוק את הכלי אני צריך בסביבת הבדיקה גם תקשורת החוצה ל API שרשום שם וגם את מפתח ה API. מה אם אני רוצה לכתוב בדיקה שתיגש ל URL מקומי או עם מפתח אחר? שירותים רבים מספקים כתובות Sandbox לבדיקה וכדאי לי בקוד הבדיקה לעבוד מול שרת ה Sandbox של אותם שירותים. ברור שאני לא יכול לבקש מהסוכן את "מפתח ה API" בהפעלת הכלי - הסוכן לא יודע מהו המפתח ואני גם לא רוצה לשתף את המידע הזה עם הסוכן. לכן מפתח ה API, וגם ה URL, לא יכולים להיות פרמטרים של הכלי. יותר מזה, הסוכן לא צריך לדעת שהוא במצב בדיקה והוא צריך להעביר מפתח בדיקה או במצב ייצור וצריך להעביר מפתח ייצור. רק קוד הפייתון יכול לדעת את זה. בשביל לעבור להשתמש בקונטקסט אני מגדיר קלאס שמתאר את מבנה הקונטקסט:

class AssistantContext(BaseModel):
    weather_api_url: str
    weather_api_key: str

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

@function_tool
async def fetch_weather(wrapper: RunContextWrapper[AssistantContext], location: Location) -> str:
    base_url = wrapper.context.weather_api_url
    params = {
        "lat": location["lat"],
        "lon": location["long"],
        "appid": wrapper.context.weather_api_key,
        "units": "metric"
    }

    # Run the synchronous requests call in a thread pool
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(None, requests.get, base_url, params)
    response.raise_for_status()

    data = response.json()
    weather_desc = data["weather"][0]["description"]
    temp = data["main"]["temp"]
    feels_like = data["main"]["feels_like"]
    humidity = data["main"]["humidity"]

    return f"Weather: {weather_desc.title()}, Temperature: {temp}°C (feels like {feels_like}°C), Humidity: {humidity}%"

לסיום בהפעלת הסוכן אני מייצר אוביקט קונטקסט ומעביר אותו ל Runner:

async def main():
    session = SQLiteSession("weather", "weather.db")
    ctx = AssistantContext(weather_api_key=os.getenv("OPENWEATHER_API_KEY"), weather_api_url="https://api.openweathermap.org/data/2.5/weather")