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

Grok Voice Think Fast 2.0 di SpaceXAI è un modello speech-to-speech. Gli invii audio passano via WebSocket e tornerà audio, e nel frattempo può ragionare e continuare a parlare mentre è già in esecuzione una chiamata a funzione che ha deciso di effettuare. Niente passaggio separato di speech-to-text, niente passaggio separato 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 fare turni rigidi) e chiamate a strumenti che partono presto nel turno. Tengo breve il discorso sui benchmark, perché per un tutorial conta cosa cambia nel tuo codice.

Costruiremo un agente vocale di assistenza clienti per un negozio online. Chi chiama può chiedere di un ordine, cambiare un’istruzione 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 illustrato dal nostro tutorial su Grok Voice Agent Builder. 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 assorbita in SpaceX e rinominata SpaceXAI il 6 luglio 2026. L’API non ha seguito il rebranding, quindi ogni identificatore qui sotto continua a dire xai, dalla variabile XAI_API_KEY all’host api.x.ai.

Uno stack vocale tradizionale concatena tre servizi: speech-to-text, un modello linguistico, poi text-to-speech, e ogni passaggio aggiunge latenza e un punto in cui il contesto può perdersi. Think Fast 2.0 comprime tutto in un unico modello che accetta audio o testo in ingresso e produce audio o testo in uscita 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 girino in parallelo. SpaceXAI dice che le chiamate a strumenti "di solito" iniziano a essere eseguite prima che l’agente finisca la sua prima frase, e quel "di solito" pesa davvero.

Nei benchmark citati da SpaceXAI di 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 tempo al primo audio da 1,25 secondi a 0,70 secondi. I numeri del vendor 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 a Think Fast 2.0 per il giorno successivo. Quel cambio è tanto di prezzo quanto di modello, 0,08 $ al minuto di audio contro 0,05 $ per la 1.0, quindi un alias non bloccato diventa più costoso senza che cambi una riga del tuo codice. Fissa la stringa versionata in tutto ciò che distribuisci.

Cosa costruiremo

L’agente copre ciò che una linea di supporto si sente chiedere: cercare un ordine, trovarne uno da un’email quando chi chiama non ha il numero, cambiare un’istruzione 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 pugno di file piccoli anziché uno script unico, perché ogni pezzo ha un lavoro diverso e vorrai testarli separatamente. Ecco la struttura:

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

• voice_client.py incapsula il WebSocket, traccia la 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 della sessione e il loop di eventi 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 una chiave API, un accordo di fatturazione finanziato (non c’è un livello gratuito permanente e i crediti promozionali per nuovi account non basteranno), e abbastanza dimestichezza con asyncio e WebSocket per seguire senza una spiegazione riga per riga di await.

Gli esempi quick start di SpaceXAI usano il pacchetto grezzo websockets anziché un SDK dedicato, e lo facciamo anche noi. La documentazione non indica una versione minima di Python. Io ho testato su 3.11.

Tieni la chiave API sul server. Se un’app browser o mobile parla direttamente con la Voice API, riceve un token effimero al posto della tua chiave reale, come spiegato nella sezione sicurezza sotto.

Impostare il progetto

Ogni file qui sotto vive nel repo del progetto, così puoi clonarlo invece di copiare frammenti:

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 per i token e la demo 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 Realtime API di Grok Voice

Grok Voice è il nome di prodotto. Quello a cui effettivamente scrivi codice è un endpoint WebSocket su wss://api.x.ai/v1/realtime, e l’intera conversazione avviene come flusso 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. Ho eseguito questo contro una chiave live e l’ordine ha corrisposto esattamente alla documentazione.

• 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) trasmettono la risposta man mano che viene generata

• response.done (server) chiude il turno

Due cose traggono in inganno. La pagina della documentazione Speech to Speech che ho linkato sopra menziona un evento conversation.item.created durante la ripresa della sessione, ma il riferimento eventi canonico elenca solo conversation.item.added, ed è quello che ho ricevuto in ogni test, quindi fai riferimento a quello. Vedrai anche un evento non documentato ping dopo qualche secondo nella 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 sulla rete:

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

• binary invia i byte grezzi del codec come frame binari WebSocket, evitando l’overhead del base64 al costo di un loop di ricezione che deve diramarsi in base al tipo di messaggio

Inizia con JSON. Ogni esempio nella documentazione lo usa, è banale da ispezionare e l’overhead del base64 non è ciò che limita una build per agenti di supporto. Passa al binario se misuri un motivo per farlo.

Compatibilità con l’OpenAI Realtime API

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

Le trascrizioni qui arrivano come conversation.item.input_audio_transcription.updated 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 query del 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 prompt di sistema, 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 anziché portarli alla lettera. Il mio dice all’agente di mantenere risposte brevi, fare una domanda alla volta e leggere qualsiasi scrittura prima di agire. Una conferma parlata è una cortesia UX, non un controllo di sicurezza. La tua applicazione fa ancora rispettare l’autorizzazione sull’operazione di scrittura.

Una cosa che mi ha sorpreso: una stringa di modello non riconosciuta non solleva errore al momento della connessione, ricade silenziosamente su grok-voice-think-fast-1.0. Fare il downgrade di una richiesta a pagamento per un refuso, senza dirlo, è un predefinito strano. Logga il campo session.model di session.created una volta all’avvio e controlla di aver ottenuto ciò che hai richiesto.

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

Trasmettere l’audio dell’utente

