Aller au contenu principal
Leçon 17 sur 18 · API Temps Réel et Agents Vocaux

Latence et optimisation temps réel

Leçon 17 sur 18 Mis à jour le 3 novembre 2026 Développeur
Sur cette page
  1. Latence et optimisation temps réel
  2. Anatomie de la latence
  3. Optimiser la détection VAD
  4. Optimiser les instructions système
  5. Réduire la latence réseau
  6. Optimiser le function calling
  7. Monitoring de la latence
  8. Points clés à retenir

Mise à jour du 2 novembre 2026 — modèles audio. Plusieurs modèles de ce cours sont dépréciés : tts-1, tts-1-hd et gpt-4o-mini-tts s’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-realtime et 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-transcribe et /v1/audio/translations le 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. Les durées attribuées à chaque étape (« VAD 300-700 ms », « réseau 20-100 ms », « modèle 200-800 ms ») et les seuils de perception (« au-delà de 500 ms ») n’avaient pas de source : OpenAI ne publie pas de tels chiffres, et la leçon apprend désormais à les mesurer. La consigne de placer son serveur « dans la même région que les serveurs OpenAI (US East) » a été retirée, faute de source. Ajouts documentés : semantic_vad, WARP pour accélérer l’ouverture d’une session WebRTC. Sources : Voice activity detection · WebRTC with WARP · Cost optimization.

Latence et optimisation temps réel

La latence est l’ennemie numéro un d’un agent vocal : le silence qui suit la fin d’une question se remarque vite au téléphone comme dans une application. Cette leçon détaille les sources de latence, la façon de les mesurer et les techniques pour les réduire.

Anatomie de la latence

La latence totale d’un agent vocal est la somme de plusieurs composantes :

VAD
Attente du silence qui clôt le tour (`silence_duration_ms`)
Réseau
Allers-retours entre l'utilisateur, votre serveur et l'API
Modèle
Traitement du contexte, puis premier morceau d'audio

S’y ajoutent, selon votre architecture, les outils que l’agent appelle et le temps d’ouverture de la session. Aucune de ces durées n’est fixe : elles dépendent de vos réglages, de la longueur de la conversation et du trajet réseau. La seule façon de savoir où part le temps est de chronométrer chaque étape, comme le fait le moniteur présenté plus bas.

Optimiser la détection VAD

Avec server_vad, la réponse ne peut pas partir avant la fin du silence que fixe silence_duration_ms : c’est donc un réglage direct de la latence perçue, à arbitrer contre le risque de couper la parole. Les valeurs ci-dessous sont des points de départ à tester, pas des recommandations d’OpenAI :

# Configuration agressive (réponse rapide, risque de couper l'utilisateur)
fast_vad = {
    "type": "server_vad",
    "threshold": 0.6,           # Seuil plus élevé = moins de faux positifs
    "prefix_padding_ms": 200,
    "silence_duration_ms": 400  # Réaction rapide
}

# Configuration patiente (moins de coupures, plus de latence)
patient_vad = {
    "type": "server_vad",
    "threshold": 0.4,
    "prefix_padding_ms": 400,
    "silence_duration_ms": 800
}

# Alternative : le VAD sémantique, qui juge la fin de phrase sur les mots ;
# eagerness "high" répond au plus tôt, "low" laisse l'utilisateur finir
semantic_vad = {"type": "semantic_vad", "eagerness": "high"}

# Configuration adaptative selon le contexte
def get_vad_config(context: str) -> dict:
    configs = {
        "menu_ivr": {"threshold": 0.6, "silence_duration_ms": 400},
        "conversation": {"threshold": 0.5, "silence_duration_ms": 600},
        "dictation": {"threshold": 0.3, "silence_duration_ms": 1200},
    }
    config = configs.get(context, configs["conversation"])
    return {
        "type": "server_vad",
        "prefix_padding_ms": 300,
        **config
    }

Optimiser les instructions système

Des instructions concises réduisent le contexte à traiter. Rappelez-vous aussi que la Realtime API applique la mise en cache des prompts automatiquement : des instructions et des outils stables, placés en tête et jamais modifiés en cours de session, se réutilisent d’un tour à l’autre.

# Instructions verbeuses (plus de tokens à traiter)
bad_instructions = """
Vous êtes un assistant vocal intelligent et compétent qui travaille
pour la société Corsen AI, une entreprise française spécialisée dans
l'intelligence artificielle. Vous devez répondre aux questions des
utilisateurs de manière professionnelle, concise et en français.
Vous avez accès à différents outils pour consulter les rendez-vous,
vérifier les disponibilités et effectuer des réservations.
N'hésitez pas à demander des précisions si la demande n'est pas claire.
"""

# Instructions optimisées (même effet, moins de tokens)
good_instructions = """
Assistant vocal Corsen AI. Français, concis, professionnel.
Utilisez les outils pour les RDV et réservations.
Demandez des précisions si nécessaire.
"""

Réduire la latence réseau

Mesurer l’aller-retour

