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 de voix à voix. Vous lui envoyez de l’audio via un WebSocket et il renvoie de l’audio ; entre les deux, il peut raisonner et continuer à parler pendant qu’un appel d’outil qu’il a décidé d’exécuter est déjà en cours. Pas d’étape séparée de reconnaissance vocale, pas d’étape séparée de synthèse vocale.

SpaceXAI a annoncé Think Fast 2.0 le 29 juillet 2026 : premier audio plus rapide, duplex intégral plus stable (il écoute tout en parlant plutôt que d’alterner strictement), et des appels d’outils qui démarrent tôt dans le tour. Je vais rester bref 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 à partir d’un e‑mail 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. Nous empruntons la voie API, et non le builder no‑code présenté dans notre tutoriel Grok Voice Agent Builder. Commencez par celui‑ci pour une version orientée 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 tout le monde appelle simplement Grok Voice. Si vous pensez encore à l’entreprise comme à xAI, c’est la même entité : elle a été intégrée à SpaceX et rebaptisée SpaceXAI le 6 juillet 2026. L’API n’a pas suivi le rebranding : tous les identifiants ci‑dessous restent xai, de la variable XAI_API_KEY à l’hôte api.x.ai.

Une pile vocale traditionnelle enchaîne trois services : reconnaissance vocale, modèle de langage, puis synthèse vocale. 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 produit de l’audio ou du texte en sortie sur la même connexion.

WebSocket voix à voix versus architecture en pipeline à 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 s’exécutent en parallèle. SpaceXAI indique que les appels d’outils « commencent généralement » à s’exécuter avant que l’agent n’ait terminé sa première phrase, un « généralement » qui compte réellement.

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

Vous verrez trois identifiants de modèle : grok-voice-latest, grok-voice-think-fast-2.0 et grok-voice-think-fast-1.0. L’alias est pratique en phase de prototypage et 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 basculement est autant un changement de modèle qu’un changement de prix : 0,08 $ par minute d’audio contre 0,05 $ pour la 1.0. Avec un alias non épinglé, la facture augmente sans qu’aucune ligne de votre code ne change. Épinglez la chaîne versionnée dans tout ce que vous déployez.

Ce que nous allons construire

L’agent couvre l’éventail classique d’une ligne de support : rechercher une commande, en retrouver une à partir d’un e‑mail quand l’appelant n’a pas de numéro, modifier une instruction de livraison, annuler, ouvrir ou vérifier un ticket, et passer l’appel à une personne. Au passage, nous traiterons les interruptions et une connexion coupée.

C’est une poignée de petits fichiers plutôt qu’un seul script, car chaque pièce a un rôle distinct et vous voudrez les tester séparément. Voici l’organisation :

• config.py charge la clé API et contient la chaîne du modèle, la fréquence d’échantillonnage et les URLs des 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 magasin en mémoire faisant office de 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 génère des tokens éphémères

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

Le parcours d’apprentissage se fait depuis un terminal. La démo ajoute le microphone.

Prérequis

Vous avez besoin d’un compte SpaceXAI avec une clé API, d’un mode de facturation approvisionné (il n’y a pas de palier gratuit permanent, et les crédits promotionnels de nouveau compte ne suffiront pas), et d’une aisance suffisante avec asyncio et les WebSockets pour suivre sans explication ligne par ligne de await.

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

Gardez la clé API côté serveur. Si une application web ou mobile parle directement à la Voice API, elle obtient un token éphémère à la place de votre vraie clé, comme expliqué dans la section sécurité ci‑dessous.

Mise en place du projet

Tous les fichiers ci‑dessous sont dans le référentiel du projet ; vous pouvez donc le cloner plutôt que 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 gère la connexion temps réel et python-dotenv lit votre clé. Le reste couvre l’endpoint de token et la démo navigateur. Placez votre clé dans .env :

XAI_API_KEY=xai-your-key-here

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

Comprendre l’API temps réel Grok Voice

Grok Voice est le nom du produit. Ce à quoi vous vous connectez réellement est un endpoint WebSocket à wss://api.x.ai/v1/realtime, et toute la conversation se déroule sous forme de flux d’événements JSON sur ce seul socket.

Cycle de vie des événements

La connexion suit une séquence fixe : le serveur envoie session.created et conversation.created dès que vous vous connectez, vous envoyez session.update pour configurer la voix et les outils, le serveur confirme par session.updated, puis vous créez des éléments de conversation et demandez des réponses. J’ai testé avec une clé active et l’ordre correspondait 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 ceci pour vous automatiquement

• response.output_audio.delta et response.output_audio_transcript.delta (serveur) diffusent la réponse au fur et à mesure de sa génération

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

