Tutorial API Grok Voice Think Fast 2.0: Construiește un agent vocal în timp real în Python

Grok Voice Think Fast 2.0 de la SpaceXAI este un model speech-to-speech. Îi trimiți audio printr-un WebSocket și îți trimite audio înapoi, iar între timp poate raționa și continua să vorbească în timp ce o apelare de funcție pe care a decis s-o facă rulează deja. Fără pas separat de speech-to-text, fără pas separat de text-to-speech.

SpaceXAI a anunțat Think Fast 2.0 pe 29 iulie 2026: primul audio mai rapid, comportament full-duplex mai stabil (ascultă în timp ce vorbește, în loc să ia pe rând strict), și apelări de unelte care pornesc devreme în tură. O să păstrez partea de benchmark scurtă, fiindcă, pentru un tutorial, contează ce se schimbă în codul tău.

Vom construi un agent vocal de suport clienți pentru un magazin online. Un apelant poate întreba despre o comandă, poate schimba o instrucțiune de livrare, poate anula, poate întrerupe agentul la jumătatea propoziției și poate relua conversația după o conexiune întreruptă. Aceasta este calea prin API, nu constructorul no-code Voice Agent Builder prezentat în tutorialul nostru Grok Voice Agent Builder. Începe de acolo pentru varianta axată pe consolă.

Ce este Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 este cel mai nou model SpaceXAI pentru Speech to Speech API, numele de produs din spatele a ceea ce majoritatea numesc Grok Voice. Dacă încă te gândești la companie ca la xAI, aceeași echipă: a fost integrată în SpaceX și rebranduită SpaceXAI pe 6 iulie 2026. API-ul nu a urmat rebrandul, așa că fiecare identificator de mai jos tot spune xai, de la variabila XAI_API_KEY la host-ul api.x.ai.

Un stack vocal tradițional leagă trei servicii: speech-to-text, un model de limbaj, apoi text-to-speech, și fiecare salt adaugă latență și un loc unde contextul se poate pierde. Think Fast 2.0 comprimă asta într-un singur model care primește audio sau text și produce audio sau text pe aceeași conexiune.

WebSocket speech-to-speech versus arhitectură pipeline cu trei servicii. Imagine de autor.

Pentru un agent care acționează, nu doar vorbește, contează că raționarea și vorbirea rulează în paralel. SpaceXAI spune că apelările de unelte „de obicei” încep să se execute înainte ca agentul să își termine prima propoziție, și acel „de obicei” chiar contează.

Pe benchmark-urile citate de SpaceXAI de la Artificial Analysis, Think Fast 2.0 obține 82,9% pe Speech to Speech Index față de 75,7% pentru 1.0 și reduce timpul până la primul audio de la 1,25 secunde la 0,70 secunde. Cifrele furnizorului pe un benchmark general sunt o ipoteză despre fluxul tău de apelanți, nu un plan de testare.

Sunt trei șiruri de model pe care le vei vedea: grok-voice-latest, grok-voice-think-fast-2.0 și grok-voice-think-fast-1.0. Aliasul e util la prototipare și nu e suficient de stabil pentru altceva.

Când am testat pe 4 august 2026, grok-voice-latest încă rezolva la grok-voice-think-fast-1.0, iar notele de lansare ale SpaceXAI programau trecerea la Think Fast 2.0 pentru ziua următoare. Acea schimbare e tot atât despre preț cât e despre model, 0,08 $ pe minut de audio față de 0,05 $ pentru 1.0, deci un alias nefixat devine mai scump fără să schimbi o linie de cod. Fixează șirul versiuni în orice implementezi.

Ce vom construi

Agentul acoperă ce se cere pe o linie de suport: caută o comandă, găsește una dintr-un email când apelantul nu are număr, schimbă o instrucțiune de livrare, anulează, deschide sau verifică un tichet și transferă apelul către o persoană. Pe parcurs apar întreruperi și o conexiune căzută.

E un pumn de fișiere mici, nu un singur script, pentru că fiecare piesă are un rol diferit și vei vrea să le testezi separat. Iată structura:

