Grok Voice Think Fast 2.0 API-tutorial: Bouw een real-time voice-agent in Python

SpaceXAI's Grok Voice Think Fast 2.0 is een speech-to-speechmodel. Je stuurt audio via een WebSocket en het stuurt audio terug. Ondertussen kan het redeneren en doorpraten terwijl een functieaanroep die het heeft gekozen al draait. Geen aparte speech-to-textstap, geen aparte text-to-speechstap.

SpaceXAI kondigde Think Fast 2.0 aan op 29 juli 2026: sneller eerste audio, stabieler full-duplexgedrag (luisteren terwijl het praat in plaats van strikt beurten nemen), en toolaanroepen die vroeg in de beurt afgaan. Ik houd het benchmarkpraatje kort; voor een tutorial telt wat er in je code verandert.

We gaan een klantenservice-voice-agent bouwen voor een webwinkel. Een beller kan vragen naar een bestelling, een bezorginstructie wijzigen, annuleren, de agent midden in een zin onderbreken en het gesprek hervatten na een verbroken verbinding. Dit is het API-pad, niet de no-code Voice Agent Builder die onze Grok Voice Agent Builder-tutorial doorloopt. Begin daar voor de console-firstversie.

Wat is Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 is SpaceXAI's nieuwste model voor de Speech to Speech API, de productnaam achter wat de meesten gewoon Grok Voice noemen. Als je het bedrijf nog als xAI kent: hetzelfde team; het is op 6 juli 2026 in SpaceX geïntegreerd en omgedoopt tot SpaceXAI. De API is niet meegegaan met de rebrand, dus elke identifier hieronder zegt nog steeds xai, van de variabele XAI_API_KEY tot de host api.x.ai.

Een traditionele voicestack schakelt drie services achter elkaar: speech-to-text, een taalmodel en dan text-to-speech, en elke stap voegt latentie toe en een plek waar context kan wegvallen. Think Fast 2.0 vouwt dat samen tot één model dat audio of tekst inneemt en audio of tekst uitstoot over dezelfde verbinding.

Speech-to-speech-WebSocket versus drievoudige service-architectuur. Afbeelding door auteur.

Voor een agent die handelt in plaats van alleen praat, telt dat redeneren en spraak parallel lopen. SpaceXAI zegt dat toolaanroepen "meestal" al starten voordat de agent zijn eerste zin heeft afgemaakt, en dat woord betekent hier echt iets.

Op de benchmarks die SpaceXAI aanhaalt van Artificial Analysis, scoort Think Fast 2.0 82,9% op de Speech to Speech Index tegen 75,7% voor 1.0, en verkort het de tijd tot eerste audio van 1,25 seconden naar 0,70 seconden. Leverancierscijfers op een algemene benchmark zijn een hypothese over jouw belstroom, geen testplan.

Je komt drie modelstrings tegen: grok-voice-latest, grok-voice-think-fast-2.0 en grok-voice-think-fast-1.0. De alias is handig tijdens prototyping en niet stabiel genoeg voor iets anders.

Toen ik op 4 augustus 2026 testte, verwees grok-voice-latest nog steeds naar grok-voice-think-fast-1.0, met SpaceXAI's releasenotes die de overgang naar Think Fast 2.0 voor de volgende dag inplanden. Die switch is net zo goed een prijswijziging als een modelwijziging, $0,08 per minuut audio tegenover $0,05 voor 1.0, dus een niet-vaste alias wordt duurder zonder dat er een regel code verandert. Pin de versiestring in alles wat je uitrolt.

Wat we gaan bouwen

De agent dekt wat een supportlijn gevraagd krijgt: een bestelling opzoeken, er eentje vinden via een e-mailadres als de beller geen nummer heeft, een bezorginstructie wijzigen, annuleren, een ticket openen of checken, en het gesprek doorzetten naar een persoon. Onderbrekingen en een verbroken verbinding komen onderweg voorbij.

Het is een handvol kleine bestanden in plaats van één script, omdat elk onderdeel een andere taak heeft en je ze apart wilt testen. Dit is de indeling:

• config.py laadt de API-sleutel en bevat de modelstring, samplefrequentie en endpoint-URL's

