Samouczek API Grok Voice Think Fast 2.0: zbuduj agenta głosowego czasu rzeczywistego w Pythonie

Grok Voice Think Fast 2.0 od SpaceXAI to model mowa–mowa. Wysyłasz do niego audio przez WebSocket, a on odsyła audio z powrotem — i w międzyczasie potrafi rozumować oraz mówić dalej, gdy wywołana przez niego funkcja już się wykonuje. Bez osobnego kroku rozpoznawania mowy i bez osobnego syntezatora mowy.

SpaceXAI ogłosiło Think Fast 2.0 29 lipca 2026 r.: szybsze pierwsze audio, stabilniejsze pełnodupleksowe zachowanie (słuchanie podczas mówienia zamiast sztywnych tur) i wywołania narzędzi uruchamiane wcześnie w turze. Zostawię benchmarki w skrócie — w samouczku liczy się to, co zmienisz w swoim kodzie.

Zbudujemy agenta głosowego do obsługi klienta dla sklepu internetowego. Dzwoniący może zapytać o zamówienie, zmienić instrukcję dostawy, anulować, przerwać agentowi w połowie zdania i wznowić rozmowę po zerwaniu połączenia. To ścieżka przez API, a nie no-code Voice Agent Builder, który omawia nasz samouczek Grok Voice Agent Builder. Zacznij tam od wersji konsolowej.

Czym jest Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 to najnowszy model SpaceXAI dla Speech to Speech API, nazwy produktu stojącej za tym, co większość nazywa po prostu Grok Voice. Jeśli wciąż kojarzysz firmę jako xAI — to ta sama ekipa: została włączona do SpaceX i przemianowana na SpaceXAI 6 lipca 2026 r. API nie podążyło za rebrandingiem, więc każdy identyfikator poniżej nadal ma xai — od zmiennej XAI_API_KEY po hosta api.x.ai.

Tradycyjny stos głosowy łączy trzy usługi: rozpoznawanie mowy, model językowy i syntezę mowy, a każdy skok dodaje opóźnienie i miejsce, w którym można zgubić kontekst. Think Fast 2.0 łączy to w jeden model, który przyjmuje audio lub tekst i zwraca audio lub tekst tym samym połączeniem.

WebSocket mowa–mowa kontra architektura trzech usług. Obraz: autor.

Dla agenta, który działa, a nie tylko mówi, kluczowe jest równoległe działanie rozumowania i mowy. SpaceXAI mówi, że wywołania narzędzi „zwykle” startują, zanim agent skończy pierwsze zdanie — i to „zwykle” ma znaczenie.

Na benchmarkach, które SpaceXAI cytuje z Artificial Analysis, Think Fast 2.0 zdobywa 82,9% w Speech to Speech Index wobec 75,7% dla 1.0 i skraca czas do pierwszego audio z 1,25 s do 0,70 s. Dane dostawcy na ogólnym benchmarku to hipoteza o przepływie twoich rozmów, nie plan testów.

Zobaczysz trzy ciągi nazw modeli: grok-voice-latest, grok-voice-think-fast-2.0 i grok-voice-think-fast-1.0. Alias jest wygodny przy prototypowaniu i za mało stabilny do czegokolwiek więcej.

Gdy testowałem 4 sierpnia 2026 r., grok-voice-latest wciąż wskazywał na grok-voice-think-fast-1.0, a informacje o wydaniach SpaceXAI planowały przełączenie na Think Fast 2.0 na następny dzień. Ta zmiana to tyleż zmiana modelu, co ceny: $0,08 za minutę audio wobec $0,05 dla 1.0, więc nieprzypięty alias drożeje bez zmiany linijki twojego kodu. Przypinaj wersję w każdym wdrożeniu.

Co zbudujemy

Agent ogarnia typowe pytania na infolinii: sprawdzenie zamówienia, znalezienie po e-mailu, gdy dzwoniący nie ma numeru, zmiana instrukcji dostawy, anulowanie, otwarcie lub sprawdzenie zgłoszenia oraz przekazanie rozmowy człowiekowi. Po drodze pojawią się przerwania i zerwane połączenie.

To kilka małych plików zamiast jednego skryptu — każdy element ma inne zadanie i chcesz je testować osobno. Układ wygląda tak:

