Samouczek API Grok Voice Think Fast 2.0: zbuduj agenta głosowego w Pythonie w czasie rzeczywistym

Grok Voice Think Fast 2.0 od SpaceXAI to model mowa‑do‑mowy. Wysyłasz do niego audio przez WebSocket, a on odsyła audio z powrotem, i po drodze potrafi rozumować oraz mówić dalej, gdy funkcja, którą zdecydował się wywołać, już się wykonuje. Bez osobnego kroku STT i bez osobnego TTS.

SpaceXAI ogłosiło Think Fast 2.0 29 lipca 2026: 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. Skrócę część o benchmarkach — w samouczku liczy się to, co zmienia się w twoim kodzie.

Zbudujemy głosowego agenta obsługi klienta dla sklepu online. Dzwoniący może zapytać o zamówienie, znaleźć je po e‑mailu, gdy nie ma numeru, zmienić instrukcje dostawy, anulować, przerwać agentowi w pół zdania i wznowić rozmowę po zerwaniu połączenia. To ścieżka przez API, nie no‑code Voice Agent Builder, który omawia nasz samouczek Grok Voice Agent Builder. Od tego zacznij wersję konsolową.

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, który większość osób nazywa po prostu Grok Voice. Jeśli nadal kojarzysz firmę jako xAI, to ta sama ekipa: 6 lipca 2026 została włączona do SpaceX i przemianowana na SpaceXAI. API nie poszło za rebrandingiem, więc poniżej każdy identyfikator nadal mówi xai — od zmiennej XAI_API_KEY po hosta api.x.ai.

Tradycyjny stos głosowy łączy trzy usługi: STT, model językowy i TTS — każda „krok” to opóźnienie i miejsce na utratę kontekstu. Think Fast 2.0 składa to w jeden model, który przyjmuje audio lub tekst i zwraca audio lub tekst po tym samym połączeniu.

WebSocket mowa‑do‑mowy kontra architektura trzech usług. Obraz: autor.

Dla agenta, który działa, a nie tylko mówi, liczy się to, że rozumowanie i mowa biegną równolegle. SpaceXAI mówi, że wywołania narzędzi „zwykle” startują, zanim agent skończy pierwsze zdanie — i to „zwykle” jest tu istotne.

Na benchmarkach SpaceXAI cytowanych z Artificial Analysis Think Fast 2.0 osiąga 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 producenta na ogólnym benchmarku to hipoteza o twoim przepływie rozmów, a nie plan testów.

Spotkasz trzy nazwy modeli: grok-voice-latest, grok-voice-think-fast-2.0 i grok-voice-think-fast-1.0. Alias jest wygodny w prototypowaniu i za mało stabilny do czegokolwiek więcej.

Gdy testowałem 4 sierpnia 2026, grok-voice-latest nadal wskazywał na grok-voice-think-fast-1.0, a release notes SpaceXAI zapowiadały przejście na Think Fast 2.0 na następny dzień. Ta zmiana to tyleż model, co cena — $0,08/min audio wobec $0,05 dla 1.0 — więc nieprzypięty alias robi się droższy bez zmiany linijki kodu. Przypinaj wersję w każdym wdrożeniu.

Co zbudujemy

Agent ogarnia to, o co pyta linia wsparcia: sprawdzenie zamówienia, znalezienie po e‑mailu, gdy brak numeru, zmiana instrukcji dostawy, anulowanie, otwarcie/sprawdzenie zgłoszenia oraz przekazanie rozmowy do człowieka. Po drodze pojawiają się przerwania i zerwane połączenie.

To kilka małych plików zamiast jednego skryptu, bo każda część ma inne zadanie i będziesz chcieć je testować osobno. Struktura:

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

• voice_client.py opakowuje WebSocket, śledzi rozliczenia i wystawia helpery do wysyłania/odbierania

• tools.py definiuje funkcje zamówień i mały pamięciowy store zamówień w zastępstwie bazy

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

• token_server.py to mały endpoint FastAPI do wystawiania efemerycznych tokenów

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

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

Wymagania wstępne

