Grok Voice Think Fast 2.0 API-Tutorial: Baue einen Echtzeit-Voice-Agenten in Python

SpaceXAI's Grok Voice Think Fast 2.0 ist ein Speech-to-Speech-Modell. Du sendest Audio über einen WebSocket und bekommst Audio zurück, während das Modell zwischendurch argumentiert und weiter spricht, obwohl ein von ihm gestarteter Funktionsaufruf bereits läuft. Kein separater Speech-to-Text-Schritt, kein separater Text-to-Speech-Schritt.

SpaceXAI kündigte Think Fast 2.0 am 29. Juli 2026 an: schnelleres erstes Audio, stabileres Vollduplex-Verhalten (zuhören, während es spricht, statt strikter Redewechsel) und Toolaufrufe, die früh in der Runde starten. Benchmarks halte ich kurz – in einem Tutorial zählt, was du am Code änderst.

Wir bauen einen Voice-Agenten für den Kundensupport eines Onlineshops. Anrufende können nach einer Bestellung fragen, Lieferhinweise ändern, stornieren, den Agenten mitten im Satz unterbrechen und nach einer unterbrochenen Verbindung weitersprechen. Das ist der API-Weg, nicht der No-Code Voice Agent Builder, den unser Grok Voice Agent Builder Tutorial zeigt. Starte dort für die Console-First-Variante.

Was ist Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 ist SpaceXAI's neuestes Modell für die Speech to Speech API, der Produktname hinter dem, was die meisten einfach Grok Voice nennen. Wenn du das Unternehmen noch als xAI kennst: Es wurde am 6. Juli 2026 in SpaceX integriert und zu SpaceXAI umbenannt. Die API hat die Umbenennung nicht mitgemacht, daher heißen alle Bezeichner unten weiterhin xai, von der XAI_API_KEY-Variable bis zum api.x.ai-Host.

Ein klassischer Voice-Stack kettet drei Dienste: Speech-to-Text, ein Sprachmodell, dann Text-to-Speech. Jeder Hop erhöht die Latenz und birgt Kontextverlust. Think Fast 2.0 führt das zu einem einzigen Modell zusammen, das Audio oder Text entgegennimmt und über dieselbe Verbindung Audio oder Text ausgibt.

Speech-to-Speech-WebSocket versus Drei-Dienste-Architektur. Bild: Autor.

Für einen Agenten, der handelt statt nur zu reden, zählt, dass Reasoning und Sprache parallel laufen. SpaceXAI sagt, Toolaufrufe starten "üblicherweise", bevor der Agent seinen ersten Satz beendet – und dieses Wort trägt Gewicht.

In den von SpaceXAI zitierten Benchmarks von Artificial Analysis erreicht Think Fast 2.0 82,9 % im Speech-to-Speech-Index gegenüber 75,7 % für 1.0 und senkt die Time-to-First-Audio von 1,25 Sekunden auf 0,70 Sekunden. Anbieterzahlen auf einem allgemeinen Benchmark sind eine Hypothese zu deinem Callflow, kein Testplan.

Du siehst drei Modellstrings: grok-voice-latest, grok-voice-think-fast-2.0 und grok-voice-think-fast-1.0. Der Alias ist beim Prototyping bequem und für alles andere zu instabil.

Am 4. August 2026 hat grok-voice-latest noch auf grok-voice-think-fast-1.0 gezeigt, mit Release Notes für den Wechsel auf Think Fast 2.0 am Folgetag. Dieser Wechsel ist Preis- und Modelländerung zugleich: $0,08 pro Audiominute statt $0,05 für 1.0. Ein unfixierter Alias wird teurer, ohne dass sich eine Zeile deines Codes ändert. Pinniere den versionierten String für alles, was du deployest.

Was wir bauen

Der Agent deckt typische Support-Anliegen ab: Bestellung nachschlagen, per E-Mail finden, wenn keine Nummer vorliegt, Lieferhinweis ändern, stornieren, Ticket öffnen/prüfen und an eine Person übergeben. Unterbrechungen und eine getrennte Verbindung kommen unterwegs vor.

Das Projekt besteht aus ein paar kleinen Dateien statt einem Monoskript – jedes Teil hat seine Aufgabe, und du willst sie getrennt testen. So ist das Layout:

• config.py lädt den API-Schlüssel und hält Modellstring, Samplerate und Endpunkt-URLs

• voice_client.py kapselt den WebSocket, trackt Kosten und bietet Send/Receive-Helfer

• tools.py definiert die Bestellfunktionen und einen kleinen In-Memory-Orderspeicher als Platzhalter für eine echte Datenbank

• assistant.py enthält den Systemprompt, die Sitzungs-Config und die Eventschleife, die alles zusammenbindet

• token_server.py ist ein kleines FastAPI-Endpoint, das kurzlebige Tokens erstellt

• app_streamlit.py setzt denselben Client hinter einen Live-Browser-Call, darauf komme ich nach dem Test-Teil zurück

Die Lernstrecke läuft im Terminal. Das Demo bringt das Mikrofon dazu.

Voraussetzungen

Du brauchst ein SpaceXAI-Konto mit API-Schlüssel, eine hinterlegte Zahlungsart (es gibt keine dauerhafte Free-Tier, und Neukunden-Guthaben tragen dich nicht) und genug Routine mit asyncio und WebSockets, um ohne Zeile-für-Zeile-Erklärung von await zu folgen.

SpaceXAI's Quick-Start-Beispiele nutzen das rohe websockets-Paket statt eines dedizierten SDK – wir auch. Die Doku nennt keine erforderliche Python-Version. Getestet habe ich mit 3.11.

Behalte den API-Schlüssel auf dem Server. Wenn ein Browser oder eine Mobile App direkt mit der Voice-API spricht, bekommt sie ein kurzlebiges Token statt deines echten Schlüssels – mehr dazu im Security-Teil unten.

Projekt einrichten

Alle Dateien unten liegen im Projekt-Repo, du kannst also klonen statt Schnipsel zu kopieren:

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 trägt die Realtime-Verbindung und python-dotenv liest deinen Schlüssel. Der Rest deckt den Token-Endpoint und das Browser-Demo ab. Pack deinen Schlüssel in .env:

XAI_API_KEY=xai-your-key-here

Das war fast alles Setup. Die Verbindung ist der spannende Teil.

Die Grok Voice Realtime-API verstehen

Grok Voice ist der Produktname. Der eigentliche Code spricht gegen einen WebSocket-Endpunkt unter wss://api.x.ai/v1/realtime, und die ganze Unterhaltung läuft als Strom von JSON-Events über genau diesen Socket.

Der Event-Lebenszyklus

Eine Verbindung folgt einem festen Muster: Der Server schickt session.created und conversation.created direkt nach dem Connect, du sendest session.update zur Konfiguration von Stimme und Tools, der Server bestätigt mit session.updated, und ab da erzeugst du Conversation-Items und forderst Antworten an. Ich habe das gegen einen Live-Schlüssel gefahren, die Reihenfolge entsprach exakt der Doku.

• session.update (Client) konfiguriert Stimme, Instruktionen, Tools und Audioformat

• conversation.item.create (Client) fügt eine User-Message, eine Assistant-Message oder ein Toolresultat hinzu

• response.create (Client) fordert das Modell zum Sprechen auf; Server-VAD sendet das automatisch für dich

• response.output_audio.delta und response.output_audio_transcript.delta (Server) streamen die Antwort während der Generierung

• response.done (Server) schließt den Turn ab

Zwei Stolpersteine: Die verlinkte Speech-to-Speech-Seite nennt ein conversation.item.created Event bei Sitzungsfortsetzung, aber die kanonische Event-Referenz listet nur conversation.item.added – und genau das kam in all meinen Tests an. Code also darauf. Zudem siehst du ein undokumentiertes ping ein paar Sekunden nach Verbindungsaufbau – nur erwähnt, damit du es nicht als Fehler liest.

Audioformate und Transport

Codec und Transport wählst du getrennt. Den Codec stellst du unter audio.input.format und audio.output.format ein: audio/pcm (Linear16, Standard 24000 Hz), audio/pcmu oder audio/pcma (G.711 bei 8 kHz, für Telefonie), oder audio/opus (24 kHz). Transport bestimmt, wie die Bytes übertragen werden:

• json (Standard) schickt Audio als Base64-Text in input_audio_buffer.append und response.output_audio.delta, leicht zu loggen und zu debuggen