• config.py ładuje klucz API i przechowuje nazwę modelu, próbkującą częstotliwość i adresy endpointów

• voice_client.py owija WebSocket, śledzi rozliczenia i wystawia helpery send/receive

• tools.py definiuje funkcje zamówień i mały pamięciowy magazyn zamówień zamiast prawdziwej bazy

• assistant.py trzyma prompt systemowy, konfigurację sesji i pętlę zdarzeń spinającą całość

• token_server.py to mały endpoint FastAPI, który wybija efemeryczne tokeny

• app_streamlit.py stawia tego samego klienta za żywą rozmową w przeglądarce — wrócę do tego po sekcji testów

Ścieżka nauki idzie z terminala. Demo dodaje mikrofon.

Wymagania wstępne

Potrzebujesz konta SpaceXAI z kluczem API, zasilonego rozliczania (nie ma stałej darmowej warstwy, a promocyjne kredyty dla nowych kont cię nie poniosą) i wystarczającej swobody z asyncio i WebSocketami, by nadążać bez linijka-po-linijce tłumaczenia await.

Przykłady szybkiego startu SpaceXAI używają surowego pakietu websockets, a nie dedykowanego SDK — my też. Dokumentacja nie podaje wymaganego Pythona. Testowałem na 3.11.

Trzymaj klucz API na serwerze. Jeśli przeglądarka lub aplikacja mobilna rozmawia z Voice API bezpośrednio, dostaje zamiast twojego prawdziwego klucza efemeryczny token — szczegóły w sekcji bezpieczeństwa poniżej.

Konfiguracja projektu

Każdy plik poniżej jest w repozytorium projektu, więc możesz je sklonować zamiast przepisywać fragmenty:

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 niesie połączenie czasu rzeczywistego, a python-dotenv czyta twój klucz. Reszta obsługuje endpoint tokena i demo w przeglądarce. Wstaw klucz do .env:

XAI_API_KEY=xai-your-key-here

To większość konfiguracji. Ciekawa jest sama komunikacja.

Zrozumieć Grok Voice Realtime API

Grok Voice to nazwa produktu. Z czym faktycznie piszesz kod, to endpoint WebSocket pod wss://api.x.ai/v1/realtime, a cała rozmowa to strumień zdarzeń JSON przez to jedno gniazdo.

Cykl życia zdarzeń

Połączenie ma stały schemat: serwer wysyła session.created i conversation.created zaraz po połączeniu, ty wysyłasz session.update by skonfigurować głos i narzędzia, serwer potwierdza session.updated, a dalej tworzysz elementy rozmowy i prosisz o odpowiedzi. Testowałem to na żywym kluczu i kolejność idealnie pokryła się z dokumentacją.

• session.update (klient) konfiguruje głos, instrukcje, narzędzia i format audio

• conversation.item.create (klient) dodaje wiadomość użytkownika, asystenta lub wynik narzędzia

• response.create (klient) prosi model, by mówił; serwerowe VAD wysyła to za ciebie automatycznie

• response.output_audio.delta i response.output_audio_transcript.delta (serwer) strumieniują odpowiedź w trakcie generowania

• response.done (serwer) domyka turę

Dwie rzeczy potrafią wywrócić plany. Strona dokumentacji Speech to Speech, którą linkowałem, wspomina o zdarzeniu conversation.item.created przy wznawianiu sesji, ale kanoniczne odniesienie zdarzeń wymienia tylko conversation.item.added — i to właśnie przychodziło we wszystkich moich testach, więc pod to pisz kod. Zobaczysz też niedokumentowane ping kilka sekund po większości połączeń — wspominam, byś nie brał tego za błąd.

Formaty audio i transport

Kodek i transport to osobne wybory. Kodek, ustawiany pod audio.input.format i audio.output.format, to audio/pcm (Linear16, domyślnie 24000 Hz), audio/pcmu lub audio/pcma (G.711 przy 8 kHz, do telefonii), albo audio/opus (24 kHz). Transport to sposób przesyłu bajtów po drucie:

• json (domyślnie) wysyła audio jako tekst base64 wewnątrz input_audio_buffer.append i response.output_audio.delta, łatwe do logowania i debugowania

• binary wysyła surowe bajty kodeka jako ramki binarne WebSocket, omijając narzut base64 kosztem pętli odbioru, która musi rozgałęziać się po typie wiadomości

