Grok Voice Think Fast 2.0 API-tutorial: bouw een realtime voice-agent in Python

Grok Voice Think Fast 2.0 van SpaceXAI is een speech-to-speech-model. Je stuurt audio via een WebSocket en het stuurt audio terug, en tussendoor kan het redeneren en blijven praten terwijl een functieaanroep die het heeft gestart al draait. Geen aparte speech-to-text-stap, geen aparte text-to-speech-stap.

SpaceXAI kondigde Think Fast 2.0 aan op 29 juli 2026: snellere eerste audio, stabieler full-duplex gedrag (luisteren terwijl het praat in plaats van strikt beurten nemen), en tool-calls die vroeg in de beurt afgaan. Ik houd de benchmarkpraat kort, want wat telt voor een tutorial is wat er in je code verandert.

We gaan een klantenservice-voice-agent bouwen voor een online winkel. Een beller kan vragen naar een bestelling, een bezorginstructie wijzigen, annuleren, de agent halverwege een zin onderbreken en het gesprek weer oppakken 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-first versie.

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 meeste mensen gewoon Grok Voice noemen. Als je het bedrijf nog als xAI ziet: hetzelfde team—het is op 6 juli 2026 in SpaceX opgegaan en hernoemd naar SpaceXAI. De API is die rebrand niet gevolgd, dus elke identifier hieronder zegt nog steeds xai, van de variabele XAI_API_KEY tot de host api.x.ai.

Een traditionele voicestack koppelt drie diensten: speech-to-text, een taalmodel, en dan text-to-speech, en elke stap voegt latency toe en een plek waar context kan verdwijnen. Think Fast 2.0 vouwt dat samen tot één model dat audio of tekst inneemt en audio of tekst uitstuurt over dezelfde verbinding.

Speech-to-speech-WebSocket versus pijplijnarchitectuur met drie services. Afbeelding door auteur.

Voor een agent die handelt in plaats van alleen praat, is het belangrijk dat redeneren en spraak parallel lopen. SpaceXAI zegt dat tool-calls "meestal" al starten voordat de agent zijn eerste zin afrondt, en dat woord doet echt werk.

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 daalt de tijd tot eerste audio van 1,25 seconden naar 0,70 seconden. Leverancierscijfers op een algemene benchmark zijn een hypothese over je belstroom, geen testplan.

Er zijn drie modelstrings die je zult zien: grok-voice-latest, grok-voice-think-fast-2.0 en grok-voice-think-fast-1.0. Het 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 plannen. 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-gepinde alias wordt duurder zonder dat er een regel code van jou verandert. Pin de versie-string in alles wat je uitrolt.

Wat we gaan bouwen

De agent dekt wat een supportlijn krijgt: een bestelling opzoeken, er een vinden op basis van 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 zijn een handvol kleine bestanden in plaats van één script, omdat elk onderdeel een andere taak heeft en je ze los wilt kunnen testen. Dit is de indeling:

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

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

• tools.py definieert de bestellingsfuncties en een kleine in-memory orderstore als vervanger van een echte database

• assistant.py bevat de system prompt, sessieconfiguratie en de eventloop die alles aan elkaar knoopt

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

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

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

Vereisten

Je hebt een SpaceXAI-account nodig met een API-sleutel, een gefinancierde betaalregeling (er is geen permanente gratis laag, en promotionele credits voor nieuwe accounts dragen je niet), en genoeg comfort met asyncio en WebSockets om mee te kunnen zonder een regel-voor-regel uitleg van await.

De quick-startvoorbeelden van SpaceXAI gebruiken het ruwe pakket websockets in plaats van een dedicated SDK, en wij ook. De docs noemen geen vereiste Python-versie. 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 clonen in plaats van snippets 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 meeste van de setup. De verbinding is het interessante deel.

De Grok Voice Realtime API begrijpen

Grok Voice is de productnaam. Waar je daadwerkelijk tegenaan codeert 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 event-lifecycle

Een verbinding volgt een vaste vorm: de server stuurt session.created en conversation.created zodra je verbindt, jij stuurt session.update om voice 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 kwam exact overeen met de docs.

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