• binary sendet rohe Codec-Bytes als WebSocket-Binary-Frames, spart Base64-Overhead auf Kosten einer Empfangsschleife mit Branching nach Nachrichtentyp

Starte mit JSON. Alle Doku-Beispiele nutzen es, es ist trivial zu inspizieren, und Base64-Overhead ist nicht der Engpass für einen Support-Agenten. Wechsle zu Binary, wenn Messungen einen Grund liefern.

Kompatibilität mit der OpenAI Realtime-API

Überspringe das, wenn du OpenAI's Realtime-API nie genutzt hast. Für alle anderen: Die Speech-to-Speech-API folgt der OpenAI Realtime-API eng genug, dass die meiste Client-Logik mit Base-URL- und Schlüsselwechsel portiert. Ein perfekter Drop-in ist sie nicht.

Transkripte kommen hier als conversation.item.input_audio_transcription.updated statt OpenAI's delta, ein paar OpenAI-Events fehlen, und SpaceXAI bringt eigene Erweiterungen: force_message für eine vorgeschriebene Offenlegung, resumption für Reconnects, und replace zum Korrigieren falsch ausgesprochener Markennamen vor TTS.

Den Echtzeit-Voice-Agenten bauen

Genug Protokoll. Hier ist der Client, der damit spricht.

Verbinden und Sitzung konfigurieren

Die Verbindung öffnet mit Bearer-Token und einem Modell-Query-Parameter, und die erste Nachricht konfiguriert das Verhalten des Agenten:

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 ist der Systemprompt, und dieses Modell mag kurze. SpaceXAI's Migrationshinweise raten, Prompts aus der älteren GPT-Voice-Ära zu vereinfachen statt 1:1 zu portieren. Meiner sagt dem Agenten: kurz antworten, eine Frage nach der anderen, und alles, was er schreibt, vorher vorlesen. Eine gesprochene Bestätigung ist UX, kein Security-Mechanismus. Deine Anwendung prüft die Berechtigung beim Schreiben ohnehin selbst.

Eine Sache hat mich überrascht: Ein unbekannter Modellstring wirft beim Connect keinen Fehler, sondern fällt still auf grok-voice-think-fast-1.0 zurück. Einen bezahlten Request wegen eines Tippfehlers downzugraden, ohne es zu sagen, ist ein seltsamer Standard. Logge session.created's Feld session.model einmal beim Start und prüfe, ob du bekommen hast, was du wolltest.

Terminalausgabe mit session.created nach dem Connect. Bild: Autor.

User-Audio streamen

Mit turn_detection.type auf server_vad fügst du nur kontinuierlich Audio an. Der Server entscheidet, wann der Anrufende aufgehört hat zu sprechen, und triggert die Antwort für dich. Setze es auf null, dann triffst du die Entscheidung selbst und übergibst den Puffer explizit, wenn der Turn vorbei ist.

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 hat drei Regler, und falsches Tuning lässt einen Voice-Agenten am häufigsten "kaputt" wirken, während Logs nichts zeigen. Keiner erscheint standardmäßig im Echo von session.updated, also prüfe in der Doku statt zu raten.

• threshold (0,1 bis 0,9, Standard 0,85): wie laut Audio sein muss, um als Sprache zu zählen; erhöhe in lauten Umgebungen, senke bei sehr leisen Sprecher:innen

• silence_duration_ms: wie lange Stille vergeht, bevor der Server den Turn beendet; zu kurz schneidet Gedanken ab, zu lang wirkt träge

• prefix_padding_ms (Standard 333): ein Audioschnipsel direkt vor der erkannten Sprache, damit die erste Silbe nicht abgeschnitten wird

Stelle zuerst silence_duration_ms ein, wenn Anrufende beim Nachdenken stets abgeschnitten werden. Das ist mein erster Griff vor den anderen beiden.

Antwort empfangen und abspielen

Audio kommt stückweise als response.output_audio.delta. Sinn von Streaming ist, jedes Stück sofort abzuspielen statt auf response.done zu warten.

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)

Behalte das Transkript auch in Produktion. Es ist dein günstigstes Debugging-Tool, wenn jemand sagt, der Agent habe "komisch geantwortet".

Tools zum Voice-Agent hinzufügen

Ein Voice-Agent, der nur redet, ist ein Chatbot mit Mikrofon.

Order-Tools erstellen