Zacznij od JSON. Każdy przykład w dokumentacji go używa, inspekcja jest banalna, a narzut base64 nie jest wąskim gardłem w agencie wsparcia. Przechodź na binary tylko, jeśli masz na to mierzalny powód.

Zgodność z OpenAI Realtime API

Pomiń, jeśli nigdy nie dotykałeś OpenAI Realtime API. Dla reszty: Speech to Speech API naśladuje OpenAI Realtime API na tyle blisko, że większość kodu klienta przenosisz, zmieniając bazowy URL i klucz — ale to nie jest idealny drop-in.

Transkrypty przychodzą tu jako conversation.item.input_audio_transcription.updated zamiast openaiowego delta, kilka zdarzeń OpenAI nie jest wspieranych, a SpaceXAI dodaje swoje rozszerzenia: force_message do obowiązkowej kwestii ujawnienia, resumption do wznowień i replace do poprawy źle wymawianych nazw marek przed TTS.

Budowa agenta głosowego czasu rzeczywistego

Dość o protokole. Oto klient, który z nim rozmawia.

Łączenie i konfiguracja sesji

Połączenie otwierasz tokenem bearer i parametrem zapytania z modelem, a pierwsza twoja wiadomość konfiguruje wszystko w zachowaniu agenta:

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 to prompt systemowy, a ten model lubi krótkie. Notatki migracyjne SpaceXAI mówią, by upraszczać prompty pisane dla starszych, „gpt-owych” modeli głosowych, zamiast przenosić je 1:1. Mój każe agentowi trzymać krótkie odpowiedzi, zadawać jedno pytanie naraz i odczytać zapis przed działaniem. Mówione potwierdzenie to miły UX, nie kontrola bezpieczeństwa. Twoja aplikacja i tak egzekwuje autoryzację samej operacji zapisu.

Jedno mnie zaskoczyło: nierozpoznana nazwa modelu nie zgłasza błędu przy łączeniu — po cichu spada do grok-voice-think-fast-1.0. Degradacja płatnego żądania z powodu literówki, bez słowa komentarza, to dziwne ustawienie domyślne. Zaloguj pole session.model z session.created raz na starcie i upewnij się, że dostałeś to, o co prosiłeś.

Wyjście terminala pokazujące session.created po połączeniu. Obraz: autor.

Strumieniowanie audio użytkownika

Przy turn_detection.type ustawionym na server_vad musisz tylko dopinać kolejne fragmenty audio. Serwer decyduje, kiedy dzwoniący przestał mówić, i wyzwala odpowiedź za ciebie. Ustaw na null, a bierzesz tę decyzję na siebie — finalizujesz bufor, kiedy uznasz, że tura się skończyła.

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 ma trzy pokrętła i zły dobór to najczęstszy powód, dla którego agent głosowy „czuje się” zepsuty, mimo że w logach brak błędów. Domyślnie nie pojawiają się one w echu session.updated, więc sprawdzaj w dokumentacji, a nie zakładaj.

• threshold (0.1–0.9, domyślnie 0.85): jak głośno musi być, by uznać mowę; podnieś w hałasie, obniż, jeśli cichych mówców gubi

• silence_duration_ms: jak długo cisza kończy turę; zbyt krótko — ucinanie w pół myśli, zbyt długo — ociężałość

• prefix_padding_ms (domyślnie 333): skrawek audio tuż sprzed wykrycia mowy, by nie przyciąć pierwszej sylaby

Najpierw stroń silence_duration_ms — jeśli dzwoniący są ucinani przy pauzach. To moje pierwsze pokrętło, zanim dotknę pozostałych.

Odbieranie i odtwarzanie odpowiedzi

Audio przychodzi małymi kawałkami jako response.output_audio.delta, a sens strumieniowania polega na odtwarzaniu każdego kawałka od razu po jego nadejściu, zamiast czekać na 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)

Trzymaj transkrypt nawet na produkcji. To najtańsze narzędzie debugowania, gdy dzwoniący mówi, że agent „powiedział coś dziwnego”.

Dodawanie narzędzi do agenta głosowego

Agent głosowy, który tylko mówi, to chatbot z mikrofonem.

Tworzenie narzędzi zamówień

