Tutoriel API Grok Voice Think Fast 2.0 : créez un agent vocal temps réel en Python

Grok Voice Think Fast 2.0 de SpaceXAI est un modèle speech‑to‑speech. Vous lui envoyez de l’audio via un WebSocket, il renvoie de l’audio, et entre les deux il raisonne et continue à parler pendant qu’un appel d’outil qu’il a décidé de lancer s’exécute déjà. Pas d’étape séparée de speech‑to‑text, pas d’étape séparée de text‑to‑speech.

SpaceXAI a annoncé Think Fast 2.0 le 29 juillet 2026 : premier audio plus rapide, comportement full‑duplex plus stable (écoute pendant qu’il parle au lieu d’alterner strictement) et appels d’outil déclenchés tôt dans le tour. Je passe vite sur les benchmarks : pour un tutoriel, ce qui compte, c’est ce qui change dans votre code.

Nous allons créer un agent vocal de support client pour une boutique en ligne. L’appelant peut demander le statut d’une commande, en retrouver une via son email s’il n’a pas le numéro, modifier une instruction de livraison, annuler, interrompre l’agent en plein milieu d’une phrase, et reprendre la conversation après une coupure. C’est la voie API, pas le builder no‑code présenté dans notre tutoriel Grok Voice Agent Builder. Commencez par là si vous privilégiez la console.

Qu’est‑ce que Grok Voice Think Fast 2.0 ?

Grok Voice Think Fast 2.0 est le dernier modèle de SpaceXAI pour l’API Speech to Speech, le nom produit derrière ce que la plupart appellent simplement Grok Voice. Si vous pensez encore à l’entreprise sous le nom xAI, c’est la même équipe : elle a été intégrée à SpaceX et renommée SpaceXAI le 6 juillet 2026. L’API n’a pas suivi le rebranding, donc chaque identifiant ci‑dessous reste xai, de la variable XAI_API_KEY à l’hôte api.x.ai.

Une pile vocale traditionnelle chaîne trois services : speech‑to‑text, un modèle de langage, puis text‑to‑speech — chaque saut ajoute de la latence et un risque de perte de contexte. Think Fast 2.0 regroupe le tout dans un seul modèle qui prend de l’audio ou du texte en entrée et renvoie de l’audio ou du texte en sortie sur la même connexion.

WebSocket speech‑to‑speech versus architecture en trois services. Image par l’auteur.

Pour un agent qui agit plutôt que de seulement parler, l’essentiel est que le raisonnement et la parole tournent en parallèle. SpaceXAI indique que les appels d’outil « commencent généralement » à s’exécuter avant la fin de la première phrase de l’agent, et ce « généralement » est important.

Sur les benchmarks cités par SpaceXAI depuis Artificial Analysis, Think Fast 2.0 atteint 82,9 % sur le Speech to Speech Index contre 75,7 % pour la version 1.0, et réduit le temps jusqu’au premier audio de 1,25 s à 0,70 s. Des chiffres éditeur sur un benchmark généraliste sont une hypothèse sur votre flux d’appels, pas un plan de test.

Vous verrez trois chaînes de modèle : grok-voice-latest, grok-voice-think-fast-2.0 et grok-voice-think-fast-1.0. L’alias est pratique en prototypage, trop instable pour le reste.

Lors de mon test le 4 août 2026, grok-voice-latest pointait encore vers grok-voice-think-fast-1.0, et les notes de version de SpaceXAI programmaient le passage à Think Fast 2.0 pour le lendemain. Ce switch change autant le modèle que le prix : 0,08 $ par minute d’audio contre 0,05 $ pour la 1.0. Un alias non épinglé devient plus cher sans aucune modification de votre code. Épinglez la version dans tout ce que vous déployez.

Ce que nous allons construire

L’agent couvre les demandes classiques d’un support : consulter une commande, en retrouver une à partir d’un email quand l’appelant n’a pas de numéro, modifier une instruction de livraison, annuler, ouvrir ou vérifier un ticket, et transférer l’appel à une personne. Les interruptions et une coupure de connexion jalonnent le parcours.

C’est une poignée de petits fichiers plutôt qu’un unique script, car chaque pièce a un rôle différent et vous voudrez les tester séparément. Structure :

• config.py charge la clé API et stocke la chaîne de modèle, la fréquence d’échantillonnage et les URLs d’endpoints

• voice_client.py encapsule le WebSocket, suit la facturation et expose des helpers d’envoi/réception