• voice_client.py wrapt de WebSocket, houdt de kosten bij en biedt send/receive-helpers

• tools.py definieert de bestelfuncties en een kleine in-memory bestellingsopslag als placeholder voor een echte database

• assistant.py bevat de systeemprompt, sessieconfiguratie en de eventloop die alles samenbrengt

• token_server.py is een kleine FastAPI-endpoint dat tijdelijke tokens uitgeeft

• app_streamlit.py zet dezelfde client achter een live browsergesprek, waar ik na de testsectie op terugkom

Het leerpad loopt vanuit een terminal. De demo voegt de microfoon toe.

Vereisten

Je hebt een SpaceXAI-account met een API-sleutel nodig, een gefinancierde betaalregeling (er is geen permanente free tier, en promotietegoeden voor nieuwe accounts redden het niet), en voldoende comfort met asyncio en WebSockets om mee te kunnen zonder een regeltje-voor-regeltje uitleg van await.

De quick-startvoorbeelden van SpaceXAI gebruiken het ruwe pakket websockets in plaats van een speciale SDK, en wij ook. De docs noemen geen vereiste Pythonversie. Ik testte op 3.11.

Houd de API-sleutel op de server. Als een browser of mobiele app direct met de Voice API praat, krijgt die een tijdelijk token in plaats van je echte sleutel, behandeld in de beveiligingssectie hieronder.

Het project opzetten

Elk bestand hieronder staat in de projectrepo, dus je kunt deze clonen in plaats van snippets te kopiëren:

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 verzorgt de realtimeverbinding en python-dotenv leest je sleutel. De rest dekt het token-endpoint en de browserdemo. Zet je sleutel in .env:

XAI_API_KEY=xai-your-key-here

Dat is het grootste deel van de setup. De verbinding is het interessante deel.

De Grok Voice Realtime API begrijpen

Grok Voice is de productnaam. Waar je in code daadwerkelijk tegenaan praat is een WebSocket-endpoint op wss://api.x.ai/v1/realtime, en het hele gesprek verloopt als een stroom JSON-events over die ene socket.

De eventlevenscyclus

Een verbinding volgt een vast patroon: de server stuurt session.created en conversation.created zodra je verbindt, jij stuurt session.update om stem en tools te configureren, de server bevestigt met session.updated, en vanaf daar maak je conversation items aan en vraag je om responses. Ik draaide dit tegen een live sleutel en de volgorde klopte precies met de docs.

• session.update (client) configureert stem, instructies, tools en audioformaat

• conversation.item.create (client) voegt een bericht van gebruiker, assistent of een toolresultaat toe

• response.create (client) vraagt het model om te spreken; server-VAD stuurt dit automatisch voor je

• response.output_audio.delta en response.output_audio_transcript.delta (server) streamen het antwoord terwijl het wordt gegenereerd

• response.done (server) sluit de beurt af

Twee dingen zijn valkuilen. De Speech to Speech-docspagina die ik hierboven linkte noemt een conversation.item.created event tijdens sessiehervatting, maar de canonieke eventreferentie vermeldt alleen conversation.item.added, en dat is wat in elke test die ik draaide binnenkwam, dus codeer daartegen. Je ziet ook een ongedocumenteerd ping event een paar seconden in de meeste verbindingen, alleen genoemd zodat je het niet als fout leest.

Audioformaten en transport

Codec en transport zijn aparte keuzes. De codec, ingesteld onder audio.input.format en audio.output.format, is audio/pcm (Linear16, standaard 24.000 Hz), audio/pcmu of audio/pcma (G.711 op 8 kHz, voor telefonie), of audio/opus (24 kHz). Transport is hoe die bytes over de lijn gaan:

• json (de standaard) stuurt audio als base64-tekst binnen input_audio_buffer.append en response.output_audio.delta, makkelijk te loggen en debuggen

• binary stuurt ruwe codecbytes als WebSocket-binaire frames, zonder base64-overhead maar met een ontvanglus die moet vertakken op berichttype

Begin met JSON. Elk voorbeeld in de docs gebruikt het, inspectie is triviaal, en base64-overhead is niet wat een supportagent-bouw afremt. Ga naar binary als je daar een reden voor meet.