Deux points piègent souvent. La page de doc Speech to Speech mentionnée plus haut parle d’un événement conversation.item.created pendant la reprise de session, mais la référence des événements canonique ne liste que conversation.item.added, et c’est ce que j’ai reçu dans tous mes tests : codez pour celui‑là. Vous verrez aussi un ping non documenté quelques secondes après la connexion ; mentionné ici pour ne pas l’interpréter comme une erreur.

Formats audio et transport

Le codec et le 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 la surcharge 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, l’inspection est triviale, et la surcharge base64 n’est pas le goulet d’étranglement d’un agent de support. Passez en binaire si vous mesurez une raison de le faire.

Compatibilité avec l’API OpenAI Realtime

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

Les transcriptions arrivent ici comme conversation.item.input_audio_transcription.updated au lieu du delta d’OpenAI, quelques événements OpenAI ne sont pas pris en charge, et SpaceXAI ajoute ses extensions : force_message pour une phrase obligatoire scriptée, resumption pour les reconnexions, et replace pour corriger des marques mal prononcées avant la synthèse.

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 modèle. Le premier message que vous envoyez configure le comportement de l’agent :

import asyncio
import json
import os
import websockets

MODEL = "grok-voice-think-fast-2.0"  # pin the version, not 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 recommandent de simplifier les prompts écrits pour les anciens modèles vocaux de l’ère GPT au lieu de les porter tels quels. Le mien demande des réponses brèves, une question à la fois, et une relecture à voix haute de toute écriture avant d’agir. Une confirmation orale est un atout UX, pas un contrôle de sécurité. Votre application doit toujours appliquer l’autorisation côté écriture.

Un point m’a surpris : une chaîne de modèle non reconnue ne renvoie pas d’erreur à la connexion, elle retombe silencieusement sur grok-voice-think-fast-1.0. Déclasser une requête payante à cause d’une faute de frappe, sans le dire, est un défaut étonnant. Journalisez le champ session.model de session.created au démarrage et vérifiez que vous obtenez bien ce que vous avez demandé.

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

Diffuser l’audio utilisateur

Avec turn_detection.type défini sur server_vad, vous n’avez qu’à continuer d’empiler l’audio. Le serveur décide quand l’appelant a fini de parler et déclenche la réponse pour vous. Définissez‑le à null si vous voulez piloter vous‑même la fin de tour, en validant 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 mal les configurer est la façon la plus fréquente de rendre un agent vocal « bizarre » alors que les logs ne montrent aucune erreur. Par défaut, ils n’apparaissent pas dans l’écho de session.updated ; référez‑vous à la doc plutôt que supposer.

• threshold (0,1 à 0,9, 0,85 par défaut) : niveau pour considérer qu’il s’agit de parole ; augmentez‑le en environnement bruyant, baissez‑le si des voix faibles sont manquées

• silence_duration_ms : durée de silence avant que le serveur ne termine 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 d’abord silence_duration_ms si les appelants se font couper lorsqu’ils marquent une pause. C’est mon premier levier avant de toucher aux deux autres.

Recevoir et lire la réponse

L’audio arrive par petits morceaux via response.output_audio.delta, et l’intérêt du streaming est de lire chaque morceau dès qu’il arrive, sans 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)  # your playback call goes here
        elif event["type"] == "response.output_audio_transcript.delta":
            print(event["delta"], end="", flush=True)

Conservez la transcription même en production. C’est l’outil de débogage le plus économique quand un appelant dit que l’agent « a dit un truc étrange ».

Ajouter des outils à l’agent vocal

Un agent vocal qui ne fait que parler n’est qu’un chatbot avec un micro.

Créer les outils de commande

Chaque outil est un schéma JSON plus une simple fonction Python de notre côté. Le modèle n’accède jamais à la base ; il ne voit que ce que notre 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 sûres à retenter en cas de délai dépassé. Les écritures ne le sont pas : relancer update_delivery_instructions après un timeout ambigu peut appliquer deux fois le même changement. Une ligne de confirmation dans le prompt ne l’empêche pas ; donnez aux écritures une clé d’idempotence ou un contrôle de doublon.

Mettez aussi les refus dans la fonction. cancel_order renvoie un motif et une alternative au lieu d’annuler une commande expédiée, car un prompt « n’annulez jamais une commande expédiée » est une suggestion, alors qu’une fonction qui refuse, non.

Gérer la boucle d’appels d’outil

Quatre étapes, et l’ordre compte plus qu’il n’y paraît. Le modèle envoie response.function_call_arguments.done, votre code exécute la fonction, vous renvoyez le résultat comme élément function_call_output , et seulement ensuite vous demandez au modèle de poursuivre.

async def handle_tool_call(ws, event):
    args = json.loads(event["arguments"])
    result = execute(event["name"], args)  # never raises; errors come back as {"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 plus d’un outil pour une requête, il émet plusieurs événements function_call_arguments.done avant toute lecture audio. Résolvez‑les tous et renvoyez chaque résultat avant un seul response.create. Envoyé trop tôt, le modèle répond sans le contexte des appels encore en cours.