• conversation.item.create (client) voegt een gebruikersbericht, een assistentbericht 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 laten mensen struikelen. 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 arriveerde, dus codeer daartegen. Je ziet ook een ongedocumenteerd ping -event een paar seconden in de meeste verbindingen, alleen genoemd zodat je het niet als een 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 24000 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 rauwe codecbytes als WebSocket-binaire frames, waarbij de base64-overhead wordt overgeslagen ten koste van een ontvangstlus die op berichttype moet vertakken

Begin met JSON. Elk voorbeeld in de docs gebruikt het, het is triviaal te inspecteren, en base64-overhead is niet wat een supportagent-build afremt. Ga naar binary als je daar een gemeten reden voor hebt.

Compatibiliteit met de OpenAI Realtime API

Sla dit over als je nog nooit de Realtime API van OpenAI hebt aangeraakt. Voor iedereen anders volgt de Speech to Speech API de OpenAI Realtime API nauw genoeg dat de meeste clientcode overzet door de base-URL en de sleutel te wijzigen, 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 disclosureregel, resumption voor reconnects, en replace om verkeerd uitgesproken merknamen te corrigeren vóór text-to-speech.

De realtime 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 system prompt, en dit model wil korte prompts. SpaceXAI's migratienotities zeggen dat je prompts die voor oudere GPT-tijdperk-voicemodellen zijn geschreven moet vereenvoudigen in plaats van ze letterlijk over te zetten. De mijne zegt de agent kort te antwoorden, één vraag tegelijk te stellen en elke schrijfactie terug te lezen vóórdat er gehandeld wordt. Een gesproken bevestiging is een UX-attentiepunt, geen beveiligingscontrole. Je applicatie handhaaft nog steeds autorisatie op de schrijfactie zelf.

Eén ding dat me verraste: een niet-herkende modelstring geeft geen fout bij het verbinden, maar valt stilzwijgend terug naar grok-voice-think-fast-1.0. Een betaalde aanvraag downgraden door een typo, zonder dat te melden, is een vreemde standaard. Log het veld session.model van session.created één keer bij startup en controleer dat je kreeg wat je vroeg.

Terminaloutput met session.created na verbinding maken. 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 gestopt met praten en triggert de response voor je. Zet het in plaats daarvan op null en jij neemt die beslissing zelf, waarbij je de buffer expliciet commit als jij vindt 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 faalde. Geen ervan verschijnt standaard in de echo van session.updated, dus check ze tegen de docs in plaats van aan te nemen.

• threshold (0,1 tot 0,9, standaard 0,85), hoe hard audio moet zijn om als spraak te tellen; verhoog in rumoerige 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 hun gedachte af, te lang voelt traag

• prefix_padding_ms (standaard 333), een stukje audio net vóórdat spraak werd gedetecteerd dat behouden blijft, zodat de eerste lettergreep niet wordt weggeknipt

Stel silence_duration_ms eerst af als bellers blijven worden afgekapt terwijl ze nadenken. Dat is degene waar ik naar grijp vóór de andere twee.

De response ontvangen en afspelen

Audio komt in kleine stukjes binnen als response.output_audio.delta, en het punt van streamen is dat je elk stukje meteen afspeelt zodra het landt 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 debugtool dat je hebt wanneer een beller zegt dat de agent "iets raars 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 eenvoudige Python-functie aan onze kant. Het model raakt de database nooit aan; het ziet alleen wat onze functie terugstuurt.

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
]

Leesbewerkingen zoals check_order_status zijn veilig om te retrypen als iets time-out. Schrijfbewerkingen niet: het retrypen van update_delivery_instructions na een dubbelzinnige timeout kan dezelfde wijziging twee keer toepassen. Een bevestigingsregel in de prompt voorkomt dat niet, dus geef writes een idempotency key of een duplicaatcheck.

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

De tool-calllus afhandelen

Vier stappen, en de volgorde doet er meer toe dan het lijkt. 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 vóór een enkele response.create. Stuur die te vroeg en het model antwoordt zonder de context van de nog lopende calls.

