Construire un agent vocal complet
Sur cette page
Mise à jour du 2 novembre 2026 — modèles audio. Plusieurs modèles de ce cours sont dépréciés :
tts-1,tts-1-hdetgpt-4o-mini-ttss’arrêtent le 6 janvier 2027 (remplaçant indiqué :gpt-realtime-2.1-mini, que le guide TTS d’OpenAI déclare non pris en charge par/v1/audio/speech: la migration de la synthèse vocale passe par l’API Realtime) ;gpt-realtime,gpt-4o-realtimeet les anciens modèles audio le 20 janvier 2027 (gpt-realtime-2.1,gpt-realtime-2.1-mini,gpt-audio-1.5) ;whisper-1,gpt-4o-transcribeet/v1/audio/translationsle 26 février 2027 (gpt-transcribe,gpt-live-transcribe,v1/realtime/translations). Pour la voix full-duplex, GPT-Live 1 est disponible dans l’API depuis le 10 septembre 2026.
Corrigé le 3 novembre 2026. La transcription de l’entrée utilisait
gpt-live-transcribe, qui ne fonctionne que dans une session de transcription et ne prend en charge niserver_vadnisemantic_vad. Dans une session de conversation comme celle-ci, où le VAD valide chaque tour, la documentation désignegpt-transcribe, qui transcrit les tours validés et se sert des tours précédents comme contexte. Sources : Realtime transcription · GPT-Transcribe · GPT-Live-Transcribe.
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-transcribe", "languages": ["fr"]},
"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