Compatibiliteit met de OpenAI Realtime API

Sla dit over als je nooit met OpenAI's Realtime API hebt gewerkt. Voor de rest: de Speech to Speech API volgt de OpenAI Realtime API voldoende dat de meeste clientcode werkt na het wijzigen van de base-URL en de sleutel, maar het is geen perfecte drop-in.

Transcripties komen hier binnen als conversation.item.input_audio_transcription.updated in plaats van OpenAI's delta, een paar OpenAI-events worden niet ondersteund, en SpaceXAI voegt eigen extensies toe: force_message voor een gescripte meldingszin, resumption voor opnieuw verbinden, en replace voor het corrigeren van verkeerd uitgesproken merknamen vóór text-to-speech.

De real-time voice-agent bouwen

Genoeg protocol. Hier is de client die ermee praat.

Verbinden en de sessie configureren

De verbinding opent met een bearer-token en een modelqueryparameter, en het eerste bericht dat je stuurt configureert alles over hoe de agent zich gedraagt:

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 is de systeemprompt, en dit model wil korte prompts. SpaceXAI's migratienotities zeggen dat je prompts die voor oudere GPT-achtige voicemodellen zijn geschreven moet vereenvoudigen in plaats van ze letterlijk over te zetten. De mijne vertelt de agent om kort te antwoorden, één vraag tegelijk te stellen en alles wat wordt weggeschreven eerst hardop voor te lezen. Een uitgesproken bevestiging is een UX-aardigheidje, geen beveiligingscontrole. Je applicatie handhaaft autorisatie nog steeds op de schrijfoperatie zelf.

Eén ding verbaasde me: een niet-herkende modelstring veroorzaakt geen fout bij het verbinden, maar valt stilletjes terug naar grok-voice-think-fast-1.0. Een betaalde aanvraag downgraden door een typefout, zonder dat te melden, is een vreemde standaard. Log het veld session.model van session.created één keer bij het opstarten en check of je kreeg wat je vroeg.

Terminaloutput die session.created toont na verbinden. Afbeelding door auteur.

Gebruikersaudio streamen

Met turn_detection.type ingesteld op server_vad, hoef je alleen maar audio te blijven toevoegen. De server beslist wanneer de beller is uitgepraat en triggert dan de respons voor je. Zet dit in plaats daarvan op null en jij neemt die beslissing zelf, waarbij je de buffer expliciet commit wanneer jij denkt dat de beurt voorbij is.

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(),
    }))

Server-VAD heeft drie knoppen, en ze verkeerd instellen is de meest voorkomende manier waarop ik een voice-agent gebroken heb zien aanvoelen terwijl er niets in de logs fout ging. Geen van deze verschijnt standaard in de echo van session.updated, dus check ze tegen de docs in plaats van uit te gaan van aannames.

• threshold (0,1 tot 0,9, standaard 0,85), hoe hard audio moet zijn om als spraak te tellen; verhoog in lawaaiige ruimtes, verlaag als stille sprekers gemist worden

• silence_duration_ms, hoe lang de beller stil is voordat de server hun beurt beëindigt; te kort kapt mensen halverwege een gedachte af, te lang voelt traag

• prefix_padding_ms (standaard 333), een stukje audio bewaard van net vóór spraak werd gedetecteerd, zodat de eerste lettergreep niet wordt weggeknipt

Stel silence_duration_ms als eerste af als bellers steeds worden afgekapt wanneer ze even nadenken. Dat is degene waar ik als eerste aan draai, vóór de andere twee.

De respons ontvangen en afspelen

Audio komt in kleine stukjes binnen als response.output_audio.delta, en het punt van streamen is dat je elk stuk afspeelt zodra het binnenkomt in plaats van te wachten op 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)

Bewaar de transcriptie ook in productie. Het is het goedkoopste debughulpmiddel dat je hebt wanneer een beller zegt dat de agent "iets geks zei".

Tools toevoegen aan de voice-agent

Een voice-agent die alleen praat is een chatbot met een microfoon.

De ordertools maken