• tools.py définit les fonctions de commande et un petit store en mémoire simulant une base de données

• assistant.py contient le prompt système, la configuration de session et la boucle d’événements qui relie le tout

• token_server.py est un petit endpoint FastAPI qui émet des jetons éphémères

• app_streamlit.py place le même client derrière un appel live depuis le navigateur, sur lequel je reviendrai après la section test

Le parcours pédagogique se fait au terminal. La démo ajoute le micro.

Prérequis

Il vous faut un compte SpaceXAI avec clé API, un mode de facturation approvisionné (pas de palier gratuit permanent, et les crédits promotionnels de nouveau compte ne suffiront pas), et assez de maîtrise de asyncio et des WebSockets pour suivre sans explication ligne par ligne de await.

Les exemples quick‑start de SpaceXAI utilisent le package websockets brut plutôt qu’un SDK dédié, et nous aussi. La doc n’indique pas de version minimale de Python. J’ai testé en 3.11.

Gardez la clé API côté serveur. Si une app web ou mobile appelle directement l’API Voice, elle obtient un jeton éphémère au lieu de votre vraie clé, comme détaillé dans la section sécurité ci‑dessous.

Mise en place du projet

Tous les fichiers ci‑dessous sont dans le repo du projet, vous pouvez donc le cloner au lieu de copier des extraits :

git clone https://github.com/KhalidAbdelaty/grok-voice-think-fast-2.0.git
cd grok-voice-think-fast-2.0
pip install -r requirements.txt

websockets porte la connexion temps réel et python-dotenv lit votre clé. Le reste couvre l’endpoint de jeton et la démo navigateur. Placez votre clé dans .env :

XAI_API_KEY=xai-your-key-here

L’essentiel de la configuration est là. La connexion est la partie intéressante.

Comprendre l’API Realtime Grok Voice

Grok Voice est le nom produit. Concrètement, vous codez contre un endpoint WebSocket à wss://api.x.ai/v1/realtime, et toute la conversation transite sous forme d’événements JSON sur ce seul socket.

Le cycle de vie des événements

Une connexion suit un schéma fixe : le serveur envoie session.created et conversation.created dès l’ouverture ; vous envoyez session.update pour configurer la voix et les outils ; le serveur confirme avec session.updated, et ensuite vous créez des items de conversation et demandez des réponses. J’ai testé en live : l’ordre correspond exactement à la doc.

• session.update (client) configure la voix, les instructions, les outils et le format audio

• conversation.item.create (client) ajoute un message utilisateur, un message assistant ou un résultat d’outil

• response.create (client) demande au modèle de parler ; la VAD serveur envoie cela automatiquement pour vous

• response.output_audio.delta et response.output_audio_transcript.delta (serveur) streament la réponse au fil de la génération

• response.done (serveur) clôt le tour

Deux points piégeux. La page de doc Speech to Speech citée plus haut mentionne un événement conversation.item.created pendant la reprise de session, mais la référence d’événements canonique ne liste que conversation.item.added, et c’est bien ce que j’ai reçu dans tous mes tests ; codez donc contre celui‑là. Vous verrez aussi un ping non documenté quelques secondes après la connexion ; c’est normal, pas une erreur.

Formats audio et transport

Codec et transport sont deux choix distincts. Le codec, défini sous audio.input.format et audio.output.format, peut être audio/pcm (Linear16, 24000 Hz par défaut), audio/pcmu ou audio/pcma (G.711 à 8 kHz, pour la téléphonie), ou audio/opus (24 kHz). Le transport détermine comment ces octets voyagent sur le fil :

• json (par défaut) envoie l’audio en base64 dans input_audio_buffer.append et response.output_audio.delta, facile à journaliser et à déboguer

• binary envoie les octets bruts du codec en trames binaires WebSocket, sans l’overhead base64, au prix d’une boucle de réception qui doit brancher selon le type de message

Commencez en JSON. Tous les exemples de la doc l’utilisent, c’est trivial à inspecter, et l’overhead base64 n’est pas le goulot d’étranglement d’un agent de support. Passez en binaire si vous mesurez une bonne raison de le faire.

Compatibilité avec l’API Realtime d’OpenAI

Passez cette section si vous n’avez jamais touché l’API Realtime d’OpenAI. Pour les autres, l’API Speech to Speech suit de près l’API Realtime d’OpenAI, assez pour porter la plupart des clients en changeant l’URL de base et la clé, mais ce n’est pas un remplacement parfait.