Attention à ce point, documenté par SpaceXAI et que j’ai tout de même rencontré : envoyer response.create dès que votre résultat d’outil part peut se chevaucher avec la phrase d’introduction encore en cours de lecture. Dans un essai, 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 se serait superposée à son propre début.

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 la reprise de réponse. Image par l’auteur.

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

Deux problèmes distincts. Un appelant parle par‑dessus l’agent en pleine réponse, et un WebSocket se coupe et doit être repris.

Gérer des 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 signale input_audio_buffer.speech_started et stoppe la génération de l’ancienne réponse. Votre rôle est le pendant côté client : vider l’audio déjà en file d’attente pour que l’agent se taise au lieu de finir une phrase que personne ne veut entendre.

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

Pour les sessions manuelles sans VAD, response.cancel fait la même chose à la 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 mais pas le moment exact à déclencher pendant un barge‑in en direct ; testez le timing.

Je l’ai testé avec une modification d’instruction de livraison en pleine réponse : lancer la requête, interrompre avec une autre adresse au milieu de la confirmation de l’agent. Ce qui compte est que l’agent applique l’instruction corrigée au lieu de terminer silencieusement l’ancienne, pas seulement que l’audio se soit arrêté. Vérifiez l’enregistrement de commande, pas le silence. La démo navigateur à la fin vous le fait entendre.

Reprendre une session déconnectée

La reprise de session est en opt‑in et ce n’est pas de la mémoire. Réglez resumption.enabled: true sur session.update, récupérez l’ID depuis l’événement conversation.created, et si le socket tombe, reconnectez‑vous avec ?conversation_id=<id> dans l’URL et réactivez l’opt‑in 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 et résultats d’outils mis en cache rejouent avant votre prochaine question, et le cache disparaît après 30 minutes d’inactivité. Je l’ai testé en demandant une commande, en coupant la connexion puis en me reconnectant pour enchaîner ; l’agent a repris l’ETA correctement.

Un écueil non documenté : le replay n’arrive pas instantanément, donc une question envoyée à l’instant où le socket s’ouvre peut passer devant et revenir sans mémoire du tour précédent. Accordez une seconde avant d’incriminer la reprise.

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

N’utilisez pas cela à la place d’un enregistrement d’état de commande dans votre propre base. Si le cache expire ou si l’appelant rappelle demain, vous repartez sans contexte, et 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, générez un token de courte durée :

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 transmet donc le token via l’en‑tête sec-websocket-protocol, préfixé par xai-client-secret..

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

La facturation repose sur deux compteurs. L’audio, envoyé ou reçu, est facturé au tarif de 0,08 $ par minute cité plus haut, soit 4,80 $ de l’heure, et chaque conversation.item.create qui n’est pas de l’audio et pas un function_call_output coûte un forfait de 0,004 $. response.create n’est pas facturé. Chaque response.done contient un objet usage qui, dans mon test, reportait output_audio_seconds et séparément billable_audio_seconds. Basez votre facturation sur ceux‑là, pas sur des estimations.

Les limites documentées de l’API Speech to Speech sont de 10 sessions concurrentes par équipe et une session plafonnée à 120 minutes, toutes deux en us-east-1. Ne planifiez pas votre capacité à partir des chiffres de la Voice Agent API, 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 permission, et que les équipes peuvent activer la Zero Data Retention, sachant que ZDR supprime l’historique persistant des conversations d’agent vocal et ne fonctionne donc pas avec la reprise.

Si vous annoncez que l’appel est enregistré ou pris en charge par l’IA, c’est le rôle de l’extension force_message mentionnée plus haut. La phrase est lue telle quelle, et non paraphrasée par le modèle.

Tester l’agent vocal

Un statut 200 sur le handshake WebSocket ne dit rien sur le fait que l’agent a fait la bonne chose. Testez le résultat, pas seulement la connexion.

• Une recherche de commande nominale, en vérifiant la réponse orale avec 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 une confirmation, vérifiée dans l’enregistrement de commande
• Un refus, comme l’annulation d’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 s’assurer que l’agent le dit au lieu d’inventer un statut
• Un outil qui renvoie une erreur, en vérifiant que l’agent l’énonce au lieu de se figer
• Reconnexion et reprise, y compris la fenêtre de replay évoquée plus haut
• Audio bruité, parole rapide, et un appelant qui épelle chiffres et adresses

J’ai exécuté la plupart de ces tests avec une clé active en rédigeant. 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é — le genre de chose qui part en prod discrètement cassée si vous ne testez que le chemin heureux. Ajoutez aussi un test multilingue, et voyez la FAQ pour une subtilité sur la façon de nommer la langue.

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

streamlit run app_streamlit.py