Con turn_detection.type impostato a server_vad, devi solo continuare ad aggiungere audio in coda. Il server decide quando chi chiama ha smesso di parlare e attiva la risposta per te. Impostalo su null e ti assumi tu la decisione, confermando 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 senza errori nei log. Per impostazione predefinita nessuna appare nell’eco di session.updated, quindi confrontale con la documentazione invece di assumere.

• threshold (0,1–0,9, predefinito 0,85), quanto deve essere forte l’audio per contare come parlato; alzalo in ambienti rumorosi, abbassalo se gli oratori più pacati vengono persi

• silence_duration_ms, quanto a lungo chi chiama resta in silenzio prima che il server chiuda il turno; troppo corto tronca le frasi, troppo lungo dà sensazione di lentezza

• prefix_padding_ms (predefinito 333), una fetta di audio tenuta da poco prima che fosse rilevato il parlato, così non si taglia la prima sillaba

Regola silence_duration_ms per primo se chi chiama viene continuamente tagliato mentre pensa. È quella a cui ricorro prima delle altre due.

Ricevere e riprodurre la risposta

L’audio arriva in piccoli pezzi come response.output_audio.delta, e il punto dello streaming è riprodurre ogni pezzo nel momento in cui 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 la trascrizione anche in produzione. È lo strumento di debug più economico che hai 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 sugli ordini

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

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 la stessa modifica due volte. 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 ordini spediti" è un suggerimento, 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 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 il modello ha bisogno di più di uno strumento per una richiesta, genera 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 qui che SpaceXAI documenta e che ho comunque preso al primo giro: inviare response.create nell’istante in cui esce il risultato del tuo 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 il proprio incipit.

Aspetta che l’audio del turno corrente finisca e mostra un breve stato di "sto pensando" nel frattempo.

Flusso della chiamata a strumento prima di continuare la risposta. Immagine dell’autore.

Gestire interruzioni e stato della conversazione

Qui ci sono due problemi distinti. 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 riprende a parlare, 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 così che l’agente taccia invece di finire una frase che nessuno vuole più sentire.

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

Per le sessioni manuali, senza VAD, response.cancel fa lo stesso lavoro su richiesta. C’è anche conversation.item.truncate per tagliare un elemento dell’assistente a ciò che è stato effettivamente ascoltato. La documentazione conferma che esiste ma non quando attivarlo durante un barge-in live, quindi prova tu il timing.

L’ho testato con una modifica alle istruzioni di consegna a metà risposta: avvia la richiesta, interrompi con un indirizzo diverso a metà della conferma dell’agente. Ciò che conta è se l’agente applica l’istruzione corretta invece di finire in silenzio quella vecchia, non se l’audio si è fermato. Verifica sul record dell’ordine, non sul silenzio. La demo nel browser alla fine ti fa sentire questo caso.

Riprendere una sessione disconnessa

La ripresa della 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, riconnettiti con ?conversation_id=<id> nell’URL e riattiva l’opzione anche 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, le trascrizioni, le chiamate a strumenti e i loro risultati vengono riprodotti prima della tua prossima domanda, e la cache scompare dopo 30 minuti di inattività. L’ho testato chiedendo di un ordine, interrompendo la connessione e riconnettendomi per un seguito senza ripetermi; l’agente ha ripreso correttamente l’ETA.

Una nota non documentata: il replay non arriva all’istante, quindi una domanda inviata nell’attimo in cui si apre il socket può precederlo e tornare senza memoria del turno precedente. Dagli un secondo prima di incolpare la ripresa.

Log da terminale di una sessione ripresa. Immagine dell’autore.

Non usare questo al posto di salvare lo stato dell’ordine 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 inserire 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 durata:

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 header Authorization personalizzato sull’handshake WebSocket, quindi passa il token tramite l’header sec-websocket-protocol, con prefisso xai-client-secret..

Il server genera il token, il browser entra nella chiamata. Immagine dell’autore.

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

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

Sulla privacy, sii preciso. Le FAQ di sicurezza di SpaceXAI dicono che richieste e risposte API vengono conservate criptate per 30 giorni per il monitoraggio abusi e non usate per training senza permesso, e che i team possono attivare lo Zero Data Retention, anche se ZDR elimina la cronologia conversazioni dell’agente vocale e quindi non funziona con la ripresa.

Se stai comunicando che una chiamata è registrata o gestita da AI, è per questo che serve l’estensione force_message che ho citato. La riga viene riprodotta esattamente com’è scritta invece di come il modello la parafrasa.

Testare l’agente vocale

Uno status 200 sull’handshake WebSocket non ti dice nulla sul fatto che l’agente abbia fatto la cosa giusta. Testa l’esito, non solo la connessione.

• Una ricerca ordine pulita, confrontando la risposta parlata con il record, non solo che sia arrivata una risposta
• Una risposta interrotta, confermando che la riproduzione si ferma e l’agente affronta la nuova richiesta
• Un aggiornamento di consegna che richiede conferma, verificato rispetto al record dell’ordine
• Un rifiuto, come annullare 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, incluso quel margine di replay che ho incontrato prima
• Audio rumoroso, parlato veloce e chi chiama che scandisce numeri e indirizzi

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

Due di questi non puoi testarli digitando. app_streamlit.py è una pagina Streamlit che mette una chiamata live nel browser: il microfono trasmette nello stesso WebSocket via WebRTC, la voce dell’agente torna in streaming e il socket resta aperto per tutta la durata.

streamlit run app_streamlit.py