Architecture WebSocket de la Realtime API

Architecture WebSocket de la Realtime API

La Realtime API d’OpenAI représente un changement de paradigme fondamental dans l’interaction avec les modèles de langage. Contrairement aux API REST classiques qui fonctionnent en requête-réponse, la Realtime API utilise le protocole WebSocket pour établir une connexion bidirectionnelle persistante entre votre application et le modèle.

Cette architecture permet des échanges audio en temps réel : vous envoyez de l’audio, le modèle le traite et répond vocalement, le tout sans latence perceptible. C’est la brique fondamentale pour construire des agents vocaux conversationnels.

WebSocket vs REST : pourquoi ce choix ?

Dans une architecture REST traditionnelle, chaque interaction nécessite une requête HTTP complète : ouverture de connexion TCP, négociation TLS, envoi des headers, attente de la réponse, fermeture. Pour de l’audio en temps réel, cette surcharge est rédhibitoire.

Le protocole WebSocket maintient un canal ouvert en permanence. Une fois la connexion établie, client et serveur peuvent s’envoyer des messages à tout moment, sans attendre que l’autre ait terminé.

Établir une connexion WebSocket

Voici comment établir une connexion à la Realtime API en Python :

import asyncio
import websockets
import json

OPENAI_API_KEY = "sk-..."
URL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"

async def connect_realtime():
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(URL, extra_headers=headers) as ws:
        print("Connecté à la Realtime API")

        # Recevoir l'événement de bienvenue
        response = await ws.recv()
        event = json.loads(response)
        print(f"Type: {event['type']}")
        # → "session.created"

        # Maintenir la connexion ouverte
        await asyncio.Future()

asyncio.run(connect_realtime())

L’URL de connexion inclut le modèle comme paramètre de requête. Le header OpenAI-Beta est obligatoire pour accéder à cette API encore en phase d’évolution.

Architecture d’une application Realtime

Une application utilisant la Realtime API suit une architecture en couches :

Le rôle du serveur proxy

En développement, vous pouvez connecter directement votre application à la Realtime API. En production, un serveur proxy est indispensable pour :

• Protéger votre clé API — ne jamais l’exposer côté client
• Authentifier vos utilisateurs — vérifier les tokens avant d’ouvrir la session
• Contrôler les coûts — limiter la durée des sessions et le nombre de tokens
• Logger les interactions — pour le débogage et la conformité

# Exemple simplifié d'un proxy WebSocket avec FastAPI
from fastapi import FastAPI, WebSocket
import websockets

app = FastAPI()

@app.websocket("/ws/realtime")
async def proxy_realtime(client_ws: WebSocket):
    await client_ws.accept()

    headers = {"Authorization": f"Bearer {OPENAI_API_KEY}",
               "OpenAI-Beta": "realtime=v1"}

    async with websockets.connect(URL, extra_headers=headers) as openai_ws:
        async def forward_to_openai():
            async for message in client_ws.iter_text():
                await openai_ws.send(message)

        async def forward_to_client():
            async for message in openai_ws:
                await client_ws.send_text(message)

        await asyncio.gather(forward_to_openai(), forward_to_client())

Points clés à retenir

• La Realtime API utilise WebSocket pour une communication bidirectionnelle à faible latence
• La connexion est persistante : une seule ouverture pour toute la session conversationnelle
• L’architecture en couches sépare audio, transport, logique métier et proxy serveur
• En production, un serveur proxy est indispensable pour la sécurité et le contrôle des coûts
• Le protocole est événementiel : chaque message est un objet JSON avec un champ type