Tutorial API Grok Voice Think Fast 2.0: Creează 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 funcție pe care a decis să o apeleze 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 strict pe rând) și apeluri de tool-uri care pornesc devreme în tură. Voi ține partea de benchmark scurtă, fiindcă, pentru un tutorial, contează ce se schimbă în codul tău.

Vom construi un agent vocal pentru suport clienți al unui 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 deconectare. Aceasta este calea prin API, nu constructorul no-code Voice Agent Builder pe care îl parcurge tutorialul nostru Grok Voice Agent Builder. Începe de acolo pentru versiunea 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 simplu Grok Voice. Dacă încă te gândești la companie ca la xAI, este aceeași echipă: a fost integrată în SpaceX și rebranduită SpaceXAI pe 6 iulie 2026. API-ul nu a urmat rebranduirea, așa că fiecare identificator de mai jos încă spune xai, de la variabila XAI_API_KEY până la hostul api.x.ai.

Un stack vocal tradițional înlănțuie trei servicii: speech-to-text, un model de limbaj, apoi text-to-speech, iar fiecare hop adaugă latență și un loc unde se poate pierde contextul. Think Fast 2.0 comprimă totul î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 în 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ă apelurile de tool-uri „de obicei” încep să se execute înainte ca agentul să-și termine prima propoziție, iar acel „de obicei” chiar contează.

Pe benchmarkurile 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.

Există trei stringuri 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 comod în prototipare și prea instabil pentru orice altceva.

Când am testat pe 4 august 2026, grok-voice-latest încă rezolva la grok-voice-think-fast-1.0, iar notițele de lansare SpaceXAI programau trecerea la Think Fast 2.0 pentru a doua zi. Acea schimbare e la fel de mult o schimbare de preț cât e de model: 0,08 $ pe minut de audio față de 0,05 $ pentru 1.0, așa că un alias nefixat devine mai scump fără să-ți atingi codul. Fixează stringul versiunii în orice desfășori.

Ce vom construi

Agentul acoperă ce primește de obicei o linie de suport: caută o comandă, găsește una după 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 deconectare.

Este un pumn de fișiere mici, nu un singur script, fiindcă fiecare piesă are alt rol și vei vrea să le testezi separat. Iată structura:

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

• voice_client.py împachetează WebSocketul, urmărește costurile și expune helperi de trimitere/primire

• tools.py definește funcțiile pentru comenzi și un mic store in-memory pentru comenzi în locul unei baze de date reale

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

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

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

Parcursul didactic pornește din terminal. Demo-ul adaugă microfonul.

Cerințe preliminare

Ai nevoie de un cont SpaceXAI cu o cheie API, o metodă de plată finanțată (nu există un free tier permanent, iar creditele promoționale pentru cont nou nu te vor ține) și suficientă familiaritate cu asyncio și WebSocketuri ca să urmărești fără o explicație linie cu linie a lui await.

Exemplele de pornire rapidă SpaceXAI folosesc pachetul brut websockets, nu un SDK dedicat, și la fel facem și noi. Documentația nu specifică o versiune minimă de Python. Eu am testat pe 3.11.

Păstrează cheia API pe server. Dacă un browser sau o aplicație 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 există în repo-ul proiectului, așa că î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 în timp real și python-dotenv îți citește cheia. Restul acoperă endpointul pentru token și demo-ul în browser. Pune cheia în .env:

XAI_API_KEY=xai-your-key-here

Asta e aproape tot setup-ul. Conexiunea e partea interesantă.

Înțelegerea Grok Voice Realtime API

Grok Voice este numele produsului. Cei împotriva căruia chiar scrii cod este un endpoint WebSocket la wss://api.x.ai/v1/realtime, iar întreaga conversație are loc ca un flux de evenimente JSON pe 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 tool-urile, 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 a coincis exact cu documentația.

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

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