Elke tool is een JSON-schema plus een simpele Pythonfunctie aan onze kant. Het model raakt de database nooit; het ziet alleen wat onze functie teruggeeft.

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
]

Leesoperaties zoals check_order_status zijn veilig te retrypen als er iets time-out. Schrijfoperaties niet: het retrypen van update_delivery_instructions na een onduidelijke timeout kan dezelfde wijziging twee keer toepassen. Een bevestigingszin in de prompt stopt dat niet, dus geef writes een idempotency key of een duplicaatcheck.

Zet de weigeringen ook in de functie. cancel_order geeft een reden en alternatief terug in plaats van een verzonden bestelling te annuleren, want een prompt met "annuleer nooit verzonden bestellingen" is een suggestie en een functie die weigert is dat niet.

De tool-aanroep lus afhandelen

Vier stappen, en de volgorde doet er meer toe dan je zou denken. Het model stuurt response.function_call_arguments.done, jouw code draait de functie, jij stuurt het resultaat terug als een function_call_output item, en pas dan vraag je het model om door te gaan.

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),
        },
    }))

Als het model meer dan één tool nodig heeft voor een verzoek, vuurt het meerdere function_call_arguments.done events af voordat er audio speelt. Los ze allemaal op en stuur elk resultaat terug vóór één enkele response.create. Stuur die te vroeg en het model antwoordt zonder de context van de nog lopende aanroepen.

Hier zit een addertje onder het gras dat SpaceXAI documenteert en dat ik toch in de eerste run raakte: response.create meteen sturen zodra je toolresultaat eruit gaat, kan overlappen met de inleidende zin die de agent nog afspeelt. In één run opende het met "Ik check meteen de status van bestelling ORD-1042" en riep halverwege de zin de tool aan, dus een onmiddellijke response zou over zijn eigen opening heen hebben gepraat.

Wacht tot de audio van de huidige beurt is afgelopen en toon een korte "denkt..."-status ertussen.

Toolaanroep-flow vóór het vervolgen van de response. Afbeelding door auteur.

Onderbrekingen en gespreksstatus beheren

Hier spelen twee losse problemen. Een beller praat over de agent heen midden in een antwoord, en een WebSocket valt weg en moet worden hervat.

Natuurlijke onderbrekingen ondersteunen

Met server_vad aan is barge-in automatisch aan de serverkant: zodra het detecteert dat de beller weer praat, signaleert het input_audio_buffer.speech_started en stopt de generatie van het oude antwoord. Jouw taak is de clienthelft van die handshake: leeg de audio die al in de wachtrij staat zodat de agent stilvalt in plaats van een zin af te maken waar niemand nog om vroeg.

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

Voor handmatige, niet-VAD-sessies doet response.cancel hetzelfde op verzoek. Er is ook conversation.item.truncate om een assistant-item terug te snoeien tot wat daadwerkelijk is gehoord. De docs bevestigen dat het bestaat maar niet wanneer je het moet afvuren tijdens een live barge-in, dus test de timing zelf.

Ik testte dit met een wijziging van de bezorginstructie halverwege een antwoord: start het verzoek, onderbreek met een ander adresdeel terwijl de agent nog bevestigt. Wat telt is of de agent de gecorrigeerde instructie toepast in plaats van stilletjes de oude af te ronden, niet of de audio stopte. Check het orderrecord, niet de stilte. De browserdemo aan het eind laat je dit horen.

Een verbroken sessie hervatten

Sessieresumptie is opt-in en het is geen geheugen. Zet resumption.enabled: true op session.update, pak het ID uit het event conversation.created, en als de socket wegvalt, verbind opnieuw met ?conversation_id=<id> in de URL en kies weer voor opt-in op de nieuwe verbinding.

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

De gecachte beurten, transcripties, toolaanroepen en toolresultaten worden teruggespeeld voordat je volgende vraag komt, en de cache verdwijnt na 30 minuten inactiviteit. Ik testte het door naar een bestelling te vragen, de verbinding te verbreken en opnieuw te verbinden voor een vervolg zonder mezelf te herhalen; de agent pakte de ETA correct op.