• config.py încarcă cheia API și păstrează șirul modelului, rata de eșantionare și URL-urile endpoint-urilor

• voice_client.py împachetează WebSocket-ul, urmărește costurile și expune helperi de trimitere/recepție

• tools.py definește funcțiile pentru comenzi și un mic magazin de comenzi în memorie, în locul unei baze reale

• assistant.py ține promptul de sistem, configurația sesiunii și bucla de evenimente care leagă totul

• token_server.py este un endpoint FastAPI mic care emite tokenuri efemere

• app_streamlit.py pune același client în spatele unui apel live în browser, la care voi reveni după secțiunea de testare

Parcursul didactic rulează dintr-un terminal. Demo-ul adaugă microfonul.

Cerințe preliminare

Ai nevoie de un cont SpaceXAI cu o cheie API, un aranjament de facturare finanțat (nu există un nivel gratuit permanent, iar creditele promoționale pentru conturi noi nu te vor duce departe) și suficientă familiaritate cu asyncio și WebSocket-uri ca să urmărești fără o explicație linie cu linie a lui await.

Exemplele de quick-start ale SpaceXAI folosesc pachetul brut websockets în locul unui SDK dedicat, și la fel facem și noi. Documentația nu specifică o versiune de Python necesară. Eu am testat pe 3.11.

Păstrează cheia API pe server. Dacă o aplicație de browser sau mobil vorbește direct cu Voice API, primește un token efemer în locul cheii tale reale, acoperit în secțiunea de securitate de mai jos.

Configurarea proiectului

Fiecare fișier de mai jos trăiește în repo-ul proiectului, deci îl poți clona în loc să copiezi fragmente:

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 poartă conexiunea realtime și python-dotenv îți citește cheia. Restul acoperă endpoint-ul de token și demo-ul de browser. Pune cheia în .env:

XAI_API_KEY=xai-your-key-here

Asta e cea mai mare parte a configurării. Conexiunea e partea interesantă.

Înțelegerea Grok Voice Realtime API

Grok Voice e numele produsului. Lucrul la care scrii efectiv cod este un endpoint WebSocket la wss://api.x.ai/v1/realtime, iar toată conversația are loc ca un flux de evenimente JSON peste acel unic socket.

Ciclul de viață al evenimentelor

O conexiune urmează o formă fixă: serverul trimite session.created și conversation.created imediat ce te conectezi, tu trimiți session.update pentru a configura vocea și uneltele, serverul confirmă cu session.updated, iar de acolo creezi elemente de conversație și soliciți răspunsuri. Am rulat asta cu o cheie live și ordinea s-a potrivit exact cu documentația.

• session.update (client) configurează vocea, instrucțiunile, uneltele și formatul audio

• conversation.item.create (client) adaugă un mesaj de utilizator, un mesaj al asistentului sau un rezultat de unealtă

• response.create (client) cere modelului să vorbească; VAD-ul serverului trimite asta pentru tine automat

• response.output_audio.delta și response.output_audio_transcript.delta (server) transmit răspunsul pe măsură ce se generează

• response.done (server) închide tura

Două lucruri îi încurcă pe oameni. Pagina de documentație Speech to Speech la care am făcut link mai sus menționează un eveniment conversation.item.created în timpul reluării sesiunii, dar referința canonică a evenimentelor listează doar conversation.item.added, și asta a sosit în fiecare test pe care l-am rulat, deci codează pentru el. Vei vedea și un ping nedocumentat la câteva secunde în majoritatea conexiunilor, menționat doar ca să nu-l citești ca pe o eroare.

Formate audio și transport

Codec și transport sunt alegeri separate. Codec-ul, setat sub audio.input.format și audio.output.format, este audio/pcm (Linear16, implicit 24000 Hz), audio/pcmu sau audio/pcma (G.711 la 8 kHz, pentru telefonie), sau audio/opus (24 kHz). Transportul este modul în care acei bytes călătoresc pe fir:

• json (implicit) trimite audio ca text base64 în input_audio_buffer.append și response.output_audio.delta, ușor de logat și depanat