Potrzebujesz konta SpaceXAI z kluczem API, opłaconego rozliczania (nie ma stałego darmowego progu, a promocyjne kredyty nowego konta nie wystarczą) oraz komfortu z asyncio i WebSocketami, by nadążyć bez linijka‑po‑linijce objaśniania await.

Przykłady quick‑start SpaceXAI używają surowego pakietu websockets zamiast dedykowanego SDK — my też. Dokumentacja nie podaje wymaganej wersji 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 efemeryczny token zamiast prawdziwego klucza — omówione w sekcji bezpieczeństwa poniżej.

Konfiguracja projektu

Każdy plik poniżej jest w repozytorium projektu, więc możesz je sklonować zamiast kopiować 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 realtime, a python-dotenv czyta twój klucz. Reszta obejmuje endpoint tokena i demo przeglądarkowe. Umieść klucz w .env:

XAI_API_KEY=xai-your-key-here

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

Jak działa Grok Voice Realtime API

Grok Voice to nazwa produktu. Tym, z czym faktycznie piszesz kod, jest endpoint WebSocket pod wss://api.x.ai/v1/realtime, a cała rozmowa to strumień zdarzeń JSON po jednym gnieździe.

Cykl życia zdarzeń

Połączenie ma stały kształt: 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 i stamtąd dodajesz elementy rozmowy i prosisz o odpowiedzi. Przetestowałem to na żywym kluczu i kolejność dokładnie zgadzał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 o wypowiedź; serwerowy VAD wysyła to za ciebie automatycznie

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

• response.done (serwer) zamyka turę

Dwie rzeczy potrafią zaskoczyć. Strona dokumentacji Speech to Speech, do której linkowałem, wspomina o zdarzeniu conversation.item.created przy wznawianiu sesji, ale kanoniczny spis zdarzeń wymienia tylko conversation.item.added i to właśnie przychodziło we wszystkich moich testach — pod to pisz kod. Zobaczysz też nieudokumentowane ping kilka sekund po starcie większości połączeń — wspominam tylko, byś nie czytał tego jako błąd.

Formaty audio i transport

Kodek i transport to osobne wybory. Kodek, ustawiany w 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, dla telefonii), albo audio/opus (24 kHz). Transport to sposób przenoszenia 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 binarne ramki 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. Przejdź na binary, jeśli pomiar wskaże powód.

Zgodność z OpenAI Realtime API

Pomiń, jeśli nigdy nie dotykałeś OpenAI Realtime API. Dla reszty: Speech to Speech API śledzi OpenAI Realtime API na tyle blisko, że większość kodu klienta przeniesiesz po zmianie bazowego URL i klucza, ale to nie jest idealny drop‑in.

Transkrypty przychodzą jako conversation.item.input_audio_transcription.updated zamiast openaiowego delta, kilka zdarzeń OpenAI nie jest wspieranych, a SpaceXAI dodaje własne rozszerzenia: force_message do podania obowiązkowej formułki, resumption do wznawiania i replace do poprawiania źle wymawianych nazw marek przed TTS.

Budowa agenta głosowego w czasie rzeczywistym

Dość protokołu. Oto klient, który z nim rozmawia.

Łączenie i konfigurowanie sesji

Połączenie otwierasz tokenem bearer i parametrem modelu w zapytaniu, a pierwsza wiadomość konfiguruje zachowanie 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 pod starsze, „gpt‑owe” modele głosowe, zamiast przenosić je słowo w słowo. Mój każe agentowi mówić krótko, zadawać po jednym pytaniu i odczytać zmianę przed działaniem. Słowna prośba o potwierdzenie to miły element UX, nie kontrola bezpieczeństwa. Twoja aplikacja i tak egzekwuje autoryzację na samym zapisie.

Zaskoczyło mnie jedno: nieznana nazwa modelu nie zgłasza błędu przy łączeniu — po cichu spada na grok-voice-think-fast-1.0. Degradacja płatnego żądania przez literówkę, bez informacji, to dziwny domyślny tryb. Zaloguj pole session.model z session.created przy starcie i sprawdź, czy dostałeś to, o co prosiłeś.

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

Streamowanie audio użytkownika

Przy turn_detection.type ustawionym na server_vad wystarczy, że dokładasz audio. Serwer decyduje, kiedy rozmówca przestał mówić, i sam wyzwala odpowiedź. Ustaw na null, a decyzja należy do ciebie — jawnie zatwierdzasz bufor, gdy 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(),
    }))

