Construire un agent vocal complet

Construire un agent vocal complet

Cette leçon rassemble tous les concepts vus précédemment pour construire un agent vocal fonctionnel de bout en bout. Vous allez créer un assistant qui écoute l’utilisateur, comprend sa demande, exécute des actions et répond vocalement — le tout en temps réel via la Realtime API.

Architecture de l’agent

Un agent vocal complet se compose de quatre modules :

La classe VoiceAgent

Voici la structure complète d’un agent vocal :

import asyncio
import json
import base64
import websockets
import sounddevice as sd
import numpy as np
import pyaudio

class VoiceAgent:
    """Agent vocal complet basé sur la Realtime API."""

    def __init__(self, api_key: str, instructions: str, tools: list = None):
        self.api_key = api_key
        self.instructions = instructions
        self.tools = tools or []
        self.ws = None
        self.is_playing = False
        self.sample_rate = 24000
        self.chunk_size = 2400  # 100ms

    async def connect(self):
        """Établit la connexion et configure la session."""
        url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "OpenAI-Beta": "realtime=v1"
        }

        self.ws = await websockets.connect(url, extra_headers=headers)

        # Attendre session.created
        event = json.loads(await self.ws.recv())
        print(f"Session : {event['session']['id']}")

        # Configurer
        await self.ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": self.instructions,
                "voice": "nova",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {"model": "whisper-1"},
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 600
                },
                "tools": self.tools,
                "tool_choice": "auto",
                "temperature": 0.7
            }
        }))

        await self.ws.recv()  # session.updated
        print("Agent vocal prêt.")

    async def start(self):
        """Lance l'agent vocal."""
        await self.connect()
        await asyncio.gather(
            self._capture_audio(),
            self._process_events()
        )

Module de capture audio

    async def _capture_audio(self):
        """Capture le microphone et envoie à la Realtime API."""
        loop = asyncio.get_event_loop()
        queue = asyncio.Queue()

        def callback(indata, frames, time_info, status):
            pcm = (indata[:, 0] * 32767).astype(np.int16)
            loop.call_soon_threadsafe(queue.put_nowait, pcm)

        stream = sd.InputStream(
            samplerate=self.sample_rate,
            channels=1,
            dtype="float32",
            blocksize=self.chunk_size,
            callback=callback
        )

        with stream:
            while True:
                pcm_data = await queue.get()
                audio_b64 = base64.b64encode(pcm_data.tobytes()).decode()
                await self.ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": audio_b64
                }))

Module de traitement des événements

    async def _process_events(self):
        """Traite tous les événements du serveur."""
        pa = pyaudio.PyAudio()
        player = pa.open(
            format=pyaudio.paInt16, channels=1,
            rate=self.sample_rate, output=True
        )

        try:
            async for raw in self.ws:
                event = json.loads(raw)
                t = event["type"]

                if t == "response.audio.delta":
                    audio = base64.b64decode(event["delta"])
                    player.write(audio)
                    self.is_playing = True

                elif t == "response.audio.done":
                    self.is_playing = False

                elif t == "response.audio_transcript.delta":
                    print(event["delta"], end="", flush=True)

                elif t == "input_audio_buffer.speech_started":
                    if self.is_playing:
                        await self.ws.send(json.dumps({
                            "type": "response.cancel"
                        }))
                        self.is_playing = False

                elif t == "response.done":
                    await self._handle_response_done(event)

                elif t == "error":
                    print(f"\nErreur : {event['error']['message']}")

        finally:
            player.close()
            pa.terminate()

Module de gestion des outils

    async def _handle_response_done(self, event):
        """Traite les appels de fonctions dans la réponse."""
        response = event["response"]
        function_calls = [
            o for o in response.get("output", [])
            if o["type"] == "function_call"
        ]

        if not function_calls:
            print()  # Nouvelle ligne après la transcription
            return

        for call in function_calls:
            name = call["name"]
            args = json.loads(call["arguments"])
            call_id = call["call_id"]

            print(f"\n→ {name}({args})")

            # Exécuter la fonction (à adapter selon votre logique métier)
            result = await self._execute_function(name, args)

            await self.ws.send(json.dumps({
                "type": "conversation.item.create",
                "item": {
                    "type": "function_call_output",
                    "call_id": call_id,
                    "output": json.dumps(result)
                }
            }))

        # Demander au modèle de continuer
        await self.ws.send(json.dumps({"type": "response.create"}))

    async def _execute_function(self, name: str, args: dict) -> dict:
        """Dispatche l'appel de fonction."""
        # À implémenter selon votre logique métier
        return {"error": f"Fonction {name} non implémentée"}

Lancer l’agent

async def main():
    agent = VoiceAgent(
        api_key="sk-...",
        instructions=(
            "Vous êtes un assistant vocal de la société Corsen AI. "
            "Vous aidez les utilisateurs à gérer leurs rendez-vous "
            "et à obtenir des informations sur nos services. "
            "Répondez en français, de manière concise et professionnelle."
        ),
        tools=[
            {
                "type": "function",
                "name": "check_availability",
                "description": "Vérifie les disponibilités pour un créneau",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "date": {"type": "string"},
                        "service": {"type": "string"}
                    },
                    "required": ["date"]
                }
            }
        ]
    )

    await agent.start()

asyncio.run(main())

Points clés à retenir

• Un agent vocal complet combine quatre modules : audio, session, outils et gestion d’état
• La classe VoiceAgent encapsule toute la logique dans une interface simple
• asyncio.gather permet la capture micro et le traitement des événements en parallèle
• La gestion des interruptions (annulation de réponse quand l’utilisateur parle) est indispensable
• Les outils transforment l’agent d’un simple chatbot vocal en un assistant capable d’agir