• binary trimite bytes raw ai codec-ului ca frame-uri binare WebSocket, sărind peste overhead-ul base64 cu costul unei bucle de recepție care trebuie să ramifice pe tipul mesajului

Începe cu JSON. Fiecare exemplu din documentație îl folosește, e trivial de inspectat, iar overhead-ul base64 nu îți limitează un build de agent de suport. Treci la binary dacă măsori un motiv.

Compatibilitate cu OpenAI Realtime API

Sari peste dacă n-ai atins niciodată OpenAI Realtime API. Pentru toți ceilalți, Speech to Speech API urmărește OpenAI Realtime API îndeaproape, suficient cât majoritatea codului de client se portează prin schimbarea bazei URL și a cheii, dar nu e un substitut perfect drop-in.

Transcrierile sosesc ca conversation.item.input_audio_transcription.updated aici în loc de delta la OpenAI, câteva evenimente OpenAI nu sunt suportate, iar SpaceXAI adaugă extensii proprii: force_message pentru o linie de dezvăluire scriptată, resumption pentru reconectări, și replace pentru corectarea pronunțiilor de brand înainte de text-to-speech.

Construirea agentului vocal în timp real

Destul protocol. Iată clientul care vorbește cu el.

Conectarea și configurarea sesiunii

Conexiunea se deschide cu un bearer token și un parametru de interogare pentru model, iar primul mesaj pe care îl trimiți configurează totul despre cum se comportă agentul:

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 este promptul de sistem, iar acest model preferă unele scurte. Notele de migrare ale SpaceXAI spun să simplifici prompturile scrise pentru modelele vocale din era GPT, nu să le portezi mot-à-mot. Al meu îi spune agentului să păstreze răspunsurile scurte, să pună o singură întrebare pe rând și să citească orice scriere înapoi înainte de a acționa. O confirmare vorbită e o finețe de UX, nu un control de securitate. Aplicația ta tot impune autorizarea la operația de scriere în sine.

Ce m-a surprins: un șir de model nerecunoscut nu ridică o eroare la conectare, revine în liniște la grok-voice-think-fast-1.0. Degradarea unei cereri plătite din cauza unui typo, fără s-o spună, e un implicit ciudat. Loghează câmpul session.model din session.created o dată la pornire și verifică dacă ai primit ce ai cerut.

Ieșire în terminal arătând session.created după conectare. Imagine de autor.

Transmiterea în flux a audio-ului utilizatorului

Cu turn_detection.type setat la server_vad, trebuie doar să continui să adaugi audio. Serverul decide când apelantul a încetat să vorbească și declanșează răspunsul pentru tine. Setează-l la null și preiei tu decizia, confirmând bufferul explicit când crezi că tura s-a încheiat.

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

VAD-ul serverului are trei reglaje, iar setarea lor greșită e cel mai comun mod în care am văzut un agent vocal părând stricat în timp ce nimic din loguri nu dădea eroare. Niciunul nu apare în ecoul lui session.updated implicit, așa că verifică-le în documentație, nu presupune.

• threshold (0,1 la 0,9, implicit 0,85), cât de tare trebuie să fie audio ca să conteze ca vorbire; ridică-l în încăperi zgomotoase, scade-l dacă vorbitorii tăcuți sunt ratați

• silence_duration_ms, cât timp apelantul tace înainte ca serverul să încheie tura; prea scurt taie oamenii la jumătatea gândului, prea lung se simte greoi

• prefix_padding_ms (implicit 333), o felie de audio păstrată chiar dinainte de detectarea vorbirii, ca să nu fie tăiată prima silabă

Reglează silence_duration_ms mai întâi dacă apelanții tot sunt tăiați când fac pauze să gândească. La el ajung înaintea celorlalți doi.

Recepția și redarea răspunsului

Audio sosește în bucăți mici ca response.output_audio.delta, iar rostul streamingului e să redai fiecare bucată în clipa în care ajunge, în loc să aștepți 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)

Păstrează transcrierea chiar și în producție. E cel mai ieftin instrument de depanare când un apelant spune că agentul „a zis ceva ciudat”.

Adăugarea de unelte agentului vocal

Un agent vocal care doar vorbește e un chatbot cu microfon.

