Tutorial API Grok Voice Think Fast 2.0: crea un agente vocale realtime in Python

Grok Voice Think Fast 2.0 di SpaceXAI è un modello speech-to-speech. Gli invii audio tramite WebSocket e ricevi audio in risposta; nel mezzo può ragionare e continuare a parlare mentre una funzione che ha deciso di chiamare è già in esecuzione. Niente passaggi separati di speech-to-text o di text-to-speech.

SpaceXAI ha annunciato Think Fast 2.0 il 29 luglio 2026: primo audio più rapido, comportamento full-duplex più stabile (ascolta mentre parla invece di alternarsi rigidamente) e chiamate agli strumenti che partono presto nel turno. Terrò i benchmark brevi: per un tutorial conta cosa cambia nel tuo codice.

Costruiremo un agente vocale per l'assistenza clienti di un negozio online. Chi chiama può chiedere di un ordine, cambiare le istruzioni di consegna, annullare, interrompere l'agente a metà frase e riprendere la conversazione dopo una disconnessione. Questo è il percorso via API, non il Voice Agent Builder no-code che il nostro tutorial sul Grok Voice Agent Builder illustra. Parti da lì per la versione console-first.

Cos'è Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 è il modello più recente di SpaceXAI per la Speech to Speech API, il nome di prodotto dietro a ciò che la maggior parte chiama semplicemente Grok Voice. Se pensi ancora all'azienda come xAI, è la stessa: è stata inglobata in SpaceX e rinominata SpaceXAI il 6 luglio 2026. L'API non ha seguito il rebrand, quindi ogni identificatore qui sotto dice ancora xai, dalla variabile XAI_API_KEY all'host api.x.ai.

Uno stack vocale tradizionale collega tre servizi: speech-to-text, un modello linguistico, poi text-to-speech, e ogni passaggio aggiunge latenza e punti in cui si può perdere contesto. Think Fast 2.0 riduce tutto a un unico modello che riceve audio o testo e produce audio o testo sulla stessa connessione.

WebSocket speech-to-speech contro architettura a tre servizi. Immagine dell'autore.

Per un agente che agisce invece di limitarsi a parlare, conta che ragionamento e parlato vadano in parallelo. SpaceXAI afferma che le chiamate agli strumenti "di solito" iniziano a eseguire prima che l'agente finisca la sua prima frase, e quel "di solito" conta davvero.

Sui benchmark citati da SpaceXAI su Artificial Analysis, Think Fast 2.0 ottiene l'82,9% sullo Speech to Speech Index contro il 75,7% della 1.0, e riduce il time to first audio da 1,25 secondi a 0,70 secondi. I numeri del fornitore su un benchmark generale sono un'ipotesi sul tuo flusso di chiamata, non un piano di test.

Vedrai tre stringhe di modello: grok-voice-latest, grok-voice-think-fast-2.0 e grok-voice-think-fast-1.0. L'alias è comodo in fase di prototipazione e non abbastanza stabile per altro.

Quando ho testato il 4 agosto 2026, grok-voice-latest risolveva ancora in grok-voice-think-fast-1.0, con le note di rilascio di SpaceXAI che pianificavano il passaggio alla Think Fast 2.0 per il giorno successivo. Quel passaggio è un cambiamento di prezzo oltre che di modello, 0,08 $ al minuto di audio contro 0,05 $ della 1.0, quindi un alias non fissato diventa più costoso senza che il tuo codice cambi di una riga. Fissa la stringa versionata in tutto ciò che distribuisci.

Cosa costruiremo

L'agente copre le richieste tipiche di una linea di assistenza: cercare un ordine, trovarlo a partire da un'email quando chi chiama non ha il numero, cambiare le istruzioni di consegna, annullare, aprire o controllare un ticket e passare la chiamata a una persona. Lungo la strada compaiono interruzioni e una connessione che cade.

È un insieme di piccoli file, non un unico script, perché ogni parte ha un compito diverso e vorrai testarle separatamente. Ecco la struttura:

• config.py carica la chiave API e contiene la stringa del modello, il sample rate e gli endpoint

• voice_client.py incapsula il WebSocket, tiene traccia della fatturazione ed espone helper di invio/ricezione

• tools.py definisce le funzioni sugli ordini e un piccolo archivio ordini in memoria al posto di un database reale

• assistant.py contiene il prompt di sistema, la configurazione di sessione e l'event loop che lega tutto

• token_server.py è un piccolo endpoint FastAPI che genera token effimeri

• app_streamlit.py mette lo stesso client dietro una chiamata live dal browser, a cui tornerò dopo la sezione test

Il percorso didattico parte dal terminale. La demo aggiunge il microfono.

Prerequisiti

Ti serve un account SpaceXAI con chiave API, un metodo di fatturazione attivo (non esiste un free tier permanente e i crediti promozionali dei nuovi account non bastano) e sufficiente dimestichezza con asyncio e i WebSocket per seguire senza una spiegazione riga per riga di await.

Gli esempi quick-start di SpaceXAI usano il pacchetto websockets grezzo invece di un SDK dedicato, e lo facciamo anche noi. I documenti non indicano una versione di Python richiesta. Io ho testato con la 3.11.

Tieni la chiave API sul server. Se un browser o un'app mobile parla direttamente con la Voice API, riceve un token effimero invece della tua chiave reale, trattato nella sezione sicurezza qui sotto.

Configurazione del progetto

Ogni file qui sotto è nel repo del progetto, così puoi clonarlo invece di copiare gli snippet:

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 gestisce la connessione realtime e python-dotenv legge la tua chiave. Il resto copre l'endpoint dei token e la demo nel browser. Metti la tua chiave in .env:

XAI_API_KEY=xai-your-key-here

Questa è gran parte della configurazione. La connessione è la parte interessante.

Capire la Grok Voice Realtime API

Grok Voice è il nome del prodotto. A livello di codice, parli con un endpoint WebSocket su wss://api.x.ai/v1/realtime, e l'intera conversazione avviene come uno stream di eventi JSON su quell'unico socket.

Il ciclo di vita degli eventi

Una connessione segue una forma fissa: il server invia session.created e conversation.created appena ti connetti, tu invii session.update per configurare voce e strumenti, il server conferma con session.updated, e da lì crei elementi di conversazione e richiedi risposte. L'ho provato con una chiave live e l'ordine ha corrisposto esattamente ai documenti.

• session.update (client) configura voce, istruzioni, strumenti e formato audio

• conversation.item.create (client) aggiunge un messaggio utente, un messaggio assistente o un risultato di uno strumento

• response.create (client) chiede al modello di parlare; il VAD del server lo invia automaticamente per te

• response.output_audio.delta e response.output_audio_transcript.delta (server) streammano la risposta mentre viene generata

• response.done (server) chiude il turno

Due cose ingannano spesso. La pagina delle Speech to Speech docs che ho linkato sopra cita un evento conversation.item.created durante la ripresa della sessione, ma il riferimento eventi canonico elenca solo conversation.item.added, ed è quello che è arrivato in ogni test che ho fatto, quindi programma su quello. Vedrai anche un ping non documentato pochi secondi dopo la maggior parte delle connessioni, menzionato solo per non scambiarlo per un errore.

Formati audio e trasporto

Codec e trasporto sono scelte separate. Il codec, impostato sotto audio.input.format e audio.output.format, è audio/pcm (Linear16, predefinito 24000 Hz), audio/pcmu o audio/pcma (G.711 a 8 kHz, per la telefonia), oppure audio/opus (24 kHz). Il trasporto è come quei byte viaggiano sul filo:

• json (predefinito) invia l'audio come testo base64 dentro input_audio_buffer.append e response.output_audio.delta, facile da registrare e fare debug

• binary invia i byte grezzi del codec come frame binari WebSocket, evitando l'overhead base64 al costo di un ciclo di ricezione che deve diramarsi sul tipo di messaggio