Serwerowy VAD ma trzy pokrętła i ich złe ustawienie to najczęstszy powód, dla którego agent wydaje się zepsuty, mimo braku błędów w logach. Domyślnie nie pojawiają się w echo session.updated, więc sprawdzaj z dokumentacją, nie zakładaj.

• threshold (0,1–0,9, domyślnie 0,85) — jak głośno musi być, by zaliczyć do mowy; podnieś w hałasie, obniż, jeśli cisi mówiący są pomijani

• silence_duration_ms — jak długą ciszę serwer uznaje za koniec tury; zbyt krótko — ucina w pół myśli, zbyt długo — ociężale

• prefix_padding_ms (domyślnie 333) — kawałek audio sprzed wykrycia mowy, by nie uciąć pierwszej sylaby

Zacznij od silence_duration_ms jeśli rozmówcy są urywani, gdy robią pauzy. To pierwsze, po co sięgam.

Odbieranie i odtwarzanie odpowiedzi

Audio przychodzi małymi porcjami jako response.output_audio.delta, a sens streamingu jest taki, że odtwarzasz porcje od razu po ich 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 rozmówca mówi, że agent „powiedział coś dziwnego”.

Dodawanie narzędzi do agenta głosowego

Agent, 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 w Pythonie 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, można bezpiecznie ponawiać po timeoutach. Zapisu — nie: ponowienie update_delivery_instructions po niejednoznacznym timeoutcie może zastosować zmianę dwa razy. Linijka potwierdzenia w promcie tego nie zatrzyma — daj zapisom klucz idempotencji lub kontrolę duplikatów.

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

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

Cztery kroki i kolejność ważniejsza, niż się wydaje. 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, wyśle wiele function_call_arguments.done zanim popłynie audio. Rozwiąż wszystkie i wyślij każdy wynik przed pojedynczym response.create. Wyślij zbyt wcześnie — model odpowie bez kontekstu z trwających wywołań.

Jest tu pułapka, którą SpaceXAI dokumentuje, a ja i tak w nią wpadłem: wysłanie response.create w chwili wysyłki wyniku narzędzia może nałożyć się na zdanie wstępne, które agent jeszcze odtwarza. W jednej próbie zaczął „Zaraz sprawdzę 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ż obecne audio się skończy, i pokaż krótkie „myślenie” pomiędzy.

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

Zarządzanie przerwaniami i stanem rozmowy

Dwa osobne problemy: rozmówca wchodzi w słowo agentowi oraz zrywa się WebSocket i trzeba go podnieść.

Wspieranie naturalnych przerwań

Z server_vad barge‑in dzieje się serwerowo: w chwili wykrycia, że rozmówca znów mówi, sygnalizuje input_audio_buffer.speech_started i przestaje generować starą odpowiedź. Twoja rola to część kliencka: czyścisz zakolejkowane audio, by agent ucichł, zamiast dokańczać zdanie, którego nikt nie chce słyszeć.

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

Dla sesji manualnych, bez VAD, response.cancel robi to samo na żądanie. Jest też conversation.item.truncate do przycięcia elementu asystenta do tego, co faktycznie zostało usłyszane. Dokumentacja potwierdza istnienie, ale nie mówi, kiedy go wysyłać podczas żywego barge‑in — przetestuj timing samodzielnie.

Testowałem to na zmianie instrukcji dostawy w połowie odpowiedzi: rozpocznij prośbę, przerwij innym adresem w trakcie potwierdzania. Liczy się to, czy agent zastosuje poprawkę, zamiast po cichu dokończyć starą — nie to, czy audio ustało. Asercje rób na rekordzie zamówienia, nie na ciszy. Demo w przeglądarce na końcu pozwala to usłyszeć.

Wznawianie rozłączonej sesji

Wznawianie jest opt‑in i to nie jest pamięć. Ustaw resumption.enabled: true w session.update, weź ID z zdarzenia conversation.created, a jeśli gniazdo padnie, połącz ponownie z ?conversation_id=<id> w URL i znów włącz 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 są odtwarzane przed twoim następnym pytaniem, a cache znika po 30 minutach bezczynności. Sprawdziłem: zapytałem o zamówienie, zerwałem połączenie i wznowiłem pytaniem uzupełniającym bez powtarzania — agent poprawnie podchwycił ETA.