• response.create (client) cere modelului să vorbească; VAD de pe server îți trimite asta 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 împiedică pe oameni. Pagina Speech to Speech din docs pe care am legat-o mai sus menționează un eveniment conversation.item.created în timpul reluării sesiunii, dar referința de evenimente canonică listează doar conversation.item.added, iar acesta a sosit în fiecare test pe care l-am rulat, așa că codează pentru el. Vei vedea și un eveniment ping nedocumentat la câteva secunde în majoritatea conexiunilor, menționat doar ca să nu-l citești ca eroare.

Formate audio și transport

Codecul și transportul sunt alegeri separate. Codecul, 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 cum circulă acei bytes pe fir:

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

• binary trimite bytes de codec brut ca cadre binare WebSocket, sărind peste overheadul base64 cu prețul unei bucle de recepție care trebuie să branșeze pe tipul mesajului

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

Compatibilitate cu OpenAI Realtime API

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

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 divulgare scriptată, resumption pentru reconectări și replace pentru corectarea pronunției brandurilor înainte de text-to-speech.

Construirea agentului vocal în timp real

Gata cu protocolul. Iată clientul care vorbește cu el.

Conectarea și configurarea sesiunii

Conexiunea se deschide cu un bearer token și un parametru de query 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. Notițele de migrare ale SpaceXAI spun să simplifici prompturile scrise pentru modelele vocale din era GPT, nu să le portezi ad litteram. Al meu îi spune agentului să țină răspunsurile scurte, să pună o singură întrebare pe rând și să citească înapoi orice scriere înainte de a acționa. O confirmare vorbită este o finețe de UX, nu un control de securitate. Aplicația ta tot impune autorizarea pe scrierea propriu-zisă.

Un lucru care m-a surprins: un string de model nerecunoscut nu ridică o eroare la conectare, ci revine în tăcere la grok-voice-think-fast-1.0. Degradarea unei cereri plătite din cauza unui typo, fără să 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 cu session.created după conectare. Imagine de autor.

Transmiterea audio-ului utilizatorului

Cu turn_detection.type setat la server_vad, tot ce trebuie să faci este să continui să adaugi audio. Serverul decide când apelantul a încetat să vorbească și declanșează răspunsul pentru tine. Pune-l pe null și decizia îți aparține, confirmând explicit bufferul 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(),
    }))

Server VAD are trei reglaje, iar setarea greșită este cel mai comun mod în care am văzut un agent vocal să pară stricat fără erori în loguri. Niciunul nu apare în ecoul lui session.updated implicit, așa că verifică-le în documentație, nu presupune.

• threshold (0,1–0,9, implicit 0,85), cât de tare trebuie să fie audio ca să fie considerat vorbire; ridică-l în camere zgomotoase, coboară-l dacă vorbitorii tăcuți nu sunt detectați

• silence_duration_ms, cât timp tace apelantul înainte ca serverul să încheie tura; prea scurt taie oamenii la mijlocul ideii, 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ă mai întâi silence_duration_ms dacă apelanții sunt tot tăiați când fac pauze să gândească. E primul la care apelez înainte să ating ceilalți doi.

Primirea și redarea răspunsului

Audio sosește în bucăți mici ca response.output_audio.delta, iar scopul transmiterii pe flux este să redai fiecare bucată în momentul în care ajunge, nu 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ă transcriptul și în producție. Este cel mai ieftin instrument de debugging când un apelant spune că agentul „a zis ceva ciudat”.

Adăugarea de tool-uri agentului vocal

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

Crearea tool-urilor pentru comenzi

Fiecare tool 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țiunile de citire precum check_order_status sunt sigure de retry dacă ceva expiră. Operațiunile de scriere nu sunt: retry pe 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, așa că dă scrierilor o cheie de idempotentă sau un control de duplicate.

Pune refuzurile și în funcție. cancel_order returnează un motiv și o alternativă în loc să anuleze o comandă expediată, fiindcă 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 tool-urilor