Inizia con JSON. Ogni esempio nei documenti lo usa, è banale da ispezionare e l'overhead base64 non è il collo di bottiglia per un agente di supporto. Passa al binario solo se misuri un motivo per farlo.

Compatibilità con la Realtime API di OpenAI

Salta avanti se non hai mai toccato la Realtime API di OpenAI. Per tutti gli altri, la Speech to Speech API segue la Realtime API di OpenAI abbastanza da poter portare la maggior parte del codice client cambiando base URL e chiave, ma non è un drop-in perfetto.

I transcript arrivano come conversation.item.input_audio_transcription.updated qui invece del delta di OpenAI, alcuni eventi OpenAI non sono supportati e SpaceXAI aggiunge estensioni proprie: force_message per una riga di disclosure pre-scritta, resumption per le riconnessioni, e replace per correggere nomi di brand pronunciati male prima del text-to-speech.

Costruire l'agente vocale in tempo reale

Basta protocollo. Ecco il client che ci parla.

Connessione e configurazione della sessione

La connessione si apre con un bearer token e un parametro di query per il modello, e il primo messaggio che invii configura tutto il comportamento dell'agente:

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 è il system prompt, e questo modello preferisce prompt brevi. Le note di migrazione di SpaceXAI dicono di semplificare i prompt scritti per i vecchi modelli vocali dell'era GPT invece di portarli pari pari. Il mio dice all'agente di tenere le risposte corte, fare una domanda alla volta e leggere qualsiasi scrittura prima di agire. Una conferma parlata è una gentilezza UX, non un controllo di sicurezza. La tua applicazione fa comunque rispettare l'autorizzazione sulla scrittura.

Una cosa che mi ha sorpreso: una stringa di modello non riconosciuta non genera errore in fase di connessione, ricade silenziosamente su grok-voice-think-fast-1.0. Declassare una richiesta a pagamento per un refuso, senza dirlo, è un predefinito strano. Registra il campo session.model di session.created una volta all'avvio e verifica di aver ottenuto ciò che hai chiesto.

Output del terminale che mostra session.created dopo la connessione. Immagine dell'autore.

Inviare in streaming l'audio dell'utente

Con turn_detection.type impostato su server_vad, devi solo continuare ad aggiungere audio. Il server decide quando chi chiama ha smesso di parlare e attiva per te la risposta. Impostalo su null e la decisione spetta a te, impegnando esplicitamente il buffer quando ritieni che il turno sia finito.

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

Il VAD del server ha tre manopole, e impostarle male è il modo più comune in cui ho visto un agente vocale sembrare rotto pur senza errori nei log. Per impostazione predefinita, nessuna di queste appare nell'eco di session.updated, quindi controllale con i documenti invece di supporre.

• threshold (da 0,1 a 0,9, predefinito 0,85), quanto deve essere forte l'audio per contare come parlato; alzalo in ambienti rumorosi, abbassalo se oratori con voce bassa vengono persi

• silence_duration_ms, per quanto tempo chi chiama resta in silenzio prima che il server termini il suo turno; troppo breve taglia le persone a metà pensiero, troppo lungo risulta lento

• prefix_padding_ms (predefinito 333), una porzione di audio mantenuta appena prima del rilevamento della voce, così la prima sillaba non viene tagliata

Regola prima silence_duration_ms se chi chiama viene spesso interrotto mentre fa una pausa per pensare. È la prima che tocco prima delle altre due.

Ricevere e riprodurre la risposta

L'audio arriva in piccoli pezzi come response.output_audio.delta, e lo streaming serve a riprodurre ogni pezzo appena arriva, invece di aspettare 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)

Tieni il transcript anche in produzione. È lo strumento di debug più economico quando chi chiama dice che l'agente "ha detto qualcosa di strano".

Aggiungere strumenti all'agente vocale

Un agente vocale che parla soltanto è un chatbot con un microfono.

Creare gli strumenti per gli ordini