Les transcriptions arrivent en conversation.item.input_audio_transcription.updated ici au lieu du delta d’OpenAI, quelques événements OpenAI ne sont pas pris en charge, et SpaceXAI ajoute ses propres extensions : force_message pour une mention légale scriptée, resumption pour les reconnexions, et replace pour corriger des noms de marque mal prononcés avant la synthèse vocale.

Construire l’agent vocal temps réel

Assez de protocole. Voici le client qui lui parle.

Connexion et configuration de la session

La connexion s’ouvre avec un bearer token et un paramètre de requête pour le modèle, et votre premier message configure le comportement de l’agent :

import asyncio
import json
import os
import websockets

MODEL = "grok-voice-think-fast-2.0"  # épinglez la version, pas grok-voice-latest

async def connect():
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}"
    ws = await websockets.connect(
        url, additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}
    )
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "voice": "eve",
            "instructions": SYSTEM_PROMPT,
            "turn_detection": {"type": "server_vad"},
            "tools": ORDER_TOOLS,
            "resumption": {"enabled": True},
        }
    }))

return ws

instructions est le prompt système, et ce modèle préfère les prompts courts. Les notes de migration de SpaceXAI suggèrent de simplifier les prompts écrits pour les anciens modèles vocaux de l’ère GPT plutôt que de les porter tels quels. Le mien demande des réponses brèves, une question à la fois, et une relecture à voix haute avant toute action d’écriture. La confirmation orale est un confort UX, pas un contrôle de sécurité. Votre application doit toujours faire respecter l’autorisation lors de l’écriture elle‑même.

Point surprenant : une chaîne de modèle non reconnue ne lève pas d’erreur à la connexion, elle retombe silencieusement sur grok-voice-think-fast-1.0. Rétrograder une requête payante à cause d’une faute de frappe, sans le dire, est un choix discutable. Logguez le champ session.model de session.created au démarrage pour vérifier que vous avez bien le modèle demandé.

Sortie terminal montrant session.created après connexion. Image par l’auteur.

Streamer l’audio de l’utilisateur

Avec turn_detection.type réglé sur server_vad, vous n’avez qu’à empiler les chunks audio. Le serveur décide quand l’appelant a cessé de parler et déclenche la réponse pour vous. Réglez‑le sur null pour gérer cela vous‑même et valider explicitement le buffer quand vous jugez le tour terminé.

async def send_audio_chunk(ws, pcm_bytes: bytes):
    await ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": base64.b64encode(pcm_bytes).decode(),
    }))

La VAD serveur a trois réglages, et un mauvais calibrage est la cause la plus courante d’un agent qui « semble cassé » sans aucune erreur dans les logs. Par défaut, ils n’apparaissent pas dans l’écho de session.updated : fiez‑vous à la doc, pas aux suppositions.

• threshold (0,1 à 0,9, 0,85 par défaut) : niveau requis pour compter comme de la parole ; augmentez en environnement bruyant, baissez si des voix faibles sont ratées

• silence_duration_ms : durée de silence avant de clore le tour ; trop court coupe les gens en pleine réflexion, trop long donne une impression de lenteur

• prefix_padding_ms (333 par défaut) : tranche d’audio conservée juste avant la détection de parole pour éviter de rogner la première syllabe

Ajustez silence_duration_ms en premier si les appelants sont coupés lorsqu’ils marquent une pause. C’est mon premier levier avant les deux autres.

Recevoir et jouer la réponse

L’audio arrive en petits morceaux via response.output_audio.delta, et l’intérêt du streaming est de jouer chaque morceau dès réception plutôt que d’attendre response.done.

async def play_response(ws):
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            chunk = base64.b64decode(event["delta"])
            speaker.write(chunk)  # votre appel de lecture ici
        elif event["type"] == "response.output_audio_transcript.delta":
            print(event["delta"], end="", flush=True)

Conservez la transcription même en production. C’est votre outil de debug le plus économique lorsque quelqu’un dit que l’agent « a dit un truc bizarre ».

Ajouter des outils à l’agent vocal

Un agent vocal qui ne fait que parler, c’est un chatbot avec un micro.

Créer les outils de commande

Chaque outil est un schéma JSON plus une fonction Python côté serveur. Le modèle ne touche jamais la base : il ne voit que ce que votre fonction renvoie.