Crearea uneltelor pentru comenzi

Fiecare unealtă este o schemă JSON plus o funcție Python simplă de partea noastră. Modelul nu atinge niciodată baza de date, vede doar ce returnează funcția noastră.

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
]

Operațiile de citire precum check_order_status sunt sigure la retry dacă ceva expiră. Operațiile de scriere nu sunt: retry-ul pentru update_delivery_instructions după un timeout ambiguu poate aplica aceeași schimbare de două ori. O linie de confirmare în prompt nu oprește asta, deci dă scrierilor o cheie de idempotentă sau un check de duplicat.

Pune refuzurile și în funcție. cancel_order returnează un motiv și o alternativă în loc să anuleze o comandă expediată, pentru că un prompt care spune „nu anula niciodată comenzile expediate” e o sugestie, iar o funcție care refuză nu e.

Gestionarea buclei de apelare a uneltelor

Patru pași, iar ordinea contează mai mult decât pare. Modelul trimite response.function_call_arguments.done, codul tău rulează funcția, tu trimiți rezultatul înapoi ca un element function_call_output și abia apoi ceri modelului să continue.

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

Dacă modelul are nevoie de mai mult de o unealtă pentru o cerere, declanșează multiple evenimente function_call_arguments.done înainte de a reda vreun audio. Rezolvă-le pe toate și trimite fiecare rezultat înainte de un singur response.create. Dacă îl trimiți prea devreme, modelul răspunde fără contextul din apelurile încă în curs.

Există un mic hop pe care SpaceXAI îl documentează și tot l-am lovit din prima: trimiterea lui response.create imediat ce rezultatul uneltei pleacă se poate suprapune cu propoziția introductivă pe care agentul încă o redă. Într-o rulare a deschis cu „Verific imediat statusul comenzii ORD-1042” și a apelat unealta la jumătatea propoziției, așa că un răspuns imediat ar fi vorbit peste propriul început.

Așteaptă să se termine audio-ul turei curente și arată un scurt status de „gândește” între timp.

Fluxul apelării uneltei înainte de continuarea răspunsului. Imagine de autor.

Gestionarea întreruperilor și a stării conversației

Două probleme separate aici. Un apelant vorbește peste agent la mijlocul răspunsului și un WebSocket cade și trebuie reluat.

Suportarea întreruperilor naturale

Cu server_vad activ, barge-in-ul e automat pe server: în clipa în care detectează că apelantul vorbește din nou, semnalează input_audio_buffer.speech_started și oprește generarea răspunsului vechi. Treaba ta e jumătatea de client a strângerii de mână, curățând orice audio e deja în coadă ca agentul să tacă în loc să termine o propoziție pe care nimeni n-a cerut-o.

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

Pentru sesiuni manuale, non-VAD, response.cancel face același lucru la cerere. Există și conversation.item.truncate pentru a tăia un element al asistentului până la ce s-a auzit efectiv. Documentația confirmă că există dar nu și când să-l declanșezi în timpul unui barge-in live, așa că testează tu sincronizarea.

Am testat asta cu o schimbare de instrucțiuni de livrare la mijlocul răspunsului: începe cererea, întrerupe cu o altă adresă la jumătatea confirmării agentului. Ce contează e dacă agentul aplică instrucțiunea corectată în loc să termine în liniște pe cea veche, nu dacă audio s-a oprit. Verifică înregistrarea comenzii, nu liniștea. Demo-ul din browser de la final îți permite să auzi asta.

Reluarea unei sesiuni deconectate

Reluarea sesiunii este opțională și nu e memorie. Setează resumption.enabled: true pe session.update, ia ID-ul din evenimentul conversation.created, iar dacă socketul cade, reconectează-te cu ?conversation_id=<id> în URL și optează din nou pe noua conexiune.

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

Turele, transcrierile, apelările de unelte și rezultatele de unelte din cache se redau înainte de următoarea ta întrebare, iar cache-ul dispare după 30 de minute de inactivitate. Am testat întrebând despre o comandă, întrerupând conexiunea și reconectând pentru un follow-up fără să mă repet; agentul a reluat corect ETA-ul.