Ogni strumento è uno schema JSON più una normale funzione Python dal nostro lato. Il modello non tocca mai il database, vede solo ciò che restituisce la nostra funzione.

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
]

Le operazioni di lettura come check_order_status sono sicure da ritentare se qualcosa va in timeout. Le operazioni di scrittura no: ritentare update_delivery_instructions dopo un timeout ambiguo può applicare due volte la stessa modifica. Una riga di conferma nel prompt non lo impedisce, quindi dai alle scritture una chiave di idempotenza o un controllo duplicati.

Metti i rifiuti anche nella funzione. cancel_order restituisce un motivo e un'alternativa invece di annullare un ordine spedito, perché un prompt che dice "non annullare mai ordini spediti" è un suggerimento, mentre una funzione che rifiuta non lo è.

Gestire il loop di chiamata agli strumenti

Quattro passaggi, e l'ordine conta più di quanto sembri. Il modello invia response.function_call_arguments.done, il tuo codice esegue la funzione, invii il risultato indietro come elemento function_call_output e solo allora chiedi al modello di continuare.

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

Se al modello servono più di uno strumento per una richiesta, invia più eventi function_call_arguments.done prima che parta qualsiasi audio. Risolvi tutti e invia ogni risultato prima di un solo response.create. Se lo invii troppo presto, il modello risponde senza il contesto delle chiamate ancora in corso.

C'è una trappola che SpaceXAI documenta e che ho comunque incontrato al primo tentativo: inviare response.create nell'istante in cui invii il risultato dello strumento può sovrapporsi alla frase introduttiva che l'agente sta ancora riproducendo. In un'esecuzione ha iniziato con "Controllo subito lo stato dell'ordine ORD-1042" e ha chiamato lo strumento a metà frase, quindi una risposta immediata avrebbe parlato sopra al proprio incipit.

Aspetta che l'audio del turno corrente finisca e mostra nel frattempo un breve stato di "elaborazione".

Flusso di chiamata allo strumento prima di continuare la risposta. Immagine dell'autore.

Gestire interruzioni e stato della conversazione

Due problemi separati. Chi chiama parla sopra l'agente a metà risposta, e un WebSocket cade e va ripreso.

Supportare interruzioni naturali

Con server_vad attivo, il barge-in è automatico lato server: nel momento in cui rileva che chi chiama parla di nuovo, segnala input_audio_buffer.speech_started e smette di generare la vecchia risposta. Il tuo compito è la metà client di quella stretta di mano, svuotando qualsiasi audio già in coda perché l'agente stia zitto invece di finire una frase che nessuno vuole sentire.

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

Per le sessioni manuali, senza VAD, response.cancel fa lo stesso su richiesta. C'è anche conversation.item.truncate per ridurre un elemento dell'assistente a ciò che è stato effettivamente ascoltato. I documenti confermano che esiste ma non quando attivarlo durante un barge-in live, quindi testa tu stesso la tempistica.

L'ho testato con un cambio di istruzioni di consegna a metà risposta: avvia la richiesta, interrompi con un indirizzo diverso a metà della conferma dell'agente. Conta se l'agente applica l'istruzione corretta invece di finire in silenzio quella vecchia, non se l'audio si è fermato. Verifica contro il record dell'ordine, non contro il silenzio. La demo nel browser alla fine ti permette di sentirlo.

Riprendere una sessione disconnessa

La ripresa di sessione è opt-in e non è memoria. Imposta resumption.enabled: true su session.update, prendi l'ID dall'evento conversation.created, e se il socket cade, riconnetti con ?conversation_id=<id> nell'URL e attiva di nuovo sulla nuova connessione.

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

I turni memorizzati, i transcript, le chiamate agli strumenti e i risultati vengono riprodotti prima della tua prossima domanda, e la cache scompare dopo 30 minuti di inattività. L'ho testato chiedendo di un ordine, facendo cadere la connessione e riconnettendomi per un follow-up senza ripetermi; l'agente ha ripreso correttamente l'ETA.