Jedes Tool ist ein JSON-Schema plus eine einfache Python-Funktion auf unserer Seite. Das Modell fasst die Datenbank nie an – es sieht nur, was unsere Funktion zurückgibt.

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
]

Reads wie check_order_status darfst du bei Timeout gefahrlos wiederholen. Writes nicht: Ein erneuter update_delivery_instructions nach zweifelhaftem Timeout kann dieselbe Änderung doppelt anwenden. Ein Prompt mit Bestätigungszeile verhindert das nicht – nutze Idempotenzschlüssel oder Duplikatsprüfungen.

Packe Ablehnungen in die Funktion. cancel_order gibt einen Grund und eine Alternative zurück, statt eine versandte Bestellung zu stornieren. Ein Prompt mit "storniere nie versandte Bestellungen" ist eine Bitte – eine Funktion, die verweigert, ist verbindlich.

Die Tool-Call-Schleife handhaben

Vier Schritte, und die Reihenfolge ist wichtiger, als sie wirkt. Das Modell sendet response.function_call_arguments.done, dein Code führt die Funktion aus, du schickst das Ergebnis als function_call_output -Item zurück und erst dann bittest du das Modell, fortzufahren.

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

Braucht das Modell mehr als ein Tool für eine Anfrage, feuert es mehrere function_call_arguments.done -Events, bevor Audio spielt. Löse alle und sende jedes Ergebnis, bevor du ein einziges response.create sendest. Zu früh gesendet, antwortet das Modell ohne den Kontext der noch laufenden Aufrufe.

Ein dokumentierter Fallstrick, den ich trotzdem erwischt habe: Ein response.create direkt nach dem Toolresultat kann sich mit dem Einleitungssatz überlappen, der noch läuft. In einem Lauf begann es mit "Ich prüfe gleich den Status von ORD-1042" und rief das Tool mitten im Satz – eine sofortige Antwort hätte sich selbst übergangen.

Warte, bis das Audio des aktuellen Turns fertig ist, und zeige dazwischen kurz einen "Denke..."-Status.

Tool-Call-Flow vor der Fortsetzung der Antwort. Bild: Autor.

Unterbrechungen und Konversationszustand managen

Zwei verschiedene Probleme: Anrufende sprechen dem Agenten in die Antwort, und ein WebSocket fällt ab und muss fortgesetzt werden.

Natürliche Unterbrechungen unterstützen

Mit server_vad ist Barge-in serverseitig automatisch: Sobald es erneute Sprache erkennt, signalisiert es input_audio_buffer.speech_started und stoppt die alte Antwort. Dein Part ist die Client-Hälfte: den bereits gequeueten Sound leeren, damit der Agent verstummt und nicht den Satz zu Ende spricht, den niemand mehr hören will.

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

Für manuelle, nicht-VAD-Sessions erledigt response.cancel dasselbe auf Zuruf. Es gibt auch conversation.item.truncate , um ein Assistant-Item auf den tatsächlich gehörten Teil zu kürzen. Die Doku bestätigt, dass es existiert, aber nicht, wann es während eines Live-Barge-ins zu feuern ist – das Timing musst du selbst testen.

Getestet habe ich das mit einer Änderung der Lieferhinweise mitten in der Antwort: Anfrage starten, dann während der Bestätigung mit einer anderen Adresse unterbrechen. Entscheidend ist, ob der Agent die korrigierte Anweisung anwendet statt still die alte zu Ende zu führen – nicht, ob das Audio stoppte. Prüfe gegen den Order-Record, nicht gegen die Stille. Das Browser-Demo am Ende lässt dich das hören.

Eine getrennte Sitzung fortsetzen

Sitzungsfortsetzung ist opt-in und keine Memory-Funktion. Setze resumption.enabled: true in session.update, hol dir die ID aus dem conversation.created-Event, und falls der Socket abbricht, verbinde neu mit ?conversation_id=<id> in der URL und opt-in erneut auf der neuen Verbindung.

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

Die zwischengespeicherten Turns, Transkripte, Toolaufrufe und -ergebnisse spielen vor deiner nächsten Frage erneut ab, und der Cache verfällt nach 30 Minuten Inaktivität. Ich habe gefragt, Verbindung getrennt, neu verbunden und eine Folgefrage gestellt, ohne mich zu wiederholen; der Agent hat die ETA korrekt wiederaufgenommen.