Hier zit een valkuil die SpaceXAI documenteert en die ik toch bij de eerste poging raakte: response.create sturen op het moment dat je toolresultaat eruit gaat kan overlappen met de inleidende zin die de agent nog afspeelt. In één run opende het met "Ik kijk meteen de status van order ORD-1042 na" en riep het de tool halverwege de zin aan, dus een directe response had over zijn eigen opener heen gepraat.

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

Tool-callflow vóór het vervolg van de response. Afbeelding door auteur.

Onderbrekingen en gesprekstoestand beheren

Twee aparte problemen hier. Een beller praat over de agent heen tijdens een response, en een WebSocket valt weg en moet worden hervat.

Natuurlijke onderbrekingen ondersteunen

Met server_vad aan is barge-in server-side automatisch: zodra het detecteert dat de beller weer spreekt, signaleert het input_audio_buffer.speech_started en stopt het met het genereren van de oude response. Jouw taak is de clienthelft van die handshake: wis alles wat al in de wachtrij voor audio staat zodat de agent stilvalt in plaats van een zin af te maken waar niemand om vroeg.

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

Voor handmatige, niet-VAD-sessies doet response.cancel op verzoek hetzelfde. Er is ook conversation.item.truncate om een assistentitem bij te snijden 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 bezorginstructies halverwege de response: start het verzoek, onderbreek met een ander adres halverwege de bevestiging van de agent. Wat telt is of de agent de gecorrigeerde instructie toepast in plaats van stilletjes de oude te voltooien, niet of de audio stopte. Check het orderrecord, niet de stilte. De browserdemo aan het eind laat 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 uitvalt, verbind opnieuw met ?conversation_id=<id> in de URL en kies opnieuw 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, tool-calls en toolresultaten spelen opnieuw af vóór je volgende vraag, 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 weer op.

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

Terminallog van een hervatte sessie. Afbeelding door auteur.

Gebruik dit niet als vervanging voor het opslaan van orderstatus in je eigen database. Als de cache verloopt of de beller morgen terugbelt, begin je vanaf nul context, en dat is bewust zo ontworpen.

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 die het token door via de header sec-websocket-protocol, voorafgegaan door xai-client-secret..

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

Billing loopt op twee meters. Audio, verzonden of ontvangen, loopt op 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 daarop, niet op schattingen.

Gedocumenteerde limieten op de Speech to Speech API zijn 10 gelijktijdige sessies per team en een 120-minuten sessiecap, 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 -responses 30 dagen versleuteld worden bewaard voor misbruikmonitoring en niet voor training worden gebruikt zonder toestemming, en dat teams Zero Data Retention kunnen inschakelen, hoewel ZDR bewaarde voice-agentgesprekshistorie laat vallen en daarom niet werkt met resumptie.

Als je meldt dat een gesprek wordt opgenomen of door AI wordt afgehandeld, is dat waar de force_message-extensie die ik eerder noemde voor is. De regel wordt exact zo uitgespeeld als geschreven, in plaats van wat het model ervan zou maken.

De voice-agent testen

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

• Een schone orderlookup, waarbij je het uitgesproken antwoord checkt tegen het record, niet alleen dat er een response binnenkwam
• Een onderbroken response, bevestigen dat de weergave stopt en de agent de nieuwe vraag oppakt
• Een bezorgupdate die een bevestiging vereist, gecontroleerd tegen het orderrecord
• Een weigering, zoals het annuleren van een verzonden bestelling, waarbij de agent de regel moet uitleggen in plaats van zich te verontschuldigen
• Een onbekend ordernummer, zeker weten 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
• Herverbinden en hervatten, inclusief dat replaywindow waar ik eerder tegenaan liep
• Rumoerige audio, snelle spraak en een beller die nummers en adressen spelt

Ik draaide het meeste hiervan tegen een live sleutel tijdens het schrijven. De interessante falen waren gedragsmatig, geen errors: de resumptietiming hierboven, en een VAD-drempel buiten bereik die geaccepteerd werd in plaats van afgewezen, het soort dingen dat stil gebroken live gaat als je alleen de happy path test. Voeg ook een meertalige test toe, en zie de FAQ voor een bijzonderheid in hoe je de taal noemt.

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

streamlit run app_streamlit.py