Placez votre relais WebSocket au plus près de vos utilisateurs et mesurez le trajet plutôt que de le supposer. Un événement anodin, comme vider le tampon d’entrée, suffit à chronométrer un aller-retour :

import time

async def measure_roundtrip(ws):
    """Mesure la latence réseau vers la Realtime API."""
    start = time.monotonic()
    await ws.send(json.dumps({"type": "input_audio_buffer.clear"}))
    while json.loads(await ws.recv())["type"] != "input_audio_buffer.cleared":
        pass  # ignorer les autres événements arrivés entre-temps
    roundtrip_ms = (time.monotonic() - start) * 1000
    print(f"Latence réseau : {roundtrip_ms:.0f} ms")
    return roundtrip_ms

Keepalive et reconnexion

import asyncio
import websockets

async def robust_connection(api_key: str, max_retries: int = 5):
    """Connexion robuste avec reconnexion automatique."""
    retry_count = 0

    while retry_count < max_retries:
        try:
            async with websockets.connect(
                REALTIME_URL,
                additional_headers={
                    "Authorization": f"Bearer {api_key}"
                },
                ping_interval=20,     # Ping toutes les 20s
                ping_timeout=10,       # Timeout ping 10s
                close_timeout=5        # Timeout fermeture 5s
            ) as ws:
                retry_count = 0  # Reset on successful connection
                yield ws

        except websockets.exceptions.ConnectionClosed:
            retry_count += 1
            wait = min(2 ** retry_count, 30)
            print(f"Reconnexion dans {wait}s...")
            await asyncio.sleep(wait)

Une reconnexion ouvre une nouvelle session : la conversation précédente n’y figure plus, et une session dure de toute façon au plus 60 minutes. Gardez donc votre propre trace des échanges (transcriptions, résultats d’outils) et réinjectez l’essentiel avec conversation.item.create après une reconnexion.

Accélérer l’ouverture en WebRTC : WARP

Côté navigateur ou application native, OpenAI documente WARP (WebRTC Abridged Roundtrip Protocol), qui réduit le temps d’établissement d’une session WebRTC en combinant trois optimisations de transport : DTLS 1.3, SPED et SNAP. Ce sont des brouillons de l’IETF, pris en charge par les versions récentes de libwebrtc ; vous pouvez activer celles que votre client sait faire, une à une. Le guide WebRTC with WARP détaille la configuration.

Optimiser le function calling

Les appels de fonctions ajoutent de la latence. Optimisez leur exécution :

import asyncio

async def execute_functions_parallel(function_calls: list) -> list:
    """Exécute les appels de fonctions en parallèle."""
    tasks = []
    for call in function_calls:
        task = execute_single_function(call["name"], call["arguments"])
        tasks.append(task)

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# Cache pour les fonctions fréquentes
from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_data(key: str) -> dict:
    """Cache les résultats des requêtes fréquentes."""
    # Requête base de données ou API
    return {"data": "..."}

Monitoring de la latence

Intégrez un système de mesure pour identifier les goulots d’étranglement :

import time
from dataclasses import dataclass, field

@dataclass
class LatencyMetrics:
    speech_end_time: float = 0
    response_start_time: float = 0
    first_audio_time: float = 0
    response_complete_time: float = 0
    function_call_times: list = field(default_factory=list)

    @property
    def time_to_first_audio(self) -> float:
        if self.speech_end_time and self.first_audio_time:
            return (self.first_audio_time - self.speech_end_time) * 1000
        return 0

    @property
    def total_response_time(self) -> float:
        if self.speech_end_time and self.response_complete_time:
            return (self.response_complete_time - self.speech_end_time) * 1000
        return 0

    def report(self) -> dict:
        return {
            "ttfa_ms": round(self.time_to_first_audio),
            "total_ms": round(self.total_response_time),
            "function_calls_ms": [round(t * 1000) for t in self.function_call_times]
        }

class LatencyTracker:
    def __init__(self):
        self.current = LatencyMetrics()
        self.history = []

    def on_event(self, event_type: str):
        now = time.time()

        if event_type == "input_audio_buffer.speech_stopped":
            self.current.speech_end_time = now

        elif event_type == "response.output_audio.delta":
            if not self.current.first_audio_time:
                self.current.first_audio_time = now

        elif event_type == "response.done":
            self.current.response_complete_time = now
            self.history.append(self.current.report())
            self.current = LatencyMetrics()

    def average_ttfa(self) -> float:
        if not self.history:
            return 0
        return sum(h["ttfa_ms"] for h in self.history) / len(self.history)

Points clés à retenir

  • La latence totale additionne VAD, réseau, modèle et outils : mesurez chaque étape, OpenAI ne publie pas de durées types
  • silence_duration_ms règle directement l’attente avant la réponse ; semantic_vad et son eagerness sont l’autre levier
  • Des instructions système concises réduisent le temps de traitement du modèle
  • L’exécution parallèle des appels de fonctions et le cache réduisent la latence applicative
  • Un système de monitoring avec TTFA (Time-to-First-Audio) est indispensable pour l’optimisation continue