ORDER_TOOLS = [
    {
        "type": "function",
        "name": "check_order_status",
        "description": "Look up the status, ETA, and delivery instructions for an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_number": {"type": "string", "description": "e.g. ORD-1042"},
            },
            "required": ["order_number"],
        },
    },
    # find_orders, update_delivery_instructions, cancel_order,
    # create_support_ticket, check_ticket_status and transfer_to_human
    # all follow the same shape
]

Les opérations de lecture comme check_order_status sont sans risque en cas de retry après un timeout. Les écritures ne le sont pas : relancer update_delivery_instructions après un timeout ambigu peut appliquer deux fois la même modification. Une ligne de confirmation dans le prompt ne l’empêche pas ; utilisez plutôt une clé d’idempotence ou une détection de doublon.

Placez aussi les refus dans la fonction. cancel_order renvoie une raison et une alternative au lieu d’annuler une commande expédiée, car « ne jamais annuler une commande expédiée » dans un prompt reste une suggestion, tandis qu’une fonction qui refuse, non.

Gérer la boucle d’appels d’outil

Quatre étapes, et l’ordre compte plus qu’on ne le croit. Le modèle envoie response.function_call_arguments.done, votre code exécute la fonction, vous renvoyez le résultat comme item function_call_output , puis seulement vous demandez au modèle de continuer.

async def handle_tool_call(ws, event):
    args = json.loads(event["arguments"])
    result = execute(event["name"], args)  # ne lève jamais; les erreurs reviennent sous {"error": ...}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": event["call_id"],
            "output": json.dumps(result),
        },
    }))

Si le modèle a besoin de plusieurs outils pour une demande, il déclenche plusieurs événements function_call_arguments.done avant toute lecture audio. Résolvez‑les tous et envoyez chaque résultat avant un seul response.create. L’envoyer trop tôt fait répondre le modèle sans le contexte des appels encore en cours.

Attention au piège, documenté par SpaceXAI et que j’ai tout de même croisé : envoyer response.create dès qu’un résultat d’outil part peut chevaucher la phrase d’intro encore en cours. Dans un run, il a commencé par « Je vérifie le statut de la commande ORD‑1042 tout de suite » et a appelé l’outil au milieu de la phrase ; une réponse immédiate aurait parlé par‑dessus son ouverture.

Attendez la fin de l’audio du tour en cours et affichez un court état « réflexion » entre les deux.

Flux d’appel d’outil avant reprise de la réponse. Image par l’auteur.

Gérer les interruptions et l’état de conversation

Deux sujets distincts : un appelant parle par‑dessus l’agent en pleine réponse, et un WebSocket tombe puis doit reprendre.

Prendre en charge les interruptions naturelles

Avec server_vad activé, le barge‑in est automatique côté serveur : dès qu’il détecte que l’appelant reparle, il émet input_audio_buffer.speech_started et stoppe la génération de l’ancienne réponse. À vous d’assurer la moitié cliente de la poignée de main, en vidant l’audio déjà en file pour que l’agent se taise au lieu de finir une phrase que personne n’a demandée.

if event["type"] == "input_audio_buffer.speech_started":
    playback_queue.clear()

Pour les sessions sans VAD, response.cancel joue le même rôle sur demande. Il existe aussi conversation.item.truncate pour réduire un item assistant à ce qui a réellement été entendu. La doc confirme son existence sans préciser quand le déclencher pendant un barge‑in live ; testez le timing vous‑même.

J’ai testé avec une modification d’instructions de livraison en pleine réponse : démarrer la demande, interrompre avec une adresse différente pendant la confirmation de l’agent. Ce qui compte, c’est que l’agent applique la consigne corrigée plutôt que de finir discrètement l’ancienne, pas seulement que l’audio s’arrête. Vérifiez sur l’enregistrement de commande, pas au silence. La démo navigateur en fin d’article permet de l’entendre.

Reprendre une session déconnectée

La reprise de session est opt‑in et ce n’est pas de la mémoire. Mettez resumption.enabled: true dans session.update, récupérez l’ID depuis l’événement conversation.created, et si le socket tombe, reconnectez avec ?conversation_id=<id> dans l’URL et réactivez l’option sur la nouvelle connexion.

async def reconnect(conversation_id):
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}&conversation_id={conversation_id}"
    ws = await websockets.connect(url, additional_headers=auth_header)
    await ws.send(json.dumps({"type": "session.update", "session": {"resumption": {"enabled": True}}}))
    return ws

