Latence et optimisation temps réel

Latence et optimisation temps réel

La latence est l’ennemi numéro un d’un agent vocal. Au-delà de 500 ms entre la fin de la question et le début de la réponse, l’utilisateur perçoit un délai gênant. Au-delà d’une seconde, l’expérience se dégrade fortement. Cette leçon détaille les sources de latence et les techniques pour les minimiser.

Anatomie de la latence

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

Optimiser la détection VAD

Le silence_duration_ms est le paramètre qui impacte le plus la latence perçue :

# 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
}

# 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 temps de traitement du modèle :

# 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

Localisation du serveur

Déployez votre proxy WebSocket dans la même région que les serveurs OpenAI (US East) pour minimiser le round-trip :

import time

async def measure_roundtrip(ws):
    """Mesure la latence réseau vers la Realtime API."""
    start = time.time()
    await ws.send(json.dumps({
        "type": "input_audio_buffer.clear"
    }))
    await ws.recv()  # input_audio_buffer.cleared
    roundtrip_ms = (time.time() - 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,
                extra_headers={
                    "Authorization": f"Bearer {api_key}",
                    "OpenAI-Beta": "realtime=v1"
                },
                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)

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.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 est la somme du VAD (300-700 ms), du réseau (20-100 ms) et du modèle (200-800 ms)
• silence_duration_ms est le paramètre avec le plus d’impact : 400 ms pour du réactif, 600+ ms pour du patient
• 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