Una particolarità non documentata: il replay non arriva all'istante, quindi una domanda inviata nel momento in cui il socket si apre può anticiparlo e tornare senza memoria del turno precedente. Dagli un secondo prima di incolpare la ripresa.

Log del terminale di una sessione ripresa. Immagine dell'autore.

Non usare questo al posto del salvataggio dello stato degli ordini nel tuo database. Se la cache scade o chi chiama richiama domani, riparti da zero contesto, ed è voluto.

Mettere in sicurezza e monitorare l'agente

Non mettere mai una chiave API permanente nel codice di browser o mobile. Se un client si connette direttamente invece di passare dal tuo server, genera un token a breve scadenza:

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 non può impostare un'intestazione Authorization personalizzata nell'handshake WebSocket, quindi passa il token tramite l'intestazione sec-websocket-protocol, con prefisso xai-client-secret..

Il server genera il token, il browser entra in chiamata. Immagine dell'autore.

La fatturazione ha due contatori. L'audio, inviato o ricevuto, è tariffato a 0,08 $ al minuto come detto prima, cioè 4,80 $ l'ora, e ogni conversation.item.create che non è audio e non è un function_call_output costa una tariffa fissa di 0,004 $. response.create non è fatturato. Ogni response.done include un oggetto usage che nei miei test riportava output_audio_seconds insieme a billable_audio_seconds separato. Fattura da quelli, non da stime.

I limiti documentati sulla Speech to Speech API sono 10 sessioni concorrenti per team e un tetto di 120 minuti per sessione, entrambi in us-east-1. Non pianificare la capacità in base ai numeri della Voice Agent API, che sono diversi.

Sulla privacy, sii preciso. Le FAQ sulla sicurezza di SpaceXAI dicono che le richieste e risposte API sono conservate crittografate per 30 giorni per il monitoraggio degli abusi e non usate per l'addestramento senza permesso, e che i team possono attivare la Zero Data Retention, anche se ZDR elimina la cronologia delle conversazioni degli agenti vocali e quindi non funziona con la ripresa.

Se stai dichiarando che una chiamata è registrata o gestita da IA, è ciò che fa l'estensione force_message che ho menzionato prima. La riga viene riprodotta esattamente com'è scritta invece di qualsiasi parafrasi del modello.

Testare l'agente vocale

Un 200 sull'handshake WebSocket non dice nulla sul fatto che l'agente abbia fatto la cosa giusta. Testa l'esito, non solo la connessione.

• Una consultazione pulita di un ordine, verificando la risposta parlata contro il record, non solo che sia arrivata una risposta
• Una risposta interrotta, confermando che la riproduzione si ferma e l'agente gestisce la nuova richiesta
• Un aggiornamento di consegna che richiede conferma, verificato contro il record dell'ordine
• Un rifiuto, come l'annullamento di un ordine spedito, dove l'agente deve spiegare la regola invece di scusarsi
• Un numero d'ordine sconosciuto, assicurandoti che l'agente lo dica invece di inventare uno stato
• Uno strumento che restituisce un errore, verificando che l'agente lo comunichi invece di bloccarsi
• Riconnessione e ripresa, inclusa quella finestra di replay che ho incontrato prima
• Audio rumoroso, parlato veloce e un chiamante che scandisce numeri e indirizzi

Ho eseguito la maggior parte di questi test con una chiave live durante la scrittura. I fallimenti interessanti erano comportamentali, non errori: la tempistica della ripresa sopra, e una soglia VAD fuori range accettata invece che rifiutata, il tipo di cosa che arriva in produzione silenziosamente rotta se testi solo il caso felice. Aggiungi anche un test multilingue e guarda le FAQ per una particolarità su come indicare la lingua.

Due di questi non li puoi testare digitando. app_streamlit.py è una pagina Streamlit che mette una chiamata live nel browser: il microfono streamma nello stesso WebSocket via WebRTC, la voce dell'agente torna in streaming e il socket resta aperto.

streamlit run app_streamlit.py