Patru pași, iar ordinea contează mai mult decât pare. Modelul trimite response.function_call_arguments.done, codul tău rulează funcția, 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 un tool pentru o cerere, emite multiple evenimente function_call_arguments.done înainte ca vreun audio să fie redat. Rezolvă-le pe toate și trimite fiecare rezultat înainte de un singur response.create. Trimite-l prea devreme și modelul răspunde fără contextul apelurilor încă în derulare.

Există un „gotcha” documentat de SpaceXAI pe care tot l-am nimerit din prima: trimiterea lui response.create chiar în clipa când iese rezultatul tool-ului se poate suprapune cu propoziția introductivă pe care agentul încă o redă. Într-o rulare a început cu „Verific imediat statusul comenzii ORD-1042” și a apelat tool-ul la jumătatea propoziției, așa că un răspuns imediat ar fi vorbit peste propriul opener.

Așteaptă să se termine audio-ul din tura curentă și afișează un scurt status de „gândire” între timp.

Fluxul apelului de tool î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 în mijlocul răspunsului și un WebSocket cade și trebuie reluat.

Suport pentru întreruperi naturale

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

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

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

Am testat asta cu o schimbare a instrucțiunii de livrare la mijlocul răspunsului: pornește cererea, întrerupe cu o altă adresă la jumătatea confirmării agentului. Important este dacă agentul aplică instrucțiunea corectată în loc să termine în liniște pe cea veche, nu dacă s-a oprit audio-ul. 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 este memorie. Setează resumption.enabled: true pe session.update, ia ID-ul din evenimentul conversation.created și, dacă socketul cade, reconectează-te cu ?conversation_id=<id> în URL și optează din nou pe conexiunea nouă.

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 în cache, transcripturile, apelurile de tool și rezultatele tool-urilor 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.

O capcană nedocumentată: reluarea nu sosește instantaneu, așa că o întrebare trimisă chiar când se deschide socketul o poate devansa și se întoarce fără memoria turei anterioare. Dă-i o secundă înainte să dai vina pe resumption.

Log de terminal pentru o sesiune reluată. Imagine de autor.

Nu folosi asta în loc să salvezi starea comenzii în baza ta de date. Dacă expiră cache-ul sau apelantul sună mâine, pornești de la zero context, iar asta este 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 durată 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 handshake-ul WebSocket, așa că transmite 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 contori. Audio, trimis sau primit, rulează la 0,08 $ pe minut, cum am menționat mai devreme, adică 4,80 $ pe oră, iar fiecare conversation.item.create care nu este audio și nu este function_call_output costă 0,004 $ fix. 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 separat billable_audio_seconds. Facturează din acelea, nu din estimări.

Limitele documentate pe Speech to Speech API sunt 10 sesiuni concurrente per echipă și un plafon de 120 de minute pe sesiune, ambele în us-east-1. Nu planifica capacitatea din cifrele 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 criptat timp de 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 elimină istoricul conversatiilor de agent vocal persistat și, prin urmare, nu funcționează cu reluarea.

Dacă dezvălui că un apel este înregistrat sau gestionat de AI, pentru asta este extensia force_message pe care am menționat-o mai devreme. Linia se redă exact cum e scrisă, nu cum o parafrazează 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, nu inventează un status
• Un tool care returnează o eroare, verificând că agentul o spune în loc să se blocheze
• Reconectare și reluare, inclusiv acea fereastră de replay pe care am întâlnit-o mai devreme
• Audio zgomotos, vorbire rapidă și un apelant care silabisește numere și adrese

Am rulat majoritatea acestora 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 se livrează tăcut stricat dacă testezi doar scenariul fericit. Adaugă și un test multilingv și vezi întrebările frecvente pentru o particularitate despre cum numești limba.

Două dintre acestea nu le poți testa tastând. app_streamlit.py este o pagină Streamlit care pune un apel live în browser: microfonul transmite în același WebSocket prin WebRTC, vocea agentului revine în flux, iar socketul rămâne deschis tot timpul.

streamlit run app_streamlit.py