Un aspect nedocumentat: reluarea nu sosește instant, deci o întrebare trimisă chiar în clipa în care se deschide socketul o poate lua înainte și se întoarce fără memoria turei anterioare. Dă-i o secundă înainte să dai vina pe resumption.

Jurnal în terminal al unei sesiuni reluate. Imagine de autor.

Nu folosi asta în locul salvării stării comenzii în baza ta de date. Dacă expiră cache-ul sau apelantul sună mâine, pornești de la zero context, și asta e intenționat.

Securizarea și monitorizarea agentului

Nu pune niciodată o cheie API permanentă în cod de browser sau mobil. Dacă un client se conectează direct în loc să treacă prin serverul tău, emite un token cu viață scurtă:

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

Un browser nu poate seta un header Authorization personalizat pe un handshake WebSocket, deci trece tokenul prin headerul sec-websocket-protocol, prefixat cu xai-client-secret..

Serverul emite token, browserul intră în apel. Imagine de autor.

Facturarea rulează pe doi contoare. Audio, trimis sau primit, rulează la 0,08 $ pe minut cum am menționat, adică 4,80 $ pe oră, iar fiecare conversation.item.create care nu e audio și nu e un function_call_output costă fix 0,004 $. response.create nu este taxat deloc. Fiecare response.done poartă un obiect usage, care în testul meu a raportat output_audio_seconds alături de un billable_audio_seconds separat. Facturează din acelea, nu din estimări.

Limitele documentate pe Speech to Speech API sunt 10 sesiuni concurente per echipă și un plafon de 120 de minute pe sesiune, ambele în us-east-1. Nu planifica capacitatea după cifrele din Voice Agent API, care sunt diferite.

La confidențialitate, fii precis. Întrebările frecvente de securitate ale SpaceXAI spun că cererile și răspunsurile API sunt păstrate criptate 30 de zile pentru monitorizarea abuzurilor și nu sunt folosite pentru antrenare fără permisiune, și că echipele pot activa Zero Data Retention, deși ZDR renunță la istoricul convorbirilor agentului vocal și deci nu funcționează cu reluarea.

Dacă dezvălui că un apel e înregistrat sau gestionat de AI, asta face extensia force_message menționată mai devreme. Replica se redă exact cum e scrisă, nu cum ar parafraza modelul.

Testarea agentului vocal

Un status 200 pe handshake-ul WebSocket nu-ți spune nimic despre dacă agentul a făcut ce trebuie. Testează rezultatul, nu doar conexiunea.

• O interogare curată a unei comenzi, verificând răspunsul vorbit față de înregistrare, nu doar că a sosit un răspuns
• Un răspuns întrerupt, confirmând că redarea se oprește și agentul abordează noua cerere
• O actualizare de livrare care necesită confirmare, verificată față de înregistrarea comenzii
• Un refuz, precum anularea unei comenzi expediate, unde agentul trebuie să explice regula, nu să își ceară scuze
• Un număr de comandă necunoscut, asigurându-te că agentul spune asta în loc să inventeze un status
• O unealtă care returnează o eroare, verificând că agentul o spune în loc să blocheze
• Reconectare și reluare, inclusiv acea fereastră de redare pe care am întâlnit-o mai devreme
• Audio zgomotos, vorbire rapidă și un apelant care silabisește numere și adrese

Pe majoritatea le-am rulat cu o cheie live în timp ce scriam. Eșecurile interesante au fost comportamentale, nu erori: sincronizarea reluării de mai sus și un prag VAD în afara intervalului acceptat în loc să fie respins, genul de lucru care ajunge în producție tăcut stricat dacă testezi doar calea fericită. Adaugă și un test multilingv și vezi întrebările frecvente pentru o particularitate despre cum denumești limba.

Două dintre cele de mai sus nu le poți testa tastând. app_streamlit.py e o pagină Streamlit care pune un apel live în browser: microfonul transmite în același WebSocket peste WebRTC, vocea agentului se întoarce în flux, iar socketul rămâne deschis pe tot parcursul.

streamlit run app_streamlit.py