Każde narzędzie to schemat JSON plus zwykła funkcja Pythona po naszej stronie. Model nigdy nie dotyka bazy — widzi tylko to, co zwróci nasza funkcja.

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
]

Operacje odczytu jak check_order_status są bezpieczne do ponowień przy timeoutach. Zapisy — nie: ponowienie update_delivery_instructions po niejednoznacznym timeoutcie może nałożyć tę samą zmianę dwa razy. Linijka potwierdzenia w promcie tego nie powstrzyma — daj zapisom klucz idempotencji albo sprawdzanie duplikatu.

Odmowy też umieść w funkcji. cancel_order zwraca powód i alternatywę zamiast anulować wysłane zamówienie, bo prompt „nigdy nie anuluj wysłanych zamówień” to sugestia, a funkcja, która odmawia — nie.

Obsługa pętli wywołań narzędzi

Cztery kroki i kolejność ważniejsza niż wygląda. Model wysyła response.function_call_arguments.done, twój kod uruchamia funkcję, odsyłasz wynik jako element function_call_output i dopiero wtedy prosisz model o kontynuację.

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

Jeśli model potrzebuje więcej niż jednego narzędzia do żądania, odpala wiele zdarzeń function_call_arguments.done przed zagraniem jakiegokolwiek audio. Rozwiąż wszystkie i wyślij każdy wynik, zanim padnie pojedyncze response.create. Jeśli wyślesz za wcześnie, model odpowie bez kontekstu z wywołań w locie.

Jest tu pułapka, którą SpaceXAI dokumentuje, a i tak w nią wpadłem: wysłanie response.create natychmiast po twoim wyniku narzędzia może nałożyć się na wstępne zdanie, które agent wciąż odtwarza. U mnie raz zaczął „Już sprawdzam status zamówienia ORD-1042” i wywołał narzędzie w połowie zdania — więc natychmiastowa odpowiedź zagadałaby własny wstęp.

Poczekaj, aż audio bieżącej tury się skończy, i pokaż krótkie „myślenie” międzyczasem.

Przepływ wywołania narzędzia przed kontynuacją odpowiedzi. Obraz: autor.

Zarządzanie przerwaniami i stanem rozmowy

Dwa oddzielne problemy: dzwoniący mówi agentowi w słowo w trakcie odpowiedzi oraz zerwany WebSocket, który trzeba podnieść.

Naturalne przerwania

Z server_vad barge-in dzieje się po stronie serwera: gdy tylko wykryje, że dzwoniący znów mówi, sygnalizuje input_audio_buffer.speech_started i przestaje generować starą odpowiedź. Twoja robota to część kliencka — wyczyścić kolejkę odtwarzania, by agent zamilkł zamiast kończyć zdanie, którego nikt nie chce słyszeć.

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

Przy sesjach bez VAD, ręcznych, response.cancel robi to samo na żądanie. Jest też conversation.item.truncate do przycięcia elementu asystenta do tego, co faktycznie wybrzmiało. Dokumentacja potwierdza istnienie, ale nie mówi, kiedy go odpalać przy żywym barge-in — sprawdź timing sam.

Testowałem to na zmianie instrukcji dostawy w połowie odpowiedzi: zaczynasz prośbę, przerywasz innym adresem w trakcie potwierdzenia agenta. Liczy się, czy agent zastosował poprawioną instrukcję zamiast po cichu dokończyć starą — nie to, czy audio się zatrzymało. Sprawdzaj rekord zamówienia, nie ciszę. Przeglądarkowe demo na końcu pozwala to usłyszeć.

Wznawianie rozłączonej sesji

Wznowienie sesji to opt-in i to nie jest pamięć. Ustaw resumption.enabled: true w session.update, zgarnij ID ze zdarzenia conversation.created, a gdy gniazdo padnie, połącz się ponownie z ?conversation_id=<id> w URL i znów się na to zapisz na nowym połączeniu.

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

Zbuforowane tury, transkrypty, wywołania narzędzi i ich wyniki odtwarzają się przed twoim następnym pytaniem, a cache znika po 30 minutach bezczynności. Sprawdziłem: zapytałem o zamówienie, zerwałem połączenie, połączyłem się ponownie z pytaniem uzupełniającym bez powtarzania — agent poprawnie podjął temat ETA.