Nieudokumentowany haczyk: 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 wznowionej sesji. Obraz: autor.

Nie używaj tego zamiast zapisywania stanu zamówienia w swojej bazie. Jeśli cache wygaśnie lub rozmówca oddzwoni jutro, zaczynasz od zera — i tak ma być.

Zabezpieczanie i monitorowanie agenta

Nigdy nie wkładaj stałego klucza API do kodu przeglądarki czy mobilnego. Jeśli klient łączy się bezpośrednio zamiast przez twój serwer, wystaw krótko żyjący 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ć własnego nagłówka Authorization w handshake WebSocket, więc przekazuje token przez nagłówek sec-websocket-protocol, z prefiksem xai-client-secret..

Serwer wystawia token, przeglądarka dołącza do połączenia. Obraz: autor.

Rozliczanie biegnie po dwóch licznikach. Audio — wysyłane lub odbierane — to wspomniane $0,08 za minutę, czyli $4,80 za godzinę, a każde conversation.item.create, które nie jest audio i nie jest function_call_output, kosztuje płasko $0,004. response.create nie jest w ogóle rozliczane. Każde response.done niesie obiekt usage, który w moim teście raportował output_audio_seconds obok osobnych billable_audio_seconds. Rozliczaj z tego, nie ze szacunków.

Udokumentowane limity Speech to Speech API to 10 jednoczesnych sesji na zespół i limit 120 minut na sesję, oba w us-east-1. Nie planuj pojemności na podstawie liczb Voice Agent API — są inne.

W kwestii prywatności bądź precyzyjny. FAQ bezpieczeństwa SpaceXAI mówi, że żądania i odpowiedzi API są przechowywane zaszyfrowane przez 30 dni do monitoringu nadużyć i nie są używane do trenowania bez zgody; zespoły mogą włączyć Zero Data Retention, ale ZDR kasuje historię rozmów głosowych i nie działa ze wznawianiem.

Jeśli ujawniasz nagrywanie rozmowy lub udział AI, od tego jest rozszerzenie force_message wspomniane wcześniej. Linijka odtworzy się dokładnie tak, jak napisana, zamiast parafrazy modelu.

Testowanie agenta głosowego

Status 200 na handshake WebSocket nie mówi nic o tym, czy agent zrobił właściwą rzecz. Testuj rezultat, nie tylko połączenie.

• Czyste wyszukanie zamówienia — porównuj odpowiedź mówioną z rekordem, nie tylko to, że przyszła
• Przerwana odpowiedź — potwierdź, że odtwarzanie się zatrzymuje, a agent podejmuje nowe żądanie
• Aktualizacja dostawy wymagająca potwierdzenia — sprawdź na rekordzie zamówienia
• Odmowa — np. anulowanie wysłanego zamówienia — agent ma wyjaśnić zasadę, nie przepraszać
• Nieznany numer zamówienia — upewnij się, że agent to mówi, a nie wymyśla statusu
• Narzędzie zwraca błąd — sprawdź, że agent go komunikuje, zamiast zawiesić się
• Ponowne połączenie i wznowienie — łącznie z oknem odtwarzania, o którym wspominałem
• Hałaśliwe audio, szybka mowa i rozmówca literujący liczby/adresy

Większość z tego przepuściłem na żywym kluczu, pisząc. Ciekawe porażki były behawioralne, nie błędowe: timing wznowienia powyżej i poza zakresem threshold VAD przyjęty zamiast odrzuconego — rzeczy, które cicho wypłyną, jeśli testujesz tylko happy path. Dodaj też test wielojęzyczny — w FAQ jest niuans nazewnictwa języka.

Dwóch z nich nie przetestujesz pisząc. app_streamlit.py to strona Streamlit, która stawia żywą rozmowę w przeglądarce: mikrofon streamuje do tego samego WebSocket przez WebRTC, głos agenta wraca strumieniem, a gniazdo pozostaje otwarte przez cały czas.

streamlit run app_streamlit.py