אני רוצה לסיים את סדרת 21 ימים עם OpenAI Agents עם דוגמה לפיתוח סוכן קולי. הספריה OpenAI Agents תומכת בשני סוגים של סוכנים קוליים.

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

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

1. איך יוצרים סוכן זמן אמת

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

2. איך לטפל בהודעה מהמשתמש

השורה האחרונה בקוד יצירת הסוכן הפעילה את הפונקציה _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)

השרת לוקח את הקול מההודעה ומוסיף את ההודעה הקולית ל session באמצעות פקודת send_audio. כאן משמעות המילה send היא לא לשלוח את ההודעה לדפדפן אלא לשלוח הודעה קולית ל session.

3. איך הסוכן מדווח תשובות חזרה למשתמש

אחרי שההודעה הקולית מגיעה מהמשתמש ל 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.

4. קוד צד לקוח

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

5. קוד התוכנית המלא

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

https://github.com/ynonp/learnagentssdk