Eén ongedocumenteerde kanttekening: de replay landt niet instantaan, dus een vraag die je op het moment van verbinden afvuurt, kan hem voor zijn en terugkomen zonder geheugen van de eerdere beurt. Geef het een seconde voor je resumptie de schuld geeft.

Terminallog van een hervatte sessie. Afbeelding door auteur.

Gebruik dit niet in plaats van bestelstatus in je eigen database op te slaan. Als de cache verloopt of de beller morgen terugbelt, begin je zonder context, en dat is met opzet.

De agent beveiligen en monitoren

Zet nooit een permanente API-sleutel in browser- of mobiele code. Als een client direct verbindt in plaats van via je server, geef dan een kortlevend token uit:

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": ...}

Een browser kan geen aangepaste Authorization header zetten op een WebSocket-handshake, dus geeft hij het token door via de sec-websocket-protocol-header, voorafgegaan door xai-client-secret..

Server geeft token uit, browser stapt in het gesprek. Afbeelding door auteur.

Facturering draait op twee meters. Audio, verzonden of ontvangen, loopt tegen de $0,08 per minuut die ik eerder noemde, wat $4,80 per uur is, en elke conversation.item.create die geen audio is en geen function_call_output kost een vaste $0,004. response.create wordt helemaal niet gefactureerd. Elke response.done bevat een usage-object, dat in mijn test output_audio_seconds rapporteerde naast een apart billable_audio_seconds. Factureer vanaf die waarden, niet vanaf schattingen.

Gedocumenteerde limieten op de Speech to Speech API zijn 10 gelijktijdige sessies per team en een sessielimiet van 120 minuten, beide in us-east-1. Plan geen capaciteit op basis van de cijfers van de Voice Agent API, die anders zijn.

Wees precies over privacy. SpaceXAI's security-FAQ zegt dat API-verzoeken en -antwoorden 30 dagen versleuteld bewaard blijven voor misbruikmonitoring en niet voor training worden gebruikt zonder toestemming, en dat teams Zero Data Retention kunnen aanzetten, al laat ZDR de bewaarde gespreksgeschiedenis van voice-agents vallen en werkt daarom niet met resumptie.

Als je meldt dat een gesprek wordt opgenomen of door AI wordt afgehandeld, daarvoor is de force_message-extensie die ik eerder noemde. De zin speelt exact zoals geschreven af in plaats van wat het model ervan maakt.

De voice-agent testen

Een 200-status op de WebSocket-handshake zegt niets over of de agent het juiste deed. Test het resultaat, niet alleen de verbinding.

• Een schone orderlookup, waarbij je het uitgesproken antwoord checkt tegen het record, niet alleen dat er een respons kwam
• Een onderbroken antwoord, bevestigen dat afspelen stopt en de agent het nieuwe verzoek oppakt
• Een bezorgupdate die bevestiging vereist, gecheckt tegen het orderrecord
• Een weigering, zoals het annuleren van een verzonden bestelling, waar de agent de regel moet uitleggen in plaats van zich te verontschuldigen
• Een onbekend ordernummer, zorgen dat de agent dat zegt in plaats van een status te verzinnen
• Een tool die een fout teruggeeft, checken dat de agent die uitspreekt in plaats van te blijven hangen
• Opnieuw verbinden en hervatten, inclusief dat replayvenster waar ik eerder tegenaan liep
• Rumoerige audio, snel praten en een beller die nummers en adressen spelt

Ik draaide het meeste hiervan tegen een live sleutel tijdens het schrijven. De interessante mislukkingen waren gedragsmatig, geen fouten: de resumptietiming hierboven, en een VAD-drempel buiten bereik die werd geaccepteerd in plaats van geweigerd; het soort dat stil kapot in productie belandt als je alleen de happy path test. Voeg ook een meertalige test toe, en zie de FAQ's voor een bijzonderheid in hoe je de taal benoemt.

Twee daarvan kun je niet door te typen testen. app_streamlit.py is een Streamlit-pagina die een live gesprek in de browser zet: de microfoon streamt via WebRTC naar dezelfde WebSocket, de stem van de agent stroomt terug, en de socket blijft de hele tijd open.

streamlit run app_streamlit.py