Les tours, transcriptions, appels d’outil et résultats mis en cache rejouent avant votre prochaine question, et le cache expire après 30 minutes d’inactivité. J’ai testé en demandant une commande, en coupant la connexion, puis en me reconnectant pour une relance sans me répéter ; l’agent a bien repris l’ETA.

Limite non documentée : la relecture n’arrive pas instantanément, donc une question envoyée au moment même où le socket s’ouvre peut la devancer et revenir sans mémoire du tour précédent. Patientez une seconde avant d’incriminer la reprise.

Log terminal d’une session reprise. Image par l’auteur.

Ne remplacez pas cela par une persistance d’état côté base chez vous. Si le cache expire ou si l’appelant rappelle demain, vous repartez sans contexte — c’est voulu.

Sécuriser et superviser l’agent

Ne mettez jamais une clé API permanente dans du code navigateur ou mobile. Si un client se connecte directement au lieu de passer par votre serveur, émettez un jeton court :

from fastapi import FastAPI
import httpx, os

app = FastAPI()

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.x.ai/v1/realtime/client_secrets",
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()  # {"value": "xai-realtime-client-secret-...", "expires_at": ...}

Un navigateur ne peut pas définir un en‑tête Authorization personnalisé lors du handshake WebSocket ; il passe donc le jeton via l’en‑tête sec-websocket-protocol, préfixé par xai-client-secret..

Le serveur émet un jeton, le navigateur rejoint l’appel. Image par l’auteur.

La facturation suit deux compteurs. L’audio, envoyé ou reçu, est facturé au tarif mentionné de 0,08 $ par minute, soit 4,80 $ l’heure, et chaque conversation.item.create qui n’est pas de l’audio et pas un function_call_output coûte 0,004 $ forfaitaire. response.create n’est pas facturé. Chaque response.done contient un objet usage qui, lors de mes tests, incluait output_audio_seconds et un billable_audio_seconds distinct. Billez à partir de ceux‑là, pas d’estimations.

Les limites documentées de l’API Speech to Speech sont 10 sessions concurrentes par équipe et 120 minutes par session, le tout en us-east-1. Ne planifiez pas vos capacités à partir des chiffres de l’API Voice Agent, qui sont différents.

Côté confidentialité, soyez précis. La FAQ sécurité de SpaceXAI indique que les requêtes et réponses API sont conservées chiffrées pendant 30 jours pour la surveillance des abus et non utilisées pour l’entraînement sans autorisation, et que les équipes peuvent activer la Zero Data Retention, sachant que ZDR supprime l’historique persistant des conversations et donc n’est pas compatible avec la reprise.

Si vous devez indiquer qu’un appel est enregistré ou géré par l’IA, c’est le rôle de l’extension force_message citée plus haut. La phrase joue telle quelle, sans paraphrase par le modèle.

Tester l’agent vocal

Un 200 sur le handshake WebSocket ne dit rien de la pertinence de l’agent. Testez le résultat, pas seulement la connexion.

• Une consultation de commande « heureuse », en vérifiant la réponse parlée par rapport à l’enregistrement, pas seulement qu’une réponse est arrivée
• Une réponse interrompue, en confirmant que la lecture s’arrête et que l’agent traite la nouvelle demande
• Une mise à jour de livraison nécessitant confirmation, vérifiée sur l’enregistrement de commande
• Un refus, par exemple annuler une commande expédiée, où l’agent doit expliquer la règle plutôt que s’excuser
• Un numéro de commande inconnu, pour vérifier que l’agent le dit au lieu d’inventer un statut
• Un outil qui renvoie une erreur, et l’agent qui l’énonce plutôt que de se figer
• Reconnexion et reprise, y compris la fenêtre de relecture mentionnée plus haut
• Audio bruyant, parole rapide, et un appelant qui énonce chiffres et adresses lettre par lettre

J’en ai passé la plupart en live pendant l’écriture. Les échecs intéressants étaient comportementaux, pas des erreurs : le timing de reprise ci‑dessus, et un seuil VAD hors plage accepté au lieu d’être rejeté — typiquement le genre de chose qui part en prod silencieusement si vous ne testez que la voie heureuse. Ajoutez aussi un test multilingue, et voyez la FAQ pour une subtilité sur le nommage de la langue.

Deux de ces points ne se testent pas au clavier. app_streamlit.py est une page Streamlit qui place un appel live dans le navigateur : le micro stream dans le même WebSocket via WebRTC, la voix de l’agent revient en stream, et le socket reste ouvert.

streamlit run app_streamlit.py