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
Quatre modules, et surtout une contrainte qui commande toute l’architecture : tout doit se passer en même temps. Le micro capture pendant que le modèle répond, les événements arrivent pendant que l’audio se joue, l’utilisateur peut couper la parole à n’importe quel instant. Un agent vocal n’est pas une boucle question-réponse, c’est un système où plusieurs flux avancent en parallèle — d’où l’usage massif d’asyncio dans ce qui suit.
La classe VoiceAgent
La classe rassemble l’état que ces flux parallèles doivent partager : la connexion, la session en cours, la réponse en train d’être jouée. C’est précisément parce que plusieurs tâches y accèdent simultanément qu’il vaut mieux un objet qu’un ensemble de variables globales — le jour où il faut annuler une réponse en cours, il faut savoir qu’il y en a une.
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-realtime-2.1"
headers = {
"Authorization": f"Bearer {self.api_key}"
}
self.ws = await websockets.connect(url, additional_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": {
"type": "realtime",
"output_modalities": ["audio"],
"instructions": self.instructions,
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": 24000},
"transcription": {"model": "gpt-live-transcribe"},
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 600
}
},
"output": {
"format": {"type": "audio/pcm", "rate": 24000},
"voice": "marin"
}
},
"tools": self.tools,
"tool_choice": "auto"
}
}))
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.output_audio.delta":
audio = base64.b64decode(event["delta"])
player.write(audio)
self.is_playing = True
elif t == "response.output_audio.done":
self.is_playing = False
elif t == "response.output_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
C’est ce module qui fait passer l’agent du statut de chatbot vocal à celui d’assistant capable d’agir. La différence avec le function calling classique tient à la contrainte temporelle : l’utilisateur attend en ligne, sans indication visuelle de progression. Au-delà d’une seconde ou deux, faites parler l’agent pendant l’exécution — « je vérifie cela tout de suite » — car un silence dans une conversation vocale est interprété comme une panne.
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
VoiceAgentencapsule toute la logique dans une interface simple asyncio.gatherpermet 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