Ein undokumentierter Haken: Das Replay landet nicht sofort, eine Frage direkt beim Socket-Open kann es schlagen und ohne Erinnerung an den vorherigen Turn beantwortet werden. Gib ihm einen Moment, bevor du Resumption die Schuld gibst.

Terminal-Log einer fortgesetzten Sitzung. Bild: Autor.

Nutze das nicht anstelle eigener Persistenz im Ordersystem. Wenn der Cache abläuft oder der Anruf morgen wiederkommt, startest du bewusst ohne Kontext.

Agent absichern und überwachen

Lege niemals einen permanenten API-Schlüssel in Browser- oder Mobile-Code. Wenn ein Client direkt verbindet statt über deinen Server, erzeuge ein kurzlebiges Token:

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

Ein Browser kann beim WebSocket-Handshake keinen eigenen Authorization -Header setzen, daher übergibt er das Token über den Header sec-websocket-protocol, mit dem Prefix xai-client-secret..

Server prägt Token, Browser tritt dem Call bei. Bild: Autor.

Abrechnung läuft über zwei Zähler. Audio, gesendet oder empfangen, kostet die genannten $0,08 pro Minute, also $4,80 pro Stunde, und jeder conversation.item.create, der kein Audio und kein function_call_output ist, kostet pauschal $0,004. response.create wird nicht berechnet. Jedes response.done trägt ein usage-Objekt, das in meinem Test output_audio_seconds plus separate billable_audio_seconds enthielt. Rechne daraus ab, nicht aus Schätzungen.

Dokumentierte Limits der Speech-to-Speech-API: 10 gleichzeitige Sessions pro Team und 120 Minuten Session-Cap, beides in us-east-1. Plane Kapazitäten nicht anhand der Zahlen der Voice Agent API – die sind anders.

Zur Privatsphäre: Sei präzise. SpaceXAI's Security-FAQ sagt, API-Requests und -Responses werden 30 Tage verschlüsselt für Abuse-Monitoring aufbewahrt und ohne Zustimmung nicht zum Training genutzt. Teams können Zero Data Retention aktivieren – ZDR entfernt aber persistierte Voice-Agent-Verlaufshistorie und funktioniert daher nicht mit Resumption.

Wenn du aufzeichnungs- oder KI-Hinweise gibst, dafür ist die Erweiterung force_message gedacht. Die Zeile wird genau so abgespielt, nicht paraphrasiert.

Den Voice-Agent testen

Ein 200er beim WebSocket-Handshake sagt nichts darüber, ob der Agent das Richtige tat. Teste das Ergebnis, nicht nur die Verbindung.

• Ein sauberer Order-Lookup – gesprochene Antwort gegen den Datensatz prüfen, nicht nur eine Antwort prüfen
• Eine unterbrochene Antwort – Playback stoppt, Agent geht auf die neue Bitte ein
• Ein Lieferupdate mit Bestätigung – gegen den Order-Datensatz geprüft
• Eine Verweigerung, etwa Stornieren einer versandten Bestellung – der Agent erklärt die Regel statt sich zu entschuldigen
• Eine unbekannte Bestellnummer – Agent sagt das klar statt einen Status zu erfinden
• Ein Tool mit Fehler – Agent spricht den Fehler aus statt zu hängen
• Reconnect und Resumption – inklusive des Replay-Fensters von oben
• Rauschiges Audio, schnelles Sprechen und ein Anrufer, der Zahlen/Adressen buchstabiert

Ich habe das meiste davon gegen einen Live-Schlüssel durchgespielt. Die interessanten Fehler waren Verhaltensfragen, keine Exceptions: das Resumption-Timing oben und ein VAD-Threshold außerhalb des Bereichs, der akzeptiert statt abgelehnt wurde – Dinge, die still defekt shippen, wenn du nur die Happy-Path prüfst. Füge auch einen mehrsprachigen Test hinzu und siehe die FAQs für eine Besonderheit bei der Sprachangabe.

Zwei davon kannst du nicht tippen. app_streamlit.py ist eine Streamlit-Seite für einen Live-Call im Browser: Das Mikro streamt über WebRTC in denselben WebSocket, die Stimme des Agenten kommt zurück, der Socket bleibt offen.

streamlit run app_streamlit.py