Jedna niedokumentowana rzecz: replay nie ląduje natychmiast, więc pytanie wysłane w chwili otwarcia gniazda może je wyprzedzić i wrócić bez pamięci wcześniejszej tury. Daj temu sekundę, zanim obwinisz resumption.

Log terminala wzniesionej sesji. Obraz: autor.

Nie używaj tego zamiast zapisywania stanu zamówienia we własnej bazie. Jeśli cache wygaśnie albo dzwoniący zadzwoni jutro, startujesz bez kontekstu — tak ma być.

Zabezpieczanie i monitorowanie agenta

Nigdy nie wkładaj stałego klucza API do kodu przeglądarki ani mobilnego. Jeśli klient łączy się bezpośrednio zamiast przez twój serwer, wybij krótkotrwały 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": ...}

Przeglądarka nie może ustawić niestandardowego nagłówka Authorization w uzgadnianiu WebSocket, więc przekazuje token przez nagłówek sec-websocket-protocol, z prefiksem xai-client-secret..

Serwer wybija token, przeglądarka dołącza do rozmowy. Obraz: autor.

Rozliczenia lecą po dwóch licznikach. Audio — wysłane lub odebrane — idzie po $0,08 za minutę, jak wspomniałem, czyli $4,80 za godzinę, a każde conversation.item.create, które nie jest audio i nie jest function_call_output, kosztuje płaskie $0,004. response.create nie jest w ogóle fakturowane. Każde response.done niesie obiekt usage, który u mnie raportował output_audio_seconds obok billable_audio_seconds. Fakturowanie rób z nich, nie ze szacunków.

Udokumentowane limity Speech to Speech API to 10 równoległych sesji na zespół i 120-minutowy limit sesji, oba w us-east-1. Nie planuj mocy według liczb Voice Agent API — są inne.

Jeśli chodzi o prywatność, bądź precyzyjny. FAQ bezpieczeństwa SpaceXAI mówi, że żądania i odpowiedzi API są przechowywane szyfrowane przez 30 dni do monitorowania nadużyć i nie są używane do trenowania bez zgody, a zespoły mogą włączyć Zero Data Retention — choć ZDR usuwa trwałą historię rozmów agenta głosowego, więc nie działa z wznawianiem.

Jeśli ujawniasz, że rozmowa jest nagrywana lub obsługiwana przez AI, od tego jest rozszerzenie force_message. Kwestia leci dokładnie tak, jak napisana, zamiast w parafrazie modelu.

Testowanie agenta głosowego

Status 200 na uzgadnianiu WebSocket nie mówi nic, czy agent zrobił dobrze. Testuj wynik, nie samo połączenie.

• Czyste wyszukanie zamówienia — porównanie wypowiedzianej odpowiedzi z rekordem, nie tylko, że odpowiedź nadeszła
• Przerwana odpowiedź — potwierdzenie, że odtwarzanie się zatrzymuje i agent adresuje nowe żądanie
• Aktualizacja dostawy wymagająca potwierdzenia — sprawdzona na rekordzie zamówienia
• Odmowa, np. anulowanie wysłanego zamówienia — agent ma wyjaśnić zasadę, a nie przepraszać
• Nieznany numer zamówienia — upewnij się, że agent to mówi, zamiast wymyślać status
• Narzędzie zwracające błąd — sprawdź, że agent go wypowiada, zamiast się zawiesić
• Ponowne połączenie i wznowienie — włącznie z oknem replay, o którym pisałem
• Hałaśliwe audio, szybka mowa i dzwoniący, który literuje numery i adresy

Większość z tego odpaliłem na żywym kluczu przy pisaniu. Ciekawe porażki były behawioralne, nie błędy: timing wznowienia powyżej i przyjęty zamiast odrzuconego próg VAD poza zakresem — rzeczy, które wyjdą po cichu zepsute, jeśli testujesz tylko happy path. Dodaj też test wielojęzyczny i zajrzyj do FAQ po niuans nazewnictwa języka.

Dwóch z nich nie przetestujesz pisaniem. app_streamlit.py to strona Streamlit, która stawia żywą rozmowę w przeglądarce: mikrofon strumieniuje do tego samego WebSocketa przez WebRTC, głos agenta wraca strumieniowo, a gniazdo pozostaje otwarte przez całość.

